music-assistant-server

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