music-assistant-server

1"""Timeline building and broadcasting for Plex remote control."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import time
8from typing import TYPE_CHECKING, Any
9from urllib.parse import urlparse
10
11from aiohttp import ClientTimeout
12from music_assistant_models.enums import PlaybackState, RepeatMode
13
14from .parsing import plex_key_for_item
15
16if TYPE_CHECKING:
17    from music_assistant.providers.plex import PlexProvider
18
19LOGGER = logging.getLogger(__name__)
20
21
22class TimelineMixin:
23    """Mixin providing timeline building and broadcasting."""
24
25    if TYPE_CHECKING:
26        provider: PlexProvider
27        client_id: str
28        subscriptions: dict[str, dict[str, object]]
29        play_queue_id: str | None
30        play_queue_version: int
31        play_queue_item_ids: dict[int, int]
32        _ma_player_id: str | None
33        headers: dict[str, str]
34        plex_server: Any
35
36    def _resolve_plex_state(self, player: Any, queue: Any) -> str:
37        """
38        Resolve the Plex playback state string from MA player/queue state.
39
40        The queue is the source of truth for playback: for synced or grouped
41        players (and players using a protocol output) the child player's own
42        ``playback_state`` stays idle while the active queue (on the sync
43        leader / group / protocol parent) is actually playing. Relying on
44        ``player.playback_state`` therefore reports "paused" for those players
45        even though MA is playing.
46
47        :param player: The MA player (may be a sync child / group member).
48        :param queue: The active MA queue for the player (if any).
49        :return: One of "playing", "paused" or "stopped".
50        """
51        if queue is not None:
52            state = queue.state
53        elif player is not None:
54            # player.state.playback_state is MA's resolved/final state, which already
55            # follows the active output protocol player and sync leader.
56            state = player.state.playback_state
57        else:
58            return "stopped"
59
60        if state == PlaybackState.PLAYING:
61            return "playing"
62        if state == PlaybackState.PAUSED:
63            return "paused"
64        if state == PlaybackState.IDLE:
65            has_track = bool(queue and queue.current_item and queue.current_item.media_item)
66            return "paused" if has_track else "stopped"
67        return "stopped"
68
69    def _build_timeline_attributes(
70        self,
71        track: Any,
72        state: str,
73        duration: int,
74        time_ms: int,
75        volume: int,
76        shuffle: int,
77        repeat: int,
78        controllable: str,
79        queue: Any | None,
80    ) -> list[str]:
81        """
82        Build timeline attributes for a playing track.
83
84        :param track: The current track media item.
85        :param state: Playback state (playing, paused, etc.).
86        :param duration: Track duration in milliseconds.
87        :param time_ms: Current playback time in milliseconds.
88        :param volume: Volume level (0-100).
89        :param shuffle: Shuffle state (0 or 1).
90        :param repeat: Repeat mode (0=off, 1=one, 2=all).
91        :param controllable: Controllable features string.
92        :param queue: The MA queue object.
93        :return: List of timeline attribute strings.
94        """
95        key = plex_key_for_item(track, self.provider.instance_id)
96        if not key:
97            return []
98        rating_key = key.split("/")[-1]
99
100        plex_url = urlparse(self.provider._baseurl)
101        machine_identifier = self.provider._plex_server.machineIdentifier
102        address = plex_url.hostname
103        port = plex_url.port or (443 if plex_url.scheme == "https" else 32400)
104        protocol = plex_url.scheme
105
106        attrs = [
107            f'state="{state}"',
108            f'duration="{duration}"',
109            f'time="{time_ms}"',
110            f'ratingKey="{rating_key}"',
111            f'key="{key}"',
112        ]
113
114        if self.play_queue_id and queue:
115            if queue.current_index is not None:
116                play_queue_item_id = self.play_queue_item_ids.get(
117                    queue.current_index, queue.current_index + 1
118                )
119                attrs.append(f'playQueueItemID="{play_queue_item_id}"')
120            attrs.append(f'playQueueID="{self.play_queue_id}"')
121            attrs.append(f'playQueueVersion="{self.play_queue_version}"')
122            attrs.append(f'containerKey="/playQueues/{self.play_queue_id}"')
123
124        attrs.extend(
125            [
126                'type="music"',
127                f'volume="{volume}"',
128                f'shuffle="{shuffle}"',
129                f'repeat="{repeat}"',
130                f'controllable="{controllable}"',
131                f'machineIdentifier="{machine_identifier}"',
132                f'address="{address}"',
133                f'port="{port}"',
134                f'protocol="{protocol}"',
135            ]
136        )
137
138        return attrs
139
140    async def _build_timeline_xml(
141        self, include_metadata: bool = False, command_id: str = "0"
142    ) -> str:
143        """
144        Build timeline XML from current Music Assistant player state.
145
146        :param include_metadata: Whether to include metadata in the timeline.
147        :param command_id: The command ID for the timeline response.
148        """
149        player_id = self._ma_player_id
150
151        player = self.provider.mass.players.get_player(player_id) if player_id else None
152        queue = self.provider.mass.players.get_active_queue(player) if player else None
153
154        controllable = (
155            "volume,repeat,skipPrevious,seekTo,stepBack,stepForward,stop,playPause,shuffle,skipNext"
156        )
157
158        state = self._resolve_plex_state(player, queue)
159
160        # group_volume is MA's effective, group-aware volume: for groups/syncgroups it
161        # averages the (powered) members, otherwise it returns the player's own resolved
162        # volume (following any attached volume control / protocol player).
163        volume = int(player.group_volume or 0) if player else 0
164
165        shuffle = 0
166        repeat = 0
167        if queue:
168            shuffle = 1 if queue.shuffle_enabled else 0
169            repeat = {RepeatMode.ONE: 1, RepeatMode.ALL: 2}.get(queue.repeat_mode, 0)
170
171        if (
172            state in ["playing", "paused"]
173            and queue
174            and queue.current_item
175            and queue.current_item.media_item
176        ):
177            track = queue.current_item.media_item
178            duration = round(track.duration * 1000) if track.duration else 0
179            time_ms = round(queue.corrected_elapsed_time * 1000)
180
181            attrs = self._build_timeline_attributes(
182                track, state, duration, time_ms, volume, shuffle, repeat, controllable, queue
183            )
184
185            if attrs:
186                music_timeline = f"<Timeline {' '.join(attrs)}/>"
187            else:
188                music_timeline = (
189                    f'<Timeline state="{state}" time="{time_ms}" type="music" volume="{volume}" '
190                    f'shuffle="{shuffle}" repeat="{repeat}" controllable="{controllable}"/>'
191                )
192        else:
193            time_ms = 0
194            music_timeline = (
195                f'<Timeline state="{state}" time="{time_ms}" type="music" volume="{volume}" '
196                f'shuffle="{shuffle}" repeat="{repeat}" controllable="{controllable}"/>'
197            )
198
199        video_timeline = '<Timeline type="video" state="stopped"/>'
200        photo_timeline = '<Timeline type="photo" state="stopped"/>'
201
202        return (
203            f'<MediaContainer commandID="{command_id}">'
204            f"{music_timeline}{video_timeline}{photo_timeline}"
205            f"</MediaContainer>"
206        )
207
208    async def _send_timeline(self, client_id: str) -> None:
209        """
210        Send timeline update to a specific subscribed controller.
211
212        :param client_id: The client ID to send the timeline to.
213        """
214        subscription = self.subscriptions.get(client_id)
215        if not subscription:
216            return
217
218        timeline_xml = await self._build_timeline_xml()
219
220        try:
221            async with self.provider.mass.http_session.post(
222                f"{subscription['url']}/:/timeline",
223                data=timeline_xml,
224                headers={
225                    "X-Plex-Client-Identifier": self.client_id,
226                    "Content-Type": "text/xml",
227                },
228                timeout=ClientTimeout(total=5),
229            ) as resp:
230                if resp.status < 400:
231                    subscription["last_update"] = time.time()
232        except Exception as e:
233            LOGGER.debug(f"Failed to send timeline to {client_id}: {e}")
234
235    async def _send_timeline_to_server(self) -> None:
236        """Send timeline update to Plex server for activity tracking."""
237        if not self._ma_player_id:
238            return
239
240        try:
241            player = self.provider.mass.players.get_player(self._ma_player_id)
242            queue = self.provider.mass.players.get_active_queue(player) if player else None
243
244            if (
245                not player
246                or not queue
247                or not queue.current_item
248                or not queue.current_item.media_item
249            ):
250                return
251
252            track = queue.current_item.media_item
253
254            plex_key = plex_key_for_item(track, self.provider.instance_id)
255            if not plex_key:
256                return
257
258            rating_key = plex_key.split("/")[-1]
259
260            plex_state = self._resolve_plex_state(player, queue)
261
262            position_ms = round(queue.corrected_elapsed_time * 1000)
263            duration_ms = round(track.duration * 1000) if track.duration else 0
264
265            container_key = ""
266            play_queue_item_id = ""
267            if self.play_queue_id:
268                container_key = f"/playQueues/{self.play_queue_id}"
269                if queue.current_index is not None:
270                    play_queue_item_id = str(
271                        self.play_queue_item_ids.get(queue.current_index, queue.current_index + 1)
272                    )
273
274            params: dict[str, str] = {
275                "ratingKey": rating_key,
276                "key": plex_key,
277                "state": plex_state,
278                "time": str(position_ms),
279                "duration": str(duration_ms),
280            }
281
282            if container_key:
283                params["containerKey"] = container_key
284            if play_queue_item_id:
285                params["playQueueItemID"] = play_queue_item_id
286
287            plex_server = self.plex_server
288            headers = self.headers
289
290            def send_timeline() -> None:
291                plex_server.query("/:/timeline", params=params, headers=headers)
292
293            await asyncio.to_thread(send_timeline)
294
295        except Exception as e:
296            LOGGER.debug(f"Failed to send timeline to Plex server: {e}")
297
298    async def _broadcast_timeline(self) -> None:
299        """Send timeline to all subscribed controllers."""
300        current_time = time.time()
301        stale_clients = []
302        for client_id, sub in self.subscriptions.items():
303            try:
304                last_update = float(sub["last_update"])  # type: ignore[arg-type]
305                if current_time - last_update > 90:
306                    stale_clients.append(client_id)
307            except ValueError, TypeError:
308                LOGGER.debug(f"Invalid last_update for client {client_id}, treating as stale")
309                stale_clients.append(client_id)
310
311        for client_id in stale_clients:
312            del self.subscriptions[client_id]
313
314        await asyncio.gather(
315            *(self._send_timeline(client_id) for client_id in list(self.subscriptions.keys())),
316            return_exceptions=True,
317        )
318