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