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