music-assistant-server

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