music-assistant-server

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