music-assistant-server

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