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):
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):
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(self, player: Player) -> Iterator[Player]:
2588        """
2589        Return all group players the given player is a member of.
2590
2591        :param player: The player to look up the group memberships for.
2592        """
2593        # Deliberately walks the registry instead of all_players(): this is internal
2594        # topology used to keep derived state consistent, so it must not be narrowed by
2595        # the user context of whichever command happens to trigger the lookup. A group
2596        # player mirrors its members, so it is also included while unavailable -
2597        # skipping it there is exactly how its state goes stale. Disabled players take
2598        # no part at all and players still registering are not fully set up yet.
2599        player_id = player.player_id
2600        for _player in list(self._players.values()):
2601            if _player.player_id == player_id:
2602                continue
2603            if _player.state.type != PlayerType.GROUP:
2604                continue
2605            if not _player.state.enabled or not _player.initialized.is_set():
2606                continue
2607            if player_id in _player.state.group_members:
2608                yield _player
2609
2610    # Protocol linking methods are provided by ProtocolLinkingMixin (protocol_linking.py)
2611
2612    def _repair_protocol_parent_links(self) -> None:
2613        """
2614        Repair protocol parent links in player configs on startup.
2615
2616        Scans player configs with a protocol_parent_id set and clears parent_ids
2617        that point to player configs that no longer exist (e.g., deleted universal
2618        players). A valid parent link also proves the player is a protocol child,
2619        so a stale player_type (left behind by an aborted registration) is healed.
2620        """
2621        all_player_configs = self.mass.config.get(CONF_PLAYERS, {})
2622        for player_id, player_config in all_player_configs.items():
2623            values = player_config.get("values") or {}
2624            parent_id = values.get(CONF_PROTOCOL_PARENT_ID)
2625            if not parent_id:
2626                continue
2627            # Check if parent config still exists
2628            parent_config = all_player_configs.get(parent_id)
2629            if not parent_config:
2630                self.logger.debug(
2631                    "Clearing stale protocol_parent_id %s for %s (parent config deleted)",
2632                    parent_id,
2633                    player_id,
2634                )
2635                conf_key = f"{CONF_PLAYERS}/{player_id}/values/{CONF_PROTOCOL_PARENT_ID}"
2636                self.mass.config.set(conf_key, None)
2637                continue
2638            if player_config.get("player_type") != PlayerType.PROTOCOL.value:
2639                self.logger.info(
2640                    "Repairing player type of %s - linked as protocol child of %s",
2641                    player_id,
2642                    parent_id,
2643                )
2644                self.mass.config.set_player_type(player_id, PlayerType.PROTOCOL)
2645
2646    async def _fix_group_member_configs(self) -> None:
2647        """
2648        Fix stale protocol player IDs in sync group member configs.
2649
2650        When a sync group references a protocol player ID instead of
2651        the parent player ID, correct it using the cached protocol parent mapping.
2652        """
2653        all_player_configs = self.mass.config.get(CONF_PLAYERS, {})
2654        total_fixes = 0
2655        fixed_groups: list[str] = []
2656
2657        for group_id, group_config in list(all_player_configs.items()):
2658            if group_config.get("provider") != "sync_group":
2659                continue
2660            old_members: list[str] = group_config.get("values", {}).get(CONF_GROUP_MEMBERS, [])
2661            if not old_members:
2662                continue
2663
2664            new_members: list[str] = []
2665            changes = 0
2666            for member_id in old_members:
2667                parent_id = self._get_cached_protocol_parent_id(member_id)
2668                corrected_id = parent_id or member_id
2669                if corrected_id != member_id:
2670                    changes += 1
2671                    self.logger.debug(
2672                        "Sync group %s: corrected member %s -> %s",
2673                        group_id,
2674                        member_id,
2675                        corrected_id,
2676                    )
2677                if corrected_id not in new_members:
2678                    new_members.append(corrected_id)
2679
2680            if changes:
2681                self.mass.config.set_raw_player_config_value(
2682                    group_id, CONF_GROUP_MEMBERS, new_members
2683                )
2684                total_fixes += changes
2685                fixed_groups.append(group_id)
2686
2687        for group_id in fixed_groups:
2688            if (group_player := self.get_player(group_id)) and group_player.available:
2689                await group_player.on_config_updated()
2690
2691        if total_fixes:
2692            self.logger.info(
2693                "Fixed %d stale member reference(s) across %d sync group(s)",
2694                total_fixes,
2695                len(fixed_groups),
2696            )
2697
2698    async def _poll_players(self) -> None:
2699        """Background task that polls players for updates."""
2700        while True:
2701            for player in list(self._players.values()):
2702                # if the player is playing, update elapsed time every tick
2703                # to ensure the queue has accurate details
2704                player_playing = player.state.playback_state == PlaybackState.PLAYING
2705                if player_playing and player.type != PlayerType.PROTOCOL:
2706                    self.mass.call_later(
2707                        0.5,
2708                        self.mass.player_queues.on_player_update,
2709                        player,
2710                        {"corrected_elapsed_time": (None, player.state.corrected_elapsed_time)},
2711                        task_id=f"queue_on_player_update_{player.player_id}",
2712                    )
2713                # Poll player;
2714                if not player.needs_poll:
2715                    continue
2716                try:
2717                    last_poll: float = player.extra_data[ATTR_LAST_POLL]
2718                except KeyError:
2719                    last_poll = 0.0
2720                if (self.mass.loop.time() - last_poll) < player.poll_interval:
2721                    continue
2722                player.extra_data[ATTR_LAST_POLL] = self.mass.loop.time()
2723                try:
2724                    await player.poll()
2725                except Exception as err:
2726                    self.logger.warning(
2727                        "Error while requesting latest state from player %s: %s",
2728                        player.state.name,
2729                        str(err),
2730                        exc_info=err if self.logger.isEnabledFor(10) else None,
2731                    )
2732                # Yield to event loop to prevent blocking
2733                await asyncio.sleep(0)
2734            await asyncio.sleep(1)
2735
2736    def _handle_group_dsp_change(
2737        self, player: Player, prev_group_members: list[str], new_group_members: list[str]
2738    ) -> None:
2739        """Handle DSP reload when group membership changes."""
2740        # reset cached group volume snapshot since membership changed
2741        player.extra_data.pop(ATTR_GROUP_VOLUME_SNAPSHOT, None)
2742        prev_child_count = len(prev_group_members)
2743        new_child_count = len(new_group_members)
2744        is_player_group = player.state.type == PlayerType.GROUP
2745
2746        # handle special case for PlayerGroups: since there are no leaders,
2747        # DSP still always work with a single player in the group.
2748        multi_device_dsp_threshold = 1 if is_player_group else 0
2749
2750        prev_is_multiple_devices = prev_child_count > multi_device_dsp_threshold
2751        new_is_multiple_devices = new_child_count > multi_device_dsp_threshold
2752
2753        if prev_is_multiple_devices == new_is_multiple_devices:
2754            return  # no change in multi-device status
2755
2756        supports_multi_device_dsp = (
2757            PlayerFeature.MULTI_DEVICE_DSP in player.state.supported_features
2758        )
2759
2760        dsp_enabled: bool
2761        if player.state.type == PlayerType.GROUP:
2762            # Since player groups do not have leaders, we will use the only child
2763            # that was in the group before and after the change
2764            if prev_is_multiple_devices:
2765                if childs := new_group_members:
2766                    # We shrank the group from multiple players to a single player
2767                    # So the now only child will control the DSP
2768                    dsp_enabled = self.mass.config.get_player_dsp_config(childs[0]).enabled
2769                else:
2770                    dsp_enabled = False
2771            elif childs := prev_group_members:
2772                # We grew the group from a single player to multiple players,
2773                # let's see if the previous single player had DSP enabled
2774                dsp_enabled = self.mass.config.get_player_dsp_config(childs[0]).enabled
2775            else:
2776                dsp_enabled = False
2777        else:
2778            dsp_enabled = self.mass.config.get_player_dsp_config(player.player_id).enabled
2779
2780        if dsp_enabled and not supports_multi_device_dsp:
2781            # We now know that the group configuration has changed so:
2782            # - multi-device DSP is not supported
2783            # - we switched from a group with multiple players to a single player
2784            #   (or vice versa)
2785            # - the leader has DSP enabled
2786            self.mass.create_task(self.mass.players.on_player_dsp_change(player.player_id))
2787
2788    def _check_external_source_takeover(self, player: Player) -> None:
2789        """
2790        Handle when an external source takes over playback on a player.
2791
2792        When a player has an active grouped output protocol (e.g., AirPlay group) and
2793        an external source (e.g., Spotify Connect, TV input) takes over playback,
2794        we need to clear the active output protocol and ungroup the protocol players.
2795
2796        This prevents the situation where the player appears grouped via protocol
2797        but is actually playing from a different source.
2798
2799        :param player: The player whose active_source changed.
2800        """
2801        # Only relevant for non-protocol players
2802        if player.type == PlayerType.PROTOCOL:
2803            return
2804
2805        # Not a takeover if the player is not actively playing
2806        if player.playback_state != PlaybackState.PLAYING:
2807            return
2808
2809        # Only relevant if we have an active output protocol (not native)
2810        if not player.active_output_protocol or player.active_output_protocol == "native":
2811            return
2812
2813        new_source = player.state.active_source
2814
2815        # Check if new source is external (not MA-managed)
2816        if self._is_ma_managed_source(player, new_source):
2817            return
2818
2819        # Get the active protocol player
2820        protocol_player = self.get_player(player.active_output_protocol)
2821        if not protocol_player:
2822            return
2823
2824        # If the source matches the active protocol's domain, it's expected - not a takeover
2825        # e.g., source "airplay" when using AirPlay protocol is normal
2826        if new_source and new_source.lower() == protocol_player.provider.domain.lower():
2827            return
2828
2829        if (
2830            new_source
2831            and new_source.lower() in ("airplay", "cast", "chromecast", "network")
2832            and protocol_player.provider.domain.lower() == "sendspin"
2833        ):
2834            # Special case for Sendspin bridge: if the new source matches cast or airplay and the
2835            # active protocol is Sendspin, we consider this a normal behavior and not a takeover
2836            return
2837
2838        # Confirmed external source takeover
2839        self.logger.info(
2840            "External source '%s' took over on %s while playing via protocol %s - "
2841            "clearing active output protocol and ungrouping",
2842            new_source,
2843            player.display_name,
2844            protocol_player.provider.domain,
2845        )
2846
2847        # Set active output protocol to native
2848        player.set_active_output_protocol("native")
2849
2850        # Ungroup the protocol player (async task)
2851        self.mass.create_task(protocol_player.ungroup())
2852
2853    def _is_ma_managed_source(self, player: Player, source: str | None) -> bool:
2854        """
2855        Check if a source is managed by Music Assistant.
2856
2857        MA-managed sources include:
2858        - None (=autodetect, no source explicitly set by player)
2859        - The player's own ID (MA queue)
2860        - Any active queue ID
2861        - Any plugin source ID
2862
2863        :param player: The player to check.
2864        :param source: The source ID to check.
2865        :return: True if the source is MA-managed, False if external.
2866        """
2867        if source is None:
2868            return True
2869
2870        # Player's own ID means MA queue is active
2871        if source == player.player_id:
2872            return True
2873
2874        # Check if it's a known queue ID
2875        return self.mass.player_queues.get(source) is not None
2876
2877    def _schedule_update_all_players(self, delay: float = 2.0) -> None:
2878        """
2879        Schedule a debounced update of all players' state.
2880
2881        Used when a new player is registered to ensure all existing players
2882        update their dynamic properties (like can_group_with) that may have changed.
2883
2884        :param delay: Delay in seconds before triggering updates (default 2.0).
2885        """
2886        if self.mass.closing:
2887            return
2888
2889        for player in self.all_players(
2890            return_unavailable=True,
2891            return_disabled=False,
2892            return_protocol_players=True,
2893        ):
2894            self.trigger_player_update(player.player_id, debounce_delay=delay)
2895
2896    async def _auto_ungroup_if_synced(self, player: Player, log_context: str) -> None:
2897        """
2898        Automatically ungroup a player if it's synced to another player.
2899
2900        :param player: The player to check and potentially ungroup.
2901        :param log_context: Additional context for the log message (e.g., target player name).
2902        """
2903        if not player.state.synced_to and not player.state.active_group:
2904            return
2905        self.logger.info(
2906            "Player %s is already synced to %s, ungrouping it first before %s",
2907            player.name,
2908            player.state.synced_to or player.state.active_group,
2909            log_context,
2910        )
2911        # Use internal _handle_set_members to avoid deadlocking on the play lock
2912        # (we're already inside a cmd_set_members chain that holds a play lock).
2913        synced_to = player.state.synced_to or player.state.active_group
2914        if synced_to and (parent := self.get_player(synced_to)):
2915            try:
2916                async with self.wait_for_player_update(player.player_id, timeout=5):
2917                    await self._handle_set_members(parent, player_ids_to_remove=[player.player_id])
2918            except asyncio.CancelledError:
2919                raise
2920            except Exception:
2921                self.logger.warning(
2922                    "Failed to auto-ungroup %s from %s, proceeding anyway",
2923                    player.name,
2924                    synced_to,
2925                )
2926
2927    async def _handle_set_members(
2928        self,
2929        parent_player: Player,
2930        player_ids_to_add: list[str] | None = None,
2931        player_ids_to_remove: list[str] | None = None,
2932    ) -> None:
2933        """
2934        Handle the actual set_members logic.
2935
2936        Skips permission checks and locking (internal use only).
2937
2938        :param parent_player: The parent player to add/remove members to/from.
2939        :param player_ids_to_add: List of player_id's to add to the parent player.
2940        :param player_ids_to_remove: List of player_id's to remove from the parent player.
2941        """
2942        target_player = parent_player.player_id
2943        # handle the sync leader being removed from itself: either transfer leadership
2944        # to a remaining member (keeping playback alive) or dissolve the group entirely
2945        should_stop = False
2946        if player_ids_to_remove and target_player in player_ids_to_remove:
2947            remaining_members = [
2948                m
2949                for m in parent_player.state.group_members
2950                if m != target_player
2951                and m not in player_ids_to_remove
2952                and (member := self.get_player(m))
2953                and member.state.available
2954            ]
2955            active_queue = self.get_active_queue(parent_player)
2956            if remaining_members and active_queue and active_queue.state != PlaybackState.IDLE:
2957                # transfer leadership to a remaining member instead of dissolving
2958                await self._transfer_ad_hoc_leadership(parent_player, remaining_members)
2959                return
2960            self.logger.info(
2961                "Dissolving sync group of player %s as it is being removed from itself",
2962                parent_player.name,
2963            )
2964            player_ids_to_add = None
2965            player_ids_to_remove = [
2966                x for x in parent_player.state.group_members if x != target_player
2967            ]
2968            should_stop = True
2969        # filter all player ids on compatibility and availability
2970        final_player_ids_to_add: list[str] = []
2971        for child_player_id in player_ids_to_add or []:
2972            if child_player_id == target_player:
2973                continue
2974            if child_player_id in final_player_ids_to_add:
2975                continue
2976            if (
2977                not (child_player := self.get_player(child_player_id))
2978                or not child_player.state.available
2979            ):
2980                self.logger.warning("Player %s is not available", child_player_id)
2981                continue
2982
2983            # check if player can be synced/grouped with the target player
2984            # state.can_group_with already handles all expansion and translation
2985            if child_player_id not in parent_player.state.can_group_with:
2986                self.logger.warning(
2987                    "Player %s can not be grouped with %s",
2988                    child_player.name,
2989                    parent_player.name,
2990                )
2991                continue
2992
2993            if (
2994                child_player.state.synced_to
2995                and child_player.state.synced_to == target_player
2996                and child_player_id in parent_player.state.group_members
2997            ):
2998                continue  # already synced to this target
2999
3000            # also skip if the child is part of this group via its sync leader
3001            # (e.g. synced to the sync leader of this syncgroup)
3002            if (
3003                child_player.state.active_group == target_player
3004                and child_player_id in parent_player.state.group_members
3005            ):
3006                continue
3007
3008            # handle edge case: child player is synced to a different player
3009            # automatically ungroup it first and wait for state to propagate
3010            # but not if the child is already part of this group (via its sync leader)
3011            if child_player.state.synced_to and target_player not in {
3012                child_player.state.synced_to,
3013                child_player.state.active_group,
3014            }:
3015                await self._auto_ungroup_if_synced(child_player, f"joining {parent_player.name}")
3016
3017            # power on the player if needed
3018            if (
3019                not child_player.state.powered
3020                and child_player.state.power_control != PLAYER_CONTROL_NONE
3021            ):
3022                await self._handle_cmd_power(child_player.player_id, True)
3023            # if we reach here, all checks passed
3024            final_player_ids_to_add.append(child_player_id)
3025
3026        # process player ids to remove and filter out invalid/unavailable players and edge cases
3027        final_player_ids_to_remove: list[str] = []
3028        if player_ids_to_remove:
3029            for child_player_id in player_ids_to_remove:
3030                if child_player_id in parent_player.state.group_members:
3031                    final_player_ids_to_remove.append(child_player_id)
3032                    continue
3033                # also accept the removal if the child player itself reports
3034                # being synced to this parent - handles race conditions where the
3035                # parent's group_members state is stale/not yet updated
3036                child_player = self.get_player(child_player_id)
3037                if child_player and child_player.state.synced_to == target_player:
3038                    final_player_ids_to_remove.append(child_player_id)
3039                    continue
3040
3041        # Forward command to the appropriate player after all (base) sanity checks
3042        # GROUP players (sync_group, universal_group) manage their own members internally
3043        # and don't need protocol translation - call their set_members directly
3044        if (
3045            parent_player.type == PlayerType.GROUP
3046            and PlayerFeature.SET_MEMBERS in parent_player.state.supported_features
3047        ):
3048            await parent_player.set_members(
3049                player_ids_to_add=final_player_ids_to_add,
3050                player_ids_to_remove=final_player_ids_to_remove,
3051            )
3052            return
3053        # For regular players, handle protocol selection and translation
3054        await self._handle_set_members_with_protocols(
3055            parent_player, final_player_ids_to_add, final_player_ids_to_remove
3056        )
3057
3058        if should_stop:
3059            # Stop playback on the player if it is being removed from itself
3060            await self._handle_cmd_stop(parent_player.player_id)
3061
3062    async def _handle_set_members_with_protocols(
3063        self,
3064        parent_player: Player,
3065        player_ids_to_add: list[str],
3066        player_ids_to_remove: list[str],
3067    ) -> None:
3068        """
3069        Handle set_members considering protocol and native members.
3070
3071        Skips permission checks, locking, and all redirect logic (internal use only).
3072        Translates visible player IDs to protocol player IDs when appropriate,
3073        and forwards to the correct player's set_members.
3074
3075        :param parent_player: The parent player to add/remove members to/from.
3076        :param player_ids_to_add: List of visible player IDs to add as members.
3077        :param player_ids_to_remove: List of visible player IDs to remove from members.
3078        """
3079        # Get parent's active protocol domain and player if available
3080        parent_protocol_domain = None
3081        parent_protocol_player = None
3082        if (
3083            parent_player.active_output_protocol
3084            and parent_player.active_output_protocol != "native"
3085        ):
3086            parent_protocol_player = self.get_player(parent_player.active_output_protocol)
3087            if parent_protocol_player:
3088                parent_protocol_domain = parent_protocol_player.provider.domain
3089
3090        self.logger.debug(
3091            "set_members on %s: active_protocol=%s, adding=%s, removing=%s",
3092            parent_player.state.name,
3093            parent_protocol_domain or "none",
3094            player_ids_to_add,
3095            player_ids_to_remove,
3096        )
3097
3098        # Translate members to add
3099        (
3100            protocol_members_to_add,
3101            native_members_to_add,
3102            parent_protocol_player,
3103            parent_protocol_domain,
3104        ) = self._translate_members_for_protocols(
3105            parent_player, player_ids_to_add, parent_protocol_player, parent_protocol_domain
3106        )
3107
3108        self.logger.debug(
3109            "Translated members: protocol=%s (domain=%s), native=%s",
3110            protocol_members_to_add,
3111            parent_protocol_domain,
3112            native_members_to_add,
3113        )
3114
3115        # Translate members to remove
3116        protocol_members_to_remove, native_members_to_remove = (
3117            self._translate_members_to_remove_for_protocols(
3118                parent_player, player_ids_to_remove, parent_protocol_player, parent_protocol_domain
3119            )
3120        )
3121
3122        # Forward protocol members to protocol player's set_members
3123        if (protocol_members_to_add or protocol_members_to_remove) and parent_protocol_player:
3124            await self._forward_protocol_set_members(
3125                parent_player,
3126                parent_protocol_player,
3127                protocol_members_to_add,
3128                protocol_members_to_remove,
3129            )
3130
3131        # Forward native members to parent player's set_members
3132        if native_members_to_add or native_members_to_remove:
3133            filtered_native_add = self._filter_native_members(native_members_to_add, parent_player)
3134            # For removal, allow protocol players if they're actually in the parent's group_members
3135            # This handles native protocol players (e.g., native AirPlay) where group_members
3136            # contains protocol player IDs
3137            filtered_native_remove = [
3138                pid
3139                for pid in native_members_to_remove
3140                if (p := self.get_player(pid))
3141                and (p.type != PlayerType.PROTOCOL or pid in parent_player.group_members)
3142            ]
3143            self.logger.debug(
3144                "Native grouping on %s: filtered_add=%s, filtered_remove=%s",
3145                parent_player.state.name,
3146                filtered_native_add,
3147                filtered_native_remove,
3148            )
3149            if filtered_native_add or filtered_native_remove:
3150                if PlayerFeature.SET_MEMBERS not in parent_player.state.supported_features:
3151                    return
3152                self.logger.info(
3153                    "Calling set_members on native player %s with add=%s, remove=%s",
3154                    parent_player.state.name,
3155                    filtered_native_add,
3156                    filtered_native_remove,
3157                )
3158                await parent_player.set_members(
3159                    player_ids_to_add=filtered_native_add or None,
3160                    player_ids_to_remove=filtered_native_remove or None,
3161                )
3162
3163    async def _transfer_ad_hoc_leadership(
3164        self, leader: Player, remaining_members: list[str]
3165    ) -> None:
3166        """
3167        Transfer leadership of an ad-hoc sync group to a remaining member.
3168
3169        Called when the sync leader of an ad-hoc group is unjoined while other
3170        members remain and playback is active. The queue is moved to a newly
3171        selected leader, the remaining members are regrouped under it and playback
3172        resumes at the saved position (accepting a brief audio gap).
3173
3174        :param leader: The current sync leader being removed from the group.
3175        :param remaining_members: Available group members (excluding the leader)
3176            that should keep playing under a new leader.
3177        """
3178        active_queue = self.get_active_queue(leader)
3179        was_playing = active_queue is not None and active_queue.state == PlaybackState.PLAYING
3180        new_leader_id = self._select_ad_hoc_leader(leader, remaining_members)
3181        self.logger.info(
3182            "Transferring leadership of %s to %s (%s remaining member(s))",
3183            leader.name,
3184            new_leader_id,
3185            len(remaining_members),
3186        )
3187        # Move the queue to the new leader. transfer_queue frees the new leader from
3188        # the old leader's group and stops the old leader; the playback position
3189        # survives because stop() stores it in resume_pos.
3190        await self.mass.player_queues.transfer_queue(
3191            leader.player_id, new_leader_id, auto_play=False
3192        )
3193        # regroup the other remaining members under the new leader
3194        other_members = [m for m in remaining_members if m != new_leader_id]
3195        if other_members:
3196            await self.cmd_set_members(new_leader_id, player_ids_to_add=other_members)
3197        if was_playing:
3198            await self.mass.player_queues.resume(new_leader_id)
3199
3200    def _select_ad_hoc_leader(self, leader: Player, remaining_members: list[str]) -> str:
3201        """
3202        Pick the new leader for an ad-hoc sync group leadership transfer.
3203
3204        Prefers a remaining member that supports the protocol the group is currently
3205        playing on, so the other members can be regrouped under it; falls back to the
3206        first remaining member. The members' own ``can_group_with`` is unusable here
3207        because it is empty while they are still synced to the old leader.
3208
3209        :param leader: The current sync leader being removed.
3210        :param remaining_members: Candidate member player_ids, already filtered for
3211            availability. Must not be empty.
3212        """
3213        active_domain: str | None = None
3214        if leader.active_output_protocol and leader.active_output_protocol != "native":
3215            if protocol_player := self.get_player(leader.active_output_protocol):
3216                active_domain = protocol_player.provider.domain
3217        if active_domain:
3218            for member_id in remaining_members:
3219                member = self.get_player(member_id)
3220                if member is None:
3221                    continue
3222                if member.provider.domain == active_domain or any(
3223                    protocol.protocol_domain == active_domain and protocol.available
3224                    for protocol in member.linked_output_protocols
3225                ):
3226                    return member_id
3227        return remaining_members[0]
3228
3229    def _clear_sleep_timer(self, player: Player) -> None:
3230        """
3231        Clear the active sleep timer for the player.
3232
3233        :param player: Player to clear the timer for.
3234        """
3235        self.mass.cancel_timer(self._sleep_timer_task_id(player.player_id))
3236        if player.sleep_timer_expires_at is not None:
3237            player.set_sleep_timer_expires_at(None)
3238            player.update_state()
3239            self._signal_sleep_timer_updated(player, None)
3240
3241    async def _handle_sleep_timer_expired(self, player_id: str) -> None:
3242        """
3243        Stop playback when a player's sleep timer expires.
3244
3245        :param player_id: Player ID whose sleep timer expired.
3246        """
3247        player = self.get_player(player_id)
3248        if player is None or player.sleep_timer_expires_at is None:
3249            return
3250        player.set_sleep_timer_expires_at(None)
3251        player.update_state()
3252        self._signal_sleep_timer_updated(player, None)
3253        await self.cmd_stop(player_id)
3254
3255    def _signal_sleep_timer_updated(self, player: Player, expires_at: float | None) -> None:
3256        """
3257        Signal a sleep timer change for the player on the event bus.
3258
3259        :param player: Player whose sleep timer changed.
3260        :param expires_at: New expiry timestamp, or None when the timer was cleared.
3261        """
3262        if player.state.type == PlayerType.PROTOCOL:
3263            return
3264        self.mass.signal_event(
3265            EventType.PLAYER_SLEEP_TIMER_UPDATED,
3266            object_id=player.player_id,
3267            data=expires_at,
3268        )
3269
3270    @staticmethod
3271    def _sleep_timer_task_id(player_id: str) -> str:
3272        """
3273        Return the scheduled task ID for a player's sleep timer.
3274
3275        :param player_id: Player ID to build the task ID for.
3276        """
3277        return f"player_sleep_timer_{player_id}"
3278
3279    # Private command handlers (no permission checks)
3280
3281    async def _handle_cmd_resume(
3282        self, player_id: str, source: str | None = None, media: PlayerMedia | None = None
3283    ) -> None:
3284        """
3285        Handle resume playback command.
3286
3287        Skips permission checks and locking (internal use only).
3288        """
3289        player = self._get_player_with_redirect(player_id)
3290        source = source or player.state.active_source
3291        media = media or player.state.current_media
3292        # power on the player if needed
3293        if not player.state.powered and player.state.power_control != PLAYER_CONTROL_NONE:
3294            await self._handle_cmd_power(player.player_id, True)
3295        # Redirect to queue controller if it is active
3296        if active_queue := self.mass.player_queues.get(source or player_id):
3297            await self.mass.player_queues.resume(active_queue.queue_id)
3298            return
3299        # try to handle command on player directly
3300        # TODO: check if player has an active source with native resume support
3301        active_source = next((x for x in player.state.source_list if x.id == source), None)
3302        if (
3303            player.state.playback_state in (PlaybackState.IDLE, PlaybackState.PAUSED)
3304            and active_source
3305            and active_source.can_play_pause
3306            and PlayerFeature.PAUSE in player.state.supported_features
3307        ):
3308            # player has some other source active and native resume support
3309            await player.play()
3310            return
3311        if active_source and not active_source.passive:
3312            await self.select_source(player_id, active_source.id)
3313            return
3314        if media:
3315            # try to re-play the current media item
3316            await player.play_media(media)
3317            return
3318        # fallback: just try to resume queue playback
3319        await self.mass.player_queues.resume(player.player_id)
3320
3321    async def _handle_cmd_power(
3322        self, player_id: str, powered: bool, skip_auto_play: bool = False
3323    ) -> None:
3324        """
3325        Handle player power on/off command.
3326
3327        Skips permission checks and locking (internal use only).
3328
3329        :param player_id: The player ID to power on/off.
3330        :param powered: True to power on, False to power off.
3331        :param skip_auto_play: If True, skip auto-play on power on.
3332        """
3333        player = self.get_player(player_id, True)
3334        assert player is not None  # for type checking
3335        player_state = player.state
3336
3337        if player_state.powered == powered:
3338            self.logger.debug(
3339                "Ignoring power %s command for player %s: already in state %s",
3340                "ON" if powered else "OFF",
3341                player_state.name,
3342                "ON" if player_state.powered else "OFF",
3343            )
3344            return  # nothing to do
3345
3346        # ungroup player at power off
3347        player_was_sync_child = bool(player.state.synced_to or player.state.active_group)
3348        if (
3349            (player_was_sync_child or player.group_members)
3350            and player.type == PlayerType.PLAYER
3351            and not powered
3352        ):
3353            # ungroup player if it is synced (or is a sync leader itself)
3354            await self.cmd_ungroup(player_id)
3355
3356        # always stop player at power off
3357        if (
3358            not powered
3359            and not player_was_sync_child
3360            and player_state.playback_state in (PlaybackState.PLAYING, PlaybackState.PAUSED)
3361        ):
3362            # wait for the stop command to process and prevent race conditions
3363            async with self.wait_for_player_update(player_id, timeout=5):
3364                await self._handle_cmd_stop(player_id)
3365
3366        # power off all synced childs when player is a sync leader
3367        elif not powered and player_state.type == PlayerType.PLAYER and player_state.group_members:
3368            async with TaskManager(self.mass) as tg:
3369                for member in self.iter_group_members(player, True):
3370                    if member.power_control == PLAYER_CONTROL_NONE:
3371                        continue
3372                    tg.create_task(self._handle_cmd_power(member.player_id, False))
3373
3374        # handle actual power command
3375        if player_state.power_control == PLAYER_CONTROL_NONE:
3376            self.logger.debug(
3377                "Player %s does not support power control, ignoring power command",
3378                player_state.name,
3379            )
3380            return
3381        if player_state.power_control == PLAYER_CONTROL_NATIVE:
3382            # player supports power command natively: forward to player provider
3383            await player.power(powered)
3384            if powered:
3385                await wait_for_power_on(self.logger, player)
3386        elif player_state.power_control == PLAYER_CONTROL_FAKE:
3387            # user wants to use fake power control - so we (optimistically) update the state
3388            # and store the state in the cache
3389            player.extra_data[ATTR_FAKE_POWER] = powered
3390            # Group players need to actively form/dissolve their session when the
3391            # user toggles fake power — otherwise the toggle would only update the
3392            # cosmetic state without ever capturing or releasing the members.
3393            if player_state.type == PlayerType.GROUP:
3394                await player.power(powered)
3395            player.update_state()  # trigger update of the player state
3396            if player_state.type != PlayerType.GROUP:
3397                # see register(): group fake-power is intentionally not persisted
3398                # because there is no session to restore at boot.
3399                await self.mass.cache.set(
3400                    key=player_id,
3401                    data=powered,
3402                    provider=self.domain,
3403                    category=CACHE_CATEGORY_PLAYER_POWER,
3404                )
3405        # handle external player control
3406        elif player_control := self._controls.get(player.state.power_control):
3407            control_name = player_control.name
3408            self.logger.debug("Redirecting power command to PlayerControl %s", control_name)
3409            if not player_control.supports_power:
3410                raise UnsupportedFeaturedException(
3411                    f"Player control {control_name} is not available"
3412                )
3413            if powered:
3414                assert player_control.power_on is not None  # for type checking
3415                await player_control.power_on()
3416                await wait_for_power_on(self.logger, player, player_control)
3417            else:
3418                assert player_control.power_off is not None  # for type checking
3419                await player_control.power_off()
3420        # always trigger a state update to update the UI
3421        player.refresh_state()
3422
3423        # handle 'auto play on power on' feature
3424        if (
3425            not skip_auto_play
3426            and not player_state.active_group
3427            and not player_state.synced_to
3428            and powered
3429            and player.config.get_value(CONF_AUTO_PLAY)
3430            and player_state.active_source in (None, player_id)
3431            and not player.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS)
3432        ):
3433            await self.mass.player_queues.resume(player_id)
3434
3435    async def _handle_cmd_volume_set(self, player_id: str, volume_level: int) -> None:
3436        """
3437        Handle Player volume set command.
3438
3439        Skips permission checks and locking (internal use only).
3440
3441        :param player_id: player_id of the player to handle the command.
3442        :param volume_level: logical volume level (0..100) to set on the player.
3443        """
3444        player = self.get_player(player_id, True)
3445        assert player is not None  # for type checker
3446
3447        # Clamp logical volume to 0-100
3448        volume_level = max(0, min(100, volume_level))
3449
3450        if player.type == PlayerType.GROUP:
3451            # redirect to special group volume control
3452            await self.cmd_group_volume(player_id, volume_level)
3453            return
3454
3455        # Check if player has mute lock (set when individually muted in a group)
3456        # If locked, don't auto-unmute when volume changes
3457        has_mute_lock = self._has_active_mute_lock(player)
3458        if (
3459            not has_mute_lock
3460            # the live value is what cmd_volume_mute checks, so a control change that
3461            # has not reached the player state yet may not send us into a failing unmute
3462            and player.mute_control not in (PLAYER_CONTROL_NONE, PLAYER_CONTROL_FAKE)
3463            and player.state.volume_muted
3464        ):
3465            # if player is muted and not locked, we unmute it first
3466            # skip this for fake mute since it uses volume to simulate mute
3467            self.logger.debug(
3468                "Unmuting player %s before setting volume",
3469                player.state.name,
3470            )
3471            await self.cmd_volume_mute(player_id, False)
3472
3473        if (
3474            has_mute_lock
3475            and player.mute_control == PLAYER_CONTROL_FAKE
3476            and player.extra_data.get(ATTR_FAKE_MUTE)
3477        ):
3478            # a locked player stays silent, the volume it holds is the one
3479            # that gets restored once it is unmuted again
3480            volume_level = 0
3481        else:
3482            # always reset fake mute when controlling volume
3483            player.extra_data.pop(ATTR_FAKE_MUTE, None)
3484
3485        # Scale logical volume (0-100) to device volume (min_volume-max_volume)
3486        device_volume = self.scale_volume_to_device(player_id, volume_level)
3487
3488        # Notify the active AudioSource of a volume change.
3489        # Only fire if this player is the direct owner of its queue,
3490        # not when it merely inherits active_source from a parent group —
3491        # group volume changes handle the callback once at the group level.
3492        if active := self._get_active_audio_source(player):
3493            audio_source, plugin_prov = active
3494            active_queue = self.get_active_queue(player)
3495            if active_queue is not None and active_queue.queue_id == player.player_id:
3496                await plugin_prov.on_volume_change(audio_source.item_id, volume_level)
3497
3498        # Handle native volume control support
3499        if player.volume_control == PLAYER_CONTROL_NATIVE:
3500            # player supports volume command natively: forward to player
3501            await player.volume_set(device_volume)
3502            return
3503        # Handle fake volume control support
3504        if player.volume_control == PLAYER_CONTROL_FAKE:
3505            # user wants to use fake volume control - so we (optimistically) update the state
3506            # and store the state in the cache. Fake volume uses the logical volume (no scaling).
3507            player.extra_data[ATTR_FAKE_VOLUME] = volume_level
3508            player.update_state()
3509            return
3510        # player has no volume support at all
3511        if player.volume_control == PLAYER_CONTROL_NONE:
3512            raise UnsupportedFeaturedException(
3513                f"Player {player.state.name} does not support volume control"
3514            )
3515        # handle external player control
3516        if player_control := self._controls.get(player.state.volume_control):
3517            control_name = player_control.name
3518            self.logger.debug("Redirecting volume command to PlayerControl %s", control_name)
3519            if not player_control.supports_volume:
3520                raise UnsupportedFeaturedException(
3521                    f"Player control {control_name} is not available"
3522                )
3523            assert player_control.volume_set is not None
3524            # forward the already-scaled device volume; the external control sets the
3525            # raw device volume and does not apply min/max scaling of its own
3526            await player_control.volume_set(device_volume)
3527            return
3528        if protocol_player := self.get_player(player.state.volume_control):
3529            # redirect to protocol player volume control.
3530            # forward the already-scaled device volume so the min/max limits configured
3531            # on this (user-facing) player are honored; the protocol player has no
3532            # limits of its own, so its scaling is an identity pass-through.
3533            self.logger.debug(
3534                "Redirecting volume command to protocol player %s",
3535                protocol_player.provider.manifest.name,
3536            )
3537            await self._handle_cmd_volume_set(protocol_player.player_id, device_volume)
3538            return
3539
3540    @staticmethod
3541    def _is_in_group(state: PlayerState) -> bool:
3542        """Check if the player with the given state is currently grouped with other players."""
3543        # a sync leader has neither synced_to nor active_group set, but it does lead its
3544        # own group_members, which stays empty for a player that is not grouped at all
3545        return bool(state.synced_to or state.active_group or state.group_members)
3546
3547    def _has_active_mute_lock(self, player: Player) -> bool:
3548        """
3549        Check if the given player holds a mute lock that still applies to it.
3550
3551        A lock is only earned inside a group and only holds for as long as the player
3552        is still grouped, so it can not outlive the group it was earned in.
3553
3554        :param player: The player to check, which may be a protocol player.
3555        """
3556        if player.extra_data.get(ATTR_MUTE_LOCK) and self._is_in_group(player.state):
3557            return True
3558        # cmd_volume_mute stores the lock on the parent player, while the volume command
3559        # may arrive with the protocol player ID (e.g. during group volume changes)
3560        if player.protocol_parent_id and (parent := self.get_player(player.protocol_parent_id)):
3561            return bool(parent.extra_data.get(ATTR_MUTE_LOCK)) and self._is_in_group(parent.state)
3562        return False
3563
3564    async def _mute_group_members(self, group_player: Player, muted: bool) -> None:
3565        """
3566        Mute or unmute all mute capable members of a player group or synced players.
3567
3568        :param group_player: The group player or sync leader.
3569        :param muted: bool if the group should be muted.
3570        """
3571        coros = []
3572        for child_player in self.iter_group_members(
3573            group_player, only_powered=True, exclude_self=False
3574        ):
3575            if child_player.mute_control == PLAYER_CONTROL_NONE:
3576                # members without a mute control are left alone, just like the
3577                # group mute state itself is calculated from the capable members only
3578                continue
3579            coros.append(self.cmd_volume_mute(child_player.player_id, muted))
3580        await asyncio.gather(*coros)
3581
3582    async def _handle_cmd_volume_mute(self, player: Player, mute_control: str, muted: bool) -> None:
3583        """
3584        Send the mute command to the given player's mute control.
3585
3586        Skips permission checks, locking and mute lock bookkeeping (internal use only).
3587
3588        :param player: the player to handle the command.
3589        :param mute_control: the already resolved mute control of the player.
3590        :param muted: bool if player should be muted.
3591        """
3592        if mute_control == PLAYER_CONTROL_NATIVE:
3593            # player supports mute command natively: forward to player
3594            await player.volume_mute(muted)
3595            return
3596        if mute_control == PLAYER_CONTROL_FAKE:
3597            # user wants to use fake mute control - so we use volume instead
3598            self.logger.debug(
3599                "Using volume for muting for player %s",
3600                player.state.name,
3601            )
3602            if muted:
3603                already_muted = bool(player.extra_data.get(ATTR_FAKE_MUTE))
3604                if not already_muted:
3605                    # on a repeated mute command the volume is already 0
3606                    player.extra_data[ATTR_PREVIOUS_VOLUME] = player.state.volume_level
3607                await self._handle_cmd_volume_set(player.player_id, 0)
3608                # set the flag after the volume command, as that clears it
3609                player.extra_data[ATTR_FAKE_MUTE] = True
3610                player.update_state()
3611            else:
3612                prev_volume = player.extra_data.get(ATTR_PREVIOUS_VOLUME, 1)
3613                player.extra_data[ATTR_FAKE_MUTE] = False
3614                player.update_state()
3615                await self._handle_cmd_volume_set(player.player_id, prev_volume)
3616            return
3617
3618        # handle external player control
3619        if player_control := self._controls.get(mute_control):
3620            control_name = player_control.name
3621            self.logger.debug("Redirecting mute command to PlayerControl %s", control_name)
3622            if not player_control.supports_mute:
3623                raise UnsupportedFeaturedException(
3624                    f"Player control {control_name} is not available"
3625                )
3626            assert player_control.mute_set is not None
3627            await player_control.mute_set(muted)
3628            return
3629
3630        # handle to protocol player as volume_mute control
3631        if protocol_player := self.get_player(mute_control):
3632            self.logger.debug(
3633                "Redirecting mute command to protocol player %s",
3634                protocol_player.provider.manifest.name,
3635            )
3636            await protocol_player.volume_mute(muted)
3637            return
3638
3639        # the configured control disappeared after the mute control was resolved
3640        raise UnsupportedFeaturedException(f"Player {player.state.name} does not support muting")
3641
3642    async def _handle_play_media(self, player_id: str, media: PlayerMedia) -> None:
3643        """
3644        Handle play media command without group redirect.
3645
3646        Skips permission checks, locking, and all redirect logic (internal use only).
3647
3648        :param player_id: player_id of the player to handle the command.
3649        :param media: The Media that needs to be played on the player.
3650        """
3651        player = self.get_player(player_id, raise_unavailable=True)
3652        assert player is not None
3653        # set active source if media has a source_id (e.g. plugin source or mass queue source)
3654        if media.source_id:
3655            player.set_active_mass_source(media.source_id)
3656
3657        # Determine output protocol to use:
3658        # While a session is active (playing/paused), keep using the already active
3659        # protocol so mid-session commands stay on the same output.
3660        # On a fresh start always (re)select: a leftover active protocol from a
3661        # previous session must not overrule user preference, a grouped protocol
3662        # or native playback (and it may point at a player that is gone by now).
3663        target_player: Player | None = None
3664        output_protocol: OutputProtocol | None = None
3665        if (
3666            player.state.playback_state in (PlaybackState.PLAYING, PlaybackState.PAUSED)
3667            and player.active_output_protocol
3668            and player.active_output_protocol != "native"
3669            and (protocol_player := self.get_player(player.active_output_protocol))
3670        ):
3671            # Use the already-set protocol directly
3672            output_protocol = next(
3673                (
3674                    p
3675                    for p in player.linked_output_protocols
3676                    if p.output_protocol_id == player.active_output_protocol
3677                ),
3678                None,
3679            )
3680            if output_protocol is not None:
3681                target_player = protocol_player
3682        if target_player is None:
3683            target_player, output_protocol = self._select_best_output_protocol(player)
3684
3685        if target_player.player_id != player.player_id:
3686            # Playing via linked protocol - update active output protocol
3687            # output_protocol is guaranteed to be non-None when target_player != player
3688            assert output_protocol is not None
3689            self.logger.debug(
3690                "Starting playback on %s via protocol %s (target=%s), group_members=%s",
3691                player.state.name,
3692                output_protocol.name,
3693                target_player.display_name,
3694                target_player.state.group_members,
3695            )
3696            player.set_active_output_protocol(output_protocol.output_protocol_id)
3697        elif player.type != PlayerType.GROUP:
3698            # Native playback - group players don't have output protocols of their own
3699            # (they delegate to a sync leader / member which manages its own protocol)
3700            self.logger.debug(
3701                "Starting playback on %s via native, group_members=%s",
3702                player.state.name,
3703                player.state.group_members,
3704            )
3705            player.set_active_output_protocol("native")
3706
3707        # power on the player if needed (skip auto-play since we're about to start playback)
3708        if not player.state.powered and player.state.power_control != PLAYER_CONTROL_NONE:
3709            await self._handle_cmd_power(player.player_id, True, skip_auto_play=True)
3710        await target_player.play_media(media)
3711        if target_player.player_id != player.player_id:
3712            # notify the native player that protocol playback started
3713            assert output_protocol is not None
3714            await player.on_protocol_playback(output_protocol=output_protocol)
3715
3716    async def _handle_enqueue_next_media(self, player_id: str, media: PlayerMedia) -> None:
3717        """
3718        Handle enqueue next media command without group redirect.
3719
3720        Skips permission checks, locking, and all redirect logic (internal use only).
3721
3722        :param player_id: player_id of the player to handle the command.
3723        :param media: The Media that needs to be enqueued on the player.
3724        """
3725        player = self.get_player(player_id, raise_unavailable=True)
3726        assert player is not None
3727        if target_player := self._get_control_target(
3728            player,
3729            required_feature=PlayerFeature.ENQUEUE,
3730            require_active=True,
3731        ):
3732            self.logger.debug(
3733                "Redirecting enqueue command to protocol player %s",
3734                target_player.provider.manifest.name,
3735            )
3736            await target_player.enqueue_next_media(media)
3737            return
3738
3739        if PlayerFeature.ENQUEUE not in player.state.supported_features:
3740            raise UnsupportedFeaturedException(
3741                f"Player {player.state.name} does not support enqueueing"
3742            )
3743        await player.enqueue_next_media(media)
3744
3745    async def _handle_select_source(self, player_id: str, source: str | None) -> None:
3746        """
3747        Handle select source command without group redirect.
3748
3749        Skips permission checks, locking, and all redirect logic (internal use only).
3750
3751        :param player_id: player_id of the player to handle the command.
3752        :param source: The ID of the source that needs to be activated/selected.
3753        """
3754        if source is None:
3755            source = player_id  # default to MA queue source
3756        player = self.get_player(player_id, True)
3757        assert player is not None
3758        # check if player is already playing and source is different
3759        # in that case we need to stop the player first
3760        prev_source = player.state.active_source
3761        if prev_source and source != prev_source:
3762            with suppress(PlayerCommandFailed, RuntimeError):
3763                # just try to stop (regardless of state)
3764                async with self.wait_for_player_update(player_id, timeout=5):
3765                    await self._handle_cmd_stop(player_id)
3766        # check if source is a mass queue
3767        # this can be used to restore the queue after a source switch
3768        if self.mass.player_queues.get(source):
3769            player.set_active_mass_source(source)
3770            return
3771        # Legacy compatibility: the old plugin-source API used the
3772        # plugin's instance_id directly as the source string. The refactor
3773        # moved plugin sources to first-class AudioSource MediaItems played
3774        # via player_queues.play_media. Translate a legacy plugin-instance-id
3775        # source into the new flow so old frontends, third-party scripts,
3776        # and HA automations keep working — but only when the provider
3777        # exposes EXACTLY ONE AudioSource (it was always a 1:1 mapping under
3778        # the old API; multi-source providers have to use the explicit URI).
3779        if (legacy_prov := self.mass.get_provider(source)) and isinstance(
3780            legacy_prov, PluginProvider
3781        ):
3782            if ProviderFeature.AUDIO_SOURCE not in legacy_prov.supported_features:
3783                raise PlayerCommandFailed(f"Provider {source} does not expose AudioSources")
3784            sources = await legacy_prov.get_audio_sources()
3785            if len(sources) == 1:
3786                self.logger.debug(
3787                    "Translating legacy select_source(%s) to play_media(%s)",
3788                    source,
3789                    sources[0].uri,
3790                )
3791                await self.mass.player_queues.play_media(player_id, str(sources[0].uri))
3792                return
3793            raise UnsupportedFeaturedException(
3794                f"Provider {source} exposes {len(sources)} AudioSources; the legacy "
3795                "select_source(plugin_instance_id) API only supported 1:1 mappings. "
3796                "Use player_queues.play_media with an explicit AudioSource URI."
3797            )
3798        # basic check if player supports source selection
3799        if PlayerFeature.SELECT_SOURCE not in player.state.supported_features:
3800            raise UnsupportedFeaturedException(
3801                f"Player {player.state.name} does not support source selection"
3802            )
3803        # basic check if source is valid for player
3804        if not any(x for x in player.state.source_list if x.id == source):
3805            raise PlayerCommandFailed(
3806                f"{source} is an invalid source for player {player.state.name}"
3807            )
3808        # forward to player
3809        await player.select_source(source)
3810
3811    async def _handle_cmd_stop(self, player_id: str) -> None:
3812        """
3813        Handle stop command without any redirects.
3814
3815        Skips permission checks, locking, and all redirect logic (internal use only).
3816
3817        :param player_id: player_id of the player to handle the command.
3818        """
3819        player = self.get_player(player_id, raise_unavailable=True)
3820        assert player is not None
3821        protocol_player: Player | None = None
3822        if player.active_output_protocol and player.active_output_protocol != "native":
3823            protocol_player = self.get_player(player.active_output_protocol)
3824        if player.state.playback_state == PlaybackState.IDLE:
3825            # The player already reports idle but an output protocol is still marked
3826            # active: the protocol player may never have received a stop at all
3827            # (e.g. the source stream ended on its own before this stop command
3828            # arrived). Forward an (idempotent) stop and schedule the protocol clear
3829            # so no stale session lingers on the device and the next playback
3830            # (re)selects the output protocol.
3831            if protocol_player is not None:
3832                await protocol_player.stop()
3833                if len(protocol_player.group_members) <= 1:
3834                    self.schedule_active_output_protocol_clear(player)
3835            return
3836        player.mark_stop_called()
3837        # Delegate to active protocol player if one is active
3838        target_player = player
3839        if protocol_player is not None:
3840            target_player = protocol_player
3841            if PlayerFeature.POWER in target_player.supported_features:
3842                # if protocol player supports/requires power,
3843                # we power it off instead of just stopping (which also stops playback)
3844                # this is rare as most protocols do not support power control (except for cast)
3845                await self._handle_cmd_power(target_player.player_id, False)
3846                return
3847
3848        # handle command on player(protocol) directly
3849        await target_player.stop()
3850        # Only clear active protocol if the protocol player has no remaining group members.
3851        # If there are still protocol group members, keep the protocol active so that
3852        # when playback resumes it continues on the same protocol.
3853        if target_player.player_id == player.player_id or len(target_player.group_members) <= 1:
3854            self.schedule_active_output_protocol_clear(player)
3855
3856    async def _handle_cmd_play(self, player_id: str) -> None:
3857        """
3858        Handle play command without group redirect.
3859
3860        Skips permission checks, locking, and all redirect logic (internal use only).
3861
3862        :param player_id: player_id of the player to handle the command.
3863        """
3864        player = self.get_player(player_id, raise_unavailable=True)
3865        assert player is not None
3866        if player.state.playback_state == PlaybackState.PLAYING:
3867            self.logger.info(
3868                "Ignore PLAY request to player %s: player is already playing", player.state.name
3869            )
3870            return
3871        # If an AudioSource is the active queue item, proxy play to the plugin
3872        if active := self._get_active_audio_source(player):
3873            audio_source, plugin_prov = active
3874            if audio_source.can_play_pause:
3875                await plugin_prov.on_source_control(audio_source.item_id, SourceControl.PLAY)
3876                return
3877        # handle unpause (=play if player is paused)
3878        if player.state.playback_state == PlaybackState.PAUSED:
3879            active_source = next(
3880                (x for x in player.state.source_list if x.id == player.state.active_source), None
3881            )
3882            # raise if active source does not support play/pause
3883            if active_source and not active_source.can_play_pause:
3884                msg = (
3885                    f"The active source ({active_source.name}) on player "
3886                    f"{player.state.name} does not support play/pause"
3887                )
3888                raise PlayerCommandFailed(msg)
3889            # Delegate to active protocol player if one is active
3890            if target_player := self._get_control_target(
3891                player, PlayerFeature.PAUSE, require_active=True
3892            ):
3893                await target_player.play()
3894                return
3895            # No active protocol target: if the player rendering the audio supports pause and
3896            # the active (external) source can be paused, unpause it directly instead of
3897            # restarting the source.
3898            output_player = player.resolve_output_player()
3899            if (
3900                active_source
3901                and active_source.can_play_pause
3902                and PlayerFeature.PAUSE in output_player.supported_features
3903            ):
3904                await output_player.play()
3905                return
3906
3907        # player is not paused: try to resume the player
3908        # Note: We handle resume inline here without calling _handle_cmd_resume
3909        active_source = next(
3910            (x for x in player.state.source_list if x.id == player.state.active_source), None
3911        )
3912        media = player.state.current_media
3913        # power on the player if needed
3914        if not player.state.powered and player.state.power_control != PLAYER_CONTROL_NONE:
3915            await self._handle_cmd_power(player.player_id, True)
3916        if active_source and not active_source.passive:
3917            await self._handle_select_source(player_id, active_source.id)
3918            return
3919        if media:
3920            # try to re-play the current media item
3921            await player.play_media(media)
3922            return
3923        # fallback: just send play command - which will fail if nothing can be played
3924        await player.play()
3925
3926    async def _handle_cmd_pause(self, player_id: str) -> None:
3927        """
3928        Handle pause command without any redirects.
3929
3930        Skips permission checks, locking, and all redirect logic (internal use only).
3931
3932        :param player_id: player_id of the player to handle the command.
3933        """
3934        player = self.get_player(player_id, raise_unavailable=True)
3935        assert player is not None
3936        if player.state.playback_state == PlaybackState.IDLE:
3937            return
3938        # If an AudioSource is the active queue item, proxy pause to the plugin
3939        if active := self._get_active_audio_source(player):
3940            audio_source, plugin_prov = active
3941            if audio_source.can_play_pause:
3942                await plugin_prov.on_source_control(audio_source.item_id, SourceControl.PAUSE)
3943                return
3944        # handle command on player/source directly
3945        active_source = next(
3946            (x for x in player.state.source_list if x.id == player.state.active_source), None
3947        )
3948        if active_source and not active_source.can_play_pause:
3949            # raise if active source does not support play/pause
3950            msg = (
3951                f"The active source ({active_source.name}) on player "
3952                f"{player.state.name} does not support play/pause"
3953            )
3954            raise PlayerCommandFailed(msg)
3955        # Delegate to active protocol player if one is active
3956        if target_player := self._get_control_target(
3957            player, PlayerFeature.PAUSE, require_active=True
3958        ):
3959            await target_player.pause()
3960            return
3961        # No active protocol target: if the player rendering the audio supports pause and the
3962        # active (external) source can be paused, forward the command to it instead of stopping
3963        # it (mirrors the external-source handling in cmd_seek/cmd_next_track).
3964        output_player = player.resolve_output_player()
3965        if (
3966            active_source
3967            and active_source.can_play_pause
3968            and PlayerFeature.PAUSE in output_player.supported_features
3969        ):
3970            await output_player.pause()
3971            return
3972        # player/protocol does not support pause: fall back to stop
3973        self.logger.debug(
3974            "Player/protocol %s does not support pause, using STOP instead",
3975            player.state.name,
3976        )
3977        await self._handle_cmd_stop(player.player_id)
3978