music-assistant-server

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