fix(plex): Refactor player state to allow for drift and tracking paus…

…e based on Source behavior Plex only updates player position every 15 seconds (of played track) so player state needs to be adapted to not detect this as a pause or seek. * Allow per-source drift allowed before triggering seek * Allow per-source pause detection
FoxxMD · Oct 23, 2024 · 3cc735b · 3cc735b
1 parent b74f31c
commit 3cc735b
Show file tree

Hide file tree

Showing 7 changed files with 94 additions and 12 deletions.
diff --git a/src/backend/common/infrastructure/config/source/plex.ts b/src/backend/common/infrastructure/config/source/plex.ts
@@ -1,3 +1,4 @@
+import { PollingOptions } from "../common.js";
 import { CommonSourceConfig, CommonSourceData, CommonSourceOptions } from "./index.js";
 
 export interface PlexSourceData extends CommonSourceData {
@@ -35,7 +36,7 @@ export interface PlexSourceAIOConfig extends PlexSourceConfig {
     type: 'plex'
 }
 
-export interface PlexApiData extends CommonSourceData {
+export interface PlexApiData extends CommonSourceData, PollingOptions {
     token?: string
     /**
      * http(s)://HOST:PORT of the Plex server to connect to

diff --git a/src/backend/sources/PlayerState/AbstractPlayerState.ts b/src/backend/sources/PlayerState/AbstractPlayerState.ts
@@ -57,6 +57,8 @@ export abstract class AbstractPlayerState {
     createdAt: Dayjs = dayjs();
     stateLastUpdatedAt: Dayjs = dayjs();
 
+    protected allowedDrift?: number;
+
     protected constructor(logger: Logger, platformId: PlayPlatformId, opts: PlayerStateOptions = DefaultPlayerStateOptions) {
         this.platformId = platformId;
         this.logger = childLogger(logger, `Player ${this.platformIdStr}`);
@@ -248,18 +250,18 @@ export abstract class AbstractPlayerState {
                 // and polling/network delays means we did not catch absolute beginning of track
                 usedPosition = 1;
             }
-            this.currentListenRange = new ListenRange(new ListenProgress(timestamp, usedPosition));
+            this.currentListenRange = new ListenRange(new ListenProgress(timestamp, usedPosition), undefined, this.allowedDrift);
         } else {
             const oldEndProgress = this.currentListenRange.end;
             const newEndProgress = new ListenProgress(timestamp, position);
             if (position !== undefined && oldEndProgress !== undefined) {
-                if (position === oldEndProgress.position && !['paused', 'stopped'].includes(this.calculatedStatus)) {
-                    this.calculatedStatus = this.reportedStatus === 'stopped' ? CALCULATED_PLAYER_STATUSES.stopped : CALCULATED_PLAYER_STATUSES.paused;
-                    if (this.reportedStatus !== this.calculatedStatus) {
-                        this.logger.debug(`Reported status '${this.reportedStatus}' but track position has not progressed between two updates. Calculated player status is now ${this.calculatedStatus}`);
-                    } else {
-                        this.logger.debug(`Player position is equal between current -> last update. Updated calculated status to ${this.calculatedStatus}`);
-                    }
+                if (!this.isSessionStillPlaying(position) && !['paused', 'stopped'].includes(this.calculatedStatus)) {
+                        this.calculatedStatus = this.reportedStatus === 'stopped' ? CALCULATED_PLAYER_STATUSES.stopped : CALCULATED_PLAYER_STATUSES.paused;
+                        if (this.reportedStatus !== this.calculatedStatus) {
+                            this.logger.debug(`Reported status '${this.reportedStatus}' but track position has not progressed between two updates. Calculated player status is now ${this.calculatedStatus}`);
+                        } else {
+                            this.logger.debug(`Player position is equal between current -> last update. Updated calculated status to ${this.calculatedStatus}`);
+                        }
                 } else if (position !== oldEndProgress.position && this.calculatedStatus !== 'playing') {
                     this.calculatedStatus = CALCULATED_PLAYER_STATUSES.playing;
                     if (this.reportedStatus !== this.calculatedStatus) {
@@ -275,6 +277,8 @@ export abstract class AbstractPlayerState {
         }
     }
 
+    protected abstract isSessionStillPlaying(position: number): boolean;
+
     protected currentListenSessionEnd() {
         if (this.currentListenRange !== undefined && this.currentListenRange.getDuration() !== 0) {
             this.logger.debug('Ended current Player listen range.')

diff --git a/src/backend/sources/PlayerState/GenericPlayerState.ts b/src/backend/sources/PlayerState/GenericPlayerState.ts
@@ -6,4 +6,8 @@ export class GenericPlayerState extends AbstractPlayerState {
     constructor(logger: Logger, platformId: PlayPlatformId, opts?: PlayerStateOptions) {
         super(logger, platformId, opts);
     }
+
+    protected isSessionStillPlaying(position: number): boolean {
+        return position !== this.currentListenRange.end.position;
+    }
 }
diff --git a/src/backend/sources/PlayerState/ListenRange.ts b/src/backend/sources/PlayerState/ListenRange.ts
@@ -6,13 +6,15 @@ export class ListenRange implements ListenRangeData {
 
     public start: ListenProgress;
     public end: ListenProgress;
+    protected allowedDrift: number;
 
-    constructor(start?: ListenProgress, end?: ListenProgress) {
+    constructor(start?: ListenProgress, end?: ListenProgress, allowedDrift: number = 2500) {
         const s = start ?? new ListenProgress();
         const e = end ?? s;
 
         this.start = s;
         this.end = e;
+        this.allowedDrift = allowedDrift;
     }
 
     isPositional() {
@@ -39,7 +41,7 @@ export class ListenRange implements ListenRangeData {
         const realTimeDiff = Math.max(0, reportedTS.diff(this.end.timestamp, 'ms')); // 0 max used so TS from testing doesn't cause "backward" diff
         const positionDiff = (position - this.end.position) * 1000;
         // if user is more than 2.5 seconds ahead of real time
-        if (positionDiff - realTimeDiff > 2500) {
+        if (positionDiff - realTimeDiff > this.allowedDrift) {
             return [true, position - this.end.position];
         }
 

diff --git a/src/backend/sources/PlayerState/PlexPlayerState.ts b/src/backend/sources/PlayerState/PlexPlayerState.ts
@@ -0,0 +1,15 @@
+import { Logger } from "@foxxmd/logging";
+import { PlayPlatformId, REPORTED_PLAYER_STATUSES } from "../../common/infrastructure/Atomic.js";
+import { AbstractPlayerState, PlayerStateOptions } from "./AbstractPlayerState.js";
+import { GenericPlayerState } from "./GenericPlayerState.js";
+
+export class PlexPlayerState extends GenericPlayerState {
+    constructor(logger: Logger, platformId: PlayPlatformId, opts?: PlayerStateOptions) {
+        super(logger, platformId, opts);
+        this.allowedDrift = 17000;
+    }
+
+    protected isSessionStillPlaying(position: number): boolean {
+        return this.reportedStatus === REPORTED_PLAYER_STATUSES.playing;
+    }
+}
diff --git a/src/backend/sources/PlayerState/RealtimePlayer.ts b/src/backend/sources/PlayerState/RealtimePlayer.ts
@@ -0,0 +1,48 @@
+import { childLogger, Logger } from "@foxxmd/logging";
+import { SimpleIntervalJob, Task, ToadScheduler } from "toad-scheduler";
+
+const RT_TICK = 500;
+
+abstract class RealtimePlayer {
+
+    logger: Logger;
+    scheduler: ToadScheduler = new ToadScheduler();
+
+    protected position: number = 0;
+
+    protected constructor(logger: Logger) {
+        this.logger = childLogger(logger, `RT`);
+        const job = new SimpleIntervalJob({
+            milliseconds: RT_TICK,
+            runImmediately: true
+        }, new Task('updatePos', () => this.position += RT_TICK), { id: 'rt' });
+        this.scheduler.addSimpleIntervalJob(job);
+        this.scheduler.stop();
+    }
+
+    public play(position?: number) {
+        if (position !== undefined) {
+            this.position = position;
+        }
+        this.scheduler.startById('rt');
+    }
+
+    public pause() {
+        this.scheduler.stop();
+    }
+
+    public stop() {
+        this.pause();
+        this.position = 0;
+    }
+
+    public seek(position: number) {
+        this.position = position;
+    }
+
+    public getPosition() {
+        return this.position;
+    }
+}
+
+export class GenericRealtimePlayer extends RealtimePlayer {}
diff --git a/src/backend/sources/PlexApiSource.ts b/src/backend/sources/PlexApiSource.ts
@@ -23,6 +23,9 @@ import normalizeUrl from 'normalize-url';
 import { GetTokenDetailsResponse, GetTokenDetailsUserPlexAccount } from '@lukehagar/plexjs/sdk/models/operations/gettokendetails.js';
 import { parseRegexSingle } from '@foxxmd/regex-buddy-core';
 import { Readable } from 'node:stream';
+import { PlexPlayerState } from './PlayerState/PlexPlayerState.js';
+import { PlayerStateOptions } from './PlayerState/AbstractPlayerState.js';
+import { Logger } from '@foxxmd/logging';
 
 const shortDeviceId = truncateStringToLength(10, '');
 
@@ -66,6 +69,7 @@ export default class PlexApiSource extends MemorySource {
         const {
             data: {
                 token,
+                interval = 5,
                 usersAllow = [],
                 usersBlock = [],
                 devicesAllow = [],
@@ -74,10 +78,12 @@ export default class PlexApiSource extends MemorySource {
                 librariesBlock = [],
             } = {},
             options: {
-                logFilterFailure = (parseBool(process.env.DEBUG_MODE) ? 'debug' : 'warn')
+                logFilterFailure = (parseBool(process.env.DEBUG_MODE) ? 'debug' : 'warn'),
             } = {}
         } = this.config;
 
+        this.config.data.interval = interval;
+
         if((token === undefined || token.trim() === '')) {
             throw new Error(`'token' must be specified in config data`);
         }
@@ -385,4 +391,6 @@ export default class PlexApiSource extends MemorySource {
             position: viewOffset / 1000
         }
     }
+
+    getNewPlayer = (logger: Logger, id: PlayPlatformId, opts: PlayerStateOptions) => new PlexPlayerState(logger, id, opts);
 }