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