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