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