music-assistant-server

1"""
2Unix socket server for Snapcast control script communication.
3
4This module provides a secure communication channel between the Snapcast control script
5and Music Assistant, avoiding the need to expose the WebSocket API to the control script.
6"""
7
8from __future__ import annotations
9
10import asyncio
11import inspect
12import json
13import logging
14from contextlib import suppress
15from pathlib import Path
16from typing import TYPE_CHECKING, Any
17
18from music_assistant_models.enums import EventType
19from music_assistant_models.media_items.metadata import IMAGE_PROXY_ID_RESOLVER
20
21if TYPE_CHECKING:
22    from music_assistant.mass import MusicAssistant
23
24LOGGER = logging.getLogger(__name__)
25
26LOOP_STATUS_MAP = {
27    "all": "playlist",
28    "one": "track",
29    "off": "none",
30}
31LOOP_STATUS_MAP_REVERSE = {v: k for k, v in LOOP_STATUS_MAP.items()}
32
33
34class SnapcastSocketServer:
35    """
36    Unix socket server for a single Snapcast control script connection.
37
38    Each stream gets its own socket server instance to handle control script communication.
39    The socket provides a secure IPC channel that doesn't require authentication since
40    only local processes can connect.
41    """
42
43    def __init__(
44        self,
45        mass: MusicAssistant,
46        queue_id: str,
47        socket_path: str,
48        streamserver_ip: str,
49        streamserver_port: int,
50    ) -> None:
51        """
52        Initialize the socket server.
53
54        :param mass: The MusicAssistant instance.
55        :param queue_id: The queue ID this socket serves.
56        :param socket_path: Path to the Unix socket file.
57        :param streamserver_ip: IP address of the stream server (for image proxy).
58        :param streamserver_port: Port of the stream server (for image proxy).
59        """
60        self.mass = mass
61        self.queue_id = queue_id
62        self.socket_path = socket_path
63        self.streamserver_ip = streamserver_ip
64        self.streamserver_port = streamserver_port
65        self._server: asyncio.AbstractServer | None = None
66        self._client_writer: asyncio.StreamWriter | None = None
67        self._unsub_callback: Any = None
68        self._background_tasks: set[asyncio.Task[None]] = set()
69        self._logger = LOGGER.getChild(queue_id)
70
71    async def start(self) -> None:
72        """Start the Unix socket server."""
73        # Ensure the socket file doesn't exist
74        socket_path = Path(self.socket_path)
75        socket_path.unlink(missing_ok=True)
76
77        # Create the socket server
78        self._server = await asyncio.start_unix_server(
79            self._handle_client,
80            path=self.socket_path,
81        )
82        # Set permissions so only the current user can access
83        Path(self.socket_path).chmod(0o600)
84        self._logger.debug("Started Unix socket server at %s", self.socket_path)
85
86        # Subscribe to queue events
87        self._unsub_callback = self.mass.subscribe(
88            self._handle_mass_event,
89            (EventType.QUEUE_UPDATED,),
90            self.queue_id,
91        )
92
93    async def stop(self) -> None:
94        """Stop the Unix socket server."""
95        if self._unsub_callback:
96            self._unsub_callback()
97            self._unsub_callback = None
98
99        if self._client_writer:
100            with suppress(Exception):
101                await self.notify_shutdown()
102            self._client_writer.close()
103            with suppress(Exception):
104                await self._client_writer.wait_closed()
105            self._client_writer = None
106
107        if self._server:
108            self._server.close()
109            await self._server.wait_closed()
110            self._server = None
111
112        # Clean up socket file
113        Path(self.socket_path).unlink(missing_ok=True)
114        self._logger.debug("Stopped Unix socket server")
115
116    async def notify_shutdown(self) -> None:
117        """Tell the control script to exit."""
118        await self._send_message(
119            {
120                "event": "shutdown",
121                "object_id": self.queue_id,
122            }
123        )
124
125    async def _handle_client(
126        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
127    ) -> None:
128        """Handle a control script connection."""
129        self._logger.debug("Control script connected")
130        self._client_writer = writer
131
132        try:
133            while True:
134                line = await reader.readline()
135                if not line:
136                    break
137
138                try:
139                    message = json.loads(line.decode().strip())
140                    await self._handle_message(message)
141                except json.JSONDecodeError as err:
142                    self._logger.warning("Invalid JSON from control script: %s", err)
143                except Exception:
144                    self._logger.exception("Error handling control script message")
145        except asyncio.CancelledError:
146            pass
147        except ConnectionResetError:
148            self._logger.debug("Control script connection reset")
149        finally:
150            self._client_writer = None
151            writer.close()
152            with suppress(Exception):
153                await writer.wait_closed()
154            self._logger.debug("Control script disconnected")
155
156    async def _handle_message(self, message: dict[str, Any]) -> None:
157        """
158        Handle a message from the control script.
159
160        :param message: The JSON message from the control script.
161        """
162        msg_id = message.get("message_id")
163        command = message.get("command")
164        args = message.get("args", {})
165
166        if not command:
167            await self._send_error(msg_id, "Missing command")
168            return
169
170        try:
171            result = await self._execute_command(command, args)
172            await self._send_result(msg_id, result)
173        except Exception as err:
174            self._logger.exception("Error executing command %s", command)
175            await self._send_error(msg_id, str(err))
176
177    async def _execute_command(self, command: str, args: dict[str, Any]) -> Any:
178        """
179        Execute a Music Assistant API command.
180
181        :param command: The API command to execute.
182        :param args: The arguments for the command.
183        :return: The result of the command.
184        """
185        handler = self.mass.command_handlers.get(command)
186        if handler is None:
187            raise ValueError(f"Unknown command: {command}")
188
189        # Execute the handler
190        result = handler.target(**args)
191        if inspect.iscoroutine(result):
192            result = await result
193        return result
194
195    async def _send_result(self, msg_id: str | None, result: Any) -> None:
196        """
197        Send a success result to the control script.
198
199        :param msg_id: The message ID from the request.
200        :param result: The result data.
201        """
202        if not self._client_writer:
203            return
204
205        response: dict[str, Any] = {"message_id": msg_id}
206        if result is not None:
207            response["result"] = self._serialize(result)
208
209        await self._send_message(response)
210
211    async def _send_error(self, msg_id: str | None, error: str) -> None:
212        """
213        Send an error result to the control script.
214
215        :param msg_id: The message ID from the request.
216        :param error: The error message.
217        """
218        if not self._client_writer:
219            return
220
221        response = {
222            "message_id": msg_id,
223            "error": error,
224        }
225        await self._send_message(response)
226
227    def _serialize(self, obj: Any) -> Any:
228        """
229        Serialize a result or event payload for the control script.
230
231        Sets the imageproxy resolver on the current context so nested
232        MediaItemImage serialization fills the `proxy_id` field, letting the
233        control script build canonical ``/imageproxy/<proxy_id>`` URLs.
234
235        :param obj: The value to serialize (a model with `to_dict`, or plain data).
236        """
237        if not hasattr(obj, "to_dict"):
238            return obj
239        token = IMAGE_PROXY_ID_RESOLVER.set(self.mass.metadata.compute_image_id)
240        try:
241            return obj.to_dict()
242        finally:
243            IMAGE_PROXY_ID_RESOLVER.reset(token)
244
245    async def _send_message(self, message: dict[str, Any]) -> None:
246        """
247        Send a message to the control script.
248
249        :param message: The message to send.
250        """
251        if not self._client_writer:
252            return
253
254        try:
255            data = json.dumps(message) + "\n"
256            self._client_writer.write(data.encode())
257            await self._client_writer.drain()
258        except ConnectionResetError, BrokenPipeError:
259            self._logger.debug("Failed to send message - connection closed")
260            self._client_writer = None
261
262    def _handle_mass_event(self, event: Any) -> None:
263        """
264        Handle Music Assistant events and forward to control script.
265
266        :param event: The Music Assistant event.
267        """
268        if not self._client_writer:
269            return
270
271        # Forward queue_updated events
272        if event.event == EventType.QUEUE_UPDATED and event.object_id == self.queue_id:
273            event_msg = {
274                "event": "queue_updated",
275                "object_id": event.object_id,
276                "data": self._serialize(event.data),
277            }
278            # Schedule the send in the event loop
279            task = asyncio.create_task(self._send_message(event_msg))
280            self._background_tasks.add(task)
281
282            def _on_task_done(t: asyncio.Task[None]) -> None:
283                self._background_tasks.discard(t)
284                if not t.cancelled() and (exc := t.exception()):
285                    self._logger.debug("Background task failed", exc_info=exc)
286
287            task.add_done_callback(_on_task_done)
288