music-assistant-server
75.7 KB•PY
player_remote.py
75.7 KB • 1,882 lines • python
1"""Per-player Plex remote control instances."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import platform
8import re
9import time
10import uuid
11from collections.abc import Callable
12from typing import TYPE_CHECKING, Any
13from urllib.parse import urlparse
14
15from aiohttp import ClientTimeout, web
16from music_assistant_models.enums import (
17 EventType,
18 PlayerFeature,
19 PlayerType,
20 QueueOption,
21 RepeatMode,
22)
23from plexapi.playqueue import PlayQueue
24
25from .gdm import PlexGDMAdvertiser
26
27if TYPE_CHECKING:
28 from music_assistant_models.event import MassEvent
29
30 from music_assistant.providers.plex import PlexProvider
31
32
33LOGGER = logging.getLogger(__name__)
34
35
36class PlayerRemoteInstance:
37 """Single remote control instance for one MA player."""
38
39 def __init__(
40 self,
41 plex_provider: PlexProvider,
42 ma_player_id: str,
43 player_name: str,
44 port: int,
45 device_class: str = "speaker",
46 remote_control: bool = False,
47 ) -> None:
48 """Initialize player remote instance.
49
50 :param plex_provider: Plex provider instance.
51 :param ma_player_id: Music Assistant player ID.
52 :param player_name: Display name for the player.
53 :param port: Port for the remote control server.
54 :param device_class: Device class (speaker, phone, tablet, stb, tv, pc, cloud).
55 :param remote_control: Whether to enable remote control.
56 """
57 self.plex_provider = plex_provider
58 self.plex_server = plex_provider._plex_server
59 self.ma_player_id = ma_player_id
60 self.player_name = player_name
61 self.port = port
62 self.device_class = device_class
63 self.remote_control = remote_control
64
65 self.client_id = str(
66 uuid.uuid5(
67 uuid.NAMESPACE_DNS,
68 f"music-assistant-plex-{plex_provider.instance_id}-{ma_player_id}",
69 )
70 )
71
72 if self.remote_control:
73 # Remote control server
74 self.server: PlexRemoteControlServer | None = None
75 # GDM advertiser
76 self.gdm: PlexGDMAdvertiser | None = None
77
78 async def start(self) -> None:
79 """Start this player's remote control."""
80 if self.remote_control:
81 # Create player-specific PlexServer instance with unique client identification
82 LOGGER.info(
83 f"Created PlexServer for '{self.player_name}' with client ID: {self.client_id}"
84 )
85
86 self.server = PlexRemoteControlServer(
87 plex_provider=self.plex_provider,
88 port=self.port,
89 client_id=self.client_id,
90 ma_player_id=self.ma_player_id,
91 device_class=self.device_class,
92 )
93 LOGGER.info(
94 f"Remote control server for '{self.player_name}' bound to MA player: "
95 f"{self.ma_player_id}"
96 )
97
98 await self.server.start()
99
100 # Step 4: Start GDM broadcasting
101 self.gdm = PlexGDMAdvertiser(
102 instance_id=self.client_id,
103 port=self.port,
104 publish_ip=str(self.plex_provider.mass.streams.publish_ip),
105 name=self.player_name,
106 product="Music Assistant",
107 version=self.plex_provider.mass.version
108 if self.plex_provider.mass.version != "0.0.0"
109 else "1.0.0",
110 )
111 self.gdm.start()
112
113 LOGGER.info(f"Player '{self.player_name}' is now discoverable on port {self.port}")
114
115 async def stop(self) -> None:
116 """Stop this player's remote control."""
117 if self.remote_control:
118 if self.gdm:
119 await self.gdm.stop()
120
121 if self.server:
122 await self.server.stop()
123
124 LOGGER.info(f"Stopped remote control for player '{self.player_name}'")
125
126
127class PlexRemoteControlServer:
128 """HTTP server to receive Plex remote control commands."""
129
130 def __init__(
131 self,
132 plex_provider: PlexProvider,
133 port: int = 32500,
134 client_id: str | None = None,
135 ma_player_id: str | None = None,
136 device_class: str = "speaker",
137 ) -> None:
138 """Initialize remote control server.
139
140 :param plex_provider: Plex provider instance.
141 :param port: Port for the HTTP server.
142 :param client_id: Unique client identifier.
143 :param ma_player_id: Music Assistant player ID.
144 :param device_class: Device class (speaker, phone, tablet, stb, tv, pc, cloud).
145 """
146 self.provider = plex_provider
147 self.plex_server = plex_provider._plex_server
148 self.port = port
149 self.client_id = client_id or plex_provider.instance_id
150 self.device_class = device_class
151 self.app = web.Application()
152 self.subscriptions: dict[str, dict[str, object]] = {}
153 self.runner: web.AppRunner | None = None
154 self.http_site: web.TCPSite | None = None
155
156 # Play queue tracking (Plex-specific state that doesn't exist in MA)
157 self.play_queue_id: str | None = None
158 self.play_queue_version: int = 1
159 # Map queue index to item ID
160 self.play_queue_item_ids: dict[int, int] = {}
161
162 # Track MA queue state to detect when we need to sync to Plex
163 self._last_synced_ma_queue_length: int = 0
164 self._last_synced_ma_queue_keys: list[str] = []
165
166 # Specific MA player this server controls (set by PlayerRemoteInstance)
167 self._ma_player_id = ma_player_id
168
169 # Store unsubscribe callbacks
170 self._unsub_callbacks: list[Callable[..., None]] = []
171
172 # Flag to prevent circular updates when we modify the queue ourselves
173 self._updating_from_plex = False
174
175 self.player = self.provider.mass.players.get_player(self._ma_player_id) # type: ignore[arg-type]
176
177 self.device_name = f"{self.player.display_name}" if self.player else "Music Assistant"
178
179 self.headers = {
180 "X-Plex-Device-Name": self.device_name,
181 "X-Plex-Session-Identifier": self.client_id,
182 "X-Plex-Client-Identifier": self.client_id,
183 "X-Plex-Product": "Music Assistant",
184 "X-Plex-Platform": "Music Assistant",
185 "X-Plex-Platform-Version": platform.release(),
186 }
187
188 self._setup_routes()
189
190 def _setup_routes(self) -> None:
191 """Set up all required endpoints."""
192 # Root endpoint
193 self.app.router.add_get("/", self.handle_root)
194
195 # Subscription management
196 self.app.router.add_get("/player/timeline/subscribe", self.handle_subscribe)
197 self.app.router.add_get("/player/timeline/unsubscribe", self.handle_unsubscribe)
198 self.app.router.add_get("/player/timeline/poll", self.handle_poll)
199
200 # Playback commands
201 self.app.router.add_get("/player/playback/playMedia", self.handle_play_media)
202 self.app.router.add_get("/player/playback/refreshPlayQueue", self.handle_refresh_play_queue)
203 self.app.router.add_get("/player/playback/createPlayQueue", self.handle_create_play_queue)
204 self.app.router.add_get("/player/playback/pause", self.handle_pause)
205 self.app.router.add_get("/player/playback/play", self.handle_play)
206 self.app.router.add_get("/player/playback/stop", self.handle_stop)
207 self.app.router.add_get("/player/playback/skipNext", self.handle_skip_next)
208 self.app.router.add_get("/player/playback/skipPrevious", self.handle_skip_previous)
209 self.app.router.add_get("/player/playback/stepForward", self.handle_step_forward)
210 self.app.router.add_get("/player/playback/stepBack", self.handle_step_back)
211 self.app.router.add_get("/player/playback/seekTo", self.handle_seek_to)
212 self.app.router.add_get("/player/playback/setParameters", self.handle_set_parameters)
213 self.app.router.add_get("/player/playback/skipTo", self.handle_skip_to)
214
215 # Resources endpoint
216 self.app.router.add_get("/resources", self.handle_resources)
217
218 # CORS OPTIONS handler (for all routes)
219 self.app.router.add_route("OPTIONS", "/{tail:.*}", self.handle_options)
220
221 # --- Catch-all fallback for debugging purposes ---
222 # self.app.router.add_route("*", "/{path_info:.*}", self.handle_unknown)
223
224 async def start(self) -> None:
225 """Start HTTP server and GDM advertising."""
226 self.runner = web.AppRunner(self.app)
227 await self.runner.setup()
228
229 # Start HTTP server
230 self.http_site = web.TCPSite(self.runner, "0.0.0.0", self.port)
231 await self.http_site.start()
232 LOGGER.info(f"Plex remote control server started on HTTP port {self.port}")
233
234 # Note: GDM advertising is handled by PlexProvider in __init__.py
235 # to avoid duplicate broadcasts
236
237 # Subscribe to player and queue events for state synchronization
238 if self._ma_player_id:
239 self._unsub_callbacks.append(
240 self.provider.mass.subscribe(
241 self._handle_player_event,
242 EventType.PLAYER_UPDATED,
243 id_filter=self._ma_player_id,
244 )
245 )
246 self._unsub_callbacks.append(
247 self.provider.mass.subscribe(
248 self._handle_queue_event,
249 EventType.QUEUE_UPDATED,
250 id_filter=self._ma_player_id,
251 )
252 )
253 self._unsub_callbacks.append(
254 self.provider.mass.subscribe(
255 self._handle_queue_event,
256 EventType.QUEUE_TIME_UPDATED,
257 id_filter=self._ma_player_id,
258 )
259 )
260 self._unsub_callbacks.append(
261 self.provider.mass.subscribe(
262 self._handle_queue_items_updated,
263 EventType.QUEUE_ITEMS_UPDATED,
264 id_filter=self._ma_player_id,
265 )
266 )
267
268 async def stop(self) -> None:
269 """Stop the HTTP server."""
270 # Unsubscribe from events
271 for unsub in self._unsub_callbacks:
272 unsub()
273 self._unsub_callbacks.clear()
274
275 # Stop HTTP server
276 if self.http_site:
277 await self.http_site.stop()
278 if self.runner:
279 await self.runner.cleanup()
280 LOGGER.info("Plex remote control server stopped")
281
282 async def handle_root(self, request: web.Request) -> web.Response:
283 """Handle root endpoint - return basic player info."""
284 # Get player name
285 player_name = "Music Assistant"
286 if self._ma_player_id:
287 player = self.provider.mass.players.get_player(self._ma_player_id)
288 if player:
289 player_name = player.display_name
290
291 xml = f"""<?xml version="1.0" encoding="UTF-8"?>
292<MediaContainer machineIdentifier="{self.client_id}" version="1.0">
293 <Player title="{player_name}" machineIdentifier="{self.client_id}"/>
294</MediaContainer>"""
295 return web.Response(
296 text=xml, content_type="text/xml", headers={"Access-Control-Allow-Origin": "*"}
297 )
298
299 async def handle_subscribe(self, request: web.Request) -> web.Response:
300 """Handle timeline subscription from controller."""
301 client_id = request.headers.get("X-Plex-Client-Identifier")
302 protocol = request.query.get("protocol", "http")
303 port = request.query.get("port")
304 command_id = int(request.query.get("commandID", 0))
305
306 if not client_id or not port:
307 return web.Response(status=400)
308
309 self.subscriptions[client_id] = {
310 "url": f"{protocol}://{request.remote}:{port}",
311 "command_id": command_id,
312 "last_update": time.time(),
313 }
314
315 LOGGER.info(f"Controller {client_id} subscribed for timeline updates")
316 await self._send_timeline(client_id)
317 return web.Response(status=200)
318
319 async def handle_unsubscribe(self, request: web.Request) -> web.Response:
320 """Handle unsubscribe request."""
321 client_id = request.headers.get("X-Plex-Client-Identifier")
322 if client_id in self.subscriptions:
323 del self.subscriptions[client_id]
324 LOGGER.info(f"Controller {client_id} unsubscribed")
325 return web.Response(status=200)
326
327 async def handle_poll(self, request: web.Request) -> web.Response:
328 """Handle timeline poll request."""
329 # Extract parameters
330 include_metadata = request.query.get("includeMetadata", "0") == "1"
331 command_id = request.query.get("commandID", "0")
332
333 # Update subscription timestamp if this client is subscribed
334 client_id = request.headers.get("X-Plex-Client-Identifier")
335 if client_id and client_id in self.subscriptions:
336 self.subscriptions[client_id]["last_update"] = time.time()
337
338 # Build timeline from current MA player state
339 timeline_xml = await self._build_timeline_xml(
340 include_metadata=include_metadata, command_id=command_id
341 )
342 return web.Response(
343 text=timeline_xml,
344 content_type="text/xml",
345 headers={
346 "X-Plex-Client-Identifier": self.client_id,
347 "Access-Control-Expose-Headers": "X-Plex-Client-Identifier",
348 "Access-Control-Allow-Origin": "*",
349 },
350 )
351
352 async def _ungroup_player_if_needed(self, player_id: str) -> None:
353 """Ungroup player before playback if it's part of a group/sync."""
354 player = self.provider.mass.players.get_player(player_id)
355 if not player or player.type == PlayerType.GROUP:
356 return
357
358 if not (player.state.synced_to or player.state.group_members or player.state.active_group):
359 return
360
361 LOGGER.debug("Ungrouping player %s before starting playback from Plex", player.display_name)
362 # Use set_members directly on the group to bypass static member check
363 if (
364 player.state.active_group
365 and (group := self.provider.mass.players.get_player(player.state.active_group))
366 and group.supports_feature(PlayerFeature.SET_MEMBERS)
367 ):
368 await group.set_members(player_ids_to_remove=[player_id])
369 elif (
370 player.state.synced_to
371 and (sync_leader := self.provider.mass.players.get_player(player.state.synced_to))
372 and sync_leader.supports_feature(PlayerFeature.SET_MEMBERS)
373 ):
374 await sync_leader.set_members(player_ids_to_remove=[player_id])
375 elif player.state.group_members and player.supports_feature(PlayerFeature.SET_MEMBERS):
376 await player.set_members(player_ids_to_remove=player.group_members)
377
378 async def handle_play_media(self, request: web.Request) -> web.Response:
379 """
380 Handle playMedia command from Plex controller.
381
382 Plexamp sends various parameters:
383 - key: The item to play (track, album, playlist, etc.)
384 - containerKey: The container context (play queue)
385 - offset: Starting position in milliseconds
386 - shuffle: Whether to shuffle
387 - repeat: Repeat mode
388 """
389 # Set flag to prevent circular updates
390 self._updating_from_plex = True
391 try:
392 key = request.query.get("key")
393 container_key = request.query.get("containerKey")
394 offset = int(request.query.get("offset", 0))
395 shuffle = request.query.get("shuffle", "0") == "1"
396
397 if not key:
398 return web.Response(
399 status=400, text="Missing required 'key' parameter for playMedia command"
400 )
401
402 LOGGER.info(
403 f"Received playMedia command - key: {key}, "
404 f"containerKey: {container_key}, offset: {offset}ms"
405 )
406
407 # Use the assigned player for this server instance
408 player_id = self._ma_player_id
409 if not player_id:
410 return web.Response(status=500, text="No player assigned to this server")
411
412 # Ungroup player if it's part of a group/sync
413 # User selected this specific player, so remove from any groups
414 await self._ungroup_player_if_needed(player_id)
415
416 if container_key and "/playQueues/" in container_key:
417 # Extract play queue ID from container key
418 queue_id_match = re.search(r"/playQueues/(\d+)", container_key)
419 if queue_id_match:
420 self.play_queue_id = queue_id_match.group(1)
421 self.play_queue_version = 1
422 LOGGER.info(f"Playing from queue: {container_key} starting at {key}")
423
424 await self._play_from_plex_queue(player_id, container_key, key, shuffle, offset)
425 else:
426 # Reset queue tracking if no valid queue ID found
427 self.play_queue_id = None
428 self.play_queue_item_ids = {}
429 # Fall back to single track
430 media = await self._resolve_plex_item(key)
431 await self.provider.mass.player_queues.play_media(
432 queue_id=player_id,
433 media=media, # type: ignore[arg-type]
434 option=QueueOption.REPLACE,
435 )
436 elif container_key:
437 # Playing from a regular container (album, playlist, artist) not a play queue
438 # Reset queue tracking
439 self.play_queue_id = None
440 self.play_queue_item_ids = {}
441
442 # The key is the specific track, containerKey is the collection
443 media_to_play = await self._resolve_plex_item(container_key)
444
445 # Queue the entire container
446 await self.provider.mass.player_queues.play_media(
447 queue_id=player_id,
448 media=media_to_play, # type: ignore[arg-type]
449 option=QueueOption.REPLACE,
450 )
451
452 else:
453 # Playing a single item, reset queue tracking
454 self.play_queue_id = None
455 self.play_queue_item_ids = {}
456
457 media = await self._resolve_plex_item(key)
458
459 # Replace the queue with this media
460 await self.provider.mass.player_queues.play_media(
461 queue_id=player_id,
462 media=media, # type: ignore[arg-type]
463 option=QueueOption.REPLACE,
464 )
465
466 # Set shuffle if requested
467 if shuffle:
468 await self.provider.mass.player_queues.set_shuffle(player_id, shuffle)
469
470 # Seek to offset if specified
471 if offset > 0:
472 await self._seek_to_offset_after_playback(player_id, offset)
473
474 await self._broadcast_timeline()
475 return web.Response(status=200)
476
477 except Exception as e:
478 LOGGER.exception(f"Error handling playMedia: {e}")
479 return web.Response(status=500, text=str(e))
480 finally:
481 # Clear flag after processing
482 self._updating_from_plex = False
483
484 def _reorder_tracks_for_playback(
485 self, tracks: list[Any], start_index: int
486 ) -> tuple[list[Any], dict[int, int]]:
487 """Reorder tracks to start from a specific index and update item ID mappings.
488
489 :param tracks: List of tracks to reorder.
490 :param start_index: Index of the track to start from.
491 :return: Tuple of (reordered tracks, updated item ID mappings).
492 """
493 if start_index <= 0 or start_index >= len(tracks):
494 # No reordering needed
495 return tracks, self.play_queue_item_ids
496
497 # Reorder: [selected track, tracks after it, tracks before it]
498 reordered_tracks = (
499 tracks[start_index:] # From selected to end
500 + tracks[:start_index] # From start to selected
501 )
502
503 # Update play queue item ID mappings to reflect new order
504 new_item_ids = {}
505 for new_idx, old_idx in enumerate(
506 list(range(start_index, len(tracks))) + list(range(start_index))
507 ):
508 if old_idx in self.play_queue_item_ids:
509 new_item_ids[new_idx] = self.play_queue_item_ids[old_idx]
510
511 LOGGER.info(f"Started playback from offset {start_index} (reordered queue)")
512 return reordered_tracks, new_item_ids
513
514 async def _seek_to_offset_after_playback(self, player_id: str, offset: int) -> None:
515 """Seek to the specified offset after playback starts.
516
517 :param player_id: The player ID to seek on.
518 :param offset: The offset in milliseconds.
519 """
520 # Wait for the queue to have items loaded before seeking
521 for _ in range(10): # Try up to 10 times (5 seconds total)
522 await asyncio.sleep(0.5)
523 queue = self.provider.mass.player_queues.get(player_id)
524 if queue and queue.current_item:
525 try:
526 await self.provider.mass.players.cmd_seek(player_id, offset // 1000)
527 # Wait briefly for player state to update
528 await asyncio.sleep(0.1)
529 break
530 except Exception as e:
531 LOGGER.debug(f"Could not seek to offset {offset}ms: {e}")
532 break
533 else:
534 LOGGER.warning("Queue not ready for seeking after timeout")
535
536 async def _play_from_plex_queue(
537 self,
538 player_id: str,
539 container_key: str,
540 starting_key: str | None,
541 shuffle: bool,
542 offset: int,
543 ) -> None:
544 """Fetch play queue from Plex and load tracks.
545
546 Starts playback immediately with the first track,
547 then loads remaining tracks in the background.
548 """
549 try:
550 LOGGER.info(f"Fetching play queue: {container_key}")
551
552 # Extract queue ID from container_key (e.g., "/playQueues/123" -> "123")
553 queue_id_match = re.search(r"/playQueues/(\d+)", container_key)
554 if not queue_id_match:
555 raise ValueError(f"Invalid container_key format: {container_key}")
556
557 queue_id = queue_id_match.group(1)
558
559 # Use plexapi to fetch the play queue
560 def fetch_queue() -> PlayQueue:
561 return PlayQueue.get(self.provider._plex_server, playQueueID=queue_id)
562
563 playqueue = await asyncio.to_thread(fetch_queue)
564
565 if playqueue and playqueue.items:
566 # Get selected item offset from PlayQueue - this tells us which track to start from
567 selected_offset = getattr(playqueue, "playQueueSelectedItemOffset", 0)
568 LOGGER.info(f"PlayQueue selected item offset: {selected_offset}")
569
570 # Track play queue item IDs
571 self.play_queue_item_ids = {}
572
573 # Fetch the first track to start playback immediately
574 first_item = (
575 playqueue.items[selected_offset]
576 if selected_offset < len(playqueue.items)
577 else playqueue.items[0]
578 )
579 first_track_key = first_item.key if hasattr(first_item, "key") else None
580 first_play_queue_item_id = (
581 first_item.playQueueItemID if hasattr(first_item, "playQueueItemID") else None
582 )
583
584 if not first_track_key:
585 LOGGER.error("No valid first track in play queue")
586 if starting_key:
587 track = await self.provider.get_track(starting_key)
588 await self.provider.mass.player_queues.play_media(
589 queue_id=player_id,
590 media=track,
591 option=QueueOption.REPLACE,
592 )
593 return
594
595 # Fetch and start playing the first track immediately
596 try:
597 first_track = await self.provider.get_track(first_track_key)
598 LOGGER.info(f"Starting playback with first track: {first_track.name}")
599
600 # Store first track's play queue item ID mapping
601 if first_play_queue_item_id:
602 self.play_queue_item_ids[0] = first_play_queue_item_id
603
604 # Start playback immediately with just the first track
605 await self.provider.mass.player_queues.play_media(
606 queue_id=player_id,
607 media=first_track,
608 option=QueueOption.REPLACE,
609 )
610
611 # Seek to offset if specified (do this before loading remaining tracks)
612 if offset > 0:
613 await self._seek_to_offset_after_playback(player_id, offset)
614
615 # Broadcast timeline update immediately
616 await self._broadcast_timeline()
617
618 # Now load the remaining tracks in the background
619 self.provider.mass.create_task(
620 self._load_remaining_queue_tracks(
621 player_id, playqueue, selected_offset, shuffle
622 )
623 )
624
625 except Exception as e:
626 LOGGER.exception(f"Error starting playback with first track: {e}")
627 # Fall back to single track
628 if starting_key:
629 track = await self.provider.get_track(starting_key)
630 await self.provider.mass.player_queues.play_media(
631 queue_id=player_id,
632 media=track,
633 option=QueueOption.REPLACE,
634 )
635 else:
636 LOGGER.error("Play queue is empty or could not be fetched")
637 # Fall back to single track
638 if starting_key:
639 track = await self.provider.get_track(starting_key)
640 await self.provider.mass.player_queues.play_media(
641 queue_id=player_id,
642 media=track,
643 option=QueueOption.REPLACE,
644 )
645
646 except Exception as e:
647 LOGGER.exception(f"Error playing from queue: {e}")
648 # Fall back to single track
649 if starting_key:
650 track = await self.provider.get_track(starting_key)
651 await self.provider.mass.player_queues.play_media(
652 queue_id=player_id,
653 media=track,
654 option=QueueOption.REPLACE,
655 )
656
657 async def _load_remaining_queue_tracks(
658 self,
659 player_id: str,
660 playqueue: PlayQueue,
661 selected_offset: int,
662 shuffle: bool,
663 ) -> None:
664 """Load remaining tracks from play queue in the background.
665
666 :param player_id: The Music Assistant player ID.
667 :param playqueue: The Plex play queue.
668 :param selected_offset: The offset of the track that's already playing.
669 :param shuffle: Whether shuffle is enabled.
670 """
671 try:
672 # Prepare to fetch all tracks except the first one
673 remaining_items = []
674
675 # Get items after selected track
676 for i in range(selected_offset + 1, len(playqueue.items)):
677 remaining_items.append((i, playqueue.items[i]))
678
679 # Get items before selected track (these will be added at the end)
680 for i in range(selected_offset):
681 remaining_items.append((i, playqueue.items[i]))
682
683 if not remaining_items:
684 LOGGER.debug("No remaining tracks to load")
685 return
686
687 # Fetch all remaining tracks concurrently
688 async def fetch_track(
689 plex_idx: int, item: Any
690 ) -> tuple[int, object | None, int | None]:
691 """Fetch a single track from Plex."""
692 track_key = item.key if hasattr(item, "key") else None
693 play_queue_item_id = (
694 item.playQueueItemID if hasattr(item, "playQueueItemID") else None
695 )
696
697 if track_key:
698 try:
699 track = await self.provider.get_track(track_key)
700 return plex_idx, track, play_queue_item_id
701 except Exception as e:
702 LOGGER.debug(f"Could not fetch track {track_key}: {e}")
703
704 return plex_idx, None, None
705
706 # Fetch all tracks in parallel
707 fetch_tasks = [fetch_track(idx, item) for idx, item in remaining_items]
708 results = await asyncio.gather(*fetch_tasks, return_exceptions=True)
709
710 # Process results and build track list
711 tracks_to_add: list[object] = []
712 for result in results:
713 if isinstance(result, Exception):
714 LOGGER.debug(f"Error fetching track: {result}")
715 continue
716
717 # result is guaranteed to be a tuple here after the Exception check
718 _plex_idx, track, play_queue_item_id = result # type: ignore[misc]
719 if track:
720 ma_idx = len(tracks_to_add) + 1 # +1 because first track is already queued
721 tracks_to_add.append(track)
722
723 # Store play queue item ID mapping
724 if play_queue_item_id:
725 self.play_queue_item_ids[ma_idx] = play_queue_item_id
726
727 if tracks_to_add:
728 LOGGER.info(f"Adding {len(tracks_to_add)} remaining tracks to queue")
729
730 # Add remaining tracks to the queue
731 await self.provider.mass.player_queues.play_media(
732 queue_id=player_id,
733 media=tracks_to_add, # type: ignore[arg-type]
734 option=QueueOption.ADD,
735 )
736
737 # Update tracked state to prevent sync loop
738 queue_items = self.provider.mass.player_queues.items(player_id)
739 synced_keys = []
740 for item in queue_items:
741 if item.media_item:
742 for mapping in item.media_item.provider_mappings:
743 if mapping.provider_instance == self.provider.instance_id:
744 synced_keys.append(mapping.item_id)
745 break
746 self._last_synced_ma_queue_length = len(synced_keys)
747 self._last_synced_ma_queue_keys = synced_keys
748
749 # Apply shuffle if requested (after all tracks are loaded)
750 if shuffle:
751 await self.provider.mass.player_queues.set_shuffle(player_id, shuffle)
752
753 LOGGER.info(
754 f"Successfully loaded {len(tracks_to_add)} remaining tracks "
755 f"(total queue: {len(synced_keys)} tracks)"
756 )
757 else:
758 LOGGER.warning("No valid remaining tracks found in play queue")
759
760 except Exception as e:
761 LOGGER.exception(f"Error loading remaining queue tracks: {e}")
762
763 async def _replace_entire_queue(self, player_id: str, playqueue: PlayQueue) -> None:
764 """Replace the entire queue when nothing is currently playing.
765
766 :param player_id: The Music Assistant player ID.
767 :param playqueue: The Plex play queue to load.
768 """
769 all_tracks = []
770 self.play_queue_item_ids = {}
771
772 for i, item in enumerate(playqueue.items):
773 track_key = item.key if hasattr(item, "key") else None
774 play_queue_item_id = item.playQueueItemID if hasattr(item, "playQueueItemID") else None
775
776 if track_key:
777 try:
778 track = await self.provider.get_track(track_key)
779 all_tracks.append(track)
780
781 if play_queue_item_id:
782 self.play_queue_item_ids[len(all_tracks) - 1] = play_queue_item_id
783 except Exception as e:
784 LOGGER.debug(f"Could not fetch track {track_key}: {e}")
785 continue
786
787 if all_tracks:
788 await self.provider.mass.player_queues.play_media(
789 queue_id=player_id,
790 media=all_tracks, # type: ignore[arg-type]
791 option=QueueOption.REPLACE,
792 )
793 LOGGER.info(f"Replaced queue with {len(all_tracks)} tracks")
794
795 async def _replace_remaining_queue(
796 self, player_id: str, playqueue: PlayQueue, current_index: int
797 ) -> None:
798 """Replace only items after the current track.
799
800 :param player_id: The Music Assistant player ID.
801 :param playqueue: The Plex play queue to load.
802 :param current_index: The current track index in the MA queue.
803 """
804 # Fetch tracks that come AFTER the current track in the Plex queue
805 remaining_tracks = []
806 new_item_mappings = {}
807
808 # Start from the track after current_index
809 for i in range(current_index + 1, len(playqueue.items)):
810 item = playqueue.items[i]
811 track_key = item.key if hasattr(item, "key") else None
812 play_queue_item_id = item.playQueueItemID if hasattr(item, "playQueueItemID") else None
813
814 if track_key:
815 try:
816 track = await self.provider.get_track(track_key)
817 remaining_tracks.append(track)
818
819 # Map relative to the current position
820 if play_queue_item_id:
821 new_item_mappings[current_index + 1 + len(remaining_tracks) - 1] = (
822 play_queue_item_id
823 )
824 except Exception as e:
825 LOGGER.debug(f"Could not fetch track {track_key}: {e}")
826 continue
827
828 # Replace items after current track
829 if remaining_tracks:
830 await self.provider.mass.player_queues.play_media(
831 queue_id=player_id,
832 media=remaining_tracks, # type: ignore[arg-type]
833 option=QueueOption.REPLACE_NEXT, # Replace everything after current
834 )
835 # Update mappings for the new items
836 self.play_queue_item_ids.update(new_item_mappings)
837
838 LOGGER.info(
839 f"Replaced {len(remaining_tracks)} tracks after current track "
840 f"(index {current_index})"
841 )
842 else:
843 # No tracks after current - clear remaining queue
844 LOGGER.debug("No tracks after current track in Plex queue")
845
846 # Rebuild complete item ID mappings from Plex queue
847 # Keep mappings for tracks from index 0 to current_index unchanged
848 for i, item in enumerate(playqueue.items):
849 play_queue_item_id = item.playQueueItemID if hasattr(item, "playQueueItemID") else None
850 if play_queue_item_id:
851 self.play_queue_item_ids[i] = play_queue_item_id
852
853 async def handle_refresh_play_queue(self, request: web.Request) -> web.Response:
854 """
855 Handle refreshPlayQueue command from Plex controller.
856
857 This is called when the play queue is modified (items added, removed, reordered).
858 We need to sync the entire updated queue state to MA while preserving playback.
859 """
860 try:
861 play_queue_id = request.query.get("playQueueID")
862
863 if not play_queue_id:
864 return web.Response(status=400, text="Missing 'playQueueID' parameter")
865
866 # Log all query parameters to understand what Plex sends
867 LOGGER.info(
868 f"Received refreshPlayQueue command - playQueueID: {play_queue_id}, "
869 f"params: {dict(request.query)}"
870 )
871
872 # Verify this is our active play queue
873 if self.play_queue_id != play_queue_id:
874 LOGGER.warning(
875 f"Refresh requested for queue {play_queue_id} but active queue is "
876 f"{self.play_queue_id}"
877 )
878 return web.Response(
879 status=409,
880 text=(
881 f"Requested playQueueID {play_queue_id} does not match "
882 f"active queue {self.play_queue_id}"
883 ),
884 )
885
886 # Update the play queue version (increments on each refresh)
887 self.play_queue_version += 1
888
889 # Use plexapi to fetch the updated play queue
890 def fetch_queue() -> PlayQueue:
891 return PlayQueue.get(self.provider._plex_server, playQueueID=play_queue_id)
892
893 playqueue = await asyncio.to_thread(fetch_queue)
894
895 if not playqueue or not playqueue.items:
896 LOGGER.error("Failed to refresh play queue - queue is empty or not found")
897 return web.Response(status=404, text="Play queue not found")
898
899 # Get current MA queue state
900 player_id = self._ma_player_id
901 if not player_id:
902 LOGGER.error("No player assigned to this server")
903 return web.Response(status=500, text="No player assigned")
904
905 # disable shuffle to avoid infinite loop
906 await self.provider.mass.player_queues.set_shuffle(player_id, False)
907 ma_queue = self.provider.mass.player_queues.get(player_id)
908 if not ma_queue:
909 LOGGER.error(f"MA queue not found for player {player_id}")
910 return web.Response(status=500, text="MA queue not found")
911
912 # Get current playback state
913 current_index = ma_queue.current_index
914
915 # Get MA queue item count
916 ma_queue_items = self.provider.mass.player_queues.items(player_id)
917 ma_queue_count = len(ma_queue_items) if ma_queue_items else 0
918
919 LOGGER.debug(
920 f"Queue refresh: Current index={current_index}, "
921 f"MA has {ma_queue_count} items, Plex has {len(playqueue.items)} items"
922 )
923
924 # If nothing is playing, replace the entire queue
925 if current_index is None:
926 LOGGER.debug("No track currently playing, replacing entire queue")
927 await self._replace_entire_queue(player_id, playqueue)
928 else:
929 # Something is playing - update only the remaining queue items
930 LOGGER.debug(
931 f"Track at index {current_index} is playing, "
932 f"replacing only items after current track"
933 )
934 await self._replace_remaining_queue(player_id, playqueue, current_index)
935
936 LOGGER.info(
937 f"Refreshed play queue {play_queue_id} - now has {len(playqueue.items)} items"
938 )
939
940 # Update tracked state to prevent sync loop
941 # Get what's actually in MA queue after the refresh
942 queue_items_after = self.provider.mass.player_queues.items(player_id)
943 synced_keys = []
944 for item in queue_items_after:
945 if item.media_item:
946 for mapping in item.media_item.provider_mappings:
947 if mapping.provider_instance == self.provider.instance_id:
948 synced_keys.append(mapping.item_id)
949 break
950 self._last_synced_ma_queue_length = len(synced_keys)
951 self._last_synced_ma_queue_keys = synced_keys
952
953 return web.Response(status=200)
954
955 except Exception as e:
956 LOGGER.exception(f"Error handling refreshPlayQueue: {e}")
957 return web.Response(status=500, text=str(e))
958
959 async def handle_create_play_queue(self, request: web.Request) -> web.Response:
960 """
961 Handle createPlayQueue command from Plex controller.
962
963 Creates a new play queue from a URI (album, playlist, artist tracks, etc.)
964 and optionally applies shuffle.
965 """
966 try:
967 uri = request.query.get("uri")
968 shuffle = request.query.get("shuffle", "0") == "1"
969 continuous = request.query.get("continuous", "0") == "1"
970
971 if not uri:
972 return web.Response(status=400, text="Missing 'uri' parameter")
973
974 LOGGER.info(f"Received createPlayQueue command - uri: {uri}, shuffle: {shuffle}")
975
976 # Use the assigned player for this server instance
977 player_id = self._ma_player_id
978 if not player_id:
979 return web.Response(status=500, text="No player assigned to this server")
980
981 # Use plexapi to create play queue
982 def create_queue() -> PlayQueue:
983 # Fetch the item from URI first
984 item = self.provider._plex_server.fetchItem(uri)
985 # Create play queue from the item
986 return PlayQueue.create(
987 self.provider._plex_server,
988 item,
989 shuffle=1 if shuffle else 0,
990 continuous=1 if continuous else 0,
991 )
992
993 playqueue = await asyncio.to_thread(create_queue)
994
995 if playqueue and playqueue.items:
996 # Extract play queue ID from response
997 self.play_queue_id = str(playqueue.playQueueID)
998 self.play_queue_version = 1
999
1000 LOGGER.info(
1001 f"Created play queue {self.play_queue_id} with {len(playqueue.items)} items"
1002 )
1003
1004 # Fetch the first track to start playback immediately
1005 self.play_queue_item_ids = {}
1006 first_item = playqueue.items[0]
1007 first_track_key = first_item.key if hasattr(first_item, "key") else None
1008 first_play_queue_item_id = (
1009 first_item.playQueueItemID if hasattr(first_item, "playQueueItemID") else None
1010 )
1011
1012 if not first_track_key:
1013 LOGGER.error("No valid first track in created play queue")
1014 return web.Response(status=500, text="Failed to load tracks from play queue")
1015
1016 try:
1017 # Fetch and start playing the first track immediately
1018 first_track = await self.provider.get_track(first_track_key)
1019 LOGGER.info(f"Starting playback with first track: {first_track.name}")
1020
1021 # Store first track's play queue item ID mapping
1022 if first_play_queue_item_id:
1023 self.play_queue_item_ids[0] = first_play_queue_item_id
1024
1025 # Start playback immediately with just the first track
1026 await self.provider.mass.player_queues.play_media(
1027 queue_id=player_id,
1028 media=first_track,
1029 option=QueueOption.REPLACE,
1030 )
1031
1032 # Now load the remaining tracks in the background
1033 if len(playqueue.items) > 1:
1034 self.provider.mass.create_task(
1035 self._load_remaining_queue_tracks(
1036 player_id,
1037 playqueue,
1038 0, # Selected offset is 0 since we started from the first track
1039 shuffle,
1040 )
1041 )
1042
1043 # Broadcast timeline update
1044 await self._broadcast_timeline()
1045 return web.Response(status=200)
1046
1047 except Exception as e:
1048 LOGGER.exception(f"Error starting playback with first track: {e}")
1049 return web.Response(status=500, text=f"Failed to start playback: {e}")
1050 else:
1051 LOGGER.error("Failed to create play queue or queue is empty")
1052 return web.Response(status=500, text="Failed to create play queue")
1053
1054 except Exception as e:
1055 LOGGER.exception(f"Error handling createPlayQueue: {e}")
1056 return web.Response(status=500, text=str(e))
1057
1058 async def _resolve_plex_item(self, key: str) -> object:
1059 """Resolve a Plex key to a Music Assistant media item."""
1060 # Determine item type from the key format
1061 if "/library/metadata/" in key:
1062 # Could be track, album, or artist
1063 # Try to fetch as track first
1064 try:
1065 return await self.provider.get_track(key)
1066 except Exception as exc:
1067 LOGGER.debug(f"Failed to resolve Plex item as track for key '{key}': {exc}")
1068
1069 # Try as album
1070 try:
1071 return await self.provider.get_album(key)
1072 except Exception as exc:
1073 LOGGER.debug(f"Failed to resolve Plex item as album for key '{key}': {exc}")
1074
1075 # Try as artist
1076 try:
1077 return await self.provider.get_artist(key)
1078 except Exception:
1079 raise ValueError(f"Could not resolve Plex item: {key}") from None
1080
1081 elif "/playlists/" in key:
1082 return await self.provider.get_playlist(key)
1083 else:
1084 raise ValueError(f"Unknown Plex key format: {key}")
1085
1086 async def handle_pause(self, request: web.Request) -> web.Response:
1087 """Handle pause command (test-client.py line 98-101)."""
1088 self._updating_from_plex = True
1089 try:
1090 if self._ma_player_id:
1091 await self.provider.mass.players.cmd_pause(self._ma_player_id)
1092 await self._broadcast_timeline()
1093 return web.Response(status=200)
1094 finally:
1095 self._updating_from_plex = False
1096
1097 async def handle_play(self, request: web.Request) -> web.Response:
1098 """Handle play/resume command (test-client.py line 103-106)."""
1099 self._updating_from_plex = True
1100 try:
1101 if self._ma_player_id:
1102 # Ungroup player before resuming playback
1103 await self._ungroup_player_if_needed(self._ma_player_id)
1104 await self.provider.mass.players.cmd_play(self._ma_player_id)
1105 await self._broadcast_timeline()
1106 return web.Response(status=200)
1107 finally:
1108 self._updating_from_plex = False
1109
1110 async def handle_stop(self, request: web.Request) -> web.Response:
1111 """Handle stop command - stops playback and clears the queue."""
1112 self._updating_from_plex = True
1113 try:
1114 if self._ma_player_id:
1115 # Clear the queue (which also stops playback)
1116 self.provider.mass.player_queues.clear(self._ma_player_id)
1117
1118 # Reset play queue tracking since the queue is now cleared
1119 self.play_queue_id = None
1120 self.play_queue_item_ids = {}
1121
1122 await self._broadcast_timeline()
1123 return web.Response(status=200)
1124 finally:
1125 self._updating_from_plex = False
1126
1127 async def handle_skip_next(self, request: web.Request) -> web.Response:
1128 """Handle skip next command."""
1129 self._updating_from_plex = True
1130 try:
1131 if self._ma_player_id:
1132 await self.provider.mass.player_queues.next(self._ma_player_id)
1133 await self._broadcast_timeline()
1134 return web.Response(status=200)
1135 finally:
1136 self._updating_from_plex = False
1137
1138 async def handle_skip_previous(self, request: web.Request) -> web.Response:
1139 """Handle skip previous command."""
1140 self._updating_from_plex = True
1141 try:
1142 if self._ma_player_id:
1143 await self.provider.mass.player_queues.previous(self._ma_player_id)
1144 await self._broadcast_timeline()
1145 return web.Response(status=200)
1146 finally:
1147 self._updating_from_plex = False
1148
1149 async def handle_step_forward(self, request: web.Request) -> web.Response:
1150 """Handle step forward command (small skip forward)."""
1151 self._updating_from_plex = True
1152 try:
1153 if self._ma_player_id:
1154 queue = self.provider.mass.player_queues.get(self._ma_player_id)
1155 if queue:
1156 # Step forward 30 seconds
1157 new_position = queue.corrected_elapsed_time + 30
1158 if queue.current_item and queue.current_item.media_item:
1159 # Don't seek past the track duration
1160 max_duration = queue.current_item.media_item.duration or new_position
1161 new_position = min(new_position, max_duration)
1162 await self.provider.mass.players.cmd_seek(self._ma_player_id, int(new_position))
1163 # Wait briefly for player state to update
1164 await asyncio.sleep(0.1)
1165 await self._broadcast_timeline()
1166 return web.Response(status=200)
1167 finally:
1168 self._updating_from_plex = False
1169
1170 async def handle_step_back(self, request: web.Request) -> web.Response:
1171 """Handle step back command (small skip backward)."""
1172 self._updating_from_plex = True
1173 try:
1174 if self._ma_player_id:
1175 queue = self.provider.mass.player_queues.get(self._ma_player_id)
1176 if queue:
1177 # Step back 10 seconds
1178 new_position = max(0, queue.corrected_elapsed_time - 10)
1179 await self.provider.mass.players.cmd_seek(self._ma_player_id, int(new_position))
1180 # Wait briefly for player state to update
1181 await asyncio.sleep(0.1)
1182 await self._broadcast_timeline()
1183 return web.Response(status=200)
1184 finally:
1185 self._updating_from_plex = False
1186
1187 async def handle_skip_to(self, request: web.Request) -> web.Response:
1188 """Handle skip to specific queue item."""
1189 key = request.query.get("key")
1190 if not self._ma_player_id or not key:
1191 return web.Response(status=400, text="Missing player ID or key")
1192
1193 self._updating_from_plex = True
1194 try:
1195 ma_index = None
1196
1197 # Check if key is a play queue item ID (numeric) or a library path
1198 if key.isdigit():
1199 # Key is a play queue item ID
1200 play_queue_item_id = int(key)
1201
1202 # Find the MA queue index for this play queue item ID
1203 for idx, pq_item_id in self.play_queue_item_ids.items():
1204 if pq_item_id == play_queue_item_id:
1205 ma_index = idx
1206 break
1207
1208 if ma_index is None:
1209 LOGGER.warning(
1210 f"Could not find MA queue index for play queue item ID: "
1211 f"{play_queue_item_id}"
1212 )
1213 return web.Response(status=404, text="Queue item not found")
1214
1215 LOGGER.info(
1216 f"Skipping to queue index {ma_index} (play queue item ID: {play_queue_item_id})"
1217 )
1218 else:
1219 # Key is a library path (e.g., "/library/metadata/856761")
1220 # Find the track in the MA queue by matching the Plex key
1221 queue_items = self.provider.mass.player_queues.items(self._ma_player_id)
1222 if not queue_items:
1223 return web.Response(status=404, text="Queue is empty")
1224
1225 for idx, item in enumerate(queue_items):
1226 if not item.media_item:
1227 continue
1228
1229 # Find Plex mapping for this track
1230 for mapping in item.media_item.provider_mappings:
1231 if (
1232 mapping.provider_instance == self.provider.instance_id
1233 and mapping.item_id == key
1234 ):
1235 ma_index = idx
1236 break
1237
1238 if ma_index is not None:
1239 break
1240
1241 if ma_index is None:
1242 LOGGER.warning(f"Could not find track with key {key} in MA queue")
1243 return web.Response(status=404, text="Track not found in queue")
1244
1245 LOGGER.info(f"Skipping to queue index {ma_index} (track key: {key})")
1246
1247 # Skip to this index in the MA queue
1248 await self.provider.mass.player_queues.play_index(self._ma_player_id, ma_index)
1249
1250 await self._broadcast_timeline()
1251 return web.Response(status=200)
1252
1253 except Exception as e:
1254 LOGGER.exception(f"Error handling skipTo: {e}")
1255 return web.Response(status=500, text=str(e))
1256 finally:
1257 self._updating_from_plex = False
1258
1259 async def handle_seek_to(self, request: web.Request) -> web.Response:
1260 """Handle seek command."""
1261 self._updating_from_plex = True
1262 try:
1263 offset_ms = int(request.query.get("offset", 0))
1264 if self._ma_player_id:
1265 await self.provider.mass.players.cmd_seek(self._ma_player_id, int(offset_ms / 1000))
1266 # Wait briefly for player state to update
1267 await asyncio.sleep(0.1)
1268 await self._broadcast_timeline()
1269 return web.Response(status=200)
1270 finally:
1271 self._updating_from_plex = False
1272
1273 async def handle_set_parameters(self, request: web.Request) -> web.Response:
1274 """Handle parameter changes (volume, shuffle, repeat)."""
1275 if not self._ma_player_id:
1276 return web.Response(status=200)
1277
1278 self._updating_from_plex = True
1279 try:
1280 if "volume" in request.query:
1281 volume = int(request.query["volume"])
1282 await self.provider.mass.players.cmd_volume_set(self._ma_player_id, volume)
1283
1284 if "shuffle" in request.query:
1285 # Plex sends shuffle as "0" or "1"
1286 shuffle = request.query["shuffle"] == "1"
1287 await self.provider.mass.player_queues.set_shuffle(self._ma_player_id, shuffle)
1288
1289 if "repeat" in request.query:
1290 # Plex repeat: 0=off, 1=repeat one, 2=repeat all
1291 repeat_value = int(request.query["repeat"])
1292
1293 # Map Plex repeat to MA repeat mode
1294 if repeat_value == 0:
1295 # Repeat off
1296 self.provider.mass.player_queues.set_repeat(self._ma_player_id, RepeatMode.OFF)
1297 elif repeat_value == 1:
1298 # Repeat one track
1299 self.provider.mass.player_queues.set_repeat(self._ma_player_id, RepeatMode.ONE)
1300 elif repeat_value == 2:
1301 # Repeat all
1302 self.provider.mass.player_queues.set_repeat(self._ma_player_id, RepeatMode.ALL)
1303
1304 await self._broadcast_timeline()
1305 return web.Response(status=200)
1306 finally:
1307 self._updating_from_plex = False
1308
1309 async def handle_options(self, request: web.Request) -> web.Response:
1310 """Handle OPTIONS requests for CORS (like test-client.py)."""
1311 return web.Response(
1312 status=200,
1313 headers={
1314 "Access-Control-Allow-Origin": "*",
1315 "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
1316 "Access-Control-Allow-Headers": "*",
1317 },
1318 )
1319
1320 async def handle_resources(self, request: web.Request) -> web.Response:
1321 """Return player information (matching test-client.py format exactly)."""
1322 # Get player name
1323 player_name = "Music Assistant"
1324 if self._ma_player_id:
1325 player = self.provider.mass.players.get_player(self._ma_player_id)
1326 if player:
1327 player_name = player.display_name
1328
1329 # Get player state
1330 state = "stopped"
1331 if self._ma_player_id:
1332 player = self.provider.mass.players.get_player(self._ma_player_id)
1333 if player and player.state:
1334 state_value = (
1335 player.state.value if hasattr(player.state, "value") else str(player.state)
1336 )
1337 if state_value in ["playing", "paused"]:
1338 state = state_value
1339
1340 local_ip = self.provider.mass.streams.publish_ip
1341 version = self.provider.mass.version if self.provider.mass.version != "0.0.0" else "1.0.0"
1342
1343 # Match test-client.py format exactly
1344 xml = f"""<?xml version="1.0" encoding="UTF-8"?>
1345<MediaContainer>
1346 <Player title="{player_name}"
1347 protocol="plex"
1348 protocolVersion="1"
1349 protocolCapabilities="timeline,playback,navigation,playqueues"
1350 machineIdentifier="{self.client_id}"
1351 product="Music Assistant"
1352 platform="{platform.system()}"
1353 platformVersion="{platform.release()}"
1354 deviceClass="{self.device_class}"
1355 state="{state}"
1356 address="{local_ip}"
1357 port="{self.port}"
1358 version="{version}"
1359 provides="client,player,pubsub-player">
1360 <Connection protocol="http" address="{local_ip}" port="{self.port}"
1361 uri="http://{local_ip}:{self.port}" local="1"/>
1362 </Player>
1363</MediaContainer>"""
1364 return web.Response(
1365 text=xml, content_type="text/xml", headers={"Access-Control-Allow-Origin": "*"}
1366 )
1367
1368 def _build_timeline_attributes(
1369 self,
1370 track: Any,
1371 state: str,
1372 duration: int,
1373 time: int,
1374 volume: int,
1375 shuffle: int,
1376 repeat: int,
1377 controllable: str,
1378 queue: Any | None,
1379 ) -> list[str]:
1380 """Build timeline attributes for a playing track.
1381
1382 :param track: The current track media item.
1383 :param state: Playback state (playing, paused, etc.).
1384 :param duration: Track duration in milliseconds.
1385 :param time: Current playback time in milliseconds.
1386 :param volume: Volume level (0-100).
1387 :param shuffle: Shuffle state (0 or 1).
1388 :param repeat: Repeat mode (0=off, 1=one, 2=all).
1389 :param controllable: Controllable features string.
1390 :param queue: The MA queue object.
1391 :return: List of timeline attribute strings.
1392 """
1393 # Get Plex key and ratingKey
1394 key = None
1395 rating_key = None
1396 for mapping in track.provider_mappings:
1397 if mapping.provider_instance == self.provider.instance_id:
1398 key = mapping.item_id
1399 rating_key = key.split("/")[-1]
1400 break
1401
1402 if not key:
1403 return []
1404
1405 # Server identification
1406 plex_url = urlparse(self.provider._baseurl)
1407 machine_identifier = self.provider._plex_server.machineIdentifier
1408 address = plex_url.hostname
1409 port = plex_url.port or (443 if plex_url.scheme == "https" else 32400)
1410 protocol = plex_url.scheme
1411
1412 # Build timeline attributes
1413 attrs = [
1414 f'state="{state}"',
1415 f'duration="{duration}"',
1416 f'time="{time}"',
1417 f'ratingKey="{rating_key}"',
1418 f'key="{key}"',
1419 ]
1420
1421 # Add play queue info if available
1422 if self.play_queue_id and queue:
1423 if queue.current_index is not None:
1424 play_queue_item_id = self.play_queue_item_ids.get(
1425 queue.current_index, queue.current_index + 1
1426 )
1427 attrs.append(f'playQueueItemID="{play_queue_item_id}"')
1428 attrs.append(f'playQueueID="{self.play_queue_id}"')
1429 attrs.append(f'playQueueVersion="{self.play_queue_version}"')
1430 attrs.append(f'containerKey="/playQueues/{self.play_queue_id}"')
1431
1432 # Add standard attributes
1433 attrs.extend(
1434 [
1435 'type="music"',
1436 f'volume="{volume}"',
1437 f'shuffle="{shuffle}"',
1438 f'repeat="{repeat}"',
1439 f'controllable="{controllable}"',
1440 f'machineIdentifier="{machine_identifier}"',
1441 f'address="{address}"',
1442 f'port="{port}"',
1443 f'protocol="{protocol}"',
1444 ]
1445 )
1446
1447 return attrs
1448
1449 async def _build_timeline_xml(
1450 self, include_metadata: bool = False, command_id: str = "0"
1451 ) -> str:
1452 """Build timeline XML from current Music Assistant player state."""
1453 player_id = self._ma_player_id
1454
1455 # Get MA player and queue
1456 player = self.provider.mass.players.get_player(player_id) if player_id else None
1457 queue = self.provider.mass.player_queues.get(player_id) if player_id else None
1458
1459 # Controllable features for music
1460 controllable = (
1461 "volume,repeat,skipPrevious,seekTo,stepBack,stepForward,stop,playPause,shuffle,skipNext"
1462 )
1463
1464 # Map MA playback state to Plex state (stopped, paused, playing, buffering, error)
1465 state = "stopped"
1466 if player and player.playback_state:
1467 state_value = (
1468 player.playback_state.value
1469 if hasattr(player.playback_state, "value")
1470 else str(player.playback_state)
1471 )
1472
1473 # Map MA states to Plex states
1474 if state_value == "playing":
1475 state = "playing"
1476 elif state_value == "paused":
1477 state = "paused"
1478 elif state_value == "buffering":
1479 state = "buffering"
1480 elif state_value == "idle":
1481 # Idle with a current track = paused, idle without track = stopped
1482 state = (
1483 "paused"
1484 if queue and queue.current_item and queue.current_item.media_item
1485 else "stopped"
1486 )
1487 else:
1488 state = "stopped"
1489
1490 # Get volume (0-100) - use group_volume for groups, volume_level for others
1491 volume = 0
1492 if player:
1493 volume = (
1494 int(player.group_volume)
1495 if (player.type == PlayerType.GROUP or player.group_members)
1496 else (int(player.volume_level) if player.volume_level else 0)
1497 )
1498
1499 # Get shuffle (0/1) and repeat (0=off, 1=one, 2=all)
1500 shuffle = 0
1501 repeat = 0
1502 if queue:
1503 shuffle = 1 if queue.shuffle_enabled else 0
1504 if hasattr(queue, "repeat_mode"):
1505 repeat_mode = queue.repeat_mode
1506 if hasattr(repeat_mode, "value"):
1507 repeat_value = repeat_mode.value
1508 if repeat_value == "one":
1509 repeat = 1
1510 elif repeat_value == "all":
1511 repeat = 2
1512
1513 # Build music timeline
1514 if (
1515 state in ["playing", "paused"]
1516 and queue
1517 and queue.current_item
1518 and queue.current_item.media_item
1519 ):
1520 track = queue.current_item.media_item
1521
1522 # Duration in milliseconds
1523 duration = round(track.duration * 1000) if track.duration else 0
1524
1525 # Current playback time in milliseconds
1526 time = round(queue.corrected_elapsed_time * 1000)
1527
1528 # Build timeline attributes
1529 attrs = self._build_timeline_attributes(
1530 track, state, duration, time, volume, shuffle, repeat, controllable, queue
1531 )
1532
1533 if attrs:
1534 music_timeline = f"<Timeline {' '.join(attrs)}/>"
1535 else:
1536 # No Plex mapping, send basic timeline with actual state
1537 music_timeline = (
1538 f'<Timeline state="{state}" time="{time}" type="music" volume="{volume}" '
1539 f'shuffle="{shuffle}" repeat="{repeat}" controllable="{controllable}"/>'
1540 )
1541 else:
1542 # No current track - send stopped state with time=0
1543 time = 0
1544 music_timeline = (
1545 f'<Timeline state="{state}" time="{time}" type="music" volume="{volume}" '
1546 f'shuffle="{shuffle}" repeat="{repeat}" controllable="{controllable}"/>'
1547 )
1548
1549 # Video and photo timelines (always stopped for music player)
1550 video_timeline = '<Timeline type="video" state="stopped"/>'
1551 photo_timeline = '<Timeline type="photo" state="stopped"/>'
1552
1553 # Combine all timelines
1554 return (
1555 f'<MediaContainer commandID="{command_id}">'
1556 f"{music_timeline}{video_timeline}{photo_timeline}"
1557 f"</MediaContainer>"
1558 )
1559
1560 async def _handle_player_event(self, event: MassEvent) -> None:
1561 """Handle player state change events."""
1562 if not self._ma_player_id or event.object_id != self._ma_player_id:
1563 return
1564
1565 # Skip if we're the ones making the changes
1566 if self._updating_from_plex:
1567 return
1568
1569 try:
1570 # Send timeline to Plex server (for activity tracking)
1571 await self._send_timeline_to_server()
1572
1573 # Broadcast timeline to subscribed controllers
1574 # Timeline will be built from current MA player state
1575 await self._broadcast_timeline()
1576 except Exception as e:
1577 LOGGER.debug(f"Error handling player event: {e}")
1578
1579 async def _handle_queue_event(self, event: MassEvent) -> None:
1580 """Handle queue change events."""
1581 if not self._ma_player_id or event.object_id != self._ma_player_id:
1582 return
1583
1584 # Skip if we're the ones making the changes
1585 if self._updating_from_plex:
1586 return
1587
1588 try:
1589 # Send timeline to Plex server (for activity tracking)
1590 await self._send_timeline_to_server()
1591
1592 # Broadcast timeline to subscribed controllers
1593 # Timeline will be built from current MA player state
1594 await self._broadcast_timeline()
1595 except Exception as e:
1596 LOGGER.debug(f"Error handling queue event: {e}")
1597
1598 async def _handle_queue_items_updated(self, event: MassEvent) -> None:
1599 """Handle queue items being added/removed/reordered."""
1600 if not self._ma_player_id or event.object_id != self._ma_player_id:
1601 return
1602
1603 # Skip if we're the ones making the changes
1604 if self._updating_from_plex:
1605 return
1606
1607 # Get current MA queue state
1608 queue_items = self.provider.mass.player_queues.items(self._ma_player_id)
1609 if not queue_items:
1610 return
1611
1612 current_keys = []
1613 for item in queue_items:
1614 if not item.media_item:
1615 continue
1616 # Find Plex mapping
1617 for mapping in item.media_item.provider_mappings:
1618 if mapping.provider_instance == self.provider.instance_id:
1619 current_keys.append(mapping.item_id)
1620 break
1621
1622 # Check if queue actually changed from what we last synced FROM Plex
1623 if (
1624 len(current_keys) == self._last_synced_ma_queue_length
1625 and current_keys == self._last_synced_ma_queue_keys
1626 ):
1627 # Queue hasn't changed from last sync, skip
1628 LOGGER.debug("MA queue matches last synced state, skipping Plex sync")
1629 return
1630
1631 LOGGER.info(
1632 f"MA queue changed: {self._last_synced_ma_queue_length} -> {len(current_keys)} items"
1633 )
1634
1635 # (Re)create Plex PlayQueue from MA queue
1636 try:
1637 await self._create_plex_playqueue_from_ma()
1638 # Update tracked state
1639 self._last_synced_ma_queue_length = len(current_keys)
1640 self._last_synced_ma_queue_keys = current_keys
1641 except Exception as e:
1642 LOGGER.debug(f"Error creating Plex PlayQueue: {e}")
1643
1644 # Broadcast timeline update
1645 try:
1646 await self._broadcast_timeline()
1647 except Exception as e:
1648 LOGGER.debug(f"Error broadcasting timeline: {e}")
1649
1650 async def _create_plex_playqueue_from_ma(self) -> None:
1651 """Create a new Plex PlayQueue from current MA queue."""
1652 ma_queue = self.provider.mass.player_queues.get(self._ma_player_id) # type: ignore[arg-type]
1653 queue_items = self.provider.mass.player_queues.items(self._ma_player_id) # type: ignore[arg-type]
1654
1655 if not ma_queue or not queue_items:
1656 return
1657
1658 # Fetch Plex items for all tracks in MA queue
1659 async def fetch_plex_item(plex_key: str) -> object | None:
1660 """Fetch a single Plex item."""
1661 try:
1662
1663 def fetch_item() -> object:
1664 return self.plex_server.fetchItem(plex_key)
1665
1666 return await asyncio.to_thread(fetch_item)
1667 except Exception as e:
1668 LOGGER.debug(f"Failed to fetch Plex item {plex_key}: {e}")
1669 return None
1670
1671 # Collect all fetch tasks
1672 fetch_tasks = []
1673 for item in queue_items:
1674 if not item.media_item:
1675 continue
1676
1677 # Find Plex mapping
1678 plex_key = None
1679 for mapping in item.media_item.provider_mappings:
1680 if mapping.provider_instance == self.provider.instance_id:
1681 plex_key = mapping.item_id
1682 break
1683
1684 if plex_key:
1685 fetch_tasks.append(fetch_plex_item(plex_key))
1686
1687 # Fetch all items concurrently
1688 plex_items = []
1689 if fetch_tasks:
1690 fetched_items = await asyncio.gather(*fetch_tasks, return_exceptions=True)
1691 plex_items = [item for item in fetched_items if item is not None]
1692
1693 if not plex_items:
1694 LOGGER.debug("No Plex tracks in MA queue, skipping PlayQueue creation")
1695 return
1696
1697 # Determine which track should be selected (currently playing)
1698 start_item = None
1699 if ma_queue.current_index is not None and ma_queue.current_index < len(plex_items):
1700 start_item = plex_items[ma_queue.current_index]
1701
1702 # Create Plex PlayQueue - don't pass shuffle since MA queue is already in desired order
1703 def create_queue() -> PlayQueue:
1704 return PlayQueue.create(
1705 self.plex_server,
1706 items=plex_items,
1707 startItem=start_item,
1708 shuffle=0, # Don't shuffle, plex_items is already in MA queue order
1709 continuous=1,
1710 )
1711
1712 try:
1713 playqueue = await asyncio.to_thread(create_queue)
1714
1715 if playqueue:
1716 self.play_queue_id = str(playqueue.playQueueID)
1717 self.play_queue_version = playqueue.playQueueVersion
1718
1719 # Build item ID mappings
1720 self.play_queue_item_ids = {}
1721 for i, item in enumerate(playqueue.items):
1722 if hasattr(item, "playQueueItemID"):
1723 self.play_queue_item_ids[i] = item.playQueueItemID
1724
1725 LOGGER.info(
1726 f"Created Plex PlayQueue {self.play_queue_id} with {len(plex_items)} tracks"
1727 )
1728 except Exception as e:
1729 LOGGER.exception(f"Error creating Plex PlayQueue: {e}")
1730
1731 async def _send_timeline(self, client_id: str) -> None:
1732 """Send timeline update to specific controller."""
1733 subscription = self.subscriptions.get(client_id)
1734 if not subscription:
1735 return
1736
1737 timeline_xml = await self._build_timeline_xml()
1738
1739 try:
1740 await self.provider.mass.http_session.post(
1741 f"{subscription['url']}/:/timeline",
1742 data=timeline_xml,
1743 headers={
1744 "X-Plex-Client-Identifier": self.client_id,
1745 "Content-Type": "text/xml",
1746 },
1747 timeout=ClientTimeout(total=5),
1748 )
1749 # Update last_update timestamp on successful send
1750 subscription["last_update"] = time.time()
1751 except Exception as e:
1752 LOGGER.debug(f"Failed to send timeline to {client_id}: {e}")
1753
1754 async def _send_timeline_to_server(self) -> None:
1755 """Send timeline update to Plex server for activity tracking."""
1756 if not self._ma_player_id:
1757 return
1758
1759 try:
1760 player = self.provider.mass.players.get_player(self._ma_player_id)
1761 queue = self.provider.mass.player_queues.get(self._ma_player_id)
1762
1763 if (
1764 not player
1765 or not queue
1766 or not queue.current_item
1767 or not queue.current_item.media_item
1768 ):
1769 return
1770
1771 track = queue.current_item.media_item
1772
1773 # Find Plex mapping
1774 plex_key = None
1775 for mapping in track.provider_mappings:
1776 if mapping.provider_instance == self.provider.instance_id:
1777 plex_key = mapping.item_id
1778 break
1779
1780 if not plex_key:
1781 return
1782
1783 # Extract rating key from plex_key (e.g., "/library/metadata/12345" -> "12345")
1784 rating_key = plex_key.split("/")[-1]
1785
1786 # Get playback state
1787 state_value = (
1788 player.playback_state.value
1789 if hasattr(player.playback_state, "value")
1790 else str(player.playback_state)
1791 )
1792
1793 # Map to Plex state
1794 if state_value == "playing":
1795 plex_state = "playing"
1796 elif state_value == "paused":
1797 plex_state = "paused"
1798 else:
1799 plex_state = "stopped"
1800
1801 # Get position and duration in milliseconds
1802 position_ms = round(queue.corrected_elapsed_time * 1000)
1803 duration_ms = round(track.duration * 1000) if track.duration else 0
1804
1805 # Get play queue info if available
1806 container_key = ""
1807 play_queue_item_id = ""
1808 if self.play_queue_id:
1809 container_key = f"/playQueues/{self.play_queue_id}"
1810 if queue.current_index is not None:
1811 play_queue_item_id = str(
1812 self.play_queue_item_ids.get(queue.current_index, queue.current_index + 1)
1813 )
1814
1815 # Build timeline params (only Plex timeline data)
1816 params = {
1817 "ratingKey": rating_key,
1818 "key": plex_key,
1819 "state": plex_state,
1820 "time": str(position_ms),
1821 "duration": str(duration_ms),
1822 }
1823
1824 # Add play queue info if available
1825 if container_key:
1826 params["containerKey"] = container_key
1827 if play_queue_item_id:
1828 params["playQueueItemID"] = play_queue_item_id
1829
1830 def send_timeline() -> None:
1831 # Pass session headers to identify this specific player instance
1832 self.plex_server.query("/:/timeline", params=params, headers=self.headers)
1833
1834 await asyncio.to_thread(send_timeline)
1835
1836 except Exception as e:
1837 LOGGER.debug(f"Failed to send timeline to Plex server: {e}")
1838
1839 async def _broadcast_timeline(self) -> None:
1840 """Send timeline to all subscribed controllers."""
1841 current_time = time.time()
1842 stale_clients = []
1843 for client_id, sub in self.subscriptions.items():
1844 try:
1845 last_update = float(sub["last_update"]) # type: ignore[arg-type]
1846 if current_time - last_update > 90:
1847 stale_clients.append(client_id)
1848 except (ValueError, TypeError):
1849 # If conversion fails, treat client as stale
1850 LOGGER.debug(f"Invalid last_update for client {client_id}, treating as stale")
1851 stale_clients.append(client_id)
1852
1853 for client_id in stale_clients:
1854 del self.subscriptions[client_id]
1855
1856 await asyncio.gather(
1857 *(self._send_timeline(client_id) for client_id in list(self.subscriptions.keys())),
1858 return_exceptions=True, # Don't fail all if one fails
1859 )
1860
1861 # for debugging purposes only
1862 # async def handle_unknown(self, request: web.Request) -> web.Response:
1863 # """Catch-all handler for unexpected or unsupported paths."""
1864 # LOGGER.debug(
1865 # "Unhandled request: %s %s from %s",
1866 # request.method,
1867 # request.path,
1868 # request.remote,
1869 # )
1870 #
1871 # # You can log query/body if needed (be careful not to leak tokens)
1872 # if request.query:
1873 # LOGGER.debug("Query params for %s: %s", request.path, dict(request.query))
1874 # try:
1875 # data = await request.text()
1876 # if data:
1877 # LOGGER.debug("Body for %s: %s", request.path, data)
1878 # except Exception as e:
1879 # LOGGER.debug("Could not read request body: %s", e)
1880 #
1881 # return web.Response(status=404, text=f"Unhandled path: {request.path}")
1882