music-assistant-server

1"""Per-player Plex remote control instances."""
2
3from __future__ import annotations
4
5import logging
6import platform
7import time
8import uuid
9from collections.abc import Callable
10from typing import TYPE_CHECKING
11
12from aiohttp import web
13from music_assistant_models.enums import EventType
14
15from .gdm import PlexGDMAdvertiser
16from .playback import PlaybackMixin
17from .queue_commands import QueueCommandsMixin
18from .queue_sync import QueueSyncMixin
19from .timeline import TimelineMixin
20
21if TYPE_CHECKING:
22    from music_assistant.providers.plex import PlexProvider
23
24LOGGER = logging.getLogger(__name__)
25
26
27class PlayerRemoteInstance:
28    """Single remote control instance for one MA player."""
29
30    def __init__(
31        self,
32        plex_provider: PlexProvider,
33        ma_player_id: str,
34        player_name: str,
35        port: int,
36        device_class: str = "speaker",
37        remote_control: bool = False,
38    ) -> None:
39        """
40        Initialize player remote instance.
41
42        :param plex_provider: Plex provider instance.
43        :param ma_player_id: Music Assistant player ID.
44        :param player_name: Display name for the player.
45        :param port: Port for the remote control server.
46        :param device_class: Device class (speaker, phone, tablet, stb, tv, pc, cloud).
47        :param remote_control: Whether to enable remote control.
48        """
49        self.plex_provider = plex_provider
50        self.plex_server = plex_provider._plex_server
51        self.ma_player_id = ma_player_id
52        self.player_name = player_name
53        self.port = port
54        self.device_class = device_class
55        self.remote_control = remote_control
56
57        self.client_id = str(
58            uuid.uuid5(
59                uuid.NAMESPACE_DNS,
60                f"music-assistant-plex-{plex_provider.instance_id}-{ma_player_id}",
61            )
62        )
63
64        if self.remote_control:
65            self.server: PlexRemoteControlServer | None = None
66            self.gdm: PlexGDMAdvertiser | None = None
67
68    async def start(self) -> None:
69        """Start this player's remote control."""
70        if self.remote_control:
71            LOGGER.info(
72                f"Created PlexServer for '{self.player_name}' with client ID: {self.client_id}"
73            )
74
75            self.server = PlexRemoteControlServer(
76                plex_provider=self.plex_provider,
77                port=self.port,
78                client_id=self.client_id,
79                ma_player_id=self.ma_player_id,
80                device_class=self.device_class,
81            )
82            LOGGER.info(
83                f"Remote control server for '{self.player_name}' bound to MA player: "
84                f"{self.ma_player_id}"
85            )
86
87            await self.server.start()
88
89            self.gdm = PlexGDMAdvertiser(
90                instance_id=self.client_id,
91                port=self.port,
92                publish_ip=str(self.plex_provider.mass.streams.publish_ip),
93                name=self.player_name,
94                product="Music Assistant",
95                version=self.plex_provider.mass.version
96                if self.plex_provider.mass.version != "0.0.0"
97                else "1.0.0",
98                device_class=self.device_class,
99            )
100            self.gdm.start()
101
102            LOGGER.info(f"Player '{self.player_name}' is now discoverable on port {self.port}")
103
104    async def stop(self) -> None:
105        """Stop this player's remote control."""
106        if self.remote_control:
107            if self.gdm:
108                await self.gdm.stop()
109
110            if self.server:
111                await self.server.stop()
112
113            LOGGER.info(f"Stopped remote control for player '{self.player_name}'")
114
115
116class PlexRemoteControlServer(QueueCommandsMixin, PlaybackMixin, QueueSyncMixin, TimelineMixin):
117    """HTTP server implementing the Plex remote control protocol for one MA player."""
118
119    def __init__(
120        self,
121        plex_provider: PlexProvider,
122        port: int = 32500,
123        client_id: str | None = None,
124        ma_player_id: str | None = None,
125        device_class: str = "speaker",
126    ) -> None:
127        """
128        Initialize remote control server.
129
130        :param plex_provider: Plex provider instance.
131        :param port: Port for the HTTP server.
132        :param client_id: Unique client identifier.
133        :param ma_player_id: Music Assistant player ID.
134        :param device_class: Device class (speaker, phone, tablet, stb, tv, pc, cloud).
135        """
136        self.provider = plex_provider
137        self.plex_server = plex_provider._plex_server
138        self.port = port
139        self.client_id = client_id or plex_provider.instance_id
140        self.device_class = device_class
141        self.app = web.Application()
142        self.subscriptions: dict[str, dict[str, object]] = {}
143        self.runner: web.AppRunner | None = None
144        self.http_site: web.TCPSite | None = None
145
146        # Play queue tracking (Plex-specific state that doesn't exist in MA)
147        self.play_queue_id: str | None = None
148        self.play_queue_version: int = 1
149        self.play_queue_item_ids: dict[int, int] = {}
150
151        # Track MA queue state to detect when we need to sync to Plex
152        self._last_synced_ma_queue_length: int = 0
153        self._last_synced_ma_queue_keys: list[str] = []
154
155        self._ma_player_id = ma_player_id
156
157        self._unsub_callbacks: list[Callable[..., None]] = []
158
159        # Flag to prevent circular updates when we modify the queue ourselves
160        self._updating_from_plex = False
161
162        self.player = self.provider.mass.players.get_player(self._ma_player_id)  # type: ignore[arg-type]
163
164        self.device_name = f"{self.player.display_name}" if self.player else "Music Assistant"
165
166        self.headers = {
167            "X-Plex-Device-Name": self.device_name,
168            "X-Plex-Session-Identifier": self.client_id,
169            "X-Plex-Client-Identifier": self.client_id,
170            "X-Plex-Product": "Music Assistant",
171            "X-Plex-Platform": "Music Assistant",
172            "X-Plex-Platform-Version": platform.release(),
173        }
174
175        self._setup_routes()
176
177    def _setup_routes(self) -> None:
178        """Set up all HTTP endpoints."""
179        self.app.router.add_get("/", self.handle_root)
180
181        self.app.router.add_get("/player/timeline/subscribe", self.handle_subscribe)
182        self.app.router.add_get("/player/timeline/unsubscribe", self.handle_unsubscribe)
183        self.app.router.add_get("/player/timeline/poll", self.handle_poll)
184
185        self.app.router.add_get("/player/playback/playMedia", self.handle_play_media)
186        self.app.router.add_get("/player/playback/refreshPlayQueue", self.handle_refresh_play_queue)
187        self.app.router.add_get("/player/playback/createPlayQueue", self.handle_create_play_queue)
188        self.app.router.add_get("/player/playback/pause", self.handle_pause)
189        self.app.router.add_get("/player/playback/play", self.handle_play)
190        self.app.router.add_get("/player/playback/stop", self.handle_stop)
191        self.app.router.add_get("/player/playback/skipNext", self.handle_skip_next)
192        self.app.router.add_get("/player/playback/skipPrevious", self.handle_skip_previous)
193        self.app.router.add_get("/player/playback/stepForward", self.handle_step_forward)
194        self.app.router.add_get("/player/playback/stepBack", self.handle_step_back)
195        self.app.router.add_get("/player/playback/seekTo", self.handle_seek_to)
196        self.app.router.add_get("/player/playback/setParameters", self.handle_set_parameters)
197        self.app.router.add_get("/player/playback/skipTo", self.handle_skip_to)
198
199        self.app.router.add_get("/resources", self.handle_resources)
200
201        self.app.router.add_route("OPTIONS", "/{tail:.*}", self.handle_options)
202
203    async def start(self) -> None:
204        """Start HTTP server and subscribe to MA events."""
205        self.runner = web.AppRunner(self.app)
206        await self.runner.setup()
207
208        self.http_site = web.TCPSite(self.runner, "0.0.0.0", self.port)
209        await self.http_site.start()
210        LOGGER.info(f"Plex remote control server started on HTTP port {self.port}")
211
212        if self._ma_player_id:
213            self._unsub_callbacks.append(
214                self.provider.mass.subscribe(
215                    self._handle_state_event,
216                    (
217                        EventType.PLAYER_UPDATED,
218                        EventType.QUEUE_UPDATED,
219                        EventType.QUEUE_TIME_UPDATED,
220                    ),
221                    id_filter=self._ma_player_id,
222                )
223            )
224            self._unsub_callbacks.append(
225                self.provider.mass.subscribe(
226                    self._handle_queue_items_updated,
227                    EventType.QUEUE_ITEMS_UPDATED,
228                    id_filter=self._ma_player_id,
229                )
230            )
231
232            # Mirror an already-active MA queue to Plex (runs in the background so it
233            # never blocks startup on Plex network calls).
234            self.provider.mass.create_task(self._sync_initial_queue_to_plex())
235
236    async def stop(self) -> None:
237        """Stop the HTTP server and unsubscribe from events."""
238        for unsub in self._unsub_callbacks:
239            unsub()
240        self._unsub_callbacks.clear()
241
242        if self.http_site:
243            await self.http_site.stop()
244        if self.runner:
245            await self.runner.cleanup()
246        LOGGER.info("Plex remote control server stopped")
247
248    async def handle_root(self, request: web.Request) -> web.Response:
249        """Handle root endpoint - return basic player info."""
250        player_name = "Music Assistant"
251        if self._ma_player_id:
252            player = self.provider.mass.players.get_player(self._ma_player_id)
253            if player:
254                player_name = player.display_name
255
256        xml = f"""<?xml version="1.0" encoding="UTF-8"?>
257<MediaContainer machineIdentifier="{self.client_id}" version="1.0">
258    <Player title="{player_name}" machineIdentifier="{self.client_id}"/>
259</MediaContainer>"""
260        return web.Response(
261            text=xml, content_type="text/xml", headers={"Access-Control-Allow-Origin": "*"}
262        )
263
264    async def handle_subscribe(self, request: web.Request) -> web.Response:
265        """Handle timeline subscription from controller."""
266        client_id = request.headers.get("X-Plex-Client-Identifier")
267        protocol = request.query.get("protocol", "http")
268        port = request.query.get("port")
269        command_id = int(request.query.get("commandID", 0))
270
271        if not client_id or not port:
272            return web.Response(status=400)
273
274        self.subscriptions[client_id] = {
275            "url": f"{protocol}://{request.remote}:{port}",
276            "command_id": command_id,
277            "last_update": time.time(),
278        }
279
280        LOGGER.info(f"Controller {client_id} subscribed for timeline updates")
281        await self._send_timeline(client_id)
282        return web.Response(status=200)
283
284    async def handle_unsubscribe(self, request: web.Request) -> web.Response:
285        """Handle unsubscribe request."""
286        client_id = request.headers.get("X-Plex-Client-Identifier")
287        if client_id in self.subscriptions:
288            del self.subscriptions[client_id]
289            LOGGER.info(f"Controller {client_id} unsubscribed")
290        return web.Response(status=200)
291
292    async def handle_poll(self, request: web.Request) -> web.Response:
293        """Handle timeline poll request."""
294        include_metadata = request.query.get("includeMetadata", "0") == "1"
295        command_id = request.query.get("commandID", "0")
296
297        client_id = request.headers.get("X-Plex-Client-Identifier")
298        if client_id and client_id in self.subscriptions:
299            self.subscriptions[client_id]["last_update"] = time.time()
300
301        timeline_xml = await self._build_timeline_xml(
302            include_metadata=include_metadata, command_id=command_id
303        )
304        return web.Response(
305            text=timeline_xml,
306            content_type="text/xml",
307            headers={
308                "X-Plex-Client-Identifier": self.client_id,
309                "Access-Control-Expose-Headers": "X-Plex-Client-Identifier",
310                "Access-Control-Allow-Origin": "*",
311            },
312        )
313
314    async def handle_options(self, request: web.Request) -> web.Response:
315        """Handle OPTIONS requests for CORS."""
316        return web.Response(
317            status=200,
318            headers={
319                "Access-Control-Allow-Origin": "*",
320                "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
321                "Access-Control-Allow-Headers": "*",
322            },
323        )
324
325    async def handle_resources(self, request: web.Request) -> web.Response:
326        """Return player capabilities and connection information."""
327        player_name = "Music Assistant"
328        state = "stopped"
329        player = (
330            self.provider.mass.players.get_player(self._ma_player_id)
331            if self._ma_player_id
332            else None
333        )
334        if player:
335            player_name = player.display_name
336            queue = self.provider.mass.players.get_active_queue(player)
337            state = self._resolve_plex_state(player, queue)
338
339        local_ip = self.provider.mass.streams.publish_ip
340        version = self.provider.mass.version if self.provider.mass.version != "0.0.0" else "1.0.0"
341
342        xml = f"""<?xml version="1.0" encoding="UTF-8"?>
343<MediaContainer>
344    <Player title="{player_name}"
345            protocol="plex"
346            protocolVersion="1"
347            protocolCapabilities="timeline,playback,navigation,playqueues"
348            machineIdentifier="{self.client_id}"
349            product="Music Assistant"
350            platform="{platform.system()}"
351            platformVersion="{platform.release()}"
352            deviceClass="{self.device_class}"
353            state="{state}"
354            address="{local_ip}"
355            port="{self.port}"
356            version="{version}"
357            provides="client,player,pubsub-player">
358        <Connection protocol="http" address="{local_ip}" port="{self.port}"
359                    uri="http://{local_ip}:{self.port}" local="1"/>
360    </Player>
361</MediaContainer>"""
362        return web.Response(
363            text=xml, content_type="text/xml", headers={"Access-Control-Allow-Origin": "*"}
364        )
365