music-assistant-server

1"""
2MusicAssistant PlayerController.
3
4Handles all logic to control supported players,
5which are provided by Player Providers.
6
7Note that the PlayerController has a concept of a 'player' and a 'playerstate'.
8The Player is the actual object that is provided by the provider,
9which incorporates the (unaltered) state of the player (e.g. volume, state, etc)
10and functions for controlling the player (e.g. play, pause, etc).
11
12The playerstate is the (final) state of the player, including any user customizations
13and transformations that are applied to the player.
14The playerstate is the object that is exposed to the outside world (via the API).
15"""
16
17from __future__ import annotations
18
19import asyncio
20import contextlib
21import time
22import weakref
23from collections.abc import AsyncIterator
24from contextlib import suppress
25from typing import TYPE_CHECKING, Any, cast
26
27from music_assistant_models.auth import Scope
28from music_assistant_models.background_task import TaskSchedule
29from music_assistant_models.config_entries import ConfigEntry
30from music_assistant_models.constants import (
31    PLAYER_CONTROL_FAKE,
32    PLAYER_CONTROL_NATIVE,
33    PLAYER_CONTROL_NONE,
34)
35from music_assistant_models.enums import (
36    ConfigEntryType,
37    EventType,
38    IdentifierType,
39    MediaType,
40    PlaybackState,
41    PlayerFeature,
42    PlayerType,
43    ProviderFeature,
44    ProviderType,
45    RepeatMode,
46    SourceControl,
47)
48from music_assistant_models.errors import (
49    AlreadyRegisteredError,
50    InsufficientPermissions,
51    InvalidCommand,
52    InvalidDataError,
53    MusicAssistantError,
54    PlayerCommandFailed,
55    PlayerUnavailableError,
56    ProviderUnavailableError,
57    UnsupportedFeaturedException,
58)
59from music_assistant_models.media_items import AudioSource
60from music_assistant_models.player import PlayerOptionValueType  # noqa: TC002
61from music_assistant_models.player_control import PlayerControl  # noqa: TC002
62
63from music_assistant.constants import (
64    ATTR_ACTIVE_SOURCE,
65    ATTR_ANNOUNCEMENT_IN_PROGRESS,
66    ATTR_AVAILABLE,
67    ATTR_ENABLED,
68    ATTR_FAKE_MUTE,
69    ATTR_FAKE_POWER,
70    ATTR_FAKE_VOLUME,
71    ATTR_GROUP_MEMBERS,
72    ATTR_GROUP_VOLUME_SNAPSHOT,
73    ATTR_LAST_POLL,
74    ATTR_MUTE_CONTROL,
75    ATTR_MUTE_LOCK,
76    ATTR_POWER_CONTROL,
77    ATTR_POWERED,
78    ATTR_PREVIOUS_VOLUME,
79    ATTR_SUPPORTED_FEATURES,
80    ATTR_VOLUME_CONTROL,
81    ATTR_VOLUME_TARGET,
82    CONF_ANNOUNCE_TTS_ENGINE,
83    CONF_AUTO_PLAY,
84    CONF_CACHED_ARP_MAC,
85    CONF_ENTRY_MAX_VOLUME,
86    CONF_ENTRY_MIN_VOLUME,
87    CONF_GROUP_MEMBERS,
88    CONF_MAX_VOLUME,
89    CONF_MIN_VOLUME,
90    CONF_MUTE_CONTROL,
91    CONF_PLAY_MEDIA_OVERRIDES_GROUP,
92    CONF_PLAYER_DSP,
93    CONF_PLAYER_QUEUES,
94    CONF_PLAYERS,
95    CONF_POWER_CONTROL,
96    CONF_PROTOCOL_PARENT_ID,
97    CONF_REPORTED_MAC,
98    CONF_VOLUME_CONTROL,
99    CONF_VOLUME_STEP,
100    VERBOSE_LOG_LEVEL,
101)
102from music_assistant.controllers.webserver.helpers.auth_middleware import (
103    get_current_user,
104    get_sendspin_player_id,
105    has_scope,
106)
107from music_assistant.helpers.api import api_command
108from music_assistant.helpers.colors import get_palette_for_url
109from music_assistant.helpers.config_entries import PLAYBACK_TARGET_TYPES
110from music_assistant.helpers.plugin_engines import create_tts_engine_config_entries
111from music_assistant.helpers.util import (
112    TaskManager,
113    enrich_device_mac_address,
114    is_valid_mac_address,
115)
116from music_assistant.models.core_controller import CoreController
117from music_assistant.models.player import Player, PlayerMedia, PlayerState
118from music_assistant.models.player_provider import PlayerProvider
119from music_assistant.models.plugin import PluginProvider, SourceControlValue
120
121from .announcements import AnnouncementsMixin
122from .audio_sources import AudioSourceMixin, AudioSourceSession
123from .constants import PlayerLockPurpose
124from .helpers import handle_player_command, wait_for_power_on
125from .protocol_linking import ProtocolLinkingMixin
126
127if TYPE_CHECKING:
128    from collections.abc import Callable, Iterator
129
130    from music_assistant_models.config_entries import (
131        CoreConfig,
132        PlayerConfig,
133    )
134    from music_assistant_models.player import OutputProtocol
135    from music_assistant_models.player_queue import PlayerQueue
136
137    from music_assistant import MusicAssistant
138    from music_assistant.helpers.json import SerializableType
139
140CACHE_CATEGORY_PLAYER_POWER = 1
141
142# state keys that carry the current_media playback-position anchor; these only
143# change on discrete position events (play/pause/seek/track change/buffer correction)
144POSITION_ANCHOR_KEYS = frozenset(
145    {
146        "current_media.elapsed_time",
147        "current_media.elapsed_time_last_updated",
148    }
149)
150
151# How long the volume level of the last command outranks the level the player reports.
152# Long enough to cover a burst of volume nudges on a player that only reports its volume
153# back some time later, short enough for a change made on the device itself to win again.
154VOLUME_TARGET_EXPIRY = 2.0
155
156# How long a freshly started source session may wait for its first stream request
157# before it is considered never started and released.
158AUDIO_SOURCE_CLAIM_TIMEOUT = 30
159
160# How long a player must stay powered off before the queue it was playing is ended.
161# Home Assistant reports an entity that is (briefly) unavailable or unknown as off, so an
162# external power control can report a power off that comes straight back - which must not
163# stop the music. Long enough to sit out a reloading integration or a device missing a
164# poll, short enough that a real power off is not left streaming.
165EXTERNAL_POWER_OFF_STOP_DELAY = 15
166
167# Sentinel used to detect omitted optional arguments where ``None`` is a valid value.
168_SENTINEL: Any = object()
169
170
171class PlayerController(AnnouncementsMixin, AudioSourceMixin, ProtocolLinkingMixin, CoreController):
172    """Controller holding all logic to control registered players."""
173
174    domain: str = "players"
175
176    def __init__(self, mass: MusicAssistant) -> None:
177        """Initialize core controller."""
178        super().__init__(mass)
179        self._players: dict[str, Player] = {}
180        self._controls: dict[str, PlayerControl] = {}
181        self.manifest.name = "Player Controller"
182        self.manifest.description = (
183            "Music Assistant's core controller which manages all players from all providers."
184        )
185        self.manifest.icon = "speaker-multiple"
186        self._poll_task: asyncio.Task[None] | None = None
187        self._player_command_locks: dict[str, asyncio.Lock] = {}
188        # Re-entrancy tracking for get_player_lock, keyed on the task object
189        # (weak ref auto-clears entries if a task is GC'd before its finally runs).
190        self._task_held_locks: weakref.WeakKeyDictionary[asyncio.Task[Any], set[str]] = (
191            weakref.WeakKeyDictionary()
192        )
193        # Lock to prevent race conditions during player registration
194        self._register_lock = asyncio.Lock()
195        # Track pending protocol player evaluations (delayed to allow all protocols to register)
196        self._pending_protocol_evaluations: dict[str, asyncio.TimerHandle] = {}
197        # Serialize delayed evaluations to prevent race conditions
198        self._delayed_evaluation_lock = asyncio.Lock()
199        # Live external AudioSource playing on a player, keyed on player_id
200        self._source_sessions: dict[str, AudioSourceSession] = {}
201        # Subscribers for player state updates (called with player + changed_values)
202        self._state_update_subscribers: list[
203            Callable[[Player, dict[str, tuple[Any, Any]]], None]
204        ] = []
205
206    @contextlib.asynccontextmanager
207    async def get_player_lock(
208        self, player_id: str, purpose: PlayerLockPurpose = PlayerLockPurpose.PLAYBACK
209    ) -> AsyncIterator[None]:
210        """
211        Acquire a purpose-scoped lock for a player, with re-entrant support.
212
213        Tracks lock ownership per asyncio Task so that nested calls within the same
214        task skip re-acquisition (preventing deadlocks), while deferred callbacks
215        (call_later / create_task) correctly acquire a fresh lock.
216
217        If the lock can't be acquired within 30s the body runs anyway, to keep
218        the player responsive when a previous holder is stuck on a hung command.
219
220        :param player_id: The player to lock.
221        :param purpose: Lock category. Commands with different purposes can run
222            concurrently on the same player.
223        """
224        lock_key = f"{purpose.value}_{player_id}"
225        task = asyncio.current_task()
226
227        if task is not None and lock_key in self._task_held_locks.get(task, set()):
228            yield
229            return
230
231        lock = self._player_command_locks.setdefault(lock_key, asyncio.Lock())
232        # Two-stage acquire: a slow-acquire log at 5s and a hard give-up at 30s.
233        # If the previous holder is stuck (e.g. on a dead provider socket), we
234        # proceed without the lock so this player stays responsive.
235        acquired = False
236        try:
237            async with asyncio.timeout(5):
238                await lock.acquire()
239            acquired = True
240        except TimeoutError:
241            self.logger.debug(
242                "Acquiring %s lock for player %s is slow (>5s)", purpose.value, player_id
243            )
244            try:
245                async with asyncio.timeout(25):
246                    await lock.acquire()
247                acquired = True
248            except TimeoutError:
249                self.logger.warning(
250                    "Timed out (30s) acquiring %s lock for player %s — "
251                    "previous holder appears stuck; proceeding without lock",
252                    purpose.value,
253                    player_id,
254                )
255
256        if acquired and task is not None:
257            self._task_held_locks.setdefault(task, set()).add(lock_key)
258        try:
259            yield
260        finally:
261            if acquired:
262                if task is not None and (held := self._task_held_locks.get(task)) is not None:
263                    held.discard(lock_key)
264                    if not held:
265                        del self._task_held_locks[task]
266                lock.release()
267
268    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
269        """Return Config Entries for the Player Controller."""
270        return (
271            ConfigEntry(
272                key=CONF_VOLUME_STEP,
273                type=ConfigEntryType.INTEGER,
274                default_value=0,
275                range=(0, 10),
276                required=False,
277                category="generic",
278            ),
279            *await create_tts_engine_config_entries(
280                self.mass, CONF_ANNOUNCE_TTS_ENGINE, category="announcements"
281            ),
282        )
283
284    async def setup(self, config: CoreConfig) -> None:
285        """Async initialize of module."""
286        self._repair_protocol_parent_links()
287        self._poll_task = self.mass.create_task(self._poll_players())
288        self.mass.tasks.register_scheduled_task(
289            task_id="fix_group_member_configs",
290            name="Fix sync group member configurations",
291            handler=self._fix_group_member_configs,
292            schedule=TaskSchedule.weekly(
293                days_of_week=[0],
294                hour=4,
295                minute=0,
296            ),
297            initial_delay=300,
298        )
299
300    async def close(self) -> None:
301        """Cleanup on exit."""
302        if self._poll_task and not self._poll_task.done():
303            self._poll_task.cancel()
304        # Cancel all pending protocol evaluations
305        for handle in self._pending_protocol_evaluations.values():
306            handle.cancel()
307        self._pending_protocol_evaluations.clear()
308        for player in self._players.values():
309            if player.sleep_timer_expires_at is not None:
310                self.mass.cancel_timer(self._sleep_timer_task_id(player.player_id))
311
312    async def get_diagnostics(self) -> dict[str, SerializableType]:
313        """Return diagnostics info for this controller to include in diagnostics reports."""
314        players = list(self._players.values())
315        return {
316            "players_synced": sum(player.state.synced_to is not None for player in players),
317            "players_with_active_group": sum(
318                player.state.active_group is not None for player in players
319            ),
320            "announcements_in_progress": sum(
321                bool(player.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS)) for player in players
322            ),
323            "pending_protocol_evaluations": len(self._pending_protocol_evaluations),
324        }
325
326    async def on_provider_loaded(self, provider: PlayerProvider) -> None:
327        """Handle logic when a provider is loaded."""
328
329    async def on_provider_unload(self, provider: PlayerProvider) -> None:
330        """Handle logic when a provider is (about to get) unloaded."""
331
332    @property
333    def providers(self) -> list[PlayerProvider]:
334        """Return all loaded/running MusicProviders."""
335        return cast("list[PlayerProvider]", self.mass.get_providers(ProviderType.PLAYER))
336
337    def iter_players(
338        self,
339        return_unavailable: bool = True,
340        return_disabled: bool = False,
341        provider_filter: str | None = None,
342        return_protocol_players: bool = False,
343    ) -> Iterator[Player]:
344        """
345        Iterate over all registered players, regardless of who is asking.
346
347        Use this for internal logic - state derivation, bookkeeping and topology
348        lookups - which must stay correct no matter which user's command happened
349        to trigger it. Use :meth:`all_players` for anything presented to a user.
350
351        :param return_unavailable [bool]: Include unavailable players.
352        :param return_disabled [bool]: Include disabled players.
353        :param provider_filter [str]: Optional filter by provider lookup key.
354        :param return_protocol_players [bool]: Include protocol players (hidden by default).
355        """
356        for player in list(self._players.values()):
357            if not (player.state.available or return_unavailable):
358                continue
359            if not (player.state.enabled or return_disabled):
360                continue
361            if not player.initialized.is_set():
362                continue
363            if provider_filter is not None and player.provider.instance_id != provider_filter:
364                continue
365            if not return_protocol_players and player.state.type == PlayerType.PROTOCOL:
366                continue
367            yield player
368
369    def all_players(
370        self,
371        return_unavailable: bool = True,
372        return_disabled: bool = False,
373        provider_filter: str | None = None,
374        return_protocol_players: bool = False,
375    ) -> list[Player]:
376        """
377        Return the registered players the current user is allowed to see.
378
379        Note that this applies user filters for players (for non admin users),
380        which makes it unsuitable for internal logic - use :meth:`iter_players` there.
381
382        :param return_unavailable [bool]: Include unavailable players.
383        :param return_disabled [bool]: Include disabled players.
384        :param provider_filter [str]: Optional filter by provider lookup key.
385        :param return_protocol_players [bool]: Include protocol players (hidden by default).
386
387        :return: List of Player objects.
388        """
389        current_user = get_current_user()
390        user_filter = (
391            current_user.player_filter
392            if current_user and not has_scope(current_user, Scope.ALL)
393            else None
394        )
395        current_sendspin_player = get_sendspin_player_id()
396        return [
397            player
398            for player in self.iter_players(
399                return_unavailable=return_unavailable,
400                return_disabled=return_disabled,
401                provider_filter=provider_filter,
402                return_protocol_players=return_protocol_players,
403            )
404            if not user_filter
405            or player.player_id in user_filter
406            or player.player_id == current_sendspin_player
407        ]
408
409    @api_command("players/all", required_scope=Scope.PLAYERS_READ)
410    def all_player_states(
411        self,
412        return_unavailable: bool = True,
413        return_disabled: bool = False,
414        provider_filter: str | None = None,
415        return_protocol_players: bool = False,
416    ) -> list[PlayerState]:
417        """
418        Return PlayerState for all registered players.
419
420        :param return_unavailable [bool]: Include unavailable players.
421        :param return_disabled [bool]: Include disabled players.
422        :param provider_filter [str]: Optional filter by provider lookup key.
423        :param return_protocol_players [bool]: Include protocol players (hidden by default).
424
425        :return: List of PlayerState objects.
426        """
427        return [
428            player.state
429            for player in self.all_players(
430                return_unavailable=return_unavailable,
431                return_disabled=return_disabled,
432                provider_filter=provider_filter,
433                return_protocol_players=return_protocol_players,
434            )
435        ]
436
437    def get_player(
438        self,
439        player_id: str,
440        raise_unavailable: bool = False,
441    ) -> Player | None:
442        """
443        Return Player by player_id.
444
445        :param player_id [str]: ID of the player.
446        :param raise_unavailable [bool]: Raise if player is unavailable.
447
448        :raises PlayerUnavailableError: If player is unavailable and raise_unavailable is True.
449        :return: Player object or None.
450        """
451        if player := self._players.get(player_id):
452            if (not player.state.available or not player.state.enabled) and raise_unavailable:
453                msg = f"Player {player_id} is not available"
454                raise PlayerUnavailableError(msg)
455            return player
456        if raise_unavailable:
457            msg = f"Player {player_id} is not available"
458            raise PlayerUnavailableError(msg)
459        return None
460
461    @api_command("players/get", required_scope=Scope.PLAYERS_READ)
462    def get_player_state(
463        self,
464        player_id: str,
465        raise_unavailable: bool = False,
466    ) -> PlayerState | None:
467        """
468        Return PlayerState by player_id.
469
470        :param player_id [str]: ID of the player.
471        :param raise_unavailable [bool]: Raise if player is unavailable.
472
473        :raises PlayerUnavailableError: If player is unavailable and raise_unavailable is True.
474        :return: Player object or None.
475        """
476        current_user = get_current_user()
477        user_filter = (
478            current_user.player_filter
479            if current_user and not has_scope(current_user, Scope.ALL)
480            else None
481        )
482        current_sendspin_player = get_sendspin_player_id()
483        if (
484            current_user
485            and user_filter
486            and player_id not in user_filter
487            and player_id != current_sendspin_player
488        ):
489            msg = f"{current_user.username} does not have access to player {player_id}"
490            raise InsufficientPermissions(msg)
491        if player := self.get_player(player_id, raise_unavailable):
492            return player.state
493        return None
494
495    def get_player_by_name(self, name: str) -> Player | None:
496        """
497        Return Player by name.
498
499        Performs case-insensitive matching against the player's state name
500        (the final name visible in clients and API).
501        If multiple players match, logs a warning and returns the first match.
502
503        :param name: Name of the player.
504        :return: Player object or None.
505        """
506        name_normalized = name.strip().lower()
507        matches: list[Player] = []
508
509        for player in list(self._players.values()):
510            if player.state.name.strip().lower() == name_normalized:
511                matches.append(player)
512
513        if not matches:
514            return None
515
516        if len(matches) > 1:
517            player_ids = [p.player_id for p in matches]
518            self.logger.warning(
519                "players/get_by_name: Multiple players found with name '%s': %s - "
520                "returning first match (%s). "
521                "Consider using the players/get API with player_id instead "
522                "for unambiguous lookups.",
523                name,
524                player_ids,
525                matches[0].player_id,
526            )
527
528        return matches[0]
529
530    @api_command("players/get_by_name", required_scope=Scope.PLAYERS_READ)
531    def get_player_state_by_name(self, name: str) -> PlayerState | None:
532        """
533        Return PlayerState by name.
534
535        :param name: Name of the player.
536        :return: PlayerState object or None.
537        """
538        current_user = get_current_user()
539        user_filter = (
540            current_user.player_filter
541            if current_user and not has_scope(current_user, Scope.ALL)
542            else None
543        )
544        current_sendspin_player = get_sendspin_player_id()
545        if player := self.get_player_by_name(name):
546            if (
547                current_user
548                and user_filter
549                and player.player_id not in user_filter
550                and player.player_id != current_sendspin_player
551            ):
552                msg = f"{current_user.username} does not have access to player {player.player_id}"
553                raise InsufficientPermissions(msg)
554            return player.state
555        return None
556
557    @api_command("players/player_controls", required_scope=Scope.PLAYERS_READ)
558    def player_controls(
559        self,
560    ) -> list[PlayerControl]:
561        """Return all registered playercontrols."""
562        return list(self._controls.values())
563
564    @api_command("players/player_control", required_scope=Scope.PLAYERS_READ)
565    def get_player_control(
566        self,
567        control_id: str,
568    ) -> PlayerControl | None:
569        """
570        Return PlayerControl by control_id.
571
572        :param control_id: ID of the player control.
573        :return: PlayerControl object or None.
574        """
575        if control := self._controls.get(control_id):
576            return control
577        return None
578
579    @api_command("players/sleep_timer/get", required_scope=Scope.PLAYERS_READ)
580    def get_sleep_timer(self, player_id: str) -> float | None:
581        """
582        Return the active sleep timer expiry timestamp for the player.
583
584        :param player_id: Player ID to check.
585        """
586        player = self._get_player_with_redirect(player_id)
587        return player.sleep_timer_expires_at
588
589    @api_command("players/sleep_timer/set", required_scope=Scope.PLAYERS_CONTROL)
590    def set_sleep_timer(self, player_id: str, seconds: int) -> float:
591        """
592        Set a sleep timer for the player.
593
594        :param player_id: Player ID to set the timer for.
595        :param seconds: Delay in seconds before playback is stopped.
596        """
597        if seconds <= 0:
598            msg = "Sleep timer duration must be greater than zero seconds"
599            raise InvalidDataError(msg)
600        player = self._get_player_with_redirect(player_id)
601        try:
602            # guard against absurd durations that overflow the float timestamp math
603            expires_at = time.time() + seconds
604        except OverflowError:
605            msg = "Sleep timer duration is too large to schedule"
606            raise InvalidDataError(msg) from None
607        player.set_sleep_timer_expires_at(expires_at)
608        player.update_state()
609        self._signal_sleep_timer_updated(player, expires_at)
610        self.mass.call_later(
611            seconds,
612            self._handle_sleep_timer_expired,
613            player.player_id,
614            task_id=self._sleep_timer_task_id(player.player_id),
615        )
616        return expires_at
617
618    @api_command("players/sleep_timer/clear", required_scope=Scope.PLAYERS_CONTROL)
619    def clear_sleep_timer(self, player_id: str) -> None:
620        """
621        Clear the active sleep timer for the player.
622
623        :param player_id: Player ID to clear the timer for.
624        """
625        player = self._get_player_with_redirect(player_id)
626        self._clear_sleep_timer(player)
627
628    # Player commands
629
630    @api_command("players/cmd/stop", required_scope=Scope.PLAYERS_CONTROL)
631    @handle_player_command
632    async def cmd_stop(self, player_id: str) -> None:
633        """
634        Send STOP command to given player.
635
636        - player_id: player_id of the player to handle the command.
637        """
638        player = self._get_player_with_redirect(player_id)
639        async with self.get_player_lock(player.player_id, PlayerLockPurpose.PLAYBACK):
640            # Redirect to queue controller if it is active (skip if already in queue command context)
641            if active_queue := self.get_active_queue(player):
642                await self.mass.player_queues.stop(active_queue.queue_id)
643                return
644            # Delegate to internal handler for actual implementation
645            await self._handle_cmd_stop(player.player_id)
646
647    @api_command("players/cmd/play", required_scope=Scope.PLAYERS_CONTROL)
648    @handle_player_command
649    async def cmd_play(self, player_id: str) -> None:
650        """
651        Send PLAY (unpause) command to given player.
652
653        - player_id: player_id of the player to handle the command.
654        """
655        player = self._get_player_with_redirect(player_id)
656        async with self.get_player_lock(player.player_id, PlayerLockPurpose.PLAYBACK):
657            if player.state.playback_state == PlaybackState.PLAYING:
658                self.logger.info(
659                    "Ignore PLAY request to player %s: player is already playing",
660                    player.state.name,
661                )
662                return
663            # player is not paused: check for queue redirect, then delegate to internal handler
664            if player.state.playback_state != PlaybackState.PAUSED:
665                source = player.state.active_source
666                if active_queue := self.mass.player_queues.get(source or player_id):
667                    await self.mass.player_queues.resume(active_queue.queue_id)
668                    return
669            # Delegate to internal handler for actual implementation
670            await self._handle_cmd_play(player.player_id)
671
672    @api_command("players/cmd/pause", required_scope=Scope.PLAYERS_CONTROL)
673    @handle_player_command
674    async def cmd_pause(self, player_id: str) -> None:
675        """
676        Send PAUSE command to given player.
677
678        - player_id: player_id of the player to handle the command.
679        """
680        player = self._get_player_with_redirect(player_id)
681        # Redirect to queue controller if it is active (skip if already in queue command context)
682        if active_queue := self.get_active_queue(player):
683            await self.mass.player_queues.pause(active_queue.queue_id)
684            return
685        # Delegate to internal handler for actual implementation
686        await self._handle_cmd_pause(player.player_id)
687
688    @api_command("players/cmd/play_pause", required_scope=Scope.PLAYERS_CONTROL)
689    async def cmd_play_pause(self, player_id: str) -> None:
690        """
691        Toggle play/pause on given player.
692
693        - player_id: player_id of the player to handle the command.
694        """
695        player = self._get_player_with_redirect(player_id)
696        if player.state.playback_state == PlaybackState.PLAYING:
697            await self.cmd_pause(player.player_id)
698        else:
699            await self.cmd_play(player.player_id)
700
701    @api_command("players/cmd/resume", required_scope=Scope.PLAYERS_CONTROL)
702    @handle_player_command
703    async def cmd_resume(
704        self, player_id: str, source: str | None = None, media: PlayerMedia | None = None
705    ) -> None:
706        """
707        Send RESUME command to given player.
708
709        Resume (or restart) playback on the player.
710
711        :param player_id: player_id of the player to handle the command.
712        :param source: Optional source to resume.
713        :param media: Optional media to resume.
714        """
715        player = self._get_player_with_redirect(player_id)
716        async with self.get_player_lock(player.player_id, PlayerLockPurpose.PLAYBACK):
717            await self._handle_cmd_resume(player.player_id, source, media)
718
719    @api_command("players/cmd/seek", required_scope=Scope.PLAYERS_CONTROL)
720    @handle_player_command
721    async def cmd_seek(self, player_id: str, position: int) -> None:
722        """
723        Handle SEEK command for given player.
724
725        - player_id: player_id of the player to handle the command.
726        - position: position in seconds to seek to in the current playing item.
727        """
728        player = self._get_player_with_redirect(player_id)
729        if await self._forward_to_external_source(player, SourceControl.SEEK, position):
730            return
731        # Redirect to queue controller if it is active
732        if active_queue := self.get_active_queue(player):
733            await self.mass.player_queues.seek(active_queue.queue_id, position)
734            return
735        # handle command on player/source directly
736        active_source = next((x for x in player.source_list if x.id == player.active_source), None)
737        if active_source and not active_source.can_seek:
738            msg = (
739                f"The active source ({active_source.name}) on player "
740                f"{player.display_name} does not support seeking"
741            )
742            raise PlayerCommandFailed(msg)
743        if PlayerFeature.SEEK not in player.supported_features:
744            msg = f"Player {player.display_name} does not support seeking"
745            raise UnsupportedFeaturedException(msg)
746        # handle command on player directly
747        await player.seek(position)
748
749    @api_command("players/cmd/shuffle", required_scope=Scope.PLAYERS_CONTROL)
750    @handle_player_command
751    async def cmd_shuffle(
752        self, player_id: str, shuffle_enabled: bool, source_id: str | None = None
753    ) -> None:
754        """
755        Handle SHUFFLE command for given player.
756
757        Applies to whatever the player is playing: a live external source orders its
758        own session, a source the device runs itself orders its own content, and
759        Music Assistant's queue orders its own items.
760
761        :param player_id: player_id of the player to handle the command.
762        :param shuffle_enabled: Whether to play the current content shuffled.
763        :param source_id: Optional source (id) the command is aimed at, as listed in the
764            player's source_list. Given one, the command is refused when that source is
765            no longer playing, so it can never land on whatever took the player since.
766        """
767        player = self._get_player_with_redirect(player_id)
768        active_source_id = self._resolve_command_target(player, source_id)
769        if await self._forward_to_external_source(player, SourceControl.SHUFFLE, shuffle_enabled):
770            return
771        if active_queue := self.get_active_queue(player):
772            await self.mass.player_queues.set_shuffle(active_queue.queue_id, shuffle_enabled)
773            return
774        if active_source := next(
775            (x for x in player.state.source_list if x.id == active_source_id), None
776        ):
777            # the source belongs to the player itself (its own Spotify Connect, a device input)
778            if not active_source.can_shuffle:
779                msg = "This action is (currently) unavailable for this source."
780                raise PlayerCommandFailed(msg)
781            await player.set_shuffle(shuffle_enabled)
782            return
783        msg = f"There is nothing playing on {player.state.name} to shuffle."
784        raise PlayerCommandFailed(msg)
785
786    @api_command("players/cmd/repeat", required_scope=Scope.PLAYERS_CONTROL)
787    @handle_player_command
788    async def cmd_repeat(
789        self, player_id: str, repeat_mode: RepeatMode, source_id: str | None = None
790    ) -> None:
791        """
792        Handle REPEAT command for given player.
793
794        Applies to whatever the player is playing: a live external source repeats
795        within its own session, a source the device runs itself repeats its own
796        content, and Music Assistant's queue repeats its own items.
797
798        :param player_id: player_id of the player to handle the command.
799        :param repeat_mode: The repeat mode to apply.
800        :param source_id: Optional source (id) the command is aimed at, as listed in the
801            player's source_list. Given one, the command is refused when that source is
802            no longer playing, so it can never land on whatever took the player since.
803        """
804        if repeat_mode == RepeatMode.UNKNOWN:
805            # not a mode to set: it is what a source reports when it cannot say
806            raise InvalidCommand("Cannot set an unknown repeat mode")
807        player = self._get_player_with_redirect(player_id)
808        active_source_id = self._resolve_command_target(player, source_id)
809        if await self._forward_to_external_source(player, SourceControl.REPEAT, repeat_mode):
810            return
811        if active_queue := self.get_active_queue(player):
812            await self.mass.player_queues.set_repeat(active_queue.queue_id, repeat_mode)
813            return
814        if active_source := next(
815            (x for x in player.state.source_list if x.id == active_source_id), None
816        ):
817            # the source belongs to the player itself (its own Spotify Connect, a device input)
818            if not active_source.can_repeat:
819                msg = "This action is (currently) unavailable for this source."
820                raise PlayerCommandFailed(msg)
821            await player.set_repeat(repeat_mode)
822            return
823        msg = f"There is nothing playing on {player.state.name} to repeat."
824        raise PlayerCommandFailed(msg)
825
826    @api_command("players/cmd/next", required_scope=Scope.PLAYERS_CONTROL)
827    @handle_player_command
828    async def cmd_next_track(self, player_id: str) -> None:
829        """Handle NEXT TRACK command for given player."""
830        player = self._get_player_with_redirect(player_id)
831        active_source_id = player.state.active_source or player.player_id
832        if await self._forward_to_external_source(player, SourceControl.NEXT):
833            return
834        # Redirect to queue controller if it is active
835        if active_queue := self.get_active_queue(player):
836            await self.mass.player_queues.next(active_queue.queue_id)
837            return
838        if PlayerFeature.NEXT_PREVIOUS in player.state.supported_features:
839            # player has some other source active and native next/previous support
840            active_source = next(
841                (x for x in player.state.source_list if x.id == active_source_id), None
842            )
843            if active_source and active_source.can_next_previous:
844                await player.next_track()
845                return
846            msg = "This action is (currently) unavailable for this source."
847            raise PlayerCommandFailed(msg)
848        # Player does not support next/previous feature
849        msg = f"Player {player.state.name} does not support skipping to the next track."
850        raise UnsupportedFeaturedException(msg)
851
852    @api_command("players/cmd/previous", required_scope=Scope.PLAYERS_CONTROL)
853    @handle_player_command
854    async def cmd_previous_track(self, player_id: str) -> None:
855        """Handle PREVIOUS TRACK command for given player."""
856        player = self._get_player_with_redirect(player_id)
857        active_source_id = player.state.active_source or player.player_id
858        if await self._forward_to_external_source(player, SourceControl.PREVIOUS):
859            return
860        # Redirect to queue controller if it is active
861        if active_queue := self.get_active_queue(player):
862            await self.mass.player_queues.previous(active_queue.queue_id)
863            return
864        if PlayerFeature.NEXT_PREVIOUS in player.state.supported_features:
865            # player has some other source active and native next/previous support
866            active_source = next(
867                (x for x in player.state.source_list if x.id == active_source_id), None
868            )
869            if active_source and active_source.can_next_previous:
870                await player.previous_track()
871                return
872            msg = "This action is (currently) unavailable for this source."
873            raise PlayerCommandFailed(msg)
874        # Player does not support next/previous feature
875        msg = f"Player {player.state.name} does not support skipping to the previous track."
876        raise UnsupportedFeaturedException(msg)
877
878    @api_command("players/cmd/power", required_scope=Scope.PLAYERS_CONTROL)
879    @handle_player_command(lock=PlayerLockPurpose.PLAYBACK)
880    async def cmd_power(self, player_id: str, powered: bool) -> None:
881        """
882        Send POWER command to given player.
883
884        :param player_id: player_id of the player to handle the command.
885        :param powered: bool if player should be powered on or off.
886        """
887        # Power is serialized with PLAYBACK because powering on a sync/group player
888        # forms the group (and powering off dissolves it) - this must not race with
889        # play_media / cmd_resume / cmd_set_members on the same player.
890        await self._handle_cmd_power(player_id, powered)
891
892    @api_command("players/cmd/volume_set", required_scope=Scope.PLAYERS_CONTROL)
893    @handle_player_command
894    async def cmd_volume_set(self, player_id: str, volume_level: int) -> None:
895        """
896        Send VOLUME_SET command to given player.
897
898        :param player_id: player_id of the player to handle the command.
899        :param volume_level: volume level (0..100) to set on the player.
900        """
901        volume_level = max(0, min(100, volume_level))
902        # record the level and invalidate the group volume state up front, before waiting
903        # for the volume lock: a command that is still queued would otherwise undo what a
904        # command issued after it already recorded.
905        # skip for group players since _handle_cmd_volume_set redirects those to
906        # set_group_volume which creates/uses the snapshot itself
907        if (player := self.get_player(player_id)) and player.type != PlayerType.GROUP:
908            self._record_volume_target(player, volume_level)
909            self._invalidate_group_volume_snapshot(player_id)
910        async with self.get_player_lock(player_id, PlayerLockPurpose.VOLUME):
911            await self._handle_cmd_volume_set(player_id, volume_level, record_target=False)
912
913    @api_command("players/cmd/volume_up", required_scope=Scope.PLAYERS_CONTROL)
914    @handle_player_command
915    async def cmd_volume_up(self, player_id: str) -> None:
916        """
917        Send VOLUME_UP command to given player.
918
919        - player_id: player_id of the player to handle the command.
920        """
921        if not (player := self.get_player(player_id)):
922            return
923        if player.type == PlayerType.GROUP:
924            await self.cmd_group_volume_up(player_id)
925            return
926        current_volume = self._volume_nudge_base(player) or 0
927        new_volume = min(100, current_volume + self._get_volume_step(current_volume))
928        await self.cmd_volume_set(player_id, new_volume)
929
930    @api_command("players/cmd/volume_down", required_scope=Scope.PLAYERS_CONTROL)
931    @handle_player_command
932    async def cmd_volume_down(self, player_id: str) -> None:
933        """
934        Send VOLUME_DOWN command to given player.
935
936        - player_id: player_id of the player to handle the command.
937        """
938        if not (player := self.get_player(player_id)):
939            return
940        if player.type == PlayerType.GROUP:
941            await self.cmd_group_volume_down(player_id)
942            return
943        current_volume = self._volume_nudge_base(player) or 0
944        new_volume = max(0, current_volume - self._get_volume_step(current_volume))
945        await self.cmd_volume_set(player_id, new_volume)
946
947    @api_command("players/cmd/group_volume", required_scope=Scope.PLAYERS_CONTROL)
948    @handle_player_command
949    async def cmd_group_volume(
950        self,
951        player_id: str,
952        volume_level: int,
953    ) -> None:
954        """
955        Handle adjusting the overall/group volume to a playergroup (or synced players).
956
957        Will set a new (overall) volume level to a group player or syncgroup.
958
959        :param player_id: Player ID of group player or syncleader to handle the command.
960        :param volume_level: Volume level (0..100) to set to the group.
961        """
962        player = self.get_player(player_id, True)
963        assert player is not None  # for type checker
964        group_player = self._resolve_group_volume_player(player)
965        if group_player is None:
966            # treat as normal player volume change
967            await self.cmd_volume_set(player_id, volume_level)
968            return
969        async with self.get_player_lock(group_player.player_id, PlayerLockPurpose.GROUP_VOLUME):
970            await self.set_group_volume(group_player, volume_level)
971
972    @api_command("players/cmd/group_volume_up", required_scope=Scope.PLAYERS_CONTROL)
973    @handle_player_command
974    async def cmd_group_volume_up(self, player_id: str) -> None:
975        """
976        Send VOLUME_UP command to given playergroup.
977
978        - player_id: player_id of the player to handle the command.
979        """
980        player = self.get_player(player_id, True)
981        assert player is not None  # for type checker
982        # step from the volume of the group as a whole, which is not the volume of the
983        # addressed player when the command is addressed to one of its synced members
984        group_player = self._resolve_group_volume_player(player) or player
985        async with self.get_player_lock(group_player.player_id, PlayerLockPurpose.GROUP_VOLUME):
986            cur_volume = self._group_volume_nudge_base(group_player)
987            if cur_volume is None:
988                return
989            new_volume = min(100, cur_volume + self._get_volume_step(cur_volume))
990            await self.cmd_group_volume(player_id, new_volume)
991
992    @api_command("players/cmd/group_volume_down", required_scope=Scope.PLAYERS_CONTROL)
993    @handle_player_command
994    async def cmd_group_volume_down(self, player_id: str) -> None:
995        """
996        Send VOLUME_DOWN command to given playergroup.
997
998        - player_id: player_id of the player to handle the command.
999        """
1000        player = self.get_player(player_id, True)
1001        assert player is not None  # for type checker
1002        group_player = self._resolve_group_volume_player(player) or player
1003        async with self.get_player_lock(group_player.player_id, PlayerLockPurpose.GROUP_VOLUME):
1004            cur_volume = self._group_volume_nudge_base(group_player)
1005            if cur_volume is None:
1006                return
1007            new_volume = max(0, cur_volume - self._get_volume_step(cur_volume))
1008            await self.cmd_group_volume(player_id, new_volume)
1009
1010    @api_command("players/cmd/group_volume_mute", required_scope=Scope.PLAYERS_CONTROL)
1011    @handle_player_command
1012    async def cmd_group_volume_mute(self, player_id: str, muted: bool) -> None:
1013        """
1014        Handle muting a playergroup (or synced players) as a whole.
1015
1016        A group player or syncleader mutes all of its members, a synced player is
1017        redirected to its syncleader and an ungrouped player is muted on its own.
1018
1019        :param player_id: Player ID of the player to handle the command.
1020        :param muted: bool if the group should be muted.
1021        """
1022        player = self.get_player(player_id, True)
1023        assert player is not None  # for type checker
1024        if player.state.type == PlayerType.GROUP or player.state.group_members:
1025            # dedicated group player or sync leader
1026            await self._mute_group_members(player, muted)
1027            return
1028        if player.state.synced_to and (sync_leader := self.get_player(player.state.synced_to)):
1029            # redirect to sync leader
1030            await self._mute_group_members(sync_leader, muted)
1031            return
1032        # treat as normal player mute
1033        await self.cmd_volume_mute(player_id, muted)
1034
1035    @api_command("players/cmd/volume_mute", required_scope=Scope.PLAYERS_CONTROL)
1036    @handle_player_command(lock=PlayerLockPurpose.VOLUME)
1037    async def cmd_volume_mute(self, player_id: str, muted: bool) -> None:
1038        """
1039        Send VOLUME_MUTE command to given player.
1040
1041        - player_id: player_id of the player to handle the command.
1042        - muted: bool if player should be muted.
1043        """
1044        player = self.get_player(player_id, True)
1045        assert player
1046
1047        if player.type == PlayerType.GROUP:
1048            # redirect to special group mute control
1049            await self.cmd_group_volume_mute(player_id, muted)
1050            return
1051
1052        # clearing the mute lock may not depend on mute support, otherwise a lock set
1053        # while the player still had a mute control would outlive a control change
1054        if not muted:
1055            player.extra_data.pop(ATTR_MUTE_LOCK, None)
1056
1057        mute_control = player.mute_control
1058        if mute_control == PLAYER_CONTROL_NONE:
1059            raise UnsupportedFeaturedException(
1060                f"Player {player.state.name} does not support muting"
1061            )
1062
1063        # Set mute lock for players in a group
1064        # This prevents auto-unmute when group volume changes
1065        had_mute_lock = ATTR_MUTE_LOCK in player.extra_data
1066        if muted and self._is_in_group(player.state):
1067            player.extra_data[ATTR_MUTE_LOCK] = True
1068
1069        try:
1070            await self._handle_cmd_volume_mute(player, mute_control, muted)
1071        except Exception:
1072            # a mute that did not happen may not leave a lock behind, but a lock
1073            # earned by an earlier successful mute must survive
1074            if not had_mute_lock:
1075                player.extra_data.pop(ATTR_MUTE_LOCK, None)
1076            raise
1077
1078    @handle_player_command
1079    async def play_media(self, player_id: str, media: PlayerMedia) -> None:
1080        """
1081        Handle PLAY MEDIA on given player.
1082
1083        :param player_id: player_id of the player to handle the command.
1084        :param media: The Media that needs to be played on the player.
1085        """
1086        # An explicit play_media on a captured player honors the player's
1087        # CONF_PLAY_MEDIA_OVERRIDES_GROUP preference (default: True) — the
1088        # player is released from its group/sync first, then plays the media
1089        # standalone. With the preference off, behavior falls back to the
1090        # legacy "redirect to group leader" path below.
1091        # Note: the release step runs outside the PLAYBACK lock to avoid an
1092        # AB-BA cycle with cmd_set_members(group), which acquires lock(group)
1093        # then lock(sync_leader) via the sync_group provider.
1094        target_player = self.get_player(player_id, True)
1095        if target_player is not None and (
1096            target_player.state.synced_to or target_player.state.active_group
1097        ):
1098            override = bool(
1099                self.mass.config.get_raw_player_config_value(
1100                    target_player.player_id,
1101                    CONF_PLAY_MEDIA_OVERRIDES_GROUP,
1102                    True,
1103                )
1104            )
1105            if override:
1106                await self._release_player_for_play_media(target_player)
1107                async with self.get_player_lock(
1108                    target_player.player_id, PlayerLockPurpose.PLAYBACK
1109                ):
1110                    await self._handle_play_media(target_player.player_id, media)
1111                return
1112        player = self._get_player_with_redirect(player_id)
1113        async with self.get_player_lock(player.player_id, PlayerLockPurpose.PLAYBACK):
1114            await self._handle_play_media(player.player_id, media)
1115
1116    @api_command("players/cmd/select_sound_mode", required_scope=Scope.PLAYERS_CONTROL)
1117    @handle_player_command
1118    async def select_sound_mode(self, player_id: str, sound_mode: str) -> None:
1119        """
1120        Handle SELECT SOUND MODE command on given player.
1121
1122        - player_id: player_id of the player to handle the command
1123        - sound_mode: The ID of the sound mode that needs to be activated/selected.
1124        """
1125        player = self.get_player(player_id, True)
1126        assert player is not None  # for type checking
1127
1128        if PlayerFeature.SELECT_SOUND_MODE not in player.supported_features:
1129            raise UnsupportedFeaturedException(
1130                f"Player {player.display_name} does not support sound mode selection"
1131            )
1132
1133        prev_sound_mode = player.active_sound_mode
1134        if sound_mode == prev_sound_mode:
1135            return
1136
1137        # basic check if sound mode is valid for player
1138        if not any(x for x in player.sound_mode_list if x.id == sound_mode):
1139            raise PlayerCommandFailed(
1140                f"{sound_mode} is an invalid sound_mode for player {player.display_name}"
1141            )
1142
1143        # forward to player
1144        await player.select_sound_mode(sound_mode)
1145
1146    @api_command("players/cmd/set_option", required_scope=Scope.PLAYERS_CONTROL)
1147    @handle_player_command
1148    async def set_option(
1149        self, player_id: str, option_key: str, option_value: PlayerOptionValueType
1150    ) -> None:
1151        """
1152        Handle SET_OPTION command on given player.
1153
1154        - player_id: player_id of the player to handle the command
1155        - option_key: The key of the player option that needs to be activated/selected.
1156        - option_value: The new value of the player option.
1157        """
1158        player = self.get_player(player_id, True)
1159        assert player is not None  # for type checking
1160
1161        if PlayerFeature.OPTIONS not in player.supported_features:
1162            raise UnsupportedFeaturedException(
1163                f"Player {player.display_name} does not support set_option"
1164            )
1165
1166        prev_player_option = next((x for x in player.options if x.key == option_key), None)
1167        if not prev_player_option:
1168            return
1169        if prev_player_option.value == option_value:
1170            return
1171
1172        if prev_player_option.read_only:
1173            raise UnsupportedFeaturedException(
1174                f"Player {player.display_name} option {option_key} is read-only"
1175            )
1176
1177        # forward to player
1178        await player.set_option(option_key=option_key, option_value=option_value)
1179
1180    @api_command("players/cmd/select_source", required_scope=Scope.PLAYERS_CONTROL)
1181    @handle_player_command
1182    async def select_source(self, player_id: str, source: str | None) -> None:
1183        """
1184        Handle SELECT SOURCE command on given player.
1185
1186        - player_id: player_id of the player to handle the command.
1187        - source: The ID of the source that needs to be activated/selected.
1188        """
1189        if source is None:
1190            source = player_id  # default to MA queue source
1191        player = self.get_player(player_id, True)
1192        assert player is not None  # for type checking
1193        # If player is currently grouped, handle it so the source switch can proceed.
1194        # This allows external sources (e.g. Spotify Connect, AirPlay) to take over a grouped player.
1195        if player.state.active_group and (
1196            group_player := self.get_player(player.state.active_group)
1197        ):
1198            if player_id in group_player.state.static_group_members:
1199                # player is a static member of a permanent group - stop the group
1200                # and power it off if supported, rather than removing the member
1201                await self._handle_cmd_stop(group_player.player_id)
1202                if group_player.state.power_control != PLAYER_CONTROL_NONE:
1203                    await self._handle_cmd_power(group_player.player_id, False)
1204            else:
1205                await self.cmd_ungroup(player_id)
1206        elif player.state.synced_to:
1207            await self.cmd_ungroup(player_id)
1208        # Delegate to internal handler for actual implementation
1209        async with self.get_player_lock(player_id, PlayerLockPurpose.PLAYBACK):
1210            await self._handle_select_source(player_id, source)
1211
1212    async def deselect_source(
1213        self,
1214        player_id: str,
1215        stop_playback: bool = True,
1216        provider_instance_id: str | None = None,
1217        source_id: str | None = None,
1218        playback_session_id: str | None = None,
1219    ) -> None:
1220        """
1221        Give up the source a player was playing, and stop it.
1222
1223        Call this from a plugin when its session ends — the player has nothing to play
1224        any more, so it goes back to reporting its own queue rather than a source that
1225        has gone. Pausing is not this: a paused source keeps the player, so that its
1226        session survives being resumed.
1227
1228        :param player_id: player_id of the player to give the source up on.
1229        :param stop_playback: Whether to stop the player as well. Pass False when the
1230            caller has already stopped it, or is about to.
1231        :param provider_instance_id: Optional provider instance that owns the source session.
1232        :param source_id: Optional provider-scoped source id that owns the source session.
1233        :param playback_session_id: Optional playback session expected to own the player.
1234        """
1235        async with self.get_player_lock(player_id, PlayerLockPurpose.PLAYBACK):
1236            player = self.get_player(player_id, raise_unavailable=False)
1237            if not player:
1238                return
1239            session = self._source_sessions.get(player_id)
1240            active_provider_instance_id = session.provider_instance_id if session else None
1241            active_source_id = session.source_id if session else None
1242            active_playback_session_id = session.playback_session_id if session else None
1243            if provider_instance_id is not None and (
1244                active_provider_instance_id != provider_instance_id
1245                or (source_id is not None and active_source_id != source_id)
1246                or playback_session_id is None
1247                or active_playback_session_id != playback_session_id
1248            ):
1249                self.logger.debug(
1250                    "Ignoring source release for provider %s source %s session %s on player %s: "
1251                    "active source is provider %s source %s session %s",
1252                    provider_instance_id,
1253                    source_id,
1254                    playback_session_id,
1255                    player_id,
1256                    active_provider_instance_id,
1257                    active_source_id,
1258                    active_playback_session_id,
1259                )
1260                return
1261            try:
1262                if stop_playback:
1263                    with suppress(PlayerCommandFailed, PlayerUnavailableError, RuntimeError):
1264                        await self._handle_cmd_stop(player_id)
1265            finally:
1266                if session is not None:
1267                    current_session = self._source_sessions.get(player_id)
1268                    if (
1269                        current_session is session
1270                        and current_session.playback_session_id == active_playback_session_id
1271                    ):
1272                        await self._release_audio_source(player_id)
1273                    else:
1274                        self.logger.debug(
1275                            "Not releasing provider %s source %s session %s on player %s: "
1276                            "the source changed while playback was stopping",
1277                            provider_instance_id,
1278                            source_id,
1279                            playback_session_id,
1280                            player_id,
1281                        )
1282
1283    async def release_provider_sources(self, provider_instance_id: str) -> None:
1284        """
1285        Give up the sources a plugin owns on every player playing one.
1286
1287        Call this when the plugin goes away: a session outliving its provider leaves
1288        the player naming a source that can no longer be streamed nor handed back,
1289        with its own queue held inactive behind it.
1290
1291        :param provider_instance_id: Instance id of the plugin that is going away.
1292        """
1293        sessions = [
1294            (player_id, session.source_id, session.playback_session_id)
1295            for player_id, session in self._source_sessions.items()
1296            if session.provider_instance_id == provider_instance_id
1297        ]
1298        for player_id, source_id, playback_session_id in sessions:
1299            self.logger.debug(
1300                "Provider %s is unloading, releasing its source on player %s",
1301                provider_instance_id,
1302                player_id,
1303            )
1304            await self.deselect_source(
1305                player_id,
1306                provider_instance_id=provider_instance_id,
1307                source_id=source_id,
1308                playback_session_id=playback_session_id,
1309            )
1310
1311    @handle_player_command(lock=PlayerLockPurpose.PLAYBACK)
1312    async def enqueue_next_media(self, player_id: str, media: PlayerMedia) -> None:
1313        """
1314        Handle enqueuing of a next media item on the player.
1315
1316        :param player_id: player_id of the player to handle the command.
1317        :param media: The Media that needs to be enqueued on the player.
1318        :raises UnsupportedFeaturedException: if the player does not support enqueueing.
1319        :raises PlayerUnavailableError: if the player is not available.
1320        """
1321        # Note: No group redirect needed here as enqueue doesn't use _get_player_with_redirect
1322        # Delegate to internal handler for actual implementation
1323        await self._handle_enqueue_next_media(player_id, media)
1324
1325    @api_command("players/cmd/set_members", required_scope=Scope.PLAYERS_CONTROL)
1326    async def cmd_set_members(
1327        self,
1328        target_player: str,
1329        player_ids_to_add: list[str] | None = None,
1330        player_ids_to_remove: list[str] | None = None,
1331    ) -> None:
1332        """
1333        Join/unjoin given player(s) to/from target player.
1334
1335        Will add the given player(s) to the target player (sync leader or group player).
1336
1337        :param target_player: player_id of the syncgroup leader or group player.
1338        :param player_ids_to_add: List of player_id's to add to the target player.
1339        :param player_ids_to_remove: List of player_id's to remove from the target player.
1340
1341        :raises UnsupportedFeaturedException: if the target player does not support grouping.
1342        :raises PlayerUnavailableError: if the target player is not available.
1343        """
1344        parent_player: Player | None = self.get_player(target_player, True)
1345        assert parent_player is not None  # for type checking
1346        if PlayerFeature.SET_MEMBERS not in parent_player.state.supported_features:
1347            msg = f"Player {parent_player.name} does not support group commands"
1348            raise UnsupportedFeaturedException(msg)
1349
1350        # if the target player is a member of an active group player (e.g. a syncgroup),
1351        # redirect the command to that group player so it can manage the member change
1352        if (
1353            parent_player.type != PlayerType.GROUP
1354            and parent_player.state.active_group
1355            and (group_player := self.get_player(parent_player.state.active_group))
1356            and group_player.type == PlayerType.GROUP
1357            and PlayerFeature.SET_MEMBERS in group_player.state.supported_features
1358        ):
1359            self.logger.debug(
1360                "Redirecting set_members from %s to its group player %s",
1361                parent_player.name,
1362                group_player.name,
1363            )
1364            await self.cmd_set_members(
1365                parent_player.state.active_group, player_ids_to_add, player_ids_to_remove
1366            )
1367            return
1368
1369        if parent_player.synced_to:
1370            # handle edge case: target player is already synced itself to another player
1371            # automatically ungroup it first and wait for state to propagate
1372            await self._auto_ungroup_if_synced(parent_player, "setting members")
1373
1374        # Use lock for playback commands to prevent protocol switches from
1375        # racing with concurrent play_media / play_index / resume calls.
1376        async with self.get_player_lock(parent_player.player_id, PlayerLockPurpose.PLAYBACK):
1377            await self._handle_set_members(parent_player, player_ids_to_add, player_ids_to_remove)
1378
1379    @api_command("players/cmd/group", required_scope=Scope.PLAYERS_CONTROL)
1380    @handle_player_command
1381    async def cmd_group(self, player_id: str, target_player: str) -> None:
1382        """
1383        Handle GROUP command for given player.
1384
1385        Join/add the given player(id) to the given (leader) player/sync group.
1386        If the target player itself is already synced to another player, this may fail.
1387        If the player can not be synced with the given target player, this may fail.
1388
1389        NOTE: This is a convenience helper for cmd_set_members.
1390
1391        :param player_id: player_id of the player to handle the command.
1392        :param target_player: player_id of the syncgroup leader or group player.
1393
1394        :raises UnsupportedFeaturedException: if the target player does not support grouping.
1395        :raises PlayerCommandFailed: if the target player is already synced to another player.
1396        :raises PlayerUnavailableError: if the target player is not available.
1397        :raises PlayerCommandFailed: if the player is already grouped to another player.
1398        """
1399        await self.cmd_set_members(target_player, player_ids_to_add=[player_id])
1400
1401    @api_command("players/cmd/group_many", required_scope=Scope.PLAYERS_CONTROL)
1402    async def cmd_group_many(self, target_player: str, child_player_ids: list[str]) -> None:
1403        """
1404        Join given player(s) to target player.
1405
1406        Will add the given player(s) to the target player (sync leader or group player).
1407        This is a (deprecated) alias for cmd_set_members.
1408        """
1409        await self.cmd_set_members(target_player, player_ids_to_add=child_player_ids)
1410
1411    @api_command("players/cmd/ungroup", required_scope=Scope.PLAYERS_CONTROL)
1412    @handle_player_command
1413    async def cmd_ungroup(self, player_id: str) -> None:
1414        """
1415        Handle UNGROUP command for given player.
1416
1417        Remove the given player from any (sync)groups it currently is synced to.
1418        If the player is not currently grouped to any other player,
1419        this will silently be ignored.
1420        """
1421        if not (player := self.get_player(player_id)):
1422            self.logger.warning("Player %s is not available", player_id)
1423            return
1424
1425        # Ungroup on a group player is interpreted as 'release the captured
1426        # session entirely'. This avoids the "Cannot remove static member"
1427        # error path when transfer_queue or HA's unjoin asks us to release a
1428        # group that has static members.
1429        if player.state.type == PlayerType.GROUP:
1430            if player.state.power_control != PLAYER_CONTROL_NONE:
1431                await self._handle_cmd_power(player.player_id, False)
1432            else:
1433                await self._handle_cmd_stop(player.player_id)
1434            return
1435
1436        if player.state.active_group:
1437            group = self.get_player(player.state.active_group)
1438            is_static_member = group is not None and player_id in group.state.static_group_members
1439            if is_static_member:
1440                # Static members can't be released individually — recurse so
1441                # the group-player branch above stops/dissolves the session.
1442                if group is not None:
1443                    await self.cmd_ungroup(group.player_id)
1444                return
1445            # dynamic or non-static member — remove just this player
1446            await self.cmd_set_members(player.state.active_group, player_ids_to_remove=[player_id])
1447            return
1448
1449        if player.state.synced_to:
1450            # player is a sync member
1451            await self.cmd_set_members(player.state.synced_to, player_ids_to_remove=[player_id])
1452            return
1453
1454        if player.state.group_members:
1455            # player is a sync leader (a non-group player with synced followers).
1456            # Remove only the leader itself: _handle_set_members will either transfer
1457            # leadership to a remaining member (keeping playback alive) or, when no
1458            # members remain / nothing is playing, dissolve the group and stop.
1459            await self.cmd_set_members(player.player_id, player_ids_to_remove=[player.player_id])
1460            return
1461        # unjoin from any dynamic sync groups if we're currently in one (edge case)
1462        # this is in particular used for the Home Assistant integration which does
1463        # not have a set_members command and only supports a single unjoin command
1464        for player in self.iter_players(False):
1465            if not player.state.group_members or player.state.synced_to:
1466                continue
1467            if PlayerFeature.SET_MEMBERS not in player.state.supported_features:
1468                continue
1469            if player_id in player.state.static_group_members:
1470                continue
1471            if player_id in player.state.group_members:
1472                await self.cmd_set_members(player.player_id, player_ids_to_remove=[player_id])
1473                return
1474
1475    @api_command("players/cmd/ungroup_many", required_scope=Scope.PLAYERS_CONTROL)
1476    async def cmd_ungroup_many(self, player_ids: list[str]) -> None:
1477        """Handle UNGROUP command for all the given players."""
1478        for player_id in list(player_ids):
1479            await self.cmd_ungroup(player_id)
1480
1481    @api_command("players/create_group_player", required_scope=Scope.CONFIG_PLAYERS_WRITE)
1482    async def create_group_player(
1483        self, provider: str, name: str, members: list[str], dynamic: bool = True
1484    ) -> Player:
1485        """
1486        Create a new (permanent) Group Player.
1487
1488        :param provider: The provider (id) to create the group player for.
1489        :param name: Name of the new group player.
1490        :param members: List of player ids to add to the group.
1491        :param dynamic: Whether the group is dynamic (members can change).
1492        """
1493        if not (provider_instance := self.mass.get_provider(provider)):
1494            raise ProviderUnavailableError(f"Provider {provider} not found")
1495        provider_instance = cast("PlayerProvider", provider_instance)
1496        if ProviderFeature.CREATE_GROUP_PLAYER not in provider_instance.supported_features:
1497            raise UnsupportedFeaturedException(
1498                f"Provider {provider} does not support creating group players"
1499            )
1500        return await provider_instance.create_group_player(name, members, dynamic)
1501
1502    @api_command("players/remove_group_player", required_scope=Scope.CONFIG_PLAYERS_WRITE)
1503    async def remove_group_player(self, player_id: str) -> None:
1504        """Remove a group player."""
1505        if not (player := self.get_player(player_id)):
1506            # we simply permanently delete the player by wiping its config
1507            self.mass.config.remove(f"players/{player_id}")
1508            return
1509        if player.state.type != PlayerType.GROUP:
1510            raise UnsupportedFeaturedException(f"Player {player.state.name} is not a group player")
1511        player.provider.check_feature(ProviderFeature.REMOVE_GROUP_PLAYER)
1512        await player.provider.remove_group_player(player_id)
1513
1514    @api_command("players/add_currently_playing_to_favorites", required_scope=Scope.LIBRARY_WRITE)
1515    async def add_currently_playing_to_favorites(self, player_id: str) -> None:
1516        """
1517        Add the currently playing item/track on given player to the favorites.
1518
1519        This tries to resolve the currently playing media to an actual media item
1520        and add that to the favorites in the library. Will raise an error if the
1521        player is not currently playing anything or if the currently playing media
1522        can not be resolved to a media item.
1523        """
1524        player = self._get_player_with_redirect(player_id)
1525        # handle mass player queue active
1526        if mass_queue := self.get_active_queue(player):
1527            if not (current_item := mass_queue.current_item) or not current_item.media_item:
1528                raise PlayerCommandFailed("No current item to add to favorites")
1529            # if we're playing a radio station, try to resolve the currently playing track
1530            if current_item.media_item.media_type == MediaType.RADIO:
1531                if not (
1532                    (streamdetails := mass_queue.current_item.streamdetails)
1533                    and (stream_title := streamdetails.stream_title)
1534                    and " - " in stream_title
1535                ):
1536                    # no stream title available, so we can't resolve the track
1537                    # this can happen if the radio station does not provide metadata
1538                    # or there's a commercial break
1539                    # Possible future improvement could be to actually detect the song with a
1540                    # shazam-like approach.
1541                    raise PlayerCommandFailed("No current item to add to favorites")
1542                # send the streamtitle into a global search query
1543                search_artist, search_title_title = stream_title.split(" - ", 1)
1544                # strip off any additional comments in the title (such as from Radio Paradise)
1545                search_title_title = search_title_title.split(" | ")[0].strip()
1546                if track := await self.mass.music.get_track_by_name(
1547                    search_title_title, search_artist
1548                ):
1549                    # we found a track, so add it to the favorites
1550                    await self.mass.music.add_item_to_favorites(track)
1551                    return
1552                # we could not resolve the track, so raise an error
1553                raise PlayerCommandFailed("No current item to add to favorites")
1554
1555            # else: any other media item, just add it to the favorites directly
1556            await self.mass.music.add_item_to_favorites(current_item.media_item)
1557            return
1558
1559        # guard for player with no active source
1560        if not player.state.active_source:
1561            raise PlayerCommandFailed("Player has no active source")
1562        # handle other source active using the current_media with uri
1563        if current_media := player.state.current_media:
1564            # prefer the uri of the current media item
1565            if current_media.uri:
1566                with suppress(MusicAssistantError):
1567                    await self.mass.music.add_item_to_favorites(current_media.uri)
1568                    return
1569            # fallback to search based on artist and title (and album if available)
1570            if current_media.artist and current_media.title:
1571                if track := await self.mass.music.get_track_by_name(
1572                    current_media.title,
1573                    current_media.artist,
1574                    current_media.album,
1575                ):
1576                    # we found a track, so add it to the favorites
1577                    await self.mass.music.add_item_to_favorites(track)
1578                    return
1579        # if we reach here, we could not resolve the currently playing item
1580        raise PlayerCommandFailed("No current item to add to favorites")
1581
1582    async def register(self, player: Player) -> None:
1583        """Register a player on the Player Controller."""
1584        if self._teardown_in_progress(player):
1585            return
1586
1587        # Use lock to prevent race conditions during concurrent player registrations
1588        async with self._register_lock:
1589            player_id = player.player_id
1590
1591            if player_id in self._players:
1592                msg = f"Player {player_id} is already registered!"
1593                raise AlreadyRegisteredError(msg)
1594
1595            # ignore disabled players
1596            if not player.state.enabled:
1597                return
1598
1599            if player.type not in (PlayerType.GROUP, PlayerType.STEREO_PAIR):
1600                await self._resolve_mac_addresses(player)
1601
1602            # restore 'fake' power state from cache if available.
1603            # Group players intentionally do NOT restore their fake-power
1604            # state across restarts: at boot there is no sync session yet, so
1605            # a restored 'powered=True' would put the group in an inconsistent
1606            # 'active without captured session' state where children appear
1607            # owned by a group that has no leader. Users who want their
1608            # 'group captured' state preserved across restarts would need
1609            # explicit session restoration which is out of scope here.
1610            if player.type != PlayerType.GROUP:
1611                cached_value = await self.mass.cache.get(
1612                    key=player.player_id,
1613                    provider=self.domain,
1614                    category=CACHE_CATEGORY_PLAYER_POWER,
1615                    default=False,
1616                )
1617                if cached_value is not None:
1618                    player.extra_data[ATTR_FAKE_POWER] = cached_value
1619
1620            # _registration_aborted below only works once the player is in the registry;
1621            # until then the unregister pass of a provider unload cannot see it, so re-check
1622            # the guard from the top of this method, which the awaits above may have staled
1623            if self._teardown_in_progress(player):
1624                return
1625
1626            # finally actually register it
1627
1628            # Despite the fact that the player is not fully ready yet
1629            # (config not loaded, protocol links not evaluated),
1630            # we already add it to the _players dict here because we
1631            # want to make sure the player is available in the controller
1632            # during the rest of the registration process
1633            # (such as when fetching config or evaluating protocol links).
1634            # We use the 'initialized' attribute to indicate that the player
1635            # is still in the process of being registered so we can filter it out where needed.
1636            self._players[player_id] = player
1637            try:
1638                # update state to ensure player.state reflects the final attributes
1639                # (e.g. player type) set after super().__init__() in the player subclass,
1640                # before we fetch config (which relies on state.type for entry resolution)
1641                player.update_state(signal_event=False)
1642                # ensure we fetch and set the latest/full config for the player
1643                player_config = await self.mass.config.get_player_config(player_id)
1644                if self._registration_aborted(player):
1645                    return
1646                player.set_config(player_config)
1647                # update state again now that config is loaded
1648                player.update_state(signal_event=False)
1649                self._save_underlying_player_id(player)
1650                # call hook after the player is registered and config is set
1651                await player.on_config_updated()
1652                if self._registration_aborted(player):
1653                    return
1654
1655                # Handle protocol linking
1656                self._evaluate_protocol_links(player)
1657            except Exception, asyncio.CancelledError:
1658                # a player whose setup failed never becomes initialized, which hides it
1659                # everywhere while it keeps blocking every later registration of the same id.
1660                # Cancellation counts too: a re-triggered provider discovery aborts the task
1661                # this runs in. Only roll back while the player is still ours: an unregister
1662                # may have dropped it already, and it unloads the player itself.
1663                if self._players.get(player_id) is player:
1664                    del self._players[player_id]
1665                    # players claim resources in their constructor (event subscriptions,
1666                    # connections) that only on_unload releases. Best-effort, so a failing
1667                    # teardown cannot mask the error that got us here.
1668                    try:
1669                        await player.on_unload()
1670                    except Exception:
1671                        self.logger.exception("Error unloading player %s", player.name)
1672                raise
1673
1674            # now we're ready to signal the player is added and available
1675            player.set_initialized()
1676            self.logger.info(
1677                "Player (type %s) registered: %s/%s",
1678                player.state.type.value,
1679                player_id,
1680                player.state.name,
1681            )
1682            # signal event that a player was added
1683            if player.state.type != PlayerType.PROTOCOL:
1684                self.mass.signal_event(
1685                    EventType.PLAYER_ADDED, object_id=player.player_id, data=player
1686                )
1687            # register playerqueue for this player (if not a protocol player)
1688            if player.state.type != PlayerType.PROTOCOL:
1689                await self.mass.player_queues.on_player_register(player)
1690                if self._registration_aborted(player):
1691                    # the queue restore outlived the unregister that already cleaned it up,
1692                    # so drop the queue we just recreated for a player that is gone
1693                    self.mass.player_queues.on_player_remove(player_id, permanent=False)
1694
1695        # Schedule debounced update of all players since can_group_with values may change
1696        # when a new player is added (provider IDs expand to include the new player)
1697        self._schedule_update_all_players(2)
1698
1699    async def register_or_update(self, player: Player) -> None:
1700        """Register a new player on the controller or update existing one."""
1701        if self._teardown_in_progress(player):
1702            return
1703
1704        # the register lock ensures a replacement is never swapped in while register()
1705        # is still setting the player up
1706        async with self._register_lock:
1707            if (existing := self._players.get(player.player_id)) is not None:
1708                # a protocol player is hidden behind its parent and owns no queue, every
1709                # other player does. Reading the role the player is leaving off that
1710                # published reality keeps it independent of when the player's state was
1711                # last recalculated, which providers cannot control (they flip the type
1712                # before this call).
1713                was_protocol = self.mass.player_queues.get(player.player_id) is None
1714                becomes_protocol = player.type == PlayerType.PROTOCOL
1715                role_changed = becomes_protocol != was_protocol
1716                if role_changed:
1717                    # release the topology of the role the player is leaving
1718                    self._cleanup_player_type_transition(
1719                        existing, becomes_protocol=becomes_protocol
1720                    )
1721                self._players[player.player_id] = player
1722                if existing is not player:
1723                    # a fresh instance starts out with a base config only, so it needs
1724                    # the config the registration resolved before it can be used
1725                    player.set_config(existing.config)
1726                    await player.on_config_updated()
1727                    if self._registration_aborted(player):
1728                        return
1729                # the replacement takes over the identity of an already registered
1730                # player, so it must be marked initialized as well
1731                player.set_initialized()
1732                player.update_state()
1733                # the derived-transport edge may have been set/revoked after the
1734                # initial registration (e.g. via a bridge claim)
1735                self._save_underlying_player_id(player)
1736                if role_changed:
1737                    await self._finish_player_type_transition(player)
1738                # Also schedule update when replacing existing player
1739                self._schedule_update_all_players()
1740                return
1741
1742        await self.register(player)
1743
1744    def trigger_player_update(
1745        self, player_id: str, force_update: bool = False, debounce_delay: float = 0.25
1746    ) -> None:
1747        """Trigger a (debounced) update for the given player."""
1748        if self.mass.closing:
1749            return
1750        if not (player := self.get_player(player_id)):
1751            return
1752        # mark dirty right away (not at execution): a trigger means state the player
1753        # derives from changed, and a direct update_state call may come in before
1754        # the debounced one runs
1755        player.mark_state_dirty()
1756        task_id = f"player_update_state_{player_id}"
1757        self.mass.call_later(
1758            debounce_delay,
1759            player.update_state,
1760            force_update=force_update,
1761            task_id=task_id,
1762        )
1763
1764    async def unregister(
1765        self,
1766        player_id: str,
1767        permanent: bool = False,
1768        replacement_player_id: str | None = None,
1769    ) -> None:
1770        """
1771        Unregister a player from the player controller.
1772
1773        Called (by a PlayerProvider) when a player is removed or no longer available
1774        (for a longer period of time). This will remove the player from the player
1775        controller and optionally remove the player's config from the mass config.
1776        If the player is not registered, this will silently be ignored.
1777
1778        :param player_id: Player ID of the player to unregister.
1779        :param permanent: If True, remove the player permanently by deleting its config.
1780                          If False, the player config will not be removed.
1781        :param replacement_player_id: Player ID that takes this player's place, only
1782                                      used for a permanent removal.
1783        """
1784        player = self._players.get(player_id)
1785        if player is None:
1786            return
1787        # a player that is going away is done with any live source it was playing,
1788        # so let the owning plugin release an upstream session pointing at us
1789        await self._release_audio_source(player_id)
1790        del self._players[player_id]
1791        # clean up all lock entries for this player
1792        for prefix in [p.value for p in PlayerLockPurpose]:
1793            self._player_command_locks.pop(f"{prefix}_{player_id}", None)
1794        if handle := self._pending_protocol_evaluations.pop(player_id, None):
1795            handle.cancel()
1796        self._clear_sleep_timer(player)
1797        self.mass.player_queues.on_player_remove(player_id, permanent=permanent)
1798        # teardown is best-effort: a provider that fails to release its player must not
1799        # strand the other players of that provider, nor the provider unload itself
1800        try:
1801            await player.on_unload()
1802        except Exception:
1803            self.logger.exception("Error unloading player %s", player.name)
1804        if permanent:
1805            # player permanent removal: cleanup protocol links, delete config
1806            # and signal PLAYER_REMOVED event.
1807            # No group detach is issued here: the player is already out of the registry,
1808            # so it is filtered out of every group's live member list, and its persisted
1809            # membership is settled by delete_player_config below.
1810            self._cleanup_protocol_links(player)
1811            self.delete_player_config(player_id, replacement_player_id)
1812            self.logger.info("Player removed: %s", player.name)
1813            if player.state.type != PlayerType.PROTOCOL:
1814                self.mass.signal_event(EventType.PLAYER_REMOVED, player_id)
1815        else:
1816            # temporary unavailable: mark player as unavailable
1817            # note: the player will be re-registered later if it comes back online
1818            player.state.available = False
1819            self.logger.info("Player unavailable: %s", player.name)
1820            if player.state.type != PlayerType.PROTOCOL:
1821                self.mass.signal_event(
1822                    EventType.PLAYER_UPDATED, object_id=player.player_id, data=player.state
1823                )
1824        # Schedule debounced update of all players since can_group_with values may change
1825        self._schedule_update_all_players()
1826
1827    @api_command("players/remove", required_scope=Scope.CONFIG_PLAYERS_WRITE)
1828    async def remove(self, player_id: str) -> None:
1829        """
1830        Remove a player from a provider.
1831
1832        Can only be called when a PlayerProvider supports ProviderFeature.REMOVE_PLAYER.
1833        """
1834        player = self.get_player(player_id)
1835        if player is None:
1836            # we simply permanently delete the player config since it is not registered
1837            self.delete_player_config(player_id)
1838            return
1839        if player.state.type == PlayerType.GROUP:
1840            # Handle group player removal
1841            player.provider.check_feature(ProviderFeature.REMOVE_GROUP_PLAYER)
1842            await player.provider.remove_group_player(player_id)
1843            return
1844        player.provider.check_feature(ProviderFeature.REMOVE_PLAYER)
1845        await player.provider.remove_player(player_id)
1846        # check for group memberships that need to be updated
1847        if player.state.active_group and (
1848            group_player := self.mass.players.get_player(player.state.active_group)
1849        ):
1850            # try to remove from the group
1851            with suppress(UnsupportedFeaturedException, PlayerCommandFailed):
1852                await group_player.set_members(
1853                    player_ids_to_remove=[player_id],
1854                )
1855        # We removed the player and can now clean up its config
1856        self.delete_player_config(player_id)
1857
1858    def delete_player_config(
1859        self, player_id: str, replacement_player_id: str | None = None
1860    ) -> None:
1861        """
1862        Permanently delete a player's configuration, including its DSP and queue settings.
1863
1864        The saved queue of a player that is no longer registered is dropped along with it,
1865        so a device that returns under the same id starts out fresh. The player itself is
1866        not unregistered.
1867        The config of a linked protocol player is wiped along with it, so the device
1868        returns as a brand new player once it is discovered again. Protocol players that
1869        are still registered or that already moved to another parent keep their config;
1870        registered ones are detached from the removed player and re-evaluated.
1871        Any group that lists the player as a member follows the replacement, or loses
1872        the member when there is none.
1873
1874        :param player_id: Player ID of the player to delete the configuration of.
1875        :param replacement_player_id: Player ID that takes this player's place, so users
1876                                      restricted to it and groups it belongs to follow
1877                                      the replacement.
1878        """
1879        self._detach_protocol_children(player_id)
1880        self._update_group_memberships(player_id, replacement_player_id)
1881        player_ids = [
1882            protocol_id
1883            for protocol_id in self.mass.config.get(CONF_PLAYERS, {})
1884            if self._get_cached_protocol_parent_id(protocol_id) == player_id
1885            and self.get_player(protocol_id) is None
1886        ]
1887        player_ids.append(player_id)
1888        for pid in player_ids:
1889            for key in (
1890                f"{CONF_PLAYERS}/{pid}",
1891                f"{CONF_PLAYER_DSP}/{pid}",
1892                f"{CONF_PLAYER_QUEUES}/{pid}",
1893            ):
1894                self.mass.config.remove(key)
1895            if self.get_player(pid) is None:
1896                self.mass.player_queues.purge_saved_queue(pid)
1897        # a user access filter is an allow-list of player ids, so it must not be left
1898        # pointing at a player whose config was just wiped: a replaced player hands its
1899        # entries over to its replacement, a removed one has them dropped
1900        if replacement_player_id:
1901            self.mass.create_task(
1902                self.mass.webserver.auth.replace_player_in_user_filters(
1903                    player_id, replacement_player_id, removed_player_ids=player_ids
1904                )
1905            )
1906        else:
1907            self.mass.create_task(
1908                self.mass.webserver.auth.remove_from_user_filters(player_ids=player_ids)
1909            )
1910
1911    def scale_volume_to_device(self, player_id: str, logical_volume: int) -> int:
1912        """Scale logical volume (0-100) to device volume (min_volume-max_volume)."""
1913        min_volume, max_volume = self._get_volume_limits(player_id)
1914        if min_volume == 0 and max_volume == 100:
1915            return logical_volume
1916        # Scale: logical 0 -> min_volume, logical 100 -> max_volume
1917        return min_volume + (logical_volume * (max_volume - min_volume)) // 100
1918
1919    def scale_volume_from_device(self, player_id: str, device_volume: int) -> int:
1920        """Scale device volume (min_volume-max_volume) to logical volume (0-100)."""
1921        min_volume, max_volume = self._get_volume_limits(player_id)
1922        if min_volume == 0 and max_volume == 100:
1923            return device_volume
1924        volume_range = max_volume - min_volume
1925        if volume_range == 0:
1926            return 0
1927        # Scale to 0-100 without clamping so that out-of-range device volumes
1928        # produce distinct logical values, ensuring state change detection triggers
1929        # volume limit enforcement
1930        return ((device_volume - min_volume) * 100) // volume_range
1931
1932    def on_player_position_jumped(self, player: Player) -> None:
1933        """
1934        Handle a discrete jump of a player's corrected playback position.
1935
1936        Called by a Player when its corrected position moved significantly
1937        outside regular playback progression (seek or buffer correction). This
1938        is not an event by itself: it re-bases the active queue's timing on the
1939        fresh position and nudges related players so derived positions stay in
1940        sync; current_media then re-anchors from the corrected queue time on
1941        the follow-up update, which emits the actual update event.
1942        """
1943        if self.mass.closing:
1944            return
1945        self.mass.player_queues.on_player_elapsed_time_corrected(player)
1946        self.trigger_player_update(player.player_id)
1947        self._forward_state_update(player, {})
1948
1949    def signal_player_state_update(
1950        self,
1951        player: Player,
1952        changed_values: dict[str, tuple[Any, Any]],
1953        force_update: bool = False,
1954        skip_forward: bool = False,
1955        media_position_jumped: bool = False,
1956    ) -> None:
1957        """
1958        Signal a player state update.
1959
1960        Called by a Player when its state has changed.
1961        This will update the player state in the controller and signal the event bus.
1962        """
1963        player_id = player.player_id
1964        if self.mass.closing:
1965            return
1966
1967        # ignore updates for disabled players
1968        if not player.state.enabled and ATTR_ENABLED not in changed_values:
1969            return
1970
1971        # The current_media position anchor only changes on discrete events
1972        # (play/pause/seek/track change/buffer correction), so a change set holding
1973        # only anchor keys represents a position correction rather than a regular
1974        # state change.
1975        non_anchor_keys = changed_values.keys() - POSITION_ANCHOR_KEYS
1976        if len(non_anchor_keys) == 0 and not force_update:
1977            if not media_position_jumped:
1978                # anchor adoption without a significant corrected-position change
1979                return
1980            # current_media's corrected position jumped (seek or buffer correction
1981            # reached the current media): emit the full player update below so
1982            # consumers see the fresh position
1983
1984        if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
1985            self.logger.log(
1986                VERBOSE_LOG_LEVEL,
1987                "Player state updated for %s: changed fields: %s",
1988                player.name,
1989                ", ".join(changed_values.keys()),
1990            )
1991
1992        # signal update to the playerqueue
1993        if player.state.type != PlayerType.PROTOCOL:
1994            self.mass.call_later(
1995                0.5,
1996                self.mass.player_queues.on_player_update,
1997                player,
1998                changed_values,
1999                task_id=f"queue_on_player_update_{player.player_id}",
2000            )
2001
2002        # Kick async palette extraction on cold cache. On transition prefetch
2003        # the next queue item too. Skip players that mirror another player's media
2004        # (grouped/synced members, protocol children): their current_media - palette
2005        # included - is taken wholesale from the owner, so resolving it per member is
2006        # wasted work that also produces duplicate state updates across the group.
2007        if (
2008            not self._mirrors_parent_media(player)
2009            and (current_media := player.state.current_media)
2010            and current_media.image_url
2011        ):
2012            if current_media.palette is None:
2013                self._schedule_palette_fetch(player_id, current_media.image_url)
2014            if "current_media.image_url" in changed_values or "current_media" in changed_values:
2015                self._schedule_next_queue_item_palette_prefetch(player_id, current_media)
2016
2017        # handle DSP reload of the leader when grouping/ungrouping
2018        if ATTR_GROUP_MEMBERS in changed_values:
2019            prev_group_members, new_group_members = changed_values[ATTR_GROUP_MEMBERS]
2020            self._handle_group_dsp_change(player, prev_group_members or [], new_group_members)
2021            # Removed group members also need to be updated since they are no longer part
2022            # of this group and are available for playback again
2023            removed_members = set(prev_group_members or []) - set(new_group_members or [])
2024            for _removed_player_id in removed_members:
2025                if removed_player := self.get_player(_removed_player_id):
2026                    removed_player.refresh_state()
2027
2028        # detect when active_source changes to
2029        # something external while we have a grouped protocol active
2030        if ATTR_ACTIVE_SOURCE in changed_values:
2031            task_id = f"external_source_takeover_{player_id}"
2032            self.mass.call_later(
2033                5,
2034                self._check_external_source_takeover,
2035                player,
2036                task_id=task_id,
2037            )
2038        # only steer into the (relatively expensive) membership cleanup when a field
2039        # that can require an unsync actually changed - this runs on every state tick
2040        if changed_values.keys() & {ATTR_AVAILABLE, ATTR_ENABLED, ATTR_POWERED}:
2041            self._handle_membership_cleanup_on_state_change(player, changed_values)
2042            self._handle_external_power_off(player, changed_values)
2043
2044        # enforce volume limits when volume changes externally
2045        if "volume_level" in changed_values:
2046            corrected = self._enforce_volume_limits(player)
2047            # a level set on the device itself makes the reference a group volume change
2048            # interpolates from obsolete. a member on its way to a level we did send
2049            # reports levels too, and a group only ever reports what its members are at,
2050            # so neither of those counts. a correction always is the device's own doing:
2051            # the levels we command never fall outside the configured range
2052            if player.state.type != PlayerType.GROUP and (
2053                corrected or self._unexpired_volume_target(player) is None
2054            ):
2055                self._invalidate_group_volume_snapshot(player_id)
2056        # dispatch to internal state update subscribers (with changed_values)
2057        self._dispatch_state_update_subscribers(player, changed_values)
2058
2059        # signal player update on the eventbus
2060        if player.state.type != PlayerType.PROTOCOL:
2061            self.mass.signal_event(EventType.PLAYER_UPDATED, object_id=player_id, data=player)
2062
2063        # signal a separate PlayerOptionsUpdated event
2064        if options := changed_values.get("options"):
2065            self.mass.signal_event(
2066                EventType.PLAYER_OPTIONS_UPDATED, object_id=player_id, data=options
2067            )
2068        # signal player config update event if playerfeatures changed
2069        # this is temporary needed for the Home Assistant integration which only
2070        # re-evalues the entity's supported features on a PLAYER_CONFIG_UPDATED event.
2071        # TODO: Remove this temporary workaround once the HA integration is updated to
2072        # also re-evaluate supported features on PLAYER_UPDATED events.
2073        if changed_values.keys() & {
2074            ATTR_SUPPORTED_FEATURES,
2075            ATTR_MUTE_CONTROL,
2076            ATTR_VOLUME_CONTROL,
2077            ATTR_POWER_CONTROL,
2078        }:
2079            self.mass.signal_event(
2080                EventType.PLAYER_CONFIG_UPDATED, object_id=player_id, data=player.config
2081            )
2082
2083        if not skip_forward or force_update:
2084            self._forward_state_update(player, changed_values)
2085
2086        # trigger update of all players in a provider if group related fields changed
2087        # this ensures that calculated fields like can_group_with are updated on all players
2088        if any(key in changed_values for key in ("group_members", "synced_to", "available")):
2089            for prov_player in player.provider.players:
2090                self.trigger_player_update(prov_player.player_id, debounce_delay=2)
2091
2092    async def register_player_control(self, player_control: PlayerControl) -> None:
2093        """Register a new PlayerControl on the controller."""
2094        if self.mass.closing:
2095            return
2096        control_id = player_control.id
2097
2098        if control_id in self._controls:
2099            msg = f"PlayerControl {control_id} is already registered"
2100            raise AlreadyRegisteredError(msg)
2101
2102        # make sure that the playercontrol's provider is set to the instance_id
2103        prov = self.mass.get_provider(player_control.provider)
2104        if not prov or prov.instance_id != player_control.provider:
2105            raise RuntimeError(f"Invalid provider ID given: {player_control.provider}")
2106
2107        self._controls[control_id] = player_control
2108
2109        self.logger.info(
2110            "PlayerControl registered: %s/%s",
2111            control_id,
2112            player_control.name,
2113        )
2114
2115        # always call update to update any attached players etc.
2116        self.update_player_control(player_control.id, include_configured=True)
2117
2118    async def register_or_update_player_control(self, player_control: PlayerControl) -> None:
2119        """Register a new playercontrol on the controller or update existing one."""
2120        if self.mass.closing:
2121            return
2122        if player_control.id in self._controls:
2123            self._controls[player_control.id] = player_control
2124            self.update_player_control(player_control.id, include_configured=True)
2125            return
2126        await self.register_player_control(player_control)
2127
2128    def update_player_control(self, control_id: str, include_configured: bool = False) -> None:
2129        """
2130        Refresh the players that use the given player control.
2131
2132        :param control_id: The control whose state or availability changed.
2133        :param include_configured: Also refresh the players that select this control in their
2134            config but do not currently resolve to it. Needed when a control (re)appears,
2135            because such a player has already fallen back to another control and would
2136            otherwise never pick this one back up.
2137        """
2138        if self.mass.closing:
2139            return
2140        # update all players that are using this control
2141        for player in list(self._players.values()):
2142            if control_id in (
2143                player.state.power_control,
2144                player.state.volume_control,
2145                player.state.mute_control,
2146            ) or (
2147                include_configured and control_id in self._configured_control_ids(player.player_id)
2148            ):
2149                self.mass.loop.call_soon(player.refresh_state)
2150
2151    def remove_player_control(self, control_id: str) -> None:
2152        """Remove a player_control from the player manager."""
2153        control = self._controls.pop(control_id, None)
2154        if control is None:
2155            return
2156        self.logger.info("PlayerControl removed: %s", control.name)
2157        # players configured to use this control still resolve to it until they are
2158        # refreshed, so let them fall back to their remaining options right away
2159        self.update_player_control(control_id)
2160
2161    def get_player_provider(self, player_id: str) -> PlayerProvider:
2162        """Return PlayerProvider for given player."""
2163        player = self._players[player_id]
2164        assert player  # for type checker
2165        return player.provider
2166
2167    def get_active_queue(self, player: Player) -> PlayerQueue | None:
2168        """Return the current active queue for a player (if any)."""
2169        # account for player that is synced (sync child)
2170        if player.state.synced_to and player.state.synced_to != player.player_id:
2171            if sync_leader := self.get_player(player.state.synced_to):
2172                return self.get_active_queue(sync_leader)
2173        # handle active group player
2174        if player.state.active_group and player.state.active_group != player.player_id:
2175            if group_player := self.get_player(player.state.active_group):
2176                return self.get_active_queue(group_player)
2177        # active_source may be filled queue id (or None)
2178        active_source = player.state.active_source or player.player_id
2179        if active_queue := self.mass.player_queues.get(active_source):
2180            return active_queue
2181        # handle active protocol player with parent player queue
2182        if player.type == PlayerType.PROTOCOL and player.protocol_parent_id:
2183            if parent_player := self.mass.players.get_player(player.protocol_parent_id):
2184                return self.get_active_queue(parent_player)
2185        return None
2186
2187    async def set_group_volume(self, group_player: Player, volume_level: int) -> None:
2188        """
2189        Set the overall volume for a player group or synced players.
2190
2191        Uses interpolation to adjust all child volumes while preserving their
2192        relative balance. A snapshot of child volumes is cached on first call and
2193        used as the reference point for subsequent adjustments.
2194
2195        :param group_player: The group player or sync leader.
2196        :param volume_level: Target volume level (0..100).
2197        """
2198        cur_volume = group_player.state.group_volume
2199        if cur_volume is None:
2200            return
2201
2202        children: list[Player] = []
2203        for child_player in self.iter_group_members(
2204            group_player, only_powered=True, exclude_self=False
2205        ):
2206            if child_player.state.volume_control == PLAYER_CONTROL_NONE:
2207                continue
2208            children.append(child_player)
2209        if not children:
2210            return
2211
2212        # cache a snapshot of child volumes on the group player as reference for interpolation.
2213        # scaling up: each child interpolates from its snapshot value toward 100.
2214        # scaling down: each child interpolates from its snapshot value toward 0.
2215        # this ensures the relative balance is preserved and all children converge
2216        # to 0 and 100 at the extremes. the snapshot is invalidated when a child's
2217        # individual volume or the group membership changes, and rebuilt when the
2218        # children it holds are no longer the ones being adjusted.
2219        # the levels a nudge steps from are the ones the members were last commanded, so
2220        # the snapshot has to read the same source, or a change a member has not confirmed
2221        # yet puts the reference above the level being set and turns a step up into one down
2222        snapshot: dict[str, int] | None = group_player.extra_data.get(ATTR_GROUP_VOLUME_SNAPSHOT)
2223        if snapshot is None or snapshot.keys() != {c.player_id for c in children}:
2224            snapshot = {c.player_id: self._volume_nudge_base(c) or 0 for c in children}
2225            group_player.extra_data[ATTR_GROUP_VOLUME_SNAPSHOT] = snapshot
2226
2227        base_group = max(snapshot.values())
2228
2229        coros = []
2230        for child_player in children:
2231            child_base = snapshot.get(child_player.player_id, 0)
2232            if volume_level >= base_group:
2233                # scaling up: interpolate each child from snapshot toward 100
2234                if base_group >= 100:
2235                    new_child_volume = child_base
2236                else:
2237                    progress = (volume_level - base_group) / (100 - base_group)
2238                    new_child_volume = round(child_base + (100 - child_base) * progress)
2239            elif base_group == 0:
2240                new_child_volume = 0
2241            else:
2242                # scaling down: interpolate each child from snapshot toward 0
2243                progress = volume_level / base_group
2244                new_child_volume = round(child_base * progress)
2245            new_child_volume = max(0, min(100, new_child_volume))
2246            coros.append(self._set_member_volume(child_player.player_id, new_child_volume))
2247        await asyncio.gather(*coros)
2248
2249        # notify active AudioSource once at the group level to prevent
2250        # feedback loops from per-child callbacks with different volume values
2251        await self._notify_source_volume_change(group_player, volume_level)
2252
2253    def iter_group_members(
2254        self,
2255        group_player: Player,
2256        only_powered: bool = False,
2257        only_playing: bool = False,
2258        active_only: bool = False,
2259        exclude_self: bool = True,
2260    ) -> Iterator[Player]:
2261        """Get (child) players attached to a group player or syncgroup."""
2262        for child_id in list(group_player.state.group_members):
2263            if child_player := self.get_player(child_id, False):
2264                if not child_player.state.available or not child_player.state.enabled:
2265                    continue
2266                if only_powered and child_player.state.powered is False:
2267                    continue
2268                if active_only and child_player.state.active_group != group_player.player_id:
2269                    continue
2270                if exclude_self and child_player.player_id == group_player.player_id:
2271                    continue
2272                if only_playing and child_player.state.playback_state not in (
2273                    PlaybackState.PLAYING,
2274                    PlaybackState.PAUSED,
2275                ):
2276                    continue
2277                yield child_player
2278
2279    def subscribe_player_state_update(
2280        self,
2281        callback: Callable[[Player, dict[str, tuple[Any, Any]]], None],
2282    ) -> Callable[[], None]:
2283        """
2284        Subscribe to player state update notifications.
2285
2286        The callback receives the Player and a dict of changed values
2287        (mapping attribute name to a (previous, new) tuple).
2288
2289        :param callback: Function to invoke for each player state update.
2290        :return: An unsubscribe function.
2291        """
2292        self._state_update_subscribers.append(callback)
2293
2294        def _unsub() -> None:
2295            with suppress(ValueError):
2296                self._state_update_subscribers.remove(callback)
2297
2298        return _unsub
2299
2300    @contextlib.asynccontextmanager
2301    async def wait_for_player_update(
2302        self,
2303        player_id: str,
2304        attribute_name: str | None = None,
2305        attribute_value: Any = _SENTINEL,
2306        timeout: float = 5.0,
2307    ) -> AsyncIterator[None]:
2308        """
2309        Async context manager that waits for a player state update.
2310
2311        Subscribes to player state updates on entry, runs the body (typically
2312        the action that triggers the expected update), then waits for a
2313        matching update on exit. If ``attribute_name`` and ``attribute_value``
2314        are both provided and the current value already matches at entry, the
2315        wait is skipped.
2316
2317        Example::
2318
2319            async with mass.players.wait_for_player_update(
2320                player_id, attribute_name="playback_state",
2321                attribute_value=PlaybackState.IDLE, timeout=5,
2322            ):
2323                await mass.players._handle_cmd_stop(player_id)
2324
2325        :param player_id: The player ID to wait for.
2326        :param attribute_name: Optional state attribute to watch for changes
2327            (e.g. ``"playback_state"``). If omitted, any state change satisfies
2328            the wait.
2329        :param attribute_value: Optional value the watched attribute must reach.
2330            Only meaningful in combination with ``attribute_name``.
2331        :param timeout: Maximum time to wait in seconds.
2332        """
2333        update_event = asyncio.Event()
2334
2335        def _on_state_update(player: Player, changed_values: dict[str, tuple[Any, Any]]) -> None:
2336            if player.player_id != player_id:
2337                return
2338            if attribute_name is None:
2339                update_event.set()
2340                return
2341            if attribute_name not in changed_values:
2342                return
2343            if attribute_value is _SENTINEL:
2344                update_event.set()
2345                return
2346            _prev, new_val = changed_values[attribute_name]
2347            if new_val == attribute_value:
2348                update_event.set()
2349
2350        # short-circuit when the desired value is already the current state
2351        already_satisfied = (
2352            attribute_name is not None
2353            and attribute_value is not _SENTINEL
2354            and (player := self.get_player(player_id)) is not None
2355            and getattr(player.state, attribute_name, _SENTINEL) == attribute_value
2356        )
2357
2358        unsub = self.subscribe_player_state_update(_on_state_update)
2359        try:
2360            yield
2361            if already_satisfied:
2362                return
2363            try:
2364                async with asyncio.timeout(timeout):
2365                    await update_event.wait()
2366            except TimeoutError:
2367                self.logger.debug(
2368                    "Timed out waiting for player update on %s (attr=%s value=%s)",
2369                    player_id,
2370                    attribute_name,
2371                    attribute_value,
2372                )
2373        finally:
2374            unsub()
2375
2376    async def on_player_config_change(self, config: PlayerConfig, changed_keys: set[str]) -> None:
2377        """Call (by config manager) when the configuration of a player changes."""
2378        min_vol_changed = f"values/{CONF_MIN_VOLUME}" in changed_keys
2379        max_vol_changed = f"values/{CONF_MAX_VOLUME}" in changed_keys
2380        if min_vol_changed or max_vol_changed:
2381            raw_min = config.get_value(CONF_MIN_VOLUME)
2382            raw_max = config.get_value(CONF_MAX_VOLUME)
2383            min_vol = int(cast("int", raw_min)) if raw_min is not None else 0
2384            max_vol = int(cast("int", raw_max)) if raw_max is not None else 100
2385            if min_vol > max_vol:
2386                msg = "Minimum volume cannot exceed maximum volume"
2387                raise InvalidDataError(msg)
2388        player = self.get_player(config.player_id)
2389        player_provider = self.mass.get_provider(config.provider)
2390        player_disabled = ATTR_ENABLED in changed_keys and not config.enabled
2391        player_enabled = ATTR_ENABLED in changed_keys and config.enabled
2392
2393        if player_disabled and player and player.state.available:
2394            # edge case: ensure that the player is powered off if the player gets disabled
2395            if player.state.power_control != PLAYER_CONTROL_NONE:
2396                await self._handle_cmd_power(config.player_id, False)
2397            elif player.state.playback_state != PlaybackState.IDLE:
2398                await self.cmd_stop(config.player_id)
2399
2400        # signal player provider that the player got enabled/disabled
2401        if (player_enabled or player_disabled) and player_provider:
2402            assert isinstance(player_provider, PlayerProvider)  # for type checking
2403            # Collect linked protocol IDs to cascade the enable/disable to.
2404            # Without this, a disabled native parent leaves its linked protocols
2405            # registered after restart; they then fail to find their parent and
2406            # get wrapped in a fresh Universal Player.
2407            cascade_protocol_ids: list[str] = []
2408            parent_is_protocol = player.state.type == PlayerType.PROTOCOL if player else False
2409            if not parent_is_protocol:
2410                if player and player.linked_output_protocols:
2411                    cascade_protocol_ids = [
2412                        link.output_protocol_id for link in player.linked_output_protocols
2413                    ]
2414                else:
2415                    cascade_protocol_ids = self._get_cached_protocol_ids(config.player_id)
2416            if player_disabled:
2417                player_provider.on_player_disabled(config.player_id)
2418            elif player_enabled:
2419                player_provider.on_player_enabled(config.player_id)
2420            for protocol_id in cascade_protocol_ids:
2421                protocol_raw = self.mass.config.get(f"{CONF_PLAYERS}/{protocol_id}")
2422                if not protocol_raw:
2423                    continue
2424                if bool(protocol_raw.get("enabled", True)) == bool(player_enabled):
2425                    continue
2426                self.mass.create_task(
2427                    self.mass.config.save_player_config(
2428                        protocol_id, {ATTR_ENABLED: bool(player_enabled)}
2429                    )
2430                )
2431            return  # enabling/disabling a player will be handled by the provider
2432
2433        if not player:
2434            return  # guard against player not being registered (yet)
2435
2436        resume_queue: PlayerQueue | None = (
2437            self.mass.player_queues.get(player.state.active_source)
2438            if player.state.active_source
2439            else None
2440        )
2441
2442        # ensure player state gets updated with any updated config
2443        player.set_config(config)
2444        await player.on_config_updated()
2445        player.update_state()
2446        # if the PlayerQueue was playing, restart playback
2447        if resume_queue and resume_queue.state == PlaybackState.PLAYING:
2448            requires_restart = any(
2449                v.requires_reload
2450                for v in config.values.values()
2451                if f"values/{v.key}" in changed_keys
2452            )
2453            if requires_restart:
2454                # always stop first to ensure the player uses the new config
2455                await self.mass.player_queues.stop(resume_queue.queue_id)
2456                self.mass.call_later(
2457                    1, self.mass.player_queues.resume, resume_queue.queue_id, False
2458                )
2459
2460    async def on_player_dsp_change(self, player_id: str) -> None:
2461        """Call (by config manager) when the DSP settings of a player change."""
2462        # signal player provider that the config changed
2463        if not (player := self.get_player(player_id)):
2464            return
2465        if player.state.playback_state == PlaybackState.PLAYING:
2466            self.logger.info("Restarting playback of Player %s after DSP change", player_id)
2467            # this will restart the queue stream/playback
2468            if self.get_active_queue(player):
2469                self.mass.call_later(
2470                    0, self.mass.player_queues.resume, player.state.active_source, False
2471                )
2472                return
2473            # if the player is not using a queue, we need to stop and start playback
2474            await self.cmd_stop(player_id)
2475            await self.cmd_play(player_id)
2476
2477    def schedule_active_output_protocol_clear(self, player: Player) -> None:
2478        """
2479        Clear the player's active output protocol once it stops playing.
2480
2481        A device may keep reporting PLAYING for a short while after a stop
2482        command, so the clear is deferred until the player reports IDLE (with a
2483        timeout as fallback). Starting a new session cancels the pending clear
2484        (see Player.set_active_output_protocol).
2485
2486        :param player: The player whose active output protocol must be cleared.
2487        """
2488        # Deduplicated per player via task_id: if a clear is already pending we
2489        # keep it, so the single tracked task stays cancellable by a new session.
2490        self.mass.create_task(
2491            self._clear_active_output_protocol_when_idle(player),
2492            task_id=f"clear_active_protocol_{player.player_id}",
2493        )
2494
2495    def __iter__(self) -> Iterator[Player]:
2496        """Iterate over all players."""
2497        return iter(self._players.values())
2498
2499    async def _resolve_mac_addresses(self, player: Player) -> None:
2500        """
2501        Resolve and persist the MAC addresses used to match the player against protocols.
2502
2503        :param player: The player to resolve the MAC address(es) for.
2504        """
2505        conf_base = f"{CONF_PLAYERS}/{player.player_id}/values"
2506        # Save the original MAC reported by the provider (before ARP enrichment)
2507        reported_mac = player.device_info.identifiers.get(IdentifierType.MAC_ADDRESS)
2508
2509        # Try to use cached ARP MAC from config for fast matching on restart.
2510        # This allows protocol linking to work immediately even if ARP is slow/fails.
2511        cached_arp_mac: str | None = self.mass.config.get(
2512            f"{conf_base}/{CONF_CACHED_ARP_MAC}", None
2513        )
2514        if cached_arp_mac and is_valid_mac_address(cached_arp_mac):
2515            player.device_info.add_identifier(IdentifierType.MAC_ADDRESS, cached_arp_mac)
2516
2517        # Enrich device MAC address via ARP if needed
2518        # (handles invalid MACs, locally-administered MACs, and missing MACs)
2519        await enrich_device_mac_address(player.device_info, self.logger)
2520
2521        # Cache the resolved MAC for fast matching on subsequent restarts
2522        current_mac = player.device_info.identifiers.get(IdentifierType.MAC_ADDRESS)
2523        if current_mac and is_valid_mac_address(current_mac) and current_mac != cached_arp_mac:
2524            self.mass.config.set(f"{conf_base}/{CONF_CACHED_ARP_MAC}", current_mac)
2525
2526        # Store original reported MAC if it differs from the resolved MAC.
2527        # This enables multi-MAC matching for devices with multiple interfaces
2528        # (e.g., WiFi + Ethernet) where ARP resolves one interface but the
2529        # protocol reports the other.
2530        if reported_mac and is_valid_mac_address(reported_mac) and current_mac:
2531            if reported_mac.upper() != current_mac.upper():
2532                player.extra_data["reported_mac"] = reported_mac
2533                self.mass.config.set(f"{conf_base}/{CONF_REPORTED_MAC}", reported_mac)
2534            else:
2535                # Provider's reported MAC matches the resolved MAC; clear any stale
2536                # stored reported MAC to avoid false-positive multi-MAC matches.
2537                self.mass.config.set(f"{conf_base}/{CONF_REPORTED_MAC}", None)
2538        elif not reported_mac or not is_valid_mac_address(reported_mac):
2539            # Restore reported MAC from config on restart only when the provider
2540            # did not supply a usable MAC address.
2541            cached_reported_mac: str | None = self.mass.config.get(
2542                f"{conf_base}/{CONF_REPORTED_MAC}", None
2543            )
2544            if cached_reported_mac and is_valid_mac_address(cached_reported_mac):
2545                if current_mac and cached_reported_mac.upper() == current_mac.upper():
2546                    # Cached value matches the resolved MAC; clear stale entry.
2547                    self.mass.config.set(f"{conf_base}/{CONF_REPORTED_MAC}", None)
2548                else:
2549                    player.extra_data["reported_mac"] = cached_reported_mac
2550
2551    def _teardown_in_progress(self, player: Player) -> bool:
2552        """
2553        Return True if the server or this player's provider is shutting down.
2554
2555        :param player: The player that is in the process of being registered.
2556        """
2557        return self.mass.closing or player.provider.unloading
2558
2559    def _registration_aborted(self, player: Player) -> bool:
2560        """
2561        Return True if the given player is no longer the registered player for its ID.
2562
2563        :param player: The player that is in the process of being registered.
2564        """
2565        # registration awaits provider I/O while the player is already in the registry,
2566        # so an unregister (e.g. a provider unload or a device disconnect) can drop or
2567        # replace it in the meantime, after which registration must stop
2568        if self._players.get(player.player_id) is player:
2569            return False
2570        self.logger.debug(
2571            "Registration of player %s aborted: it was unregistered while setting up",
2572            player.player_id,
2573        )
2574        return True
2575
2576    async def _finish_player_type_transition(self, player: Player) -> None:
2577        """
2578        Publish a registered player that moved in or out of the protocol role.
2579
2580        :param player: The player, with its new type already applied to its state.
2581        """
2582        self._evaluate_protocol_links(player)
2583        if player.state.type == PlayerType.PROTOCOL:
2584            # the player is hidden behind its parent from now on and no longer owns a queue.
2585            # only the queue is dropped, never the playback: the player either just became a
2586            # (hidden) bridge client with nothing playing on it, or is already serving its
2587            # parent, where a stop would cut that parent's stream short. A protocol player
2588            # has no active group of its own either, so there is nothing to detach here.
2589            self.mass.signal_event(EventType.PLAYER_REMOVED, player.player_id)
2590            self.mass.player_queues.on_player_remove(player.player_id, permanent=False)
2591            return
2592        # the player surfaces on its own, which leaves it unusable without a queue
2593        self.mass.signal_event(EventType.PLAYER_ADDED, object_id=player.player_id, data=player)
2594        await self.mass.player_queues.on_player_register(player)
2595        if self._registration_aborted(player):
2596            # the queue restore outlived the unregister that already cleaned it up,
2597            # so drop the queue we just recreated for a player that is gone
2598            self.mass.player_queues.on_player_remove(player.player_id, permanent=False)
2599
2600    async def _release_player_for_play_media(self, player: Player) -> None:
2601        """
2602        Release a captured player so a play_media command can target it directly.
2603
2604        :param player: The captured player to release.
2605        """
2606        # Strategy is picked from how the player is currently captured:
2607        #   synced_to            → unsync this player (cmd_ungroup)
2608        #   dynamic group member → remove from group via cmd_set_members
2609        #   static group member  → dissolve the whole group (power off if it
2610        #                          has a real power control, otherwise stop)
2611        # In every branch we wait for the relevant state attribute to actually
2612        # clear before returning. Providers (Sonos in particular) reject a
2613        # play_media on a player whose synced_to/active_group is still set
2614        # locally even though the release command has been acknowledged.
2615        if player.state.synced_to:
2616            self.logger.debug(
2617                "Unsyncing %s from %s to honor explicit play_media target",
2618                player.state.name,
2619                player.state.synced_to,
2620            )
2621            async with self.wait_for_player_update(
2622                player.player_id,
2623                attribute_name="synced_to",
2624                attribute_value=None,
2625                timeout=5,
2626            ):
2627                await self.cmd_ungroup(player.player_id)
2628            return
2629        if not player.state.active_group:
2630            return
2631        group = self.get_player(player.state.active_group)
2632        if group is None:
2633            return
2634        is_dynamic_member = (
2635            PlayerFeature.SET_MEMBERS in group.state.supported_features
2636            and player.player_id not in group.state.static_group_members
2637        )
2638        if is_dynamic_member:
2639            self.logger.debug(
2640                "Removing %s from dynamic group %s to honor explicit play_media target",
2641                player.state.name,
2642                group.state.name,
2643            )
2644            async with self.wait_for_player_update(
2645                player.player_id,
2646                attribute_name="active_group",
2647                attribute_value=None,
2648                timeout=5,
2649            ):
2650                await self.cmd_set_members(group.player_id, player_ids_to_remove=[player.player_id])
2651            return
2652        # static member: a single member can't be released, so the whole
2653        # group must dissolve. Prefer cmd_power when an explicit power
2654        # control is set so the user-visible state stays consistent.
2655        async with self.wait_for_player_update(
2656            player.player_id,
2657            attribute_name="active_group",
2658            attribute_value=None,
2659            timeout=5,
2660        ):
2661            if group.state.power_control != PLAYER_CONTROL_NONE and group.state.powered:
2662                self.logger.debug(
2663                    "Powering off %s to honor explicit play_media target on %s",
2664                    group.state.name,
2665                    player.state.name,
2666                )
2667                await self._handle_cmd_power(group.player_id, False)
2668            else:
2669                self.logger.debug(
2670                    "Stopping %s to honor explicit play_media target on %s",
2671                    group.state.name,
2672                    player.state.name,
2673                )
2674                await self._handle_cmd_stop(group.player_id)
2675
2676    def _mirrors_parent_media(self, player: Player) -> bool:
2677        """
2678        Return True if the player's current_media is taken from another player.
2679
2680        Grouped/synced members and protocol children mirror their parent's
2681        current_media (palette included), so they must not resolve it themselves.
2682
2683        :param player: The player to check.
2684        """
2685        state = player.state
2686        # a self-referential active_group/synced_to is not a real parent (mirror the
2687        # != self guard in Player.__final_current_media), so it must not skip resolution
2688        parent_id = state.active_group or state.synced_to
2689        if parent_id and parent_id != player.player_id:
2690            return True
2691        return state.type == PlayerType.PROTOCOL and player.protocol_parent_id is not None
2692
2693    def _schedule_palette_fetch(
2694        self, player_id: str, image_url: str | None, *, trigger_update: bool = True
2695    ) -> None:
2696        """
2697        Kick off an async palette extraction for an image URL.
2698
2699        :param player_id: Player the palette is scoped to (used for task dedup).
2700        :param image_url: Image URL to extract from. No-op when empty or already cached.
2701        :param trigger_update: When True, re-emit player state once palette is ready
2702                               (current track). When False, only warm the cache (prefetch).
2703        """
2704        if not image_url:
2705            return
2706        # Key the task on the image (not just the player) so a track change always
2707        # schedules a fetch for the new image instead of being dropped by an in-flight
2708        # fetch for the previous one; repeated schedules for the same image still dedupe.
2709        slot = "current" if trigger_update else "next"
2710        self.mass.create_task(
2711            self._fetch_palette(player_id, image_url, trigger_update=trigger_update),
2712            task_id=f"palette_fetch_{player_id}_{slot}_{image_url}",
2713            abort_existing=False,
2714        )
2715
2716    async def _fetch_palette(self, player_id: str, image_url: str, *, trigger_update: bool) -> None:
2717        palette = await get_palette_for_url(self.mass, image_url)
2718        if palette is None or not trigger_update:
2719            return  # prefetch only warms the cache controller; nothing to attach
2720        player = self.get_player(player_id)
2721        if player is None:
2722            return
2723        current = player.state.current_media
2724        if current is None or current.image_url != image_url:
2725            return  # media changed while fetching
2726        # Carry the palette on player state so the (sync) serialization reads it back.
2727        player.set_resolved_palette(image_url, palette)
2728        # Avoid trigger_player_update so a concurrent state-change debounce
2729        # doesn't cancel our timer via the shared player_update_state task_id.
2730        self.mass.call_later(
2731            0,
2732            player.update_state,
2733            force_update=True,
2734            task_id=f"palette_player_update_{player_id}",
2735        )
2736
2737    def _schedule_next_queue_item_palette_prefetch(
2738        self, player_id: str, current_media: PlayerMedia
2739    ) -> None:
2740        """Warm the palette cache for the next queue item so it's hot at transition."""
2741        queue_id, item_id = current_media.source_id, current_media.queue_item_id
2742        if not queue_id or not item_id:
2743            return
2744        next_item = self.mass.player_queues.get_next_item(queue_id, item_id)
2745        if next_item is None or not next_item.image:
2746            return
2747        next_url = self.mass.metadata.get_image_url(
2748            next_item.image, size=512, prefer_stream_server=True
2749        )
2750        self._schedule_palette_fetch(player_id, next_url, trigger_update=False)
2751
2752    def _configured_control_ids(self, player_id: str) -> set[str]:
2753        """Return the player control ids the given player's config selects."""
2754        return {
2755            str(value)
2756            for conf_key in (CONF_POWER_CONTROL, CONF_VOLUME_CONTROL, CONF_MUTE_CONTROL)
2757            if (value := self.mass.config.get_raw_player_config_value(player_id, conf_key))
2758        }
2759
2760    def _get_volume_step(self, current_volume: int) -> int:
2761        """
2762        Return the step size for a single volume increment at the given level.
2763
2764        A configured (non-zero) `volume_step` is a flat step. The default of 0 keeps the
2765        adaptive ladder, which takes finer steps near the ends of the range.
2766        """
2767        if configured := self.get_config_value(CONF_VOLUME_STEP, 0, return_type=int):
2768            return configured
2769        if current_volume < 10 or current_volume > 90:
2770            return 1
2771        if current_volume < 30 or current_volume > 70:
2772            return 2
2773        return 3
2774
2775    def _get_volume_limits(self, player_id: str) -> tuple[int, int]:
2776        """Get the configured min/max volume limits for a player."""
2777        min_volume = int(
2778            cast(
2779                "int",
2780                self.mass.config.get_raw_player_config_value(
2781                    player_id, CONF_MIN_VOLUME, CONF_ENTRY_MIN_VOLUME.default_value
2782                ),
2783            )
2784        )
2785        max_volume = int(
2786            cast(
2787                "int",
2788                self.mass.config.get_raw_player_config_value(
2789                    player_id, CONF_MAX_VOLUME, CONF_ENTRY_MAX_VOLUME.default_value
2790                ),
2791            )
2792        )
2793        return min_volume, max_volume
2794
2795    def _enforce_volume_limits(self, player: Player) -> bool:
2796        """
2797        Clamp device volume to min/max range when changed externally.
2798
2799        :param player: The player to check the volume of.
2800        :return: True if the volume was outside the configured range and got corrected.
2801        """
2802        player_id = player.player_id
2803        min_volume, max_volume = self._get_volume_limits(player_id)
2804        if min_volume == 0 and max_volume == 100:
2805            return False
2806        # state.volume_level is the resolved logical volume, available for all
2807        # volume control types; a device volume outside the configured range
2808        # surfaces here as a value outside 0-100 (scaling does not clamp)
2809        logical_volume = player.state.volume_level
2810        if logical_volume is None or 0 <= logical_volume <= 100:
2811            return False
2812        clamped = max(0, min(100, logical_volume))
2813        # correct via the regular volume-set path so scaling and redirection apply
2814        self.mass.create_task(self._handle_cmd_volume_set(player_id, clamped))
2815        return True
2816
2817    def _forward_state_update(
2818        self, player: Player, changed_values: dict[str, tuple[Any, Any]]
2819    ) -> None:
2820        """Forward a player state update to related players (groups, sync parent, protocols)."""
2821        # TODO: make this fan-out change-aware (skip relatives that derive nothing from
2822        # the changed fields) once reverse indexes for synced_to/active_group exist.
2823        # Propagate group or sync-leader updates to child players.
2824        if player.state.group_members:
2825            for child_player in self.iter_group_members(player, exclude_self=True):
2826                if player.type == PlayerType.GROUP:
2827                    child_player.on_group_updated(player, changed_values)
2828                else:
2829                    child_player.on_sync_parent_updated(player, changed_values)
2830        # update/signal group player(s) when a member updates. A sync leader is a member of the
2831        # group player that formed the sync group and gaining members of its own does not change
2832        # that: a group player mirrors its leader, so it depends on exactly these updates.
2833        for group_player in self._get_player_groups(player):
2834            group_player.on_group_member_updated(player, changed_values)
2835
2836        # update/signal manually sync-parent player when child updates
2837        if (_sync_parent_id := player.state.synced_to) and (
2838            _sync_parent := self.get_player(_sync_parent_id)
2839        ):
2840            self.trigger_player_update(_sync_parent.player_id)
2841        # If this is a protocol player, forward the state update to the parent player
2842        if (
2843            player.type == PlayerType.PROTOCOL
2844            and player.protocol_parent_id
2845            and (_protocol_parent := self.mass.players.get_player(player.protocol_parent_id))
2846        ):
2847            _protocol_parent.on_protocol_player_updated(player, changed_values)
2848        # If this is a parent player with linked protocols, forward state updates
2849        # to linked protocol players so their state reflects parent dependencies
2850        if player.state.type != PlayerType.PROTOCOL and player.linked_output_protocols:
2851            for linked in player.linked_output_protocols:
2852                if protocol_player := self.mass.players.get_player(linked.output_protocol_id):
2853                    protocol_player.on_protocol_parent_updated(player, changed_values)
2854
2855    def _invalidate_group_volume_snapshot(self, player_id: str) -> None:
2856        """Clear the cached group volume snapshot for all groups this player belongs to."""
2857        player = self.get_player(player_id)
2858        if not player:
2859            return
2860        if player.state.group_members:
2861            player.extra_data.pop(ATTR_GROUP_VOLUME_SNAPSHOT, None)
2862        for group_player in self._get_player_groups(player):
2863            group_player.extra_data.pop(ATTR_GROUP_VOLUME_SNAPSHOT, None)
2864        if player.state.synced_to and (leader := self.get_player(player.state.synced_to)):
2865            leader.extra_data.pop(ATTR_GROUP_VOLUME_SNAPSHOT, None)
2866
2867    def _record_volume_target(self, player: Player, volume_level: int) -> None:
2868        """Remember the volume level just commanded, as the base for the next nudge."""
2869        if self._stays_silent_on_volume_change(player):
2870            volume_level = 0
2871        player.extra_data[ATTR_VOLUME_TARGET] = (volume_level, time.monotonic())
2872
2873    def _volume_nudge_base(self, player: Player) -> int | None:
2874        """Return the volume level a volume nudge for the given player steps from."""
2875        target = self._unexpired_volume_target(player)
2876        if target is not None:
2877            return target
2878        return player.state.volume_level
2879
2880    def _group_volume_nudge_base(self, group_player: Player) -> int | None:
2881        """Return the volume level a group volume nudge for the given group steps from."""
2882        if not group_player.state.group_members:
2883            # an ungrouped player is stepped through its own volume, so it is that
2884            # volume the command lands on and that a following nudge steps from
2885            return self._volume_nudge_base(group_player)
2886        # mirrors Player.group_volume, but steps from the level last commanded to each
2887        # member instead of the level it reports, so the group is not held back by a
2888        # member that has not confirmed the previous nudge yet
2889        base: int | None = None
2890        for child_player in self.iter_group_members(
2891            group_player, only_powered=True, exclude_self=group_player.type != PlayerType.PLAYER
2892        ):
2893            if child_player.state.volume_control == PLAYER_CONTROL_NONE:
2894                continue
2895            if (child_volume := self._volume_nudge_base(child_player)) is None:
2896                continue
2897            if base is None or child_volume > base:
2898                base = child_volume
2899        return base
2900
2901    def _unexpired_volume_target(self, player: Player) -> int | None:
2902        """Return the volume level last commanded, or None once it is too old to trust."""
2903        if (target := player.extra_data.get(ATTR_VOLUME_TARGET)) is None:
2904            return None
2905        volume_level, issued_at = target
2906        if time.monotonic() - issued_at < VOLUME_TARGET_EXPIRY:
2907            return cast("int", volume_level)
2908        del player.extra_data[ATTR_VOLUME_TARGET]
2909        return None
2910
2911    def _dispatch_state_update_subscribers(
2912        self, player: Player, changed_values: dict[str, tuple[Any, Any]]
2913    ) -> None:
2914        """Notify all internal subscribers of a player state update."""
2915        for subscriber in list(self._state_update_subscribers):
2916            try:
2917                subscriber(player, changed_values)
2918            except Exception:
2919                self.logger.exception(
2920                    "Error in player state update subscriber for %s", player.player_id
2921                )
2922
2923    async def _wait_for_playback_state(
2924        self,
2925        player: Player,
2926        wanted_state: PlaybackState,
2927        timeout: float,
2928        minimal_time: float = 0,
2929    ) -> None:
2930        """Wait for a player to reach a playback state, with optional minimum wait time."""
2931        start_timestamp = time.time()
2932        async with self.wait_for_player_update(
2933            player.player_id,
2934            attribute_name="playback_state",
2935            attribute_value=wanted_state,
2936            timeout=timeout,
2937        ):
2938            pass
2939        elapsed = time.time() - start_timestamp
2940        if elapsed < minimal_time:
2941            await asyncio.sleep(minimal_time - elapsed)
2942
2943    async def _clear_active_output_protocol_when_idle(self, player: Player) -> None:
2944        """Wait for the player to stop playing, then clear its active output protocol."""
2945        await self._wait_for_playback_state(player, PlaybackState.IDLE, timeout=10)
2946        player.set_active_output_protocol(None)
2947
2948    def _handle_membership_cleanup_on_state_change(
2949        self, player: Player, changed_values: dict[str, tuple[Any, Any]]
2950    ) -> None:
2951        """Detach a player from its (sync)groups when a state change requires it."""
2952        # A player that became unavailable or disabled can no longer be commanded,
2953        # so we drop it from its parent group/leader directly.
2954        became_inactive = (
2955            ATTR_AVAILABLE in changed_values and changed_values[ATTR_AVAILABLE][1] is False
2956        ) or (ATTR_ENABLED in changed_values and changed_values[ATTR_ENABLED][1] is False)
2957        if became_inactive and (player.state.active_group or player.state.synced_to):
2958            self.mass.create_task(self._cleanup_player_memberships(player.player_id))
2959
2960        # A player whose power was turned off outside of an MA power command (e.g. its
2961        # linked power control was switched off directly) must be unsynced too. We act
2962        # only on an explicit on->off transition, leaving players without power control
2963        # (powered == None) untouched. The player is still reachable here, so we route
2964        # through cmd_ungroup which also transfers leadership when it is a sync leader.
2965        if (
2966            changed_values.get(ATTR_POWERED) == (True, False)
2967            and player.state.type == PlayerType.PLAYER
2968            and (player.state.synced_to or player.state.active_group or player.state.group_members)
2969        ):
2970            self.mass.create_task(self.cmd_ungroup(player.player_id))
2971
2972    def _handle_external_power_off(
2973        self, player: Player, changed_values: dict[str, tuple[Any, Any]]
2974    ) -> None:
2975        """End the queue of a player whose power was turned off outside of MA."""
2976        if (
2977            changed_values.get(ATTR_POWERED) != (True, False)
2978            or player.state.type not in PLAYBACK_TARGET_TYPES
2979        ):
2980            return
2981        if player.state.synced_to or player.state.active_group or player.state.group_members:
2982            # a grouped player is detached from its group instead, which ends the
2983            # group's queue through the group's own power off
2984            return
2985        # a device that powers itself off may report its stop in this very update, so
2986        # judge on the playback state as it was before it - which is also the snapshot
2987        # an MA power off works from
2988        prev_playback_state = (
2989            changed_values["playback_state"][0]
2990            if "playback_state" in changed_values
2991            else player.state.playback_state
2992        )
2993        if prev_playback_state not in (PlaybackState.PLAYING, PlaybackState.PAUSED):
2994            return
2995        self.mass.call_later(
2996            EXTERNAL_POWER_OFF_STOP_DELAY,
2997            self._stop_queue_on_external_power_off,
2998            player.player_id,
2999            task_id=f"external_power_off_stop_{player.player_id}",
3000        )
3001
3002    async def _stop_queue_on_external_power_off(self, player_id: str) -> None:
3003        """
3004        End the queue of a player that is still powered off.
3005
3006        :param player_id: The player whose power was turned off outside of MA.
3007        """
3008        # hold the playback lock across the checks: a power on that overlaps the wait
3009        # holds it while it resumes the queue, and reading the power state before that
3010        # completes would stop the playback it just started (the lock is re-entrant per
3011        # task, so the stop below re-acquiring it is a no-op)
3012        async with self.get_player_lock(player_id, PlayerLockPurpose.PLAYBACK):
3013            if not (player := self.get_player(player_id)) or player.state.powered is not False:
3014                # the power came back on while we were waiting it out
3015                return
3016            # only the player's own queue: get_active_queue resolves to another player's
3017            # queue as soon as this one is hearing someone else's audio
3018            active_queue = self.get_active_queue(player)
3019            if active_queue is None or active_queue.queue_id != player_id:
3020                return
3021            # a player gone unavailable along with its power cannot be told to stop,
3022            # but the queue teardown that matters here runs regardless
3023            with suppress(PlayerUnavailableError):
3024                await self.mass.player_queues._handle_stop(player_id)
3025
3026    async def _cleanup_player_memberships(self, player_id: str) -> None:
3027        """Ensure a player is detached from any groups or syncgroups."""
3028        if not (player := self.get_player(player_id)):
3029            return
3030        with suppress(UnsupportedFeaturedException, PlayerCommandFailed, PlayerUnavailableError):
3031            if parent_id := (player.state.active_group or player.state.synced_to):
3032                # the player is part of a (permanent) groupplayer and the user tries to ungroup
3033                if parent_player := self.get_player(parent_id):
3034                    await self._handle_set_members(parent_player, player_ids_to_remove=[player_id])
3035                return
3036
3037    def _get_player_with_redirect(self, player_id: str) -> Player:
3038        """Get player with check if playback related command should be redirected."""
3039        player = self.get_player(player_id, True)
3040        assert player is not None  # for type checking
3041        if player.state.synced_to and (sync_leader := self.get_player(player.state.synced_to)):
3042            self.logger.info(
3043                "Player %s is synced to %s and can not accept "
3044                "playback related commands itself, "
3045                "redirected the command to the sync leader.",
3046                player.name,
3047                sync_leader.name,
3048            )
3049            return sync_leader
3050        if player.state.active_group and (
3051            active_group := self.get_player(player.state.active_group)
3052        ):
3053            self.logger.info(
3054                "Player %s is part of a playergroup and can not accept "
3055                "playback related commands itself, "
3056                "redirected the command to the group leader.",
3057                player.name,
3058            )
3059            return active_group
3060        return player
3061
3062    def _get_active_audio_source(self, player: Player) -> tuple[AudioSource, PluginProvider] | None:
3063        """
3064        Return the live AudioSource a player is playing, and its owning PluginProvider.
3065
3066        A player hearing its group's or sync leader's audio is playing that player's
3067        source, so the owner is resolved the same way its active queue is. Returns
3068        None when no source is playing on it, or when the owning plugin is gone.
3069
3070        :param player: The player whose source to resolve.
3071        """
3072        return self.get_player_audio_source(self._audio_source_owner(player).player_id)
3073
3074    def _audio_source_owner(self, player: Player) -> Player:
3075        """
3076        Return the player whose source the given player is hearing.
3077
3078        Mirrors ``get_active_queue``: a sync child hears its leader, a group member
3079        hears its group, and a protocol player hears its parent.
3080
3081        :param player: The player to resolve the owner for.
3082        """
3083        if player.state.synced_to and player.state.synced_to != player.player_id:
3084            if sync_leader := self.get_player(player.state.synced_to):
3085                return self._audio_source_owner(sync_leader)
3086        if player.state.active_group and player.state.active_group != player.player_id:
3087            if group_player := self.get_player(player.state.active_group):
3088                return self._audio_source_owner(group_player)
3089        if player.type == PlayerType.PROTOCOL and player.protocol_parent_id:
3090            if parent_player := self.get_player(player.protocol_parent_id):
3091                return self._audio_source_owner(parent_player)
3092        return player
3093
3094    def _get_player_groups(self, player: Player) -> Iterator[Player]:
3095        """
3096        Return all group players the given player is a member of.
3097
3098        :param player: The player to look up the group memberships for.
3099        """
3100        # A group player mirrors its members, so it is also included while unavailable -
3101        # skipping it there is exactly how its state goes stale.
3102        player_id = player.player_id
3103        for _player in self.iter_players():
3104            if _player.player_id == player_id:
3105                continue
3106            if _player.state.type != PlayerType.GROUP:
3107                continue
3108            if player_id in _player.state.group_members:
3109                yield _player
3110
3111    # Protocol linking methods are provided by ProtocolLinkingMixin (protocol_linking.py)
3112
3113    def _repair_protocol_parent_links(self) -> None:
3114        """
3115        Repair protocol parent links in player configs on startup.
3116
3117        Scans player configs with a protocol_parent_id set and clears parent_ids
3118        that point to player configs that no longer exist (e.g., deleted universal
3119        players). A valid parent link also proves the player is a protocol child,
3120        so a stale player_type (left behind by an aborted registration) is healed.
3121        """
3122        all_player_configs = self.mass.config.get(CONF_PLAYERS, {})
3123        for player_id, player_config in all_player_configs.items():
3124            values = player_config.get("values") or {}
3125            parent_id = values.get(CONF_PROTOCOL_PARENT_ID)
3126            if not parent_id:
3127                continue
3128            # Check if parent config still exists
3129            parent_config = all_player_configs.get(parent_id)
3130            if not parent_config:
3131                self.logger.debug(
3132                    "Clearing stale protocol_parent_id %s for %s (parent config deleted)",
3133                    parent_id,
3134                    player_id,
3135                )
3136                conf_key = f"{CONF_PLAYERS}/{player_id}/values/{CONF_PROTOCOL_PARENT_ID}"
3137                self.mass.config.set(conf_key, None)
3138                continue
3139            if player_config.get("player_type") != PlayerType.PROTOCOL.value:
3140                self.logger.info(
3141                    "Repairing player type of %s - linked as protocol child of %s",
3142                    player_id,
3143                    parent_id,
3144                )
3145                self.mass.config.set_player_type(player_id, PlayerType.PROTOCOL)
3146
3147    async def _fix_group_member_configs(self) -> None:
3148        """
3149        Fix stale protocol player IDs in sync group member configs.
3150
3151        When a sync group references a protocol player ID instead of
3152        the parent player ID, correct it using the cached protocol parent mapping.
3153        """
3154        all_player_configs = self.mass.config.get(CONF_PLAYERS, {})
3155        total_fixes = 0
3156        fixed_groups: list[str] = []
3157
3158        for group_id, group_config in list(all_player_configs.items()):
3159            if group_config.get("provider") != "sync_group":
3160                continue
3161            old_members: list[str] = group_config.get("values", {}).get(CONF_GROUP_MEMBERS, [])
3162            if not old_members:
3163                continue
3164
3165            new_members: list[str] = []
3166            changes = 0
3167            for member_id in old_members:
3168                parent_id = self._get_cached_protocol_parent_id(member_id)
3169                corrected_id = parent_id or member_id
3170                if corrected_id != member_id:
3171                    changes += 1
3172                    self.logger.debug(
3173                        "Sync group %s: corrected member %s -> %s",
3174                        group_id,
3175                        member_id,
3176                        corrected_id,
3177                    )
3178                if corrected_id not in new_members:
3179                    new_members.append(corrected_id)
3180
3181            if changes:
3182                self.mass.config.set_raw_player_config_value(
3183                    group_id, CONF_GROUP_MEMBERS, new_members
3184                )
3185                total_fixes += changes
3186                fixed_groups.append(group_id)
3187
3188        for group_id in fixed_groups:
3189            if (group_player := self.get_player(group_id)) and group_player.available:
3190                await group_player.on_config_updated()
3191
3192        if total_fixes:
3193            self.logger.info(
3194                "Fixed %d stale member reference(s) across %d sync group(s)",
3195                total_fixes,
3196                len(fixed_groups),
3197            )
3198
3199    async def _poll_players(self) -> None:
3200        """Background task that polls players for updates."""
3201        while True:
3202            for player in list(self._players.values()):
3203                # if the player is playing, update elapsed time every tick
3204                # to ensure the queue has accurate details
3205                player_playing = player.state.playback_state == PlaybackState.PLAYING
3206                if player_playing and player.type != PlayerType.PROTOCOL:
3207                    self.mass.call_later(
3208                        0.5,
3209                        self.mass.player_queues.on_player_update,
3210                        player,
3211                        {"corrected_elapsed_time": (None, player.state.corrected_elapsed_time)},
3212                        task_id=f"queue_on_player_update_{player.player_id}",
3213                    )
3214                # Poll player;
3215                if not player.needs_poll:
3216                    continue
3217                try:
3218                    last_poll: float = player.extra_data[ATTR_LAST_POLL]
3219                except KeyError:
3220                    last_poll = 0.0
3221                if (self.mass.loop.time() - last_poll) < player.poll_interval:
3222                    continue
3223                player.extra_data[ATTR_LAST_POLL] = self.mass.loop.time()
3224                try:
3225                    await player.poll()
3226                except Exception as err:
3227                    self.logger.warning(
3228                        "Error while requesting latest state from player %s: %s",
3229                        player.state.name,
3230                        str(err),
3231                        exc_info=err if self.logger.isEnabledFor(10) else None,
3232                    )
3233                # Yield to event loop to prevent blocking
3234                await asyncio.sleep(0)
3235            await asyncio.sleep(1)
3236
3237    def _handle_group_dsp_change(
3238        self, player: Player, prev_group_members: list[str], new_group_members: list[str]
3239    ) -> None:
3240        """Handle DSP reload when group membership changes."""
3241        # reset cached group volume snapshot since membership changed
3242        player.extra_data.pop(ATTR_GROUP_VOLUME_SNAPSHOT, None)
3243        prev_child_count = len(prev_group_members)
3244        new_child_count = len(new_group_members)
3245        is_player_group = player.state.type == PlayerType.GROUP
3246
3247        # handle special case for PlayerGroups: since there are no leaders,
3248        # DSP still always work with a single player in the group.
3249        multi_device_dsp_threshold = 1 if is_player_group else 0
3250
3251        prev_is_multiple_devices = prev_child_count > multi_device_dsp_threshold
3252        new_is_multiple_devices = new_child_count > multi_device_dsp_threshold
3253
3254        if prev_is_multiple_devices == new_is_multiple_devices:
3255            return  # no change in multi-device status
3256
3257        supports_multi_device_dsp = (
3258            PlayerFeature.MULTI_DEVICE_DSP in player.state.supported_features
3259        )
3260
3261        dsp_enabled: bool
3262        if player.state.type == PlayerType.GROUP:
3263            # Since player groups do not have leaders, we will use the only child
3264            # that was in the group before and after the change
3265            if prev_is_multiple_devices:
3266                if childs := new_group_members:
3267                    # We shrank the group from multiple players to a single player
3268                    # So the now only child will control the DSP
3269                    dsp_enabled = self.mass.config.get_player_dsp_config(childs[0]).enabled
3270                else:
3271                    dsp_enabled = False
3272            elif childs := prev_group_members:
3273                # We grew the group from a single player to multiple players,
3274                # let's see if the previous single player had DSP enabled
3275                dsp_enabled = self.mass.config.get_player_dsp_config(childs[0]).enabled
3276            else:
3277                dsp_enabled = False
3278        else:
3279            dsp_enabled = self.mass.config.get_player_dsp_config(player.player_id).enabled
3280
3281        if dsp_enabled and not supports_multi_device_dsp:
3282            # We now know that the group configuration has changed so:
3283            # - multi-device DSP is not supported
3284            # - we switched from a group with multiple players to a single player
3285            #   (or vice versa)
3286            # - the leader has DSP enabled
3287            self.mass.create_task(self.mass.players.on_player_dsp_change(player.player_id))
3288
3289    def _check_external_source_takeover(self, player: Player) -> None:
3290        """
3291        Handle when an external source takes over playback on a player.
3292
3293        When a player has an active grouped output protocol (e.g., AirPlay group) and
3294        an external source (e.g., Spotify Connect, TV input) takes over playback,
3295        we need to clear the active output protocol and ungroup the protocol players.
3296
3297        This prevents the situation where the player appears grouped via protocol
3298        but is actually playing from a different source.
3299
3300        :param player: The player whose active_source changed.
3301        """
3302        # Only relevant for non-protocol players
3303        if player.type == PlayerType.PROTOCOL:
3304            return
3305
3306        # Not a takeover if the player is not actively playing
3307        if player.playback_state != PlaybackState.PLAYING:
3308            return
3309
3310        # Only relevant if we have an active output protocol (not native)
3311        if not player.active_output_protocol or player.active_output_protocol == "native":
3312            return
3313
3314        new_source = player.state.active_source
3315
3316        # Check if new source is external (not MA-managed)
3317        if self._is_ma_managed_source(player, new_source):
3318            return
3319
3320        # Get the active protocol player
3321        protocol_player = self.get_player(player.active_output_protocol)
3322        if not protocol_player:
3323            return
3324
3325        # If the source matches the active protocol's domain, it's expected - not a takeover
3326        # e.g., source "airplay" when using AirPlay protocol is normal
3327        if new_source and new_source.lower() == protocol_player.provider.domain.lower():
3328            return
3329
3330        if (
3331            new_source
3332            and new_source.lower() in ("airplay", "cast", "chromecast", "network")
3333            and protocol_player.provider.domain.lower() == "sendspin"
3334        ):
3335            # Special case for Sendspin bridge: if the new source matches cast or airplay and the
3336            # active protocol is Sendspin, we consider this a normal behavior and not a takeover
3337            return
3338
3339        # Confirmed external source takeover
3340        self.logger.info(
3341            "External source '%s' took over on %s while playing via protocol %s - "
3342            "clearing active output protocol and ungrouping",
3343            new_source,
3344            player.display_name,
3345            protocol_player.provider.domain,
3346        )
3347
3348        # Set active output protocol to native
3349        player.set_active_output_protocol("native")
3350
3351        # Ungroup the protocol player (async task)
3352        self.mass.create_task(protocol_player.ungroup())
3353
3354    def _is_ma_managed_source(self, player: Player, source: str | None) -> bool:
3355        """
3356        Check if a source is managed by Music Assistant.
3357
3358        MA-managed sources include:
3359        - None (=autodetect, no source explicitly set by player)
3360        - The player's own ID (MA queue)
3361        - Any active queue ID
3362        - Any live AudioSource session
3363
3364        :param player: The player to check.
3365        :param source: The source ID to check.
3366        :return: True if the source is MA-managed, False if external.
3367        """
3368        if source is None:
3369            return True
3370
3371        # Player's own ID means MA queue is active
3372        if source == player.player_id:
3373            return True
3374
3375        # A live AudioSource (e.g. Spotify Connect) is streamed by MA itself, so it is
3376        # MA that put it on the player rather than something taking the player over
3377        if self.is_live_audio_source(source):
3378            return True
3379
3380        # Check if it's a known queue ID
3381        return self.mass.player_queues.get(source) is not None
3382
3383    def _schedule_update_all_players(self, delay: float = 2.0) -> None:
3384        """
3385        Schedule a debounced update of all players' state.
3386
3387        Used when a new player is registered to ensure all existing players
3388        update their dynamic properties (like can_group_with) that may have changed.
3389
3390        :param delay: Delay in seconds before triggering updates (default 2.0).
3391        """
3392        if self.mass.closing:
3393            return
3394
3395        for player in self.all_players(
3396            return_unavailable=True,
3397            return_disabled=False,
3398            return_protocol_players=True,
3399        ):
3400            self.trigger_player_update(player.player_id, debounce_delay=delay)
3401
3402    async def _auto_ungroup_if_synced(self, player: Player, log_context: str) -> None:
3403        """
3404        Automatically ungroup a player if it's synced to another player.
3405
3406        :param player: The player to check and potentially ungroup.
3407        :param log_context: Additional context for the log message (e.g., target player name).
3408        """
3409        if not player.state.synced_to and not player.state.active_group:
3410            return
3411        self.logger.info(
3412            "Player %s is already synced to %s, ungrouping it first before %s",
3413            player.name,
3414            player.state.synced_to or player.state.active_group,
3415            log_context,
3416        )
3417        # Use internal _handle_set_members to avoid deadlocking on the play lock
3418        # (we're already inside a cmd_set_members chain that holds a play lock).
3419        synced_to = player.state.synced_to or player.state.active_group
3420        if synced_to and (parent := self.get_player(synced_to)):
3421            try:
3422                async with self.wait_for_player_update(player.player_id, timeout=5):
3423                    await self._handle_set_members(parent, player_ids_to_remove=[player.player_id])
3424            except asyncio.CancelledError:
3425                raise
3426            except Exception:
3427                self.logger.warning(
3428                    "Failed to auto-ungroup %s from %s, proceeding anyway",
3429                    player.name,
3430                    synced_to,
3431                )
3432
3433    async def _dissolve_own_group(self, player: Player, target_name: str) -> bool:
3434        """
3435        Dissolve the group a player leads before it joins another group.
3436
3437        Only a player whose members ride its own stream has a group to give up: it stops
3438        rendering that stream the moment it joins the other group, which would leave its
3439        members silent while they still show up as grouped. A group that is still playing
3440        is refused instead, so the player stays out rather than silencing its members.
3441
3442        :param player: The player about to join another group.
3443        :param target_name: Name of the group being joined, for the log messages.
3444        :return: False when the group is still in place, so the player must not join.
3445        """
3446        if not player.native_grouping_requires_own_stream:
3447            return True
3448        members = [
3449            member_id for member_id in player.state.group_members if member_id != player.player_id
3450        ]
3451        if not members:
3452            return True
3453        if player.state.playback_state in (PlaybackState.PLAYING, PlaybackState.PAUSED):
3454            # Grouping never offers a rendering leader as a target, so reaching this means
3455            # the caller worked from a can_group_with snapshot taken before the playback
3456            # started. Tearing the group down now would cut its members off mid-track.
3457            self.logger.warning(
3458                "Player %s is still serving its own group, leaving it out of %s",
3459                player.name,
3460                target_name,
3461            )
3462            return False
3463        self.logger.info(
3464            "Player %s leads a group of its own, dissolving it before it joins %s",
3465            player.name,
3466            target_name,
3467        )
3468        # Removing the members rather than the leader keeps this out of the leadership
3469        # transfer path, and the internal handler avoids deadlocking on the play lock
3470        # (we're already inside a cmd_set_members chain that holds it).
3471        try:
3472            await self._handle_set_members(player, player_ids_to_remove=members)
3473        except asyncio.CancelledError:
3474            raise
3475        except Exception:
3476            # Joining on top of a group it still leads is the silent-member state this
3477            # dissolve exists to prevent, and one the player can no longer be talked out of.
3478            self.logger.warning(
3479                "Could not dissolve the group of %s, leaving it out of %s",
3480                player.name,
3481                target_name,
3482            )
3483            return False
3484        return True
3485
3486    async def _handle_set_members(
3487        self,
3488        parent_player: Player,
3489        player_ids_to_add: list[str] | None = None,
3490        player_ids_to_remove: list[str] | None = None,
3491    ) -> None:
3492        """
3493        Handle the actual set_members logic.
3494
3495        Skips permission checks and locking (internal use only).
3496
3497        :param parent_player: The parent player to add/remove members to/from.
3498        :param player_ids_to_add: List of player_id's to add to the parent player.
3499        :param player_ids_to_remove: List of player_id's to remove from the parent player.
3500        """
3501        target_player = parent_player.player_id
3502        # handle the sync leader being removed from itself: either transfer leadership
3503        # to a remaining member (keeping playback alive) or dissolve the group entirely
3504        should_stop = False
3505        if player_ids_to_remove and target_player in player_ids_to_remove:
3506            remaining_members = [
3507                m
3508                for m in parent_player.state.group_members
3509                if m != target_player
3510                and m not in player_ids_to_remove
3511                and (member := self.get_player(m))
3512                and member.state.available
3513            ]
3514            # a new leader must be able to render audio: a group with only display,
3515            # visualizer or lighting members left has no playback heir and dissolves
3516            has_playback_heir = any(
3517                (member := self.get_player(m)) and member.state.type in PLAYBACK_TARGET_TYPES
3518                for m in remaining_members
3519            )
3520            active_queue = self.get_active_queue(parent_player)
3521            if has_playback_heir and active_queue and active_queue.state != PlaybackState.IDLE:
3522                # transfer leadership to a remaining member instead of dissolving
3523                await self._transfer_ad_hoc_leadership(parent_player, remaining_members)
3524                return
3525            self.logger.info(
3526                "Dissolving sync group of player %s as it is being removed from itself",
3527                parent_player.name,
3528            )
3529            player_ids_to_add = None
3530            player_ids_to_remove = [
3531                x for x in parent_player.state.group_members if x != target_player
3532            ]
3533            should_stop = True
3534        # filter all player ids on compatibility and availability
3535        final_player_ids_to_add: list[str] = []
3536        for child_player_id in player_ids_to_add or []:
3537            if child_player_id == target_player:
3538                continue
3539            if child_player_id in final_player_ids_to_add:
3540                continue
3541            if (
3542                not (child_player := self.get_player(child_player_id))
3543                or not child_player.state.available
3544            ):
3545                self.logger.warning("Player %s is not available", child_player_id)
3546                continue
3547
3548            # check if player can be synced/grouped with the target player
3549            # state.can_group_with already handles all expansion and translation
3550            if child_player_id not in parent_player.state.can_group_with:
3551                self.logger.warning(
3552                    "Player %s can not be grouped with %s",
3553                    child_player.name,
3554                    parent_player.name,
3555                )
3556                continue
3557
3558            if (
3559                child_player.state.synced_to
3560                and child_player.state.synced_to == target_player
3561                and child_player_id in parent_player.state.group_members
3562            ):
3563                continue  # already synced to this target
3564
3565            # also skip if the child is part of this group via its sync leader
3566            # (e.g. synced to the sync leader of this syncgroup)
3567            if (
3568                child_player.state.active_group == target_player
3569                and child_player_id in parent_player.state.group_members
3570            ):
3571                continue
3572
3573            # handle edge case: child player is synced to a different player
3574            # automatically ungroup it first and wait for state to propagate
3575            # but not if the child is already part of this group (via its sync leader)
3576            if child_player.state.synced_to and target_player not in {
3577                child_player.state.synced_to,
3578                child_player.state.active_group,
3579            }:
3580                await self._auto_ungroup_if_synced(child_player, f"joining {parent_player.name}")
3581
3582            # handle edge case: the child leads a native group of its own, which it cannot
3583            # keep serving from inside this one
3584            if not await self._dissolve_own_group(child_player, parent_player.state.name):
3585                continue
3586
3587            # power on the player if needed
3588            if (
3589                not child_player.state.powered
3590                and child_player.state.power_control != PLAYER_CONTROL_NONE
3591            ):
3592                await self._handle_cmd_power(child_player.player_id, True)
3593            # if we reach here, all checks passed
3594            final_player_ids_to_add.append(child_player_id)
3595
3596        # process player ids to remove and filter out invalid/unavailable players and edge cases
3597        final_player_ids_to_remove: list[str] = []
3598        if player_ids_to_remove:
3599            for child_player_id in player_ids_to_remove:
3600                if child_player_id in parent_player.state.group_members:
3601                    final_player_ids_to_remove.append(child_player_id)
3602                    continue
3603                # also accept the removal if the child player itself reports
3604                # being synced to this parent - handles race conditions where the
3605                # parent's group_members state is stale/not yet updated. The
3606                # native synced_to is checked as well: a protocol child's state
3607                # value is translated to the visible parent, which would reject
3608                # a removal correctly addressed at its native sync leader.
3609                child_player = self.get_player(child_player_id)
3610                if child_player and target_player in (
3611                    child_player.state.synced_to,
3612                    child_player.synced_to,
3613                ):
3614                    final_player_ids_to_remove.append(child_player_id)
3615                    continue
3616
3617        # Forward command to the appropriate player after all (base) sanity checks
3618        # GROUP players (sync_group, universal_group) manage their own members internally
3619        # and don't need protocol translation - call their set_members directly
3620        if (
3621            parent_player.type == PlayerType.GROUP
3622            and PlayerFeature.SET_MEMBERS in parent_player.state.supported_features
3623        ):
3624            await parent_player.set_members(
3625                player_ids_to_add=final_player_ids_to_add,
3626                player_ids_to_remove=final_player_ids_to_remove,
3627            )
3628            return
3629        # For regular players, handle protocol selection and translation
3630        await self._handle_set_members_with_protocols(
3631            parent_player, final_player_ids_to_add, final_player_ids_to_remove
3632        )
3633
3634        if should_stop:
3635            await self._stop_player_or_its_queue(parent_player)
3636
3637    async def _handle_set_members_with_protocols(
3638        self,
3639        parent_player: Player,
3640        player_ids_to_add: list[str],
3641        player_ids_to_remove: list[str],
3642    ) -> None:
3643        """
3644        Handle set_members considering protocol and native members.
3645
3646        Skips permission checks, locking, and all redirect logic (internal use only).
3647        Translates visible player IDs to protocol player IDs when appropriate,
3648        and forwards to the correct player's set_members.
3649
3650        :param parent_player: The parent player to add/remove members to/from.
3651        :param player_ids_to_add: List of visible player IDs to add as members.
3652        :param player_ids_to_remove: List of visible player IDs to remove from members.
3653        """
3654        # Get parent's active protocol domain and player if available
3655        parent_protocol_domain = None
3656        parent_protocol_player = None
3657        if (
3658            parent_player.active_output_protocol
3659            and parent_player.active_output_protocol != "native"
3660        ):
3661            parent_protocol_player = self.get_player(parent_player.active_output_protocol)
3662            if parent_protocol_player:
3663                parent_protocol_domain = parent_protocol_player.provider.domain
3664
3665        self.logger.debug(
3666            "set_members on %s: active_protocol=%s, adding=%s, removing=%s",
3667            parent_player.state.name,
3668            parent_protocol_domain or "none",
3669            player_ids_to_add,
3670            player_ids_to_remove,
3671        )
3672
3673        # Translate members to add
3674        (
3675            protocol_members_to_add,
3676            native_members_to_add,
3677            parent_protocol_player,
3678            parent_protocol_domain,
3679        ) = self._translate_members_for_protocols(
3680            parent_player, player_ids_to_add, parent_protocol_player, parent_protocol_domain
3681        )
3682
3683        self.logger.debug(
3684            "Translated members: protocol=%s (domain=%s), native=%s",
3685            protocol_members_to_add,
3686            parent_protocol_domain,
3687            native_members_to_add,
3688        )
3689
3690        # Translate members to remove
3691        protocol_members_to_remove, native_members_to_remove = (
3692            self._translate_members_to_remove_for_protocols(
3693                parent_player, player_ids_to_remove, parent_protocol_player, parent_protocol_domain
3694            )
3695        )
3696
3697        # Forward protocol members to protocol player's set_members
3698        if (protocol_members_to_add or protocol_members_to_remove) and parent_protocol_player:
3699            await self._forward_protocol_set_members(
3700                parent_player,
3701                parent_protocol_player,
3702                protocol_members_to_add,
3703                protocol_members_to_remove,
3704            )
3705
3706        # Forward native members to parent player's set_members
3707        if native_members_to_add or native_members_to_remove:
3708            filtered_native_add = self._filter_native_members(native_members_to_add, parent_player)
3709            # For removal, allow protocol players if they're actually in the parent's group_members
3710            # This handles native protocol players (e.g., native AirPlay) where group_members
3711            # contains protocol player IDs
3712            filtered_native_remove = [
3713                pid
3714                for pid in native_members_to_remove
3715                if (p := self.get_player(pid))
3716                and (p.type != PlayerType.PROTOCOL or pid in parent_player.group_members)
3717            ]
3718            self.logger.debug(
3719                "Native grouping on %s: filtered_add=%s, filtered_remove=%s",
3720                parent_player.state.name,
3721                filtered_native_add,
3722                filtered_native_remove,
3723            )
3724            if filtered_native_add or filtered_native_remove:
3725                if PlayerFeature.SET_MEMBERS not in parent_player.state.supported_features:
3726                    return
3727                self.logger.info(
3728                    "Calling set_members on native player %s with add=%s, remove=%s",
3729                    parent_player.state.name,
3730                    filtered_native_add,
3731                    filtered_native_remove,
3732                )
3733                await parent_player.set_members(
3734                    player_ids_to_add=filtered_native_add or None,
3735                    player_ids_to_remove=filtered_native_remove or None,
3736                )
3737
3738    async def _transfer_ad_hoc_leadership(
3739        self, leader: Player, remaining_members: list[str]
3740    ) -> None:
3741        """
3742        Transfer leadership of an ad-hoc sync group to a remaining member.
3743
3744        Called when the sync leader of an ad-hoc group is unjoined while other
3745        members remain and playback is active. The queue is moved to a newly
3746        selected leader, the remaining members are regrouped under it and playback
3747        resumes at the saved position (accepting a brief audio gap).
3748
3749        :param leader: The current sync leader being removed from the group.
3750        :param remaining_members: Available group members (excluding the leader)
3751            that should keep playing under a new leader.
3752        """
3753        active_queue = self.get_active_queue(leader)
3754        was_playing = active_queue is not None and active_queue.state == PlaybackState.PLAYING
3755        new_leader_id = self._select_ad_hoc_leader(leader, remaining_members)
3756        self.logger.info(
3757            "Transferring leadership of %s to %s (%s remaining member(s))",
3758            leader.name,
3759            new_leader_id,
3760            len(remaining_members),
3761        )
3762        # Move the queue to the new leader. transfer_queue frees the new leader from
3763        # the old leader's group and stops the old leader; the playback position
3764        # survives because stop() stores it in resume_pos.
3765        await self.mass.player_queues.transfer_queue(
3766            leader.player_id, new_leader_id, auto_play=False
3767        )
3768        # regroup the other remaining members under the new leader
3769        other_members = [m for m in remaining_members if m != new_leader_id]
3770        if other_members:
3771            await self.cmd_set_members(new_leader_id, player_ids_to_add=other_members)
3772        if was_playing:
3773            await self.mass.player_queues.resume(new_leader_id)
3774
3775    def _select_ad_hoc_leader(self, leader: Player, remaining_members: list[str]) -> str:
3776        """
3777        Pick the new leader for an ad-hoc sync group leadership transfer.
3778
3779        Prefers a remaining member that can currently be reached on the protocol the
3780        group is playing on, so the other members can be regrouped under it; falls back
3781        to the first remaining member. The members' own ``can_group_with`` is unusable
3782        here because it is empty while they are still synced to the old leader.
3783
3784        :param leader: The current sync leader being removed.
3785        :param remaining_members: Candidate member player_ids, already filtered for
3786            availability. Must contain at least one audio-capable member.
3787        """
3788        # non-audio members (display/visualizer/lighting) stay in the group as followers
3789        # but can never inherit the queue
3790        candidates = [
3791            m
3792            for m in remaining_members
3793            if (member := self.get_player(m)) and member.state.type in PLAYBACK_TARGET_TYPES
3794        ]
3795        active_domain: str | None = None
3796        if leader.active_output_protocol and leader.active_output_protocol != "native":
3797            if protocol_player := self.get_player(leader.active_output_protocol):
3798                active_domain = protocol_player.provider.domain
3799        if active_domain:
3800            for member_id in candidates:
3801                member = self.get_player(member_id)
3802                if member is None:
3803                    continue
3804                if active_domain in member.playback_domains:
3805                    return member_id
3806        return candidates[0]
3807
3808    async def _stop_player_or_its_queue(self, player: Player) -> None:
3809        """
3810        Stop the player, ending its queue when it is playing one of its own.
3811
3812        :param player: The player to stop.
3813        """
3814        # End the queue itself, exactly as a stop command does: stopping only the device
3815        # leaves the queue session open, so its preloading keeps pulling audio and a
3816        # provider streaming a live session (Spotify) stays tethered for another track or
3817        # two. Restricted to the player's own queue: get_active_queue resolves a protocol
3818        # player to its parent, and stopping that parent's queue would come straight back
3819        # here. The permission-free handler, because both callers act on behalf of the
3820        # server rather than a user that can address the player.
3821        if (
3822            active_queue := self.get_active_queue(player)
3823        ) and active_queue.queue_id == player.player_id:
3824            await self.mass.player_queues._handle_stop(player.player_id)
3825            return
3826        await self._handle_cmd_stop(player.player_id)
3827
3828    def _clear_sleep_timer(self, player: Player) -> None:
3829        """
3830        Clear the active sleep timer for the player.
3831
3832        :param player: Player to clear the timer for.
3833        """
3834        self.mass.cancel_timer(self._sleep_timer_task_id(player.player_id))
3835        if player.sleep_timer_expires_at is not None:
3836            player.set_sleep_timer_expires_at(None)
3837            player.update_state()
3838            self._signal_sleep_timer_updated(player, None)
3839
3840    async def _handle_sleep_timer_expired(self, player_id: str) -> None:
3841        """
3842        Stop playback when a player's sleep timer expires.
3843
3844        :param player_id: Player ID whose sleep timer expired.
3845        """
3846        player = self.get_player(player_id)
3847        if player is None or player.sleep_timer_expires_at is None:
3848            return
3849        player.set_sleep_timer_expires_at(None)
3850        player.update_state()
3851        self._signal_sleep_timer_updated(player, None)
3852        await self.cmd_stop(player_id)
3853
3854    def _signal_sleep_timer_updated(self, player: Player, expires_at: float | None) -> None:
3855        """
3856        Signal a sleep timer change for the player on the event bus.
3857
3858        :param player: Player whose sleep timer changed.
3859        :param expires_at: New expiry timestamp, or None when the timer was cleared.
3860        """
3861        if player.state.type == PlayerType.PROTOCOL:
3862            return
3863        self.mass.signal_event(
3864            EventType.PLAYER_SLEEP_TIMER_UPDATED,
3865            object_id=player.player_id,
3866            data=expires_at,
3867        )
3868
3869    @staticmethod
3870    def _sleep_timer_task_id(player_id: str) -> str:
3871        """
3872        Return the scheduled task ID for a player's sleep timer.
3873
3874        :param player_id: Player ID to build the task ID for.
3875        """
3876        return f"player_sleep_timer_{player_id}"
3877
3878    # Private command handlers (no permission checks)
3879
3880    async def _handle_cmd_resume(
3881        self, player_id: str, source: str | None = None, media: PlayerMedia | None = None
3882    ) -> None:
3883        """
3884        Handle resume playback command.
3885
3886        Skips permission checks and locking (internal use only).
3887        """
3888        player = self._get_player_with_redirect(player_id)
3889        source = source or player.state.active_source
3890        media = media or player.state.current_media
3891        # power on the player if needed
3892        if not player.state.powered and player.state.power_control != PLAYER_CONTROL_NONE:
3893            await self._handle_cmd_power(player.player_id, True)
3894        # Redirect to queue controller if it is active
3895        if active_queue := self.mass.player_queues.get(source or player_id):
3896            await self.mass.player_queues.resume(active_queue.queue_id)
3897            return
3898        # try to handle command on player directly
3899        # TODO: check if player has an active source with native resume support
3900        active_source = next((x for x in player.state.source_list if x.id == source), None)
3901        if (
3902            player.state.playback_state in (PlaybackState.IDLE, PlaybackState.PAUSED)
3903            and active_source
3904            and active_source.can_play_pause
3905            and PlayerFeature.PAUSE in player.state.supported_features
3906        ):
3907            # player has some other source active and native resume support
3908            await player.play()
3909            return
3910        if active_source and not active_source.passive:
3911            await self.select_source(player_id, active_source.id)
3912            return
3913        if media:
3914            # try to re-play the current media item
3915            await player.play_media(media)
3916            return
3917        # fallback: just try to resume queue playback
3918        await self.mass.player_queues.resume(player.player_id)
3919
3920    async def _handle_cmd_power(
3921        self, player_id: str, powered: bool, skip_auto_play: bool = False
3922    ) -> None:
3923        """
3924        Handle player power on/off command.
3925
3926        Skips permission checks and locking (internal use only).
3927
3928        :param player_id: The player ID to power on/off.
3929        :param powered: True to power on, False to power off.
3930        :param skip_auto_play: If True, skip auto-play on power on.
3931        """
3932        player = self.get_player(player_id, True)
3933        assert player is not None  # for type checking
3934        player_state = player.state
3935
3936        if player_state.powered == powered:
3937            self.logger.debug(
3938                "Ignoring power %s command for player %s: already in state %s",
3939                "ON" if powered else "OFF",
3940                player_state.name,
3941                "ON" if player_state.powered else "OFF",
3942            )
3943            return  # nothing to do
3944
3945        # ungroup player at power off
3946        player_was_sync_child = bool(player.state.synced_to or player.state.active_group)
3947        if (
3948            (player_was_sync_child or player.group_members)
3949            and player.type == PlayerType.PLAYER
3950            and not powered
3951        ):
3952            # ungroup player if it is synced (or is a sync leader itself)
3953            await self.cmd_ungroup(player_id)
3954
3955        # always stop player at power off
3956        if (
3957            not powered
3958            and not player_was_sync_child
3959            and player_state.playback_state in (PlaybackState.PLAYING, PlaybackState.PAUSED)
3960        ):
3961            # wait for the stop command to process and prevent race conditions
3962            async with self.wait_for_player_update(player_id, timeout=5):
3963                await self._stop_player_or_its_queue(player)
3964
3965        # power off all synced childs when player is a sync leader
3966        elif not powered and player_state.type == PlayerType.PLAYER and player_state.group_members:
3967            async with TaskManager(self.mass) as tg:
3968                for member in self.iter_group_members(player, True):
3969                    if member.power_control == PLAYER_CONTROL_NONE:
3970                        continue
3971                    tg.create_task(self._handle_cmd_power(member.player_id, False))
3972
3973        # handle actual power command
3974        if player_state.power_control == PLAYER_CONTROL_NONE:
3975            self.logger.debug(
3976                "Player %s does not support power control, ignoring power command",
3977                player_state.name,
3978            )
3979            return
3980        if player_state.power_control == PLAYER_CONTROL_NATIVE:
3981            # player supports power command natively: forward to player provider
3982            await player.power(powered)
3983            if powered:
3984                await wait_for_power_on(self.logger, player)
3985        elif player_state.power_control == PLAYER_CONTROL_FAKE:
3986            # user wants to use fake power control - so we (optimistically) update the state
3987            # and store the state in the cache
3988            player.extra_data[ATTR_FAKE_POWER] = powered
3989            # Group players need to actively form/dissolve their session when the
3990            # user toggles fake power — otherwise the toggle would only update the
3991            # cosmetic state without ever capturing or releasing the members.
3992            if player_state.type == PlayerType.GROUP:
3993                await player.power(powered)
3994            player.update_state()  # trigger update of the player state
3995            if player_state.type != PlayerType.GROUP:
3996                # see register(): group fake-power is intentionally not persisted
3997                # because there is no session to restore at boot.
3998                await self.mass.cache.set(
3999                    key=player_id,
4000                    data=powered,
4001                    provider=self.domain,
4002                    category=CACHE_CATEGORY_PLAYER_POWER,
4003                )
4004        # handle external player control
4005        elif player_control := self._controls.get(player.state.power_control):
4006            control_name = player_control.name
4007            self.logger.debug("Redirecting power command to PlayerControl %s", control_name)
4008            if not player_control.supports_power:
4009                raise UnsupportedFeaturedException(
4010                    f"Player control {control_name} is not available"
4011                )
4012            if powered:
4013                assert player_control.power_on is not None  # for type checking
4014                await player_control.power_on()
4015                await wait_for_power_on(self.logger, player, player_control)
4016            else:
4017                assert player_control.power_off is not None  # for type checking
4018                await player_control.power_off()
4019        # always trigger a state update to update the UI
4020        player.refresh_state()
4021
4022        # handle 'auto play on power on' feature
4023        if (
4024            not skip_auto_play
4025            and not player_state.active_group
4026            and not player_state.synced_to
4027            and powered
4028            and player.config.get_value(CONF_AUTO_PLAY)
4029            and player_state.active_source in (None, player_id)
4030            and not player.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS)
4031        ):
4032            await self.mass.player_queues.resume(player_id)
4033
4034    def _resolve_group_volume_player(self, player: Player) -> Player | None:
4035        """
4036        Return the player whose group a group volume command applies to.
4037
4038        Returns None if the given player is not grouped at all. Commands addressed to a
4039        synced member and to its sync leader resolve to the same player, so they read
4040        and guard one and the same group.
4041
4042        :param player: The player the command was addressed to.
4043        """
4044        # the group volume lock this resolves to may not share the VOLUME purpose:
4045        # set_group_volume sets the volume of the members concurrently and a sync leader
4046        # is a member of its own group, so a group command would wait on its own lock.
4047        if player.state.type == PlayerType.GROUP or player.state.group_members:
4048            # dedicated group player or sync leader
4049            return player
4050        if player.state.synced_to:
4051            # a synced player follows its sync leader
4052            return self.get_player(player.state.synced_to)
4053        return None
4054
4055    async def _set_member_volume(self, player_id: str, volume_level: int) -> None:
4056        """
4057        Set the volume of a single member as part of a group volume change.
4058
4059        :param player_id: player_id of the member to handle the command.
4060        :param volume_level: logical volume level (0..100) to set on the member.
4061        """
4062        # record before waiting for the lock, for the same reason as cmd_volume_set
4063        if member := self.get_player(player_id):
4064            self._record_volume_target(member, volume_level)
4065        # take the volume lock of the member itself, so a group volume change and an
4066        # individual volume command for that member can not overtake one another
4067        async with self.get_player_lock(player_id, PlayerLockPurpose.VOLUME):
4068            await self._handle_cmd_volume_set(player_id, volume_level, record_target=False)
4069
4070    async def _handle_cmd_volume_set(
4071        self, player_id: str, volume_level: int, *, record_target: bool = True
4072    ) -> None:
4073        """
4074        Handle Player volume set command.
4075
4076        Skips permission checks and locking (internal use only).
4077
4078        :param player_id: player_id of the player to handle the command.
4079        :param volume_level: logical volume level (0..100) to set on the player.
4080        :param record_target: Set to False when the caller already recorded the level as
4081            the base for the next volume nudge, before it waited for the volume lock.
4082        """
4083        player = self.get_player(player_id, True)
4084        assert player is not None  # for type checker
4085
4086        # Clamp logical volume to 0-100
4087        volume_level = max(0, min(100, volume_level))
4088
4089        if player.type == PlayerType.GROUP:
4090            # redirect to special group volume control
4091            await self.cmd_group_volume(player_id, volume_level)
4092            return
4093
4094        # A muted player stays muted: only an explicit unmute lifts it, and the level
4095        # set here is the one it plays at once that happens. Fake mute is the exception,
4096        # because it is simulated with the volume itself.
4097        if self._stays_silent_on_volume_change(player):
4098            # a locked player stays silent, the volume it holds is the one
4099            # that gets restored once it is unmuted again
4100            volume_level = 0
4101            # the lock may have been earned after the caller recorded the level it asked
4102            # for, which is then not the level this player ends up at
4103            record_target = True
4104        else:
4105            player.extra_data.pop(ATTR_FAKE_MUTE, None)
4106
4107        if record_target:
4108            self._record_volume_target(player, volume_level)
4109
4110        # Scale logical volume (0-100) to device volume (min_volume-max_volume)
4111        device_volume = self.scale_volume_to_device(player_id, volume_level)
4112
4113        await self._notify_source_volume_change(player, volume_level)
4114
4115        # Handle native volume control support
4116        if player.volume_control == PLAYER_CONTROL_NATIVE:
4117            # player supports volume command natively: forward to player
4118            await player.volume_set(device_volume)
4119            return
4120        # Handle fake volume control support
4121        if player.volume_control == PLAYER_CONTROL_FAKE:
4122            # user wants to use fake volume control - so we (optimistically) update the state
4123            # and store the state in the cache. Fake volume uses the logical volume (no scaling).
4124            player.extra_data[ATTR_FAKE_VOLUME] = volume_level
4125            player.update_state()
4126            return
4127        # player has no volume support at all
4128        if player.volume_control == PLAYER_CONTROL_NONE:
4129            raise UnsupportedFeaturedException(
4130                f"Player {player.state.name} does not support volume control"
4131            )
4132        # handle external player control
4133        if player_control := self._controls.get(player.state.volume_control):
4134            control_name = player_control.name
4135            self.logger.debug("Redirecting volume command to PlayerControl %s", control_name)
4136            if not player_control.supports_volume:
4137                raise UnsupportedFeaturedException(
4138                    f"Player control {control_name} is not available"
4139                )
4140            assert player_control.volume_set is not None
4141            # forward the already-scaled device volume; the external control sets the
4142            # raw device volume and does not apply min/max scaling of its own
4143            await player_control.volume_set(device_volume)
4144            return
4145        if protocol_player := self.get_player(player.state.volume_control):
4146            # forward the already-scaled device volume: the limits configured on this
4147            # (user-facing) player are the only ones that apply to the command
4148            self.logger.debug(
4149                "Redirecting volume command to protocol player %s",
4150                protocol_player.provider.manifest.name,
4151            )
4152            await protocol_player.volume_set(device_volume)
4153            return
4154
4155    @staticmethod
4156    def _is_in_group(state: PlayerState) -> bool:
4157        """Check if the player with the given state is currently grouped with other players."""
4158        # a sync leader has neither synced_to nor active_group set, but it does lead its
4159        # own group_members, which stays empty for a player that is not grouped at all
4160        return bool(state.synced_to or state.active_group or state.group_members)
4161
4162    def _has_active_mute_lock(self, player: Player) -> bool:
4163        """
4164        Check if the given player holds a mute lock that still applies to it.
4165
4166        A lock is only earned inside a group and only holds for as long as the player
4167        is still grouped, so it can not outlive the group it was earned in.
4168
4169        :param player: The player to check, which may be a protocol player.
4170        """
4171        if player.extra_data.get(ATTR_MUTE_LOCK) and self._is_in_group(player.state):
4172            return True
4173        # cmd_volume_mute stores the lock on the parent player, while the volume command
4174        # may arrive with the protocol player ID (e.g. during group volume changes)
4175        if player.protocol_parent_id and (parent := self.get_player(player.protocol_parent_id)):
4176            return bool(parent.extra_data.get(ATTR_MUTE_LOCK)) and self._is_in_group(parent.state)
4177        return False
4178
4179    def _stays_silent_on_volume_change(self, player: Player) -> bool:
4180        """Check if a volume command for the given player lands at 0 to keep it silent."""
4181        return (
4182            self._has_active_mute_lock(player)
4183            and player.mute_control == PLAYER_CONTROL_FAKE
4184            and bool(player.extra_data.get(ATTR_FAKE_MUTE))
4185        )
4186
4187    async def _mute_group_members(self, group_player: Player, muted: bool) -> None:
4188        """
4189        Mute or unmute all mute capable members of a player group or synced players.
4190
4191        :param group_player: The group player or sync leader.
4192        :param muted: bool if the group should be muted.
4193        """
4194        coros = []
4195        for child_player in self.iter_group_members(
4196            group_player, only_powered=True, exclude_self=False
4197        ):
4198            if child_player.mute_control == PLAYER_CONTROL_NONE:
4199                # members without a mute control are left alone, just like the
4200                # group mute state itself is calculated from the capable members only
4201                continue
4202            coros.append(self.cmd_volume_mute(child_player.player_id, muted))
4203        await asyncio.gather(*coros)
4204
4205    async def _handle_cmd_volume_mute(self, player: Player, mute_control: str, muted: bool) -> None:
4206        """
4207        Send the mute command to the given player's mute control.
4208
4209        Skips permission checks, locking and mute lock bookkeeping (internal use only).
4210
4211        :param player: the player to handle the command.
4212        :param mute_control: the already resolved mute control of the player.
4213        :param muted: bool if player should be muted.
4214        """
4215        if mute_control == PLAYER_CONTROL_NATIVE:
4216            # player supports mute command natively: forward to player
4217            await player.volume_mute(muted)
4218            return
4219        if mute_control == PLAYER_CONTROL_FAKE:
4220            # user wants to use fake mute control - so we use volume instead
4221            self.logger.debug(
4222                "Using volume for muting for player %s",
4223                player.state.name,
4224            )
4225            if muted:
4226                already_muted = bool(player.extra_data.get(ATTR_FAKE_MUTE))
4227                if not already_muted:
4228                    # on a repeated mute command the volume is already 0
4229                    player.extra_data[ATTR_PREVIOUS_VOLUME] = player.state.volume_level
4230                await self._handle_cmd_volume_set(player.player_id, 0)
4231                # set the flag after the volume command, as that clears it
4232                player.extra_data[ATTR_FAKE_MUTE] = True
4233                player.update_state()
4234            else:
4235                was_muted = bool(player.extra_data.get(ATTR_FAKE_MUTE))
4236                player.extra_data[ATTR_FAKE_MUTE] = False
4237                player.update_state()
4238                if not was_muted:
4239                    # the volume is the one the user is listening at, restoring
4240                    # anything here would turn a no-op unmute into a volume change
4241                    return
4242                stored_volume: int | None = player.extra_data.pop(ATTR_PREVIOUS_VOLUME, None)
4243                # the volume was still unknown at mute time, so pick a low volume
4244                # rather than blasting the speaker at some assumed level
4245                await self._handle_cmd_volume_set(
4246                    player.player_id, 1 if stored_volume is None else stored_volume
4247                )
4248            return
4249
4250        # handle external player control
4251        if player_control := self._controls.get(mute_control):
4252            control_name = player_control.name
4253            self.logger.debug("Redirecting mute command to PlayerControl %s", control_name)
4254            if not player_control.supports_mute:
4255                raise UnsupportedFeaturedException(
4256                    f"Player control {control_name} is not available"
4257                )
4258            assert player_control.mute_set is not None
4259            await player_control.mute_set(muted)
4260            return
4261
4262        # handle to protocol player as volume_mute control
4263        if protocol_player := self.get_player(mute_control):
4264            self.logger.debug(
4265                "Redirecting mute command to protocol player %s",
4266                protocol_player.provider.manifest.name,
4267            )
4268            await protocol_player.volume_mute(muted)
4269            return
4270
4271        # the configured control disappeared after the mute control was resolved
4272        raise UnsupportedFeaturedException(f"Player {player.state.name} does not support muting")
4273
4274    async def _handle_play_media(self, player_id: str, media: PlayerMedia) -> None:
4275        """
4276        Handle play media command without group redirect.
4277
4278        Skips permission checks, locking, and all redirect logic (internal use only).
4279
4280        :param player_id: player_id of the player to handle the command.
4281        :param media: The Media that needs to be played on the player.
4282        """
4283        player = self.get_player(player_id, raise_unavailable=True)
4284        assert player is not None
4285        # media that is not the live source itself takes the player away from it. An
4286        # announcement is the exception: it interrupts the player and hands it straight
4287        # back, so releasing the source would tear down a session that is about to
4288        # resume — and one that cannot be re-selected once its plugin has let go.
4289        if media.media_type not in (MediaType.AUDIO_SOURCE, MediaType.ANNOUNCEMENT):
4290            await self._release_audio_source(player_id)
4291        # set active source if media has a source_id (e.g. plugin source or mass queue source)
4292        if media.source_id:
4293            player.set_active_mass_source(media.source_id)
4294
4295        # Determine output protocol to use:
4296        # While a session is active (playing/paused), keep using the already active
4297        # protocol so mid-session commands stay on the same output.
4298        # On a fresh start always (re)select: a leftover active protocol from a
4299        # previous session must not overrule user preference, a grouped protocol
4300        # or native playback (and it may point at a player that is gone by now).
4301        target_player: Player | None = None
4302        output_protocol: OutputProtocol | None = None
4303        if (
4304            player.state.playback_state in (PlaybackState.PLAYING, PlaybackState.PAUSED)
4305            and player.active_output_protocol
4306            and player.active_output_protocol != "native"
4307            and (protocol_player := self.get_player(player.active_output_protocol))
4308        ):
4309            # Use the already-set protocol directly
4310            output_protocol = player.get_linked_protocol(player.active_output_protocol)
4311            if output_protocol is not None:
4312                target_player = protocol_player
4313        if target_player is None:
4314            target_player, output_protocol = self._select_best_output_protocol(player)
4315
4316        if target_player.player_id != player.player_id:
4317            # Playing via linked protocol - update active output protocol
4318            # output_protocol is guaranteed to be non-None when target_player != player
4319            assert output_protocol is not None
4320            self.logger.debug(
4321                "Starting playback on %s via protocol %s (target=%s), group_members=%s",
4322                player.state.name,
4323                output_protocol.name,
4324                target_player.display_name,
4325                target_player.state.group_members,
4326            )
4327            player.set_active_output_protocol(output_protocol.output_protocol_id)
4328        elif player.type != PlayerType.GROUP:
4329            # Native playback - group players don't have output protocols of their own
4330            # (they delegate to a sync leader / member which manages its own protocol)
4331            self.logger.debug(
4332                "Starting playback on %s via native, group_members=%s",
4333                player.state.name,
4334                player.state.group_members,
4335            )
4336            player.set_active_output_protocol("native")
4337
4338        # power on the player if needed (skip auto-play since we're about to start playback)
4339        if not player.state.powered and player.state.power_control != PLAYER_CONTROL_NONE:
4340            await self._handle_cmd_power(player.player_id, True, skip_auto_play=True)
4341        await target_player.play_media(media)
4342        if target_player.player_id != player.player_id:
4343            # notify the native player that protocol playback started
4344            assert output_protocol is not None
4345            await player.on_protocol_playback(output_protocol=output_protocol)
4346
4347    async def _handle_enqueue_next_media(self, player_id: str, media: PlayerMedia) -> None:
4348        """
4349        Handle enqueue next media command without group redirect.
4350
4351        Skips permission checks, locking, and all redirect logic (internal use only).
4352
4353        :param player_id: player_id of the player to handle the command.
4354        :param media: The Media that needs to be enqueued on the player.
4355        """
4356        player = self.get_player(player_id, raise_unavailable=True)
4357        assert player is not None
4358        if target_player := self._get_control_target(
4359            player,
4360            required_feature=PlayerFeature.ENQUEUE,
4361            require_active=True,
4362        ):
4363            self.logger.debug(
4364                "Redirecting enqueue command to protocol player %s",
4365                target_player.provider.manifest.name,
4366            )
4367            await target_player.enqueue_next_media(media)
4368            return
4369
4370        if PlayerFeature.ENQUEUE not in player.state.supported_features:
4371            raise UnsupportedFeaturedException(
4372                f"Player {player.state.name} does not support enqueueing"
4373            )
4374        await player.enqueue_next_media(media)
4375
4376    async def _notify_source_volume_change(self, player: Player, volume_level: int) -> None:
4377        """
4378        Tell the source playing on a player that its volume changed.
4379
4380        Only the player the source is actually playing on notifies, never one that
4381        merely hears it as a group member — otherwise a group volume change would
4382        fire the callback once per child, each with a different value.
4383
4384        :param player: The player whose volume changed.
4385        :param volume_level: The new volume, 0-100.
4386        """
4387        if (session := self.get_audio_source_session(player.player_id)) is None:
4388            return
4389        provider = self.mass.get_provider(session.provider_instance_id)
4390        if not isinstance(provider, PluginProvider):
4391            return
4392        await provider.on_volume_change(session.source_id, volume_level)
4393
4394    def _resolve_command_target(self, player: Player, source_id: str | None) -> str:
4395        """
4396        Return the source (id) a command issued to a player applies to.
4397
4398        :param player: The player the command was issued to.
4399        :param source_id: The source the caller aimed the command at, if it named one.
4400        :return: The id of the player's active source, which is the id of Music
4401            Assistant's own queue when nothing else is playing on it.
4402        :raises PlayerCommandFailed: When the caller named a source that is no longer
4403            the one playing.
4404        """
4405        active_source_id = player.state.active_source or player.player_id
4406        if source_id is not None and source_id != active_source_id:
4407            msg = f"The source this was meant for is no longer playing on {player.state.name}."
4408            raise PlayerCommandFailed(msg)
4409        return active_source_id
4410
4411    async def _forward_to_external_source(
4412        self,
4413        player: Player,
4414        action: SourceControl,
4415        value: SourceControlValue = None,
4416    ) -> bool:
4417        """
4418        Hand a control action to the external source playing on a player.
4419
4420        Covers the external sources Music Assistant provides itself, which own a
4421        session it can talk to. A source belonging to the player (its line-in, TV
4422        input, or its own Spotify Connect) has no such session, so this reports that
4423        it did not take the action and the caller goes on to the player itself.
4424
4425        The per-action transport flags gate what the source advertises it can do, so
4426        a client is refused rather than left waiting. Ordering is not gated here: the
4427        session decides what reordering means for its own content. Most sources do not
4428        implement it at all, though, so a client should ask the source whether it can
4429        before offering the control - handing it one that quietly does nothing is
4430        worse than not offering it.
4431
4432        :param player: The player the action was issued to.
4433        :param action: The control action to hand over.
4434        :param value: The action's argument, where it takes one.
4435        :return: True when an external source took the action.
4436        """
4437        if (active := self._get_active_audio_source(player)) is None:
4438            return False
4439        audio_source, provider = active
4440        supported = {
4441            SourceControl.PLAY: audio_source.can_play_pause,
4442            SourceControl.PAUSE: audio_source.can_play_pause,
4443            SourceControl.SEEK: audio_source.can_seek,
4444            SourceControl.NEXT: audio_source.can_next_previous,
4445            SourceControl.PREVIOUS: audio_source.can_next_previous,
4446        }.get(action, True)
4447        if not supported:
4448            msg = (
4449                f"The active source ({audio_source.name}) on player "
4450                f"{player.display_name} does not support this action"
4451            )
4452            raise PlayerCommandFailed(msg)
4453        try:
4454            await provider.on_source_control(audio_source.item_id, action, value)
4455        except NotImplementedError as err:
4456            # a source with no control surface at all (vban_receiver) reaches the base
4457            # implementation; a caller deserves a refusal rather than a server error
4458            msg = (
4459                f"The active source ({audio_source.name}) on player "
4460                f"{player.display_name} can not be controlled"
4461            )
4462            raise PlayerCommandFailed(msg) from err
4463        return True
4464
4465    async def _release_audio_source(self, player_id: str) -> None:
4466        """
4467        Let go of the live source a player was playing, if it had one.
4468
4469        Tells the owning plugin so an upstream session still pointing at Music
4470        Assistant is released. A plugin that raises must not stop the player from
4471        moving on, so failures are logged rather than propagated.
4472
4473        :param player_id: The player that is done with its source.
4474        """
4475        if (session := self._end_audio_source_session(player_id)) is None:
4476            return
4477        self.trigger_player_update(player_id)
4478        provider = self.mass.get_provider(session.provider_instance_id)
4479        if not isinstance(provider, PluginProvider):
4480            return
4481        try:
4482            await provider.on_source_released(session.source_id, player_id)
4483        except Exception:
4484            self.logger.warning(
4485                "on_source_released raised for provider %s source %s player %s",
4486                provider.instance_id,
4487                session.source_id,
4488                player_id,
4489                exc_info=True,
4490            )
4491
4492    async def _release_unclaimed_audio_source(
4493        self, player_id: str, session: AudioSourceSession, playback_session_id: str
4494    ) -> None:
4495        """
4496        Release a source whose renderer never requested the stream.
4497
4498        The play command returned without error, so the late-start release in the
4499        streams controller never fires: no stream request means no failed stream
4500        request either. Without this the player would keep publishing a source
4501        that never started, with its own queue held inactive behind it.
4502
4503        :param player_id: The player the source was started on.
4504        :param session: The session that was started for it.
4505        :param playback_session_id: Playback session active when it was started.
4506        """
4507        current = self.get_audio_source_session(player_id)
4508        if (
4509            current is not session
4510            or current.playback_session_id != playback_session_id
4511            or current.stream_session_id is not None
4512        ):
4513            return
4514        self.logger.info(
4515            "AudioSource %s was never streamed by player %s, releasing it",
4516            session.source_id,
4517            player_id,
4518        )
4519        await self.deselect_source(
4520            player_id,
4521            provider_instance_id=session.provider_instance_id,
4522            source_id=session.source_id,
4523            playback_session_id=playback_session_id,
4524        )
4525
4526    async def _resolve_audio_source_uri(
4527        self, source: str
4528    ) -> tuple[AudioSource, PluginProvider] | None:
4529        """
4530        Resolve a source string to a live AudioSource, if that is what it names.
4531
4532        :param source: The source string a select names.
4533        :return: The source and its owning plugin, or None when the string names
4534            something else (a queue, a player-native source).
4535        """
4536        if "://" not in source:
4537            return None
4538        try:
4539            item = await self.mass.music.get_item_by_uri(source)
4540        except MusicAssistantError as err:
4541            # not resolvable as media, so it is something else (a queue id, a
4542            # player-native source) — logged because a provider being unavailable
4543            # or unauthenticated also lands here
4544            self.logger.debug("Could not resolve %s as an audio source: %s", source, err)
4545            return None
4546        if not isinstance(item, AudioSource):
4547            return None
4548        provider = self.mass.get_provider(item.provider)
4549        if not isinstance(provider, PluginProvider):
4550            return None
4551        if ProviderFeature.AUDIO_SOURCE not in provider.supported_features:
4552            return None
4553        return item, provider
4554
4555    async def _start_audio_source(
4556        self, player: Player, audio_source: AudioSource, provider: PluginProvider
4557    ) -> None:
4558        """
4559        Start a live external source on a player.
4560
4561        The player's queue is left exactly as it is: it simply stops being the
4562        active source, so it is still there to resume when the source ends.
4563
4564        :param player: The player to play the source on.
4565        :param audio_source: The source that was selected.
4566        :param provider: The plugin exposing that source.
4567        """
4568        # a player outputs one source at a time, so another one already on it has to be
4569        # handed back first: replacing the session silently would leave its plugin
4570        # holding an upstream session that still points at this player
4571        if (current := self.get_audio_source_session(player.player_id)) is not None and (
4572            current.source_id != audio_source.item_id
4573            or current.provider_instance_id != provider.instance_id
4574        ):
4575            await self._release_audio_source(player.player_id)
4576        session = self._start_audio_source_session(
4577            player.player_id, audio_source, provider.instance_id
4578        )
4579        try:
4580            await self._handle_play_media(
4581                player.player_id,
4582                PlayerMedia(
4583                    uri=audio_source.uri or audio_source.item_id,
4584                    media_type=MediaType.AUDIO_SOURCE,
4585                    title=audio_source.name,
4586                    # the session's owner, which its stream url is keyed on
4587                    source_id=player.player_id,
4588                    queue_session_id=session.playback_session_id,
4589                ),
4590            )
4591        except Exception:
4592            # the source never started, so the player must not go on publishing it:
4593            # a session left behind holds the queue inactive with nothing playing it
4594            if self.get_audio_source_session(player.player_id) is session:
4595                await self._release_audio_source(player.player_id)
4596            raise
4597        # the play command returning does not mean the renderer ever fetched the
4598        # stream url: until a stream request claims the session nothing will evict
4599        # the player the source may be moving from, and nothing else would ever
4600        # clear a session that is never streamed
4601        self.mass.call_later(
4602            AUDIO_SOURCE_CLAIM_TIMEOUT,
4603            self._release_unclaimed_audio_source,
4604            player.player_id,
4605            session,
4606            session.playback_session_id,
4607            task_id=f"release_unclaimed_audio_source_{player.player_id}",
4608        )
4609
4610    async def _handle_select_source(self, player_id: str, source: str | None) -> None:
4611        """
4612        Handle select source command without group redirect.
4613
4614        Skips permission checks, locking, and all redirect logic (internal use only).
4615
4616        :param player_id: player_id of the player to handle the command.
4617        :param source: The ID of the source that needs to be activated/selected.
4618        """
4619        if source is None:
4620            source = player_id  # default to MA queue source
4621        player = self.get_player(player_id, True)
4622        assert player is not None
4623        # check if player is already playing and source is different
4624        # in that case we need to stop the player first
4625        prev_source = player.state.active_source
4626        if prev_source and source != prev_source:
4627            with suppress(PlayerCommandFailed, RuntimeError):
4628                # just try to stop (regardless of state) and let it settle, so the
4629                # new source does not race the teardown. A player that already
4630                # reports idle has nothing to tear down and does not wait at all.
4631                async with self.wait_for_player_update(
4632                    player_id,
4633                    attribute_name="playback_state",
4634                    attribute_value=PlaybackState.IDLE,
4635                    timeout=5,
4636                ):
4637                    await self._handle_cmd_stop(player_id)
4638        # an audio source uri selects the live source itself, which plays on the
4639        # player while its queue keeps its own items and goes inactive
4640        if (resolved := await self._resolve_audio_source_uri(source)) is not None:
4641            await self._start_audio_source(player, *resolved)
4642            return
4643        # anything else takes the player away from a live source it was playing
4644        await self._release_audio_source(player_id)
4645        # check if source is a mass queue
4646        # this can be used to restore the queue after a source switch
4647        if self.mass.player_queues.get(source):
4648            player.set_active_mass_source(source)
4649            return
4650        # Legacy compatibility: the old plugin-source API used the
4651        # plugin's instance_id directly as the source string. The refactor
4652        # moved plugin sources to first-class AudioSource MediaItems played
4653        # via player_queues.play_media. Translate a legacy plugin-instance-id
4654        # source into the new flow so old frontends, third-party scripts,
4655        # and HA automations keep working — but only when the provider
4656        # exposes EXACTLY ONE AudioSource (it was always a 1:1 mapping under
4657        # the old API; multi-source providers have to use the explicit URI).
4658        if (legacy_prov := self.mass.get_provider(source)) and isinstance(
4659            legacy_prov, PluginProvider
4660        ):
4661            if ProviderFeature.AUDIO_SOURCE not in legacy_prov.supported_features:
4662                raise PlayerCommandFailed(f"Provider {source} does not expose AudioSources")
4663            sources = await legacy_prov.get_audio_sources()
4664            if len(sources) == 1:
4665                self.logger.debug(
4666                    "Translating legacy select_source(%s) to play_media(%s)",
4667                    source,
4668                    sources[0].uri,
4669                )
4670                await self.mass.player_queues.play_media(player_id, str(sources[0].uri))
4671                return
4672            raise UnsupportedFeaturedException(
4673                f"Provider {source} exposes {len(sources)} AudioSources; the legacy "
4674                "select_source(plugin_instance_id) API only supported 1:1 mappings. "
4675                "Use player_queues.play_media with an explicit AudioSource URI."
4676            )
4677        # basic check if player supports source selection
4678        if PlayerFeature.SELECT_SOURCE not in player.state.supported_features:
4679            raise UnsupportedFeaturedException(
4680                f"Player {player.state.name} does not support source selection"
4681            )
4682        # basic check if source is valid for player
4683        if not any(x for x in player.state.source_list if x.id == source):
4684            raise PlayerCommandFailed(
4685                f"{source} is an invalid source for player {player.state.name}"
4686            )
4687        # forward to player
4688        await player.select_source(source)
4689
4690    async def _handle_cmd_stop(self, player_id: str) -> None:
4691        """
4692        Handle stop command without any redirects.
4693
4694        Skips permission checks, locking, and all redirect logic (internal use only).
4695
4696        :param player_id: player_id of the player to handle the command.
4697        """
4698        player = self.get_player(player_id, raise_unavailable=True)
4699        assert player is not None
4700        protocol_player: Player | None = None
4701        if player.active_output_protocol and player.active_output_protocol != "native":
4702            protocol_player = self.get_player(player.active_output_protocol)
4703        if player.state.playback_state == PlaybackState.IDLE:
4704            # The player already reports idle but an output protocol is still marked
4705            # active: the protocol player may never have received a stop at all
4706            # (e.g. the source stream ended on its own before this stop command
4707            # arrived). Forward an (idempotent) stop and schedule the protocol clear
4708            # so no stale session lingers on the device and the next playback
4709            # (re)selects the output protocol.
4710            if protocol_player is not None:
4711                await protocol_player.stop()
4712                if len(protocol_player.group_members) <= 1:
4713                    self.schedule_active_output_protocol_clear(player)
4714            return
4715        player.mark_stop_called()
4716        # Delegate to active protocol player if one is active
4717        target_player = player
4718        if protocol_player is not None:
4719            target_player = protocol_player
4720            if PlayerFeature.POWER in target_player.supported_features:
4721                # if protocol player supports/requires power,
4722                # we power it off instead of just stopping (which also stops playback)
4723                # this is rare as most protocols do not support power control (except for cast)
4724                await self._handle_cmd_power(target_player.player_id, False)
4725                return
4726
4727        # handle command on player(protocol) directly
4728        await target_player.stop()
4729        # Only clear active protocol if the protocol player has no remaining group members.
4730        # If there are still protocol group members, keep the protocol active so that
4731        # when playback resumes it continues on the same protocol.
4732        if target_player.player_id == player.player_id or len(target_player.group_members) <= 1:
4733            self.schedule_active_output_protocol_clear(player)
4734
4735    async def _handle_cmd_play(self, player_id: str) -> None:
4736        """
4737        Handle play command without group redirect.
4738
4739        Skips permission checks, locking, and all redirect logic (internal use only).
4740
4741        :param player_id: player_id of the player to handle the command.
4742        """
4743        player = self.get_player(player_id, raise_unavailable=True)
4744        assert player is not None
4745        if player.state.playback_state == PlaybackState.PLAYING:
4746            self.logger.info(
4747                "Ignore PLAY request to player %s: player is already playing", player.state.name
4748            )
4749            return
4750        # If an AudioSource is the active queue item, proxy play to the plugin
4751        if active := self._get_active_audio_source(player):
4752            audio_source, plugin_prov = active
4753            if audio_source.can_play_pause:
4754                await plugin_prov.on_source_control(audio_source.item_id, SourceControl.PLAY)
4755                return
4756        # handle unpause (=play if player is paused)
4757        if player.state.playback_state == PlaybackState.PAUSED:
4758            active_source = next(
4759                (x for x in player.state.source_list if x.id == player.state.active_source), None
4760            )
4761            # raise if active source does not support play/pause
4762            if active_source and not active_source.can_play_pause:
4763                msg = (
4764                    f"The active source ({active_source.name}) on player "
4765                    f"{player.state.name} does not support play/pause"
4766                )
4767                raise PlayerCommandFailed(msg)
4768            # Delegate to active protocol player if one is active
4769            if target_player := self._get_control_target(
4770                player, PlayerFeature.PAUSE, require_active=True
4771            ):
4772                await target_player.play()
4773                return
4774            # No active protocol target: if the player rendering the audio supports pause and
4775            # the active (external) source can be paused, unpause it directly instead of
4776            # restarting the source.
4777            output_player = player.resolve_output_player()
4778            if (
4779                active_source
4780                and active_source.can_play_pause
4781                and PlayerFeature.PAUSE in output_player.supported_features
4782            ):
4783                await output_player.play()
4784                return
4785
4786        # player is not paused: try to resume the player
4787        # Note: We handle resume inline here without calling _handle_cmd_resume
4788        active_source = next(
4789            (x for x in player.state.source_list if x.id == player.state.active_source), None
4790        )
4791        media = player.state.current_media
4792        # power on the player if needed
4793        if not player.state.powered and player.state.power_control != PLAYER_CONTROL_NONE:
4794            await self._handle_cmd_power(player.player_id, True)
4795        if active_source and not active_source.passive:
4796            await self._handle_select_source(player_id, active_source.id)
4797            return
4798        if media:
4799            # try to re-play the current media item
4800            await player.play_media(media)
4801            return
4802        # fallback: just send play command - which will fail if nothing can be played
4803        await player.play()
4804
4805    async def _handle_cmd_pause(self, player_id: str) -> None:
4806        """
4807        Handle pause command without any redirects.
4808
4809        Skips permission checks, locking, and all redirect logic (internal use only).
4810
4811        :param player_id: player_id of the player to handle the command.
4812        """
4813        player = self.get_player(player_id, raise_unavailable=True)
4814        assert player is not None
4815        if player.state.playback_state == PlaybackState.IDLE:
4816            return
4817        # If an AudioSource is the active queue item, proxy pause to the plugin
4818        if active := self._get_active_audio_source(player):
4819            audio_source, plugin_prov = active
4820            if audio_source.can_play_pause:
4821                await plugin_prov.on_source_control(audio_source.item_id, SourceControl.PAUSE)
4822                return
4823        # handle command on player/source directly
4824        active_source = next(
4825            (x for x in player.state.source_list if x.id == player.state.active_source), None
4826        )
4827        if active_source and not active_source.can_play_pause:
4828            # raise if active source does not support play/pause
4829            msg = (
4830                f"The active source ({active_source.name}) on player "
4831                f"{player.state.name} does not support play/pause"
4832            )
4833            raise PlayerCommandFailed(msg)
4834        # Delegate to active protocol player if one is active
4835        if target_player := self._get_control_target(
4836            player, PlayerFeature.PAUSE, require_active=True
4837        ):
4838            await target_player.pause()
4839            return
4840        # No active protocol target: if the player rendering the audio supports pause and the
4841        # active (external) source can be paused, forward the command to it instead of stopping
4842        # it (mirrors the external-source handling in cmd_seek/cmd_next_track).
4843        output_player = player.resolve_output_player()
4844        if (
4845            active_source
4846            and active_source.can_play_pause
4847            and PlayerFeature.PAUSE in output_player.supported_features
4848        ):
4849            await output_player.pause()
4850            return
4851        # player/protocol does not support pause: fall back to stop
4852        self.logger.debug(
4853            "Player/protocol %s does not support pause, using STOP instead",
4854            player.state.name,
4855        )
4856        await self._handle_cmd_stop(player.player_id)
4857