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