music-assistant-server

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