/
/
/
1#!/usr/bin/env python3
2"""
3Control Music Assistant Snapcast plugin.
4
5This script is a bridge between Music Assistant and Snapcast.
6It connects to Music Assistant via a Unix socket and sends metadata to Snapcast
7and listens for player commands.
8"""
9
10import json
11import logging
12import os
13import socket
14import sys
15import threading
16from collections.abc import Callable
17from contextlib import suppress
18from time import sleep
19from typing import Any
20
21import shortuuid
22
23from music_assistant.helpers.util import format_ip_for_url
24
25LOOP_STATUS_MAP = {
26 "all": "playlist",
27 "one": "track",
28 "off": "none",
29}
30LOOP_STATUS_MAP_REVERSE = {v: k for k, v in LOOP_STATUS_MAP.items()}
31
32MessageCallback = Callable[[dict[str, Any]], None]
33
34
35def send(json_msg: dict[str, Any]) -> None:
36 """Send a message to stdout."""
37 sys.stdout.write(json.dumps(json_msg))
38 sys.stdout.write("\n")
39 sys.stdout.flush()
40
41
42class MusicAssistantControl:
43 """Music Assistant Unix socket remote control Snapcast plugin."""
44
45 def __init__(
46 self,
47 queue_id: str,
48 socket_path: str,
49 streamserver_ip: str,
50 streamserver_port: int,
51 ) -> None:
52 """Initialize."""
53 self.queue_id = queue_id
54 self.socket_path = socket_path
55 self.streamserver_ip = streamserver_ip
56 self.streamserver_port = streamserver_port
57 self._metadata: dict[str, Any] = {}
58 self._properties: dict[str, Any] = {}
59 self._request_callbacks: dict[str, MessageCallback] = {}
60 self._seek_offset = 0.0
61 self._socket: socket.socket | None = None
62 self._stopped = False
63 self._shutdown_event = threading.Event()
64 self._socket_thread = threading.Thread(target=self._socket_loop, args=())
65 self._socket_thread.name = "massControl"
66 self._socket_thread.start()
67
68 def stop(self) -> None:
69 """Stop the socket thread."""
70 self._stopped = True
71 if self._socket:
72 with suppress(OSError):
73 self._socket.close()
74 if threading.current_thread() is not self._socket_thread:
75 self._socket_thread.join()
76
77 def shutdown(self) -> None:
78 """Exit the control script."""
79 logger.info("Shutdown requested by server")
80 self.stop()
81 self._shutdown_event.set()
82 # force exit; the main thread is blocked in sys.stdin.readline()
83 sys.stdout.flush()
84 os._exit(0)
85
86 def handle_snapcast_request(self, request: dict[str, Any]) -> None:
87 """Handle (JSON RPC) message from Snapcast."""
88 id: str = request["id"] # noqa: A001
89 interface, cmd = request["method"].rsplit(".", 1)
90
91 queue_id = self.queue_id
92
93 # deny invalid commands
94 if interface != "Plugin.Stream.Player" or cmd not in (
95 "Control",
96 "SetProperty",
97 "GetProperties",
98 ):
99 send(
100 {
101 "jsonrpc": "2.0",
102 "error": {"code": -32601, "message": "Method not found"},
103 "id": id,
104 }
105 )
106 return
107
108 if cmd == "Control":
109 command = request["params"]["command"]
110 params = request["params"].get("params", {})
111 logger.debug(f"Control command: {command}, params: {params}")
112 if command == "next":
113 self.send_request("player_queues/next", queue_id=queue_id)
114 elif command == "previous":
115 self.send_request("player_queues/previous", queue_id=queue_id)
116 elif command == "play":
117 self.send_request("player_queues/play", queue_id=queue_id)
118 elif command == "pause":
119 self.send_request("player_queues/pause", queue_id=queue_id)
120 elif command == "playPause":
121 self.send_request("player_queues/play_pause", queue_id=queue_id)
122 elif command == "stop":
123 self.send_request("player_queues/stop", queue_id=queue_id)
124 elif command == "setPosition":
125 position = float(params["position"])
126 self.send_request("player_queues/seek", queue_id=queue_id, position=position)
127 elif command == "seek":
128 seek_offset = float(params["offset"])
129 self.send_request("player_queues/skip", queue_id=queue_id, seconds=seek_offset)
130 elif cmd == "SetProperty":
131 properties = request["params"]
132 logger.debug(f"SetProperty: {properties}")
133 if "shuffle" in properties:
134 self.send_request(
135 "player_queues/shuffle",
136 queue_id=queue_id,
137 shuffle_enabled=properties["shuffle"],
138 )
139 if "loopStatus" in properties:
140 value = properties["loopStatus"]
141 self.send_request(
142 "player_queues/repeat",
143 queue_id=queue_id,
144 repeat_mode=LOOP_STATUS_MAP_REVERSE[value],
145 )
146 # if "volume" in properties:
147 # self.send_request("core.mixer.set_volume", {"volume": int(properties["volume"])})
148 # if "mute" in properties:
149 # self.send_request("core.mixer.set_mute", {"mute": properties["mute"]})
150 elif cmd == "GetProperties":
151
152 def handle_result(result: dict[str, Any]) -> None:
153 send(
154 {
155 "jsonrpc": "2.0",
156 "result": self._create_properties(result),
157 "id": id,
158 }
159 )
160
161 self.send_request("player_queues/get", callback=handle_result, queue_id=queue_id)
162 return
163
164 # always acknowledge the request
165 send({"jsonrpc": "2.0", "result": "ok", "id": id})
166
167 def send_snapcast_log_notification(self, message: str, severity: str = "Info") -> None:
168 """Send log message to Snapcast."""
169 send(
170 {
171 "jsonrpc": "2.0",
172 "method": "Plugin.Stream.Log",
173 "params": {"severity": severity, "message": message},
174 }
175 )
176
177 def send_snapcast_properties_notification(self, properties: dict[str, Any]) -> None:
178 """Send properties to Snapcast."""
179 send(
180 {
181 "jsonrpc": "2.0",
182 "method": "Plugin.Stream.Player.Properties",
183 "params": properties,
184 }
185 )
186
187 def send_snapcast_stream_ready_notification(self) -> None:
188 """Send stream ready notification to Snapcast."""
189 send({"jsonrpc": "2.0", "method": "Plugin.Stream.Ready"})
190
191 def _socket_loop(self) -> None:
192 logger.info("Started socket loop")
193 while not self._stopped:
194 try:
195 self._connect_and_read()
196 except (Exception, KeyboardInterrupt) as e:
197 logger.info(f"Exception in socket loop: {e!s}")
198 if not self._stopped:
199 sleep(2)
200
201 def _connect_and_read(self) -> None:
202 """Connect to the Unix socket and read messages."""
203 logger.info("Connecting to Unix socket: %s", self.socket_path)
204 self._socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
205 try:
206 self._socket.connect(self.socket_path)
207 logger.info("Connected to Unix socket")
208 self.send_snapcast_stream_ready_notification()
209
210 # Read messages from socket
211 buffer = ""
212 while not self._stopped:
213 try:
214 data = self._socket.recv(4096)
215 if not data:
216 logger.info("Socket closed by server")
217 break
218 buffer += data.decode()
219
220 # Process complete lines
221 while "\n" in buffer:
222 line, buffer = buffer.split("\n", 1)
223 if line.strip():
224 self._handle_socket_message(line)
225 except TimeoutError:
226 continue
227 except OSError as e:
228 logger.error(f"Socket error: {e}")
229 break
230 finally:
231 if self._socket:
232 self._socket.close()
233 self._socket = None
234
235 def _handle_socket_message(self, message: str) -> None:
236 """Handle a message from the Music Assistant socket."""
237 logger.debug("Socket message received: %s", message)
238 try:
239 data = json.loads(message)
240 except json.JSONDecodeError as e:
241 logger.error(f"Invalid JSON: {e}")
242 return
243
244 if data.get("event") == "shutdown":
245 self.shutdown()
246 return
247
248 # Request response
249 if "message_id" in data:
250 message_id = data["message_id"]
251 if callback := self._request_callbacks.pop(message_id, None):
252 if result := data.get("result"):
253 callback(result)
254 # TODO: handle failed requests
255 return
256
257 # Event
258 if "event" in data and data.get("object_id") == self.queue_id:
259 event = data["event"]
260 if event == "queue_updated" and data.get("data"):
261 properties = self._create_properties(data["data"])
262 self.send_snapcast_properties_notification(properties)
263 return
264
265 def _create_properties(self, mass_queue_details: dict[str, Any]) -> dict[str, Any]:
266 """Create snapcast properties from Music Assistant queue details."""
267 current_queue_item: dict[str, Any] | None = mass_queue_details.get("current_item")
268 next_queue_item: dict[str, Any] | None = mass_queue_details.get("next_item")
269 current_index: int = mass_queue_details.get("current_index") or 0
270 properties: dict[str, Any] = {
271 "canGoNext": next_queue_item is not None,
272 "canGoPrevious": current_index > 0,
273 "canPlay": current_queue_item is not None,
274 "canPause": current_queue_item is not None,
275 "canSeek": current_queue_item and current_queue_item.get("duration") is not None,
276 "canControl": True,
277 "playbackStatus": mass_queue_details["state"],
278 "loopStatus": LOOP_STATUS_MAP[mass_queue_details["repeat_mode"]],
279 "shuffle": mass_queue_details["shuffle_enabled"],
280 "volume": 0,
281 "mute": False,
282 "rate": 1.0,
283 "position": mass_queue_details["elapsed_time"],
284 }
285 image_url: str | None = None
286 if current_queue_item and (media_item := current_queue_item.get("media_item")):
287 if proxy_id := current_queue_item.get("image", {}).get("proxy_id"):
288 image_url = (
289 # we prefer the streamserver for the imageproxy because it is enabled by default
290 # where the api server is by default protected
291 f"http://{format_ip_for_url(self.streamserver_ip)}:{self.streamserver_port}"
292 f"/imageproxy/{proxy_id}?size=512"
293 )
294 properties["metadata"] = {
295 "trackId": media_item["uri"],
296 "duration": media_item["duration"],
297 "title": media_item["name"],
298 "artUrl": image_url,
299 }
300 if "artists" in media_item:
301 properties["metadata"]["artist"] = [x["name"] for x in media_item["artists"]]
302 properties["metadata"]["artistSort"] = [
303 x["sort_name"] for x in media_item["artists"]
304 ]
305 if media_item.get("album"):
306 properties["metadata"]["album"] = media_item["album"]["name"]
307 properties["metadata"]["albumSort"] = media_item["album"]["sort_name"]
308 elif current_queue_item:
309 properties["metadata"] = {
310 "title": current_queue_item["name"],
311 "trackId": current_queue_item["queue_item_id"],
312 "artUrl": image_url,
313 }
314
315 return properties
316
317 def send_request(
318 self, command: str, callback: MessageCallback | None = None, **args: str | float | bool
319 ) -> None:
320 """Send request to Music Assistant via Unix socket."""
321 if not self._socket:
322 logger.warning("Cannot send request - socket not connected")
323 return
324
325 msg_id = shortuuid.random(10)
326 command_msg = {
327 "message_id": msg_id,
328 "command": command,
329 "args": args,
330 }
331 logger.debug("send_request: %s", command_msg)
332 if callback:
333 self._request_callbacks[msg_id] = callback
334 try:
335 data = json.dumps(command_msg) + "\n"
336 self._socket.sendall(data.encode())
337 except OSError as e:
338 logger.error(f"Failed to send request: {e}")
339 self._request_callbacks.pop(msg_id, None)
340
341
342if __name__ == "__main__":
343 # Parse command line
344 queue_id = None
345 socket_path: str | None = None
346 streamserver_ip: str | None = None
347 streamserver_port: str | None = None
348 stream_id: str | None = None
349 for arg in sys.argv:
350 if arg.startswith("--stream="):
351 stream_id = arg.split("=")[1]
352 if arg.startswith("--queueid="):
353 queue_id = arg.split("=")[1]
354 if arg.startswith("--socket="):
355 socket_path = arg.split("=")[1]
356 if arg.startswith("--streamserver-ip="):
357 streamserver_ip = arg.split("=")[1]
358 if arg.startswith("--streamserver-port="):
359 streamserver_port = arg.split("=")[1]
360
361 if not queue_id or not socket_path:
362 print("Usage: --stream=<stream_id> --socket=<socket_path>") # noqa: T201
363 sys.exit()
364
365 log_format_stderr = "%(asctime)s %(module)s %(levelname)s: %(message)s"
366 log_level = logging.INFO
367 logger = logging.getLogger("meta_mass")
368 logger.propagate = False
369 logger.setLevel(log_level)
370
371 # Log to stderr
372 log_handler = logging.StreamHandler()
373 log_handler.setFormatter(logging.Formatter(log_format_stderr))
374 logger.addHandler(log_handler)
375
376 logger.debug(
377 "Initializing for stream_id %s, queue_id %s and socket %s", stream_id, queue_id, socket_path
378 )
379
380 assert streamserver_ip is not None # for type checking
381 assert streamserver_port is not None
382 ctrl = MusicAssistantControl(queue_id, socket_path, streamserver_ip, int(streamserver_port))
383
384 # keep listening for messages on stdin and forward them
385 try:
386 while not ctrl._shutdown_event.is_set():
387 line = sys.stdin.readline()
388 if not line: # EOF
389 break
390 request: Any = None
391 try:
392 request = json.loads(line)
393 ctrl.handle_snapcast_request(request)
394 except json.JSONDecodeError as e:
395 send(
396 {
397 "jsonrpc": "2.0",
398 "error": {"code": -32700, "message": "Parse error", "data": str(e)},
399 "id": None,
400 }
401 )
402 except (KeyError, TypeError, AttributeError, ValueError) as e:
403 # malformed request (e.g. missing params), reply instead of dying
404 send(
405 {
406 "jsonrpc": "2.0",
407 "error": {"code": -32603, "message": "Internal error", "data": str(e)},
408 "id": request.get("id") if isinstance(request, dict) else None,
409 }
410 )
411 finally:
412 ctrl.stop()
413 sys.exit(0)
414