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