/
/
/
1"""SnapCastProvider."""
2
3from __future__ import annotations
4
5import asyncio
6import hashlib
7import logging
8import re
9import shutil
10import socket
11from contextlib import suppress
12from pathlib import Path
13from typing import TYPE_CHECKING, cast
14
15from bidict import bidict
16from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
17from music_assistant_models.enums import ConfigEntryType, MediaType, PlaybackState
18from music_assistant_models.errors import SetupFailedError
19from music_assistant_models.helpers import create_safe_string
20from snapcast.control.server import CONTROL_PORT, Snapserver
21from zeroconf import NonUniqueNameException
22from zeroconf.asyncio import AsyncServiceInfo
23
24from music_assistant.constants import CONF_ENABLED, CONF_LOG_LEVEL, VERBOSE_LOG_LEVEL
25from music_assistant.helpers.json import SerializableType
26from music_assistant.helpers.process import AsyncProcess, check_output
27from music_assistant.helpers.util import get_ip_pton
28from music_assistant.models.player_provider import PlayerProvider
29from music_assistant.providers.snapcast.constants import (
30 CONF_CATEGORY_BUILT_IN,
31 CONF_HELP_LINK,
32 CONF_SERVER_BUFFER_SIZE,
33 CONF_SERVER_CHUNK_MS,
34 CONF_SERVER_CONTROL_PORT,
35 CONF_SERVER_HOST,
36 CONF_SERVER_INITIAL_VOLUME,
37 CONF_SERVER_SEND_AUDIO_TO_MUTED,
38 CONF_SERVER_TRANSPORT_CODEC,
39 CONF_STREAM_BIT_DEPTH,
40 CONF_STREAM_IDLE_THRESHOLD,
41 CONF_STREAM_SAMPLE_RATE,
42 CONF_USE_EXTERNAL_SERVER,
43 CONTROL_SCRIPT,
44 DEFAULT_SNAPCAST_FORMAT,
45 DEFAULT_SNAPSERVER_CONFIG_FILE,
46 DEFAULT_SNAPSERVER_IP,
47 DEFAULT_SNAPSERVER_PLUGIN_DIR,
48 DEFAULT_SNAPSERVER_PORT,
49 DEFAULT_SNAPSTREAM_IDLE_THRESHOLD,
50 MASS_ANNOUNCEMENT_POSTFIX,
51 MASS_STREAM_PREFIX,
52 SHIPPED_SNAPSERVER_CONFIG_FILE,
53 SNAPCAST_BIT_DEPTHS,
54 SNAPCAST_SAMPLE_RATES,
55 SNAPCLIENT_LIVENESS_POLL_INTERVAL,
56 SNAPWEB_DIR,
57 snapcast_stream_format,
58)
59from music_assistant.providers.snapcast.ma_stream import SnapcastMAStream
60from music_assistant.providers.snapcast.player import SnapCastPlayer
61from music_assistant.providers.universal_group.constants import UGP_PREFIX
62
63if TYPE_CHECKING:
64 from music_assistant_models.config_entries import ProviderConfig
65 from music_assistant_models.media_items import AudioFormat
66 from music_assistant_models.player import PlayerMedia
67
68 from .snap_cntrl_proto import SnapclientProto, SnapgroupProto, SnapserverProto
69
70
71async def _create_cntrl_server(
72 loop: asyncio.AbstractEventLoop,
73 host: str,
74 port: int = CONTROL_PORT,
75 reconnect: bool = False,
76) -> SnapserverProto:
77 """Server factory."""
78 server = Snapserver(loop, host, port, reconnect)
79 await server.start()
80 return cast("SnapserverProto", server)
81
82
83class SnapCastProvider(PlayerProvider):
84 """SnapCastProvider."""
85
86 reload_on_streams_network_change = True
87 _snapserver: SnapserverProto
88 _snapserver_runner: asyncio.Task[None] | None
89 _snapserver_started: asyncio.Event | None
90 _snapcast_server_host: str
91 _snapcast_server_control_port: int
92 _ids_map: bidict[str, str] # ma_id / snapclient_id
93 _use_builtin_server: bool
94 _stop_called: bool
95 _controlscript_available: bool
96 _snapcast_ma_streams: dict[str, SnapcastMAStream]
97 _snapcast_ma_streams_lock: asyncio.Lock
98 _last_status_refresh: float
99 _snapcast_stream_format: AudioFormat
100 _zc_services: dict[str, AsyncServiceInfo]
101
102 @property
103 def stream_audio_format(self) -> AudioFormat:
104 """Return the configured PCM format for Music Assistant Snapcast TCP sources."""
105 return getattr(self, "_snapcast_stream_format", DEFAULT_SNAPCAST_FORMAT)
106
107 @property
108 def queue_control_available(self) -> bool:
109 """
110 Return whether queue-based control scripts are available.
111
112 Indicates if the Snapcast control script has been successfully initialized
113 and can be used to control playback via a queue-specific control channel.
114 """
115 return (
116 self._use_builtin_server
117 and self._controlscript_available
118 and self._snapserver_started is not None
119 and self._snapserver_started.is_set()
120 )
121
122 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
123 """Return Config entries to setup this provider."""
124 returncode, output = await check_output("snapserver", "-v")
125 snapserver_version = -1
126 if returncode == 0:
127 # Parse version from output, handling potential noise from library warnings
128 # Expected format: "0.27.0" or similar version string
129 output_str = output.decode()
130 if version_match := re.search(r"(\d+)\.(\d+)\.(\d+)", output_str):
131 snapserver_version = int(version_match.group(2))
132 local_snapserver_present = snapserver_version >= 27 and snapserver_version != 30
133 if returncode == 0 and not local_snapserver_present:
134 raise SetupFailedError(
135 f"Invalid snapserver version. Expected >= 27 and != 30, got {snapserver_version}"
136 )
137
138 return (
139 ConfigEntry(
140 key=CONF_SERVER_BUFFER_SIZE,
141 type=ConfigEntryType.INTEGER,
142 range=(200, 6000),
143 default_value=1000,
144 required=False,
145 category=CONF_CATEGORY_BUILT_IN,
146 hidden=not local_snapserver_present,
147 depends_on=CONF_USE_EXTERNAL_SERVER,
148 depends_on_value_not=True,
149 help_link=CONF_HELP_LINK,
150 ),
151 ConfigEntry(
152 key=CONF_SERVER_CHUNK_MS,
153 type=ConfigEntryType.INTEGER,
154 range=(10, 100),
155 default_value=26,
156 required=False,
157 category=CONF_CATEGORY_BUILT_IN,
158 hidden=not local_snapserver_present,
159 depends_on=CONF_USE_EXTERNAL_SERVER,
160 depends_on_value_not=True,
161 help_link=CONF_HELP_LINK,
162 ),
163 ConfigEntry(
164 key=CONF_SERVER_INITIAL_VOLUME,
165 type=ConfigEntryType.INTEGER,
166 range=(0, 100),
167 default_value=25,
168 required=False,
169 category=CONF_CATEGORY_BUILT_IN,
170 hidden=not local_snapserver_present,
171 depends_on=CONF_USE_EXTERNAL_SERVER,
172 depends_on_value_not=True,
173 help_link=CONF_HELP_LINK,
174 ),
175 ConfigEntry(
176 key=CONF_SERVER_SEND_AUDIO_TO_MUTED,
177 type=ConfigEntryType.BOOLEAN,
178 default_value=False,
179 required=False,
180 category=CONF_CATEGORY_BUILT_IN,
181 hidden=not local_snapserver_present,
182 depends_on=CONF_USE_EXTERNAL_SERVER,
183 depends_on_value_not=True,
184 help_link=CONF_HELP_LINK,
185 ),
186 ConfigEntry(
187 key=CONF_SERVER_TRANSPORT_CODEC,
188 type=ConfigEntryType.STRING,
189 options=[
190 ConfigValueOption("flac"),
191 ConfigValueOption("ogg"),
192 ConfigValueOption("opus"),
193 ConfigValueOption("pcm"),
194 ],
195 default_value="flac",
196 required=False,
197 category=CONF_CATEGORY_BUILT_IN,
198 hidden=not local_snapserver_present,
199 depends_on=CONF_USE_EXTERNAL_SERVER,
200 depends_on_value_not=True,
201 help_link=CONF_HELP_LINK,
202 ),
203 ConfigEntry(
204 key=CONF_USE_EXTERNAL_SERVER,
205 type=ConfigEntryType.BOOLEAN,
206 default_value=not local_snapserver_present,
207 required=False,
208 advanced=local_snapserver_present,
209 ),
210 ConfigEntry(
211 key=CONF_SERVER_HOST,
212 type=ConfigEntryType.STRING,
213 default_value=DEFAULT_SNAPSERVER_IP,
214 required=False,
215 depends_on=CONF_USE_EXTERNAL_SERVER,
216 advanced=local_snapserver_present,
217 ),
218 ConfigEntry(
219 key=CONF_SERVER_CONTROL_PORT,
220 type=ConfigEntryType.INTEGER,
221 default_value=DEFAULT_SNAPSERVER_PORT,
222 required=False,
223 depends_on=CONF_USE_EXTERNAL_SERVER,
224 advanced=local_snapserver_present,
225 ),
226 ConfigEntry(
227 key=CONF_STREAM_IDLE_THRESHOLD,
228 type=ConfigEntryType.INTEGER,
229 default_value=DEFAULT_SNAPSTREAM_IDLE_THRESHOLD,
230 required=True,
231 advanced=local_snapserver_present,
232 ),
233 ConfigEntry(
234 key=CONF_STREAM_SAMPLE_RATE,
235 type=ConfigEntryType.INTEGER,
236 options=[ConfigValueOption(rate) for rate in SNAPCAST_SAMPLE_RATES],
237 default_value=DEFAULT_SNAPCAST_FORMAT.sample_rate,
238 required=False,
239 depends_on=CONF_USE_EXTERNAL_SERVER,
240 advanced=local_snapserver_present,
241 ),
242 ConfigEntry(
243 key=CONF_STREAM_BIT_DEPTH,
244 type=ConfigEntryType.INTEGER,
245 options=[ConfigValueOption(depth) for depth in SNAPCAST_BIT_DEPTHS],
246 default_value=DEFAULT_SNAPCAST_FORMAT.bit_depth,
247 required=False,
248 depends_on=CONF_USE_EXTERNAL_SERVER,
249 advanced=local_snapserver_present,
250 ),
251 )
252
253 async def handle_async_init(self) -> None:
254 """Handle async initialization of the provider."""
255 self._set_snapcast_log_level()
256 self._use_builtin_server = not self.config.get_value(CONF_USE_EXTERNAL_SERVER)
257 self._stop_called = False
258 self._controlscript_available = False
259 self._zc_services = {}
260 if self._use_builtin_server:
261 if Path(DEFAULT_SNAPSERVER_CONFIG_FILE).exists():
262 self._snapcast_server_config_file = DEFAULT_SNAPSERVER_CONFIG_FILE
263 else:
264 # Fallback for dev environments without a Snapserver config file.
265 # If the file is missing, Snapserver silently ignores all command-line arguments.
266 self._snapcast_server_config_file = str(SHIPPED_SNAPSERVER_CONFIG_FILE)
267
268 self._snapcast_server_host = "127.0.0.1"
269 self._snapcast_server_control_port = DEFAULT_SNAPSERVER_PORT
270 self._snapcast_server_buffer_size = cast(
271 "int", self.config.get_value(CONF_SERVER_BUFFER_SIZE)
272 )
273 self._snapcast_server_chunk_ms = self.config.get_value(CONF_SERVER_CHUNK_MS)
274 self._snapcast_server_initial_volume = self.config.get_value(CONF_SERVER_INITIAL_VOLUME)
275 self._snapcast_server_send_to_muted = self.config.get_value(
276 CONF_SERVER_SEND_AUDIO_TO_MUTED
277 )
278 self._snapcast_server_transport_codec = self.config.get_value(
279 CONF_SERVER_TRANSPORT_CODEC
280 )
281 else:
282 self._snapcast_server_host = str(self.config.get_value(CONF_SERVER_HOST))
283 self._snapcast_server_control_port = int(
284 str(self.config.get_value(CONF_SERVER_CONTROL_PORT))
285 )
286 self._snapcast_stream_idle_threshold = self.config.get_value(CONF_STREAM_IDLE_THRESHOLD)
287 # Higher rates / 24-bit are only supported with an external Snapserver for now.
288 # The built-in server always stays at the default 48 kHz / 16-bit format.
289 if self._use_builtin_server:
290 self._snapcast_stream_format = DEFAULT_SNAPCAST_FORMAT
291 else:
292 self._snapcast_stream_format = snapcast_stream_format(
293 cast("int", self.config.get_value(CONF_STREAM_SAMPLE_RATE)),
294 cast("int", self.config.get_value(CONF_STREAM_BIT_DEPTH)),
295 )
296 self._ids_map = bidict({})
297 self._last_status_refresh = 0.0
298
299 self._snapcast_ma_streams = {}
300 self._snapcast_ma_streams_lock = asyncio.Lock()
301
302 if self._use_builtin_server:
303 await self._start_builtin_server()
304 else:
305 self._snapserver_runner = None
306 self._snapserver_started = None
307 try:
308 self._snapserver = await _create_cntrl_server(
309 self.mass.loop,
310 self._snapcast_server_host,
311 port=self._snapcast_server_control_port,
312 reconnect=True,
313 )
314 self._snapserver.set_on_update_callback(self._handle_update)
315 self.logger.info(
316 "Started connection to Snapserver %s",
317 f"{self._snapcast_server_host}:{self._snapcast_server_control_port}",
318 )
319 # register callback for when the connection gets lost to the snapserver
320 self._snapserver.set_on_disconnect_callback(self._handle_disconnect)
321
322 except OSError as err:
323 msg = "Unable to start the Snapserver connection ?"
324 raise SetupFailedError(msg) from err
325
326 async def update_config(self, config: ProviderConfig, changed_keys: set[str]) -> None:
327 """Handle logic when the config is updated."""
328 await super().update_config(config, changed_keys)
329 # a log level(-only) change does not reload the provider,
330 # so realign snapcast's logger here
331 if f"values/{CONF_LOG_LEVEL}" in changed_keys:
332 self._set_snapcast_log_level()
333
334 async def loaded_in_mass(self) -> None:
335 """Call after the provider has been loaded."""
336 await super().loaded_in_mass()
337 # initial load of players
338 self._handle_update()
339
340 async def unload(self, is_removed: bool = False) -> None:
341 """Handle close/cleanup of the provider."""
342 self._stop_called = True
343
344 for snap_client in self._snapserver.clients:
345 if not (player_id := self._get_ma_id(snap_client.identifier)):
346 continue
347 if not (player := self.mass.players.get_player(player_id, raise_unavailable=False)):
348 continue
349 if player.playback_state != PlaybackState.PLAYING:
350 continue
351 await player.stop()
352
353 for stream_name in list(self._snapcast_ma_streams):
354 await self.delete_ma_stream(stream_name)
355
356 self._snapserver.stop()
357 await self._stop_builtin_server()
358
359 # unregister the snapcast mdns services
360 for info in self._zc_services.values():
361 await self.mass.discovery.aiozc.async_unregister_service(info)
362 self._zc_services.clear()
363
364 async def get_diagnostics(self) -> dict[str, SerializableType]:
365 """Return diagnostics info for this provider to include in diagnostics reports."""
366 return {
367 "builtin_server": self._use_builtin_server,
368 "builtin_server_started": (
369 self._snapserver_started.is_set() if self._snapserver_started else None
370 ),
371 "clients_total": len(self._snapserver.clients),
372 "clients_connected": sum(client.connected for client in self._snapserver.clients),
373 "groups": len(self._snapserver.groups),
374 "streams": len(self._snapserver.streams),
375 "ma_streams": len(self._snapcast_ma_streams),
376 }
377
378 async def refresh_server_status(self) -> None:
379 """
380 Refresh the full snapserver state, throttled to once per poll cycle.
381
382 Snapcast players all poll within the same controller pass; this collapses
383 that burst into a single Server.GetStatus so each client's lastSeen is
384 refreshed once per interval. Transient errors are ignored and retried.
385 """
386 now = self.mass.loop.time()
387 if now - self._last_status_refresh < SNAPCLIENT_LIVENESS_POLL_INTERVAL / 2:
388 return
389 self._last_status_refresh = now
390 try:
391 status, _ = await self._snapserver.status()
392 if isinstance(status, dict) and "server" in status:
393 self._snapserver.synchronize(status)
394 except Exception:
395 self.logger.debug("Snapserver status refresh failed", exc_info=True)
396
397 async def _start_builtin_server(self) -> None:
398 """Start the built-in Snapserver."""
399 if self._use_builtin_server:
400 self._snapserver_started = asyncio.Event()
401 self._snapserver_runner = self.mass.create_task(self._builtin_server_runner())
402 try:
403 # during startup the executor and event loop can be contended
404 # by other providers loading, which delays the start signal
405 # by several seconds
406 await asyncio.wait_for(self._snapserver_started.wait(), 30)
407 except TimeoutError as err:
408 # cancel the runner so the spawned snapserver process
409 # does not linger and occupy the ports on the next attempt
410 await self._stop_builtin_server()
411 msg = "Builtin Snapserver did not start within 30 seconds"
412 raise SetupFailedError(msg) from err
413
414 async def _stop_builtin_server(self) -> None:
415 """Stop the built-in Snapserver."""
416 self.logger.info("Stopping, built-in Snapserver")
417 if self._snapserver_runner and not self._snapserver_runner.done():
418 self._snapserver_runner.cancel()
419
420 def _setup_controlscript(self) -> str | None:
421 """
422 Copy control script to plugin directory (blocking I/O).
423
424 :return: plugin dir if successful, None otherwise.
425 """
426 logger = self.logger.getChild("snapserver")
427 if not CONTROL_SCRIPT.exists():
428 logger.warning("Control script does not exist: %s", CONTROL_SCRIPT)
429 return None
430
431 candidates = (
432 Path(DEFAULT_SNAPSERVER_PLUGIN_DIR),
433 # fallback directory for dev environments
434 Path(self.mass.storage_path) / "snapcast" / "plugins",
435 )
436 for plugin_dir in candidates:
437 control_dest = plugin_dir / "control.py"
438 try:
439 plugin_dir.mkdir(parents=True, exist_ok=True)
440 # Clean up existing file
441 control_dest.unlink(missing_ok=True)
442
443 # Copy the control script to the plugin directory
444 shutil.copy2(CONTROL_SCRIPT, control_dest)
445 # Ensure it's executable
446 control_dest.chmod(0o755)
447 logger.debug("Copied controlscript to: %s", control_dest)
448 return str(plugin_dir)
449 except (OSError, PermissionError) as err:
450 logger.debug("Could not copy controlscript to %s : %s", plugin_dir, err)
451 logger.warning("Could not copy controlscript (metadata/control disabled)")
452 return None
453
454 async def _builtin_server_runner(self) -> None:
455 """Start running the builtin snapserver."""
456 assert self._snapserver_started is not None # for type checking
457 if self._snapserver_started.is_set():
458 raise RuntimeError("Snapserver is already started!")
459 logger = self.logger.getChild("snapserver")
460 logger.info("Starting builtin Snapserver...")
461 addresses = [await get_ip_pton(self.mass.streams.publish_ip)]
462 # a DNS-SD instance name is a single DNS label, so bound the name to 63 utf-8 bytes
463 instance_name = self.mass.webserver.server_name.encode()[:63].decode("utf-8", "ignore")
464 # register the snapcast mdns services
465 for name, port in (
466 ("-http", 1780),
467 ("-jsonrpc", 1705),
468 ("-stream", 1704),
469 ("-tcp", 1705),
470 ("", 1704),
471 ):
472 zeroconf_type = f"_snapcast{name}._tcp.local."
473 try:
474 info = AsyncServiceInfo(
475 zeroconf_type,
476 name=f"{instance_name}.{zeroconf_type}",
477 properties={"is_mass": "true"},
478 addresses=addresses,
479 port=port,
480 server=f"{socket.gethostname()}.local",
481 )
482 if name in self._zc_services:
483 await self.mass.discovery.aiozc.async_update_service(info)
484 else:
485 await self.mass.discovery.aiozc.async_register_service(info, strict=False)
486 self._zc_services[name] = info
487 except NonUniqueNameException:
488 self.logger.debug(
489 "Could not register mdns record for %s as its already in use",
490 zeroconf_type,
491 )
492 except Exception:
493 self.logger.exception("Could not register mdns record for %s", zeroconf_type)
494
495 args = [
496 "snapserver",
497 # config settings taken from
498 # https://raw.githubusercontent.com/badaix/snapcast/86cd4b2b63e750a72e0dfe6a46d47caf01426c8d/server/etc/snapserver.conf
499 f"--config={self._snapcast_server_config_file}",
500 f"--server.datadir={self.mass.storage_path}",
501 "--http.enabled=true",
502 "--http.port=1780",
503 f"--http.doc_root={SNAPWEB_DIR}",
504 "--tcp-control.enabled=true",
505 f"--tcp-control.port={self._snapcast_server_control_port}",
506 (
507 f"--stream.sampleformat={self.stream_audio_format.sample_rate}"
508 f":{self.stream_audio_format.bit_depth}"
509 f":{self.stream_audio_format.channels}"
510 ),
511 f"--stream.buffer={self._snapcast_server_buffer_size}",
512 f"--stream.chunk_ms={self._snapcast_server_chunk_ms}",
513 f"--stream.codec={self._snapcast_server_transport_codec}",
514 f"--stream.send_to_muted={str(self._snapcast_server_send_to_muted).lower()}",
515 f"--streaming_client.initial_volume={self._snapcast_server_initial_volume}",
516 ]
517 loop = asyncio.get_running_loop()
518 plugin_dir = await loop.run_in_executor(None, self._setup_controlscript)
519 if plugin_dir is not None:
520 args.append(f"--stream.plugin_dir={plugin_dir}")
521 self._controlscript_available = True
522
523 started_handle: asyncio.Handle | None = None
524 async with AsyncProcess(args, stdout=True, name="snapserver") as snapserver_proc:
525 try:
526 # keep reading from stdout until exit
527 async for raw_data in snapserver_proc.iter_any():
528 text = raw_data.decode().strip()
529 for line in text.split("\n"):
530 logger.debug(line)
531 if "(Snapserver) Version 0." in line:
532 # delay init a small bit to prevent race conditions
533 # where we try to connect too soon
534 if started_handle is None:
535 started_handle = self.mass.loop.call_later(
536 2, self._snapserver_started.set
537 )
538
539 except asyncio.CancelledError:
540 # Currently, MA doesn't guarantee a defined shutdown order;
541 # Make sure to close socket servers before
542 # shutting down the snapcast server.
543 #
544 # The snapserver doesn't always cleanup the control script processes
545 # properly. We do it explicitly when closing a socket server.
546 # Should be fixed on the server side, though.
547 for stream_name in list(self._snapcast_ma_streams):
548 await self.delete_ma_stream(stream_name)
549 self._snapcast_ma_streams.clear()
550 raise
551
552 finally:
553 if started_handle is not None:
554 started_handle.cancel()
555 if self._snapserver_started is not None:
556 self._snapserver_started.clear()
557 self._controlscript_available = False
558
559 def _get_ma_id(self, snap_client_id: str) -> str | None:
560 """Return the MA player id for the given snapclient id, or None if not registered."""
561 return self._ids_map.inverse.get(snap_client_id)
562
563 def _get_snapclient_id(self, player_id: str) -> str:
564 search_dict = self._ids_map
565 snap_id = search_dict.get(player_id)
566 assert snap_id is not None # for type checking
567 return snap_id
568
569 def _generate_and_register_id(self, snap_client_id: str) -> str:
570 search_dict = self._ids_map.inverse
571 if snap_client_id not in search_dict:
572 new_id = "ma_" + str(re.sub(r"\W+", "", snap_client_id))
573 self._ids_map[new_id] = snap_client_id
574 return new_id
575 return search_dict[snap_client_id]
576
577 def _handle_player_init(self, snap_client: SnapclientProto) -> SnapCastPlayer | None:
578 """Process Snapcast add to Player controller."""
579 player_id = self._generate_and_register_id(snap_client.identifier)
580 if not self.mass.config.get_raw_player_config_value(player_id, CONF_ENABLED, True):
581 self.logger.debug("Ignoring disabled snapcast player: %s", player_id)
582 return None
583 player = self.mass.players.get_player(player_id, raise_unavailable=False)
584 if not player:
585 snap_client = self._snapserver.client(self._get_snapclient_id(player_id))
586 player = SnapCastPlayer(
587 provider=self,
588 player_id=player_id,
589 snap_client=snap_client,
590 )
591 player.setup()
592 else:
593 player = cast("SnapCastPlayer", player) # for type checking
594 asyncio.run_coroutine_threadsafe(
595 self.mass.players.register_or_update(player), loop=self.mass.loop
596 )
597 return player
598
599 def _handle_update(self) -> None:
600 """Process Snapcast init Player/Group and set callback ."""
601 for snap_client in self._snapserver.clients:
602 if not snap_client.identifier:
603 self.logger.warning(
604 "Detected Snapclient %s without identifier, skipping", snap_client.friendly_name
605 )
606 continue
607 if ma_player := self._handle_player_init(snap_client):
608 snap_client.set_callback(ma_player._handle_player_update)
609 for snap_client in self._snapserver.clients:
610 if player := self.get_snap_player(client_id=snap_client.identifier):
611 snap_client.set_callback(player._handle_player_update)
612 self._update_group_callbacks()
613
614 def poke_group_members(self, snap_group: SnapgroupProto) -> None:
615 """Process Snapcast group callback."""
616 for snap_client_id in snap_group.clients:
617 if ma_player := self.get_snap_player(client_id=snap_client_id):
618 ma_player.poke_player_update()
619
620 def _handle_disconnect(self, exc: Exception) -> None:
621 """Handle disconnect callback from snapserver."""
622 if self._stop_called or self.mass.closing:
623 # prevent auto-reconnecting of snapcast controller
624 self._snapserver.stop()
625 # we're instructed to stop/exit, so no need to restart the connection
626 return
627 self.logger.info(
628 "Connection to SnapServer lost, reason: %s. Reloading provider in 5 seconds.",
629 str(exc),
630 )
631 # schedule a reload of the provider, armed under the load path's task id so any
632 # (re)load starting before it fires cancels it
633 self.mass.call_later(
634 5,
635 self.mass.load_provider,
636 self.instance_id,
637 allow_retry=True,
638 task_id=f"load_provider_{self.instance_id}",
639 )
640
641 async def remove_player(self, player_id: str) -> None:
642 """Remove the client from the snapserver when it is deleted."""
643 success, error_msg = await self._snapserver.delete_client(
644 self._get_snapclient_id(player_id)
645 )
646 if success:
647 self.logger.debug("Snapclient removed %s", player_id)
648 else:
649 self.logger.warning("Unable to remove snapclient %s: %s", player_id, error_msg)
650
651 def _update_group_callbacks(self, poke: bool = False) -> None:
652 for grp in self._snapserver.groups:
653 grp.set_callback(self.poke_group_members)
654 if poke:
655 self.poke_group_members(grp)
656
657 async def ensure_player_owned_group(
658 self, ma_player_id: str, set_stream_id: str | None = None
659 ) -> SnapgroupProto | None:
660 """
661 Ensure a Snapcast group is owned by the given player.
662
663 This method guarantees that the returned Snapcast group is *owned* by the
664 specified Music Assistant player, meaning the group name equals the
665 player's ID and the player is the group leader.
666
667 Behavior:
668 - If the player is already the leader of its current group, that group is
669 returned unchanged.
670 - If the player is a member of another group (but not the leader), the
671 player is removed from that group, which causes Snapcast to create a new
672 single-client group for the player.
673 - The resulting group is renamed to the player's ID.
674
675 If `set_stream_id` is provided and a new group is created, the group's
676 stream is updated accordingly.
677
678 Args:
679 ma_player_id: Music Assistant player ID.
680 set_stream_id: Optional Snapcast stream ID to assign to the player's group.
681
682 Returns:
683 The Snapcast group owned by the player, or ``None`` if the player is not
684 currently part of any group.
685 """
686 player_client = self.get_snap_client(player_id=ma_player_id)
687 if player_client is None:
688 return None
689
690 curr_group = player_client.group
691
692 if curr_group is None:
693 return None
694
695 if curr_group.name == ma_player_id:
696 return curr_group
697
698 group_members = list(curr_group.clients)
699 if len(group_members) > 1 and curr_group.name:
700 # player is member of other player group, remove it, which results in a new group
701 group_members.remove(player_client.identifier)
702 res = await self._snapserver.group_clients(curr_group.identifier, group_members)
703 if not (isinstance(res, dict) and "server" in res):
704 raise RuntimeError("Couldn't remove client from group")
705 self._snapserver.synchronize(res)
706 curr_group = player_client.group
707 if curr_group is None:
708 return None
709 if set_stream_id:
710 await curr_group.set_stream(set_stream_id)
711
712 await curr_group.set_name(ma_player_id)
713 return curr_group
714
715 async def isolate_player_to_dedicated_group(
716 self,
717 target_player_id: str,
718 target_stream_id: str | None = None,
719 others_stream_id: str | None = "default",
720 ) -> None:
721 """
722 Isolate a player into a dedicated Snapcast group.
723
724 Ensures that the target player ends up in a group where it is the sole
725 member and group leader.
726
727 Behavior:
728 - The target player is first ensured to own its group.
729 - All other members of that group are removed.
730 - Each removed player is placed into its own dedicated group.
731 - Removed players' groups are optionally assigned `others_stream_id`.
732 - The target group is optionally assigned `target_stream_id`.
733
734 Callbacks for affected clients and groups are temporarily disabled during
735 the operation to avoid intermediate state updates.
736
737 Args:
738 target_player_id: Music Assistant player ID to isolate.
739 target_stream_id: Optional stream ID to assign to the target player's group.
740 others_stream_id: Stream ID assigned to newly created groups for removed players.
741 """
742 this_client_id = self._get_snapclient_id(target_player_id)
743 target_group = await self.ensure_player_owned_group(
744 target_player_id, set_stream_id=target_stream_id
745 )
746
747 if target_group is None:
748 return
749
750 target_group.set_callback(None)
751 group_members = list(target_group.clients)
752 group_members.remove(this_client_id)
753 for client_id in group_members:
754 client = self._snapserver.client(client_id)
755 client.set_callback(None)
756 if group_members:
757 res = await self._snapserver.group_clients(target_group.identifier, [this_client_id])
758 if not (isinstance(res, dict) and "server" in res):
759 raise RuntimeError("Couldn't remove client from group")
760 self._snapserver.synchronize(res)
761 for client_id in group_members:
762 if (ma_player_id := self._get_ma_id(client_id)) is None:
763 continue
764 if ma_player := cast("SnapCastPlayer", self.mass.players.get_player(ma_player_id)):
765 client = self._snapserver.client(client_id)
766 if client is not None:
767 if client.group is not None:
768 await client.group.set_name(ma_player_id)
769 if others_stream_id:
770 await client.group.set_stream(others_stream_id)
771 client.set_callback(ma_player._handle_player_update)
772
773 if target_stream_id is not None:
774 await target_group.set_stream(target_stream_id)
775
776 async def get_snapcast_media_stream(
777 self,
778 media: PlayerMedia,
779 filter_settings_owner: str | None = None,
780 existing_only: bool = False,
781 ) -> SnapcastMAStream | None:
782 """
783 Get or create a Snapcast Music Assistant stream for the given media.
784
785 Determines a deterministic Snapcast stream name based on the media type
786 and source, and either returns an existing stream or creates a new one.
787
788 Behavior:
789 - Announcement and generic media streams use a hashed name.
790 - Plugin and queue-backed sources reuse a stable stream name.
791 - Queue-backed streams may persist across playback sessions.
792 - If `existing_only` is True, no new stream will be created.
793
794 Newly created streams are registered with the Snapcast server and fully
795 set up before being returned.
796
797 Args:
798 media: Media item to stream.
799 filter_settings_owner: Optional player/entity ID used to resolve DSP filters.
800 existing_only: If True, only return an existing stream.
801
802 Returns:
803 A ``SnapcastMAStream`` instance, or ``None`` if no stream exists and
804 `existing_only` is True.
805 """
806 stream_name: str = ""
807 name_suffix: str = ""
808 queue_id: str | None = None
809 source_id: str | None = None
810 destroy_on_stop = True
811
812 if media.media_type == MediaType.ANNOUNCEMENT:
813 stream_name += hashlib.md5(media.uri.encode()).hexdigest()[:6]
814 name_suffix = MASS_ANNOUNCEMENT_POSTFIX
815 elif media.media_type == MediaType.AUDIO_SOURCE and media.source_id:
816 # AudioSource queue items: scope the stream to the queue (media.source_id).
817 # Append a short hash of the queue_item_id so a queue rapidly
818 # re-selecting the same source (AudioSource A â track â AudioSource A
819 # again) cannot collide with a half-torn-down stream of the same
820 # name (destroy_on_stop is async; the create lookup can run before
821 # the delete completes). queue_item_id is fresh per play_media so
822 # this gives us a unique key per selection.
823 uniq = (
824 hashlib.md5(media.queue_item_id.encode()).hexdigest()[:6]
825 if media.queue_item_id
826 else ""
827 )
828 stream_name += f"{media.source_id}_{uniq}" if uniq else media.source_id
829 source_id = media.source_id
830 elif media.source_id and media.source_id.startswith(UGP_PREFIX):
831 stream_name += media.source_id
832 elif media.source_id and media.queue_item_id:
833 stream_name += media.source_id
834 queue_id = media.source_id
835 source_id = media.source_id
836 destroy_on_stop = False
837 else:
838 stream_name += hashlib.md5(media.uri.encode()).hexdigest()[:6]
839
840 stream_name = create_safe_string(stream_name, lowercase=False)
841 stream_name = f"{MASS_STREAM_PREFIX}{stream_name}{name_suffix}"
842 async with self._snapcast_ma_streams_lock:
843 if not (stream := self._snapcast_ma_streams.get(stream_name)):
844 if existing_only:
845 return None
846
847 stream = SnapcastMAStream(
848 provider=self,
849 media=media,
850 stream_name=stream_name,
851 filter_settings_owner=filter_settings_owner,
852 source_id=source_id,
853 use_cntrl_script=bool(queue_id) and self.queue_control_available,
854 destroy_on_stop=destroy_on_stop,
855 )
856 self._snapcast_ma_streams[stream_name] = stream
857 else:
858 stream.update_media(media)
859 await stream.setup()
860 return stream
861
862 def get_snap_ma_stream(self, stream_name: str) -> SnapcastMAStream | None:
863 """
864 Return an existing Music Assistant Snapcast stream by name.
865
866 Args:
867 stream_name: Snapcast stream name.
868
869 Returns:
870 The corresponding ``SnapcastMAStream`` instance, or ``None`` if not found.
871 """
872 return self._snapcast_ma_streams.get(stream_name)
873
874 async def delete_ma_stream(self, stream_name: str) -> None:
875 """
876 Remove and destroy a Music Assistant Snapcast stream.
877
878 The stream is removed from internal tracking and its resources are
879 destroyed asynchronously. Errors during destruction are logged but
880 otherwise ignored.
881
882 Args:
883 stream_name: Snapcast stream name to delete.
884 """
885 async with self._snapcast_ma_streams_lock:
886 stream = self._snapcast_ma_streams.pop(stream_name, None)
887
888 if not stream:
889 return
890
891 try:
892 await stream.destroy()
893 except Exception:
894 self.logger.exception("Failed to destroy stream session %s", stream_name)
895
896 def update_stream_usage(self) -> None:
897 """
898 Update usage state for all tracked Snapcast streams.
899
900 Marks streams as "in use" if they are currently assigned to any Snapcast
901 group, and schedules unused streams for delayed shutdown.
902
903 This method should be called whenever group or stream assignments change
904 on the Snapcast server.
905 """
906 unused_streams = set(self._snapcast_ma_streams.keys())
907 for grp in self._snapserver.groups:
908 stream_id = grp.stream
909 if stream_id in self._snapcast_ma_streams:
910 ma_stream = self._snapcast_ma_streams[stream_id]
911 ma_stream.set_in_use(True)
912 unused_streams.discard(stream_id)
913
914 if not unused_streams:
915 break
916
917 for stream_id in unused_streams:
918 self._snapcast_ma_streams[stream_id].set_in_use(False)
919
920 def get_snap_client(
921 self, *, client_id: str | None = None, player_id: str | None = None
922 ) -> SnapclientProto | None:
923 """Return the snapclient for either given client_id or player_id."""
924 if player_id is not None:
925 if client_id is not None and client_id != self._get_snapclient_id(client_id):
926 raise ValueError("provided client_id and player_id do not match")
927 client_id = self._get_snapclient_id(player_id)
928
929 if client_id:
930 with suppress(KeyError):
931 return self._snapserver.client(client_id)
932
933 return None
934
935 def get_snap_player(
936 self, *, client_id: str | None = None, player_id: str | None = None
937 ) -> SnapCastPlayer | None:
938 """Return the MA SnapCastPlayer for either given client_id or player_id."""
939 if client_id is not None:
940 if (mapped_id := self._get_ma_id(client_id)) is None:
941 return None
942 if player_id is not None and player_id != mapped_id:
943 raise ValueError("provided client_id and player_id do not match")
944 player_id = mapped_id
945
946 if player_id is None:
947 return None
948
949 if ma_player := self.mass.players.get_player(player_id):
950 assert isinstance(ma_player, SnapCastPlayer) # for type checking
951 return ma_player
952
953 return None
954
955 def _set_snapcast_log_level(self) -> None:
956 """Align snapcast's log level with the provider's log level."""
957 # snapcast is very chatty at debug level, so only pass through its
958 # debug logging when verbose logging is enabled
959 if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
960 logging.getLogger("snapcast").setLevel(logging.DEBUG)
961 else:
962 logging.getLogger("snapcast").setLevel(self.logger.level + 10)
963