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