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