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