music-assistant-server

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