music-assistant-server

1"""
2Base class/model for a Player within Music Assistant.
3
4All providerspecific players should inherit from this class and implement the required methods.
5
6Note that this is NOT the final state of the player,
7as it may be overridden by (sync)group memberships, configuration options, or other factors.
8This final state will be calculated and snapshotted in the PlayerState dataclass,
9which is what is also what is sent over the API.
10The final active source can be retrieved by using the 'state' property.
11"""
12
13from __future__ import annotations
14
15import asyncio
16import time
17from abc import ABC
18from collections.abc import Callable
19from copy import deepcopy
20from typing import TYPE_CHECKING, Any, cast, final
21
22from music_assistant_models.constants import (
23    EXTRA_ATTRIBUTES_TYPES,
24    PLAYER_CONTROL_FAKE,
25    PLAYER_CONTROL_NATIVE,
26    PLAYER_CONTROL_NONE,
27)
28from music_assistant_models.enums import MediaType, PlaybackState, PlayerFeature, PlayerType
29from music_assistant_models.errors import UnsupportedFeaturedException
30from music_assistant_models.player import (
31    DeviceInfo,
32    OutputProtocol,
33    PlayerMedia,
34    PlayerOption,
35    PlayerOptionValueType,
36    PlayerSoundMode,
37    PlayerSource,
38)
39from music_assistant_models.player import Player as PlayerState
40from music_assistant_models.unique_list import UniqueList
41from propcache import under_cached_property as cached_property
42
43from music_assistant.constants import (
44    ACTIVE_PROTOCOL_FEATURES,
45    ATTR_ANNOUNCEMENT_IN_PROGRESS,
46    ATTR_FAKE_MUTE,
47    ATTR_FAKE_POWER,
48    ATTR_FAKE_VOLUME,
49    CONF_ENTRY_PLAYER_ICON,
50    CONF_EXPOSE_PLAYER_TO_HA,
51    CONF_FLOW_MODE,
52    CONF_HIDE_IN_UI,
53    CONF_LINKED_PROTOCOL_PLAYER_IDS,
54    CONF_MUTE_CONTROL,
55    CONF_PLAYERS,
56    CONF_POWER_CONTROL,
57    CONF_SMART_FADES_MODE,
58    CONF_VOLUME_CONTROL,
59    PROTOCOL_FEATURES,
60    PROTOCOL_PRIORITY,
61)
62from music_assistant.helpers.util import get_changed_dataclass_values
63
64if TYPE_CHECKING:
65    from music_assistant_models.config_entries import ConfigEntry, ConfigValueType, PlayerConfig
66    from music_assistant_models.player_queue import PlayerQueue
67
68    from .player_provider import PlayerProvider
69
70
71class Player(ABC):
72    """
73    Base representation of a Player within the Music Assistant Server.
74
75    Player Provider implementations should inherit from this base model.
76    """
77
78    _attr_type: PlayerType = PlayerType.PLAYER
79    _attr_supported_features: set[PlayerFeature]
80    _attr_group_members: list[str]
81    _attr_static_group_members: list[str]
82    _attr_device_info: DeviceInfo
83    _attr_can_group_with: set[str]
84    _attr_source_list: list[PlayerSource]
85    _attr_sound_mode_list: list[PlayerSoundMode]
86    _attr_options: list[PlayerOption]
87    _attr_available: bool = True
88    _attr_name: str | None = None
89    _attr_powered: bool | None = None
90    _attr_playback_state: PlaybackState = PlaybackState.IDLE
91    _attr_volume_level: int | None = None
92    _attr_volume_muted: bool | None = None
93    _attr_elapsed_time: float | None = None
94    _attr_elapsed_time_last_updated: float | None = None
95    _attr_active_source: str | None = None
96    _attr_active_sound_mode: str | None = None
97    _attr_current_media: PlayerMedia | None = None
98    _attr_needs_poll: bool = False
99    _attr_poll_interval: int = 30
100    _attr_hidden_by_default: bool = False
101    _attr_expose_to_ha_by_default: bool = True
102    _attr_enabled_by_default: bool = True
103
104    def __init__(self, provider: PlayerProvider, player_id: str) -> None:
105        """Initialize the Player."""
106        # set mass as public variable
107        self.mass = provider.mass
108        self.logger = provider.logger
109        # initialize mutable attributes
110        self._attr_supported_features = set()
111        self._attr_group_members = []
112        self._attr_static_group_members = []
113        self._attr_device_info = DeviceInfo()
114        self._attr_can_group_with = set()
115        self._attr_source_list = []
116        self._attr_sound_mode_list = []
117        self._attr_options = []
118        # do not override/overwrite these private attributes below!
119        self._cache: dict[str, Any] = {}  # storage dict for cached properties
120        self.__attr_linked_protocols: list[OutputProtocol] = []
121        self.__attr_protocol_parent_id: str | None = None
122        self.__attr_active_output_protocol: str | None = None
123        self._player_id = player_id
124        self._provider = provider
125        self.mass.config.create_default_player_config(
126            player_id, self.provider_id, self.type, self.name, self.enabled_by_default
127        )
128        self._config = self.mass.config.get_base_player_config(player_id, self.provider_id)
129        self._extra_data: dict[str, Any] = {}
130        self._extra_attributes: dict[str, Any] = {}
131        self._on_unload_callbacks: list[Callable[[], None]] = []
132        self.__active_mass_source: str | None = None
133        self.__initialized = asyncio.Event()
134        # The PlayerState is the (snapshotted) final state of the player
135        # after applying any config overrides and other transformations,
136        # such as the display name and player controls.
137        # the state is updated when calling 'update_state' and is what is sent over the API.
138        self._state = PlayerState(
139            player_id=self.player_id,
140            provider=self.provider_id,
141            type=self.type,
142            name=self.display_name,
143            available=self.available,
144            device_info=self.device_info,
145            supported_features=self.supported_features,
146            playback_state=self.playback_state,
147        )
148
149    @property
150    def available(self) -> bool:
151        """Return if the player is available."""
152        return self._attr_available
153
154    @property
155    def type(self) -> PlayerType:
156        """Return the type of the player."""
157        return self._attr_type
158
159    @property
160    def name(self) -> str | None:
161        """Return the name of the player."""
162        return self._attr_name
163
164    @property
165    def supported_features(self) -> set[PlayerFeature]:
166        """Return the supported features of the player."""
167        return self._attr_supported_features
168
169    @property
170    def playback_state(self) -> PlaybackState:
171        """Return the current playback state of the player."""
172        return self._attr_playback_state
173
174    @property
175    def requires_flow_mode(self) -> bool:
176        """
177        Return if the player needs flow mode.
178
179        Default implementation: True if the player does not support PlayerFeature.ENQUEUE
180        or has crossfade enabled without gapless support. Can be overridden by providers if needed.
181        """
182        if PlayerFeature.ENQUEUE not in self.supported_features:
183            # without enqueue support, flow mode is required
184            return True
185        return (
186            # player has crossfade enabled without gapless support - flow mode is required
187            PlayerFeature.GAPLESS_PLAYBACK not in self.supported_features
188            and str(self._config.get_value(CONF_SMART_FADES_MODE)) != "disabled"
189        )
190
191    @property
192    def device_info(self) -> DeviceInfo:
193        """Return the device info of the player."""
194        return self._attr_device_info
195
196    @property
197    def elapsed_time(self) -> float | None:
198        """Return the elapsed time in (fractional) seconds of the current track (if any)."""
199        return self._attr_elapsed_time
200
201    @property
202    def elapsed_time_last_updated(self) -> float | None:
203        """
204        Return when the elapsed time was last updated.
205
206        return: The (UTC) timestamp when the elapsed time was last updated,
207        or None if it was never updated (or unknown).
208        """
209        return self._attr_elapsed_time_last_updated
210
211    @property
212    def needs_poll(self) -> bool:
213        """Return if the player needs to be polled for state updates."""
214        return self._attr_needs_poll
215
216    @property
217    def poll_interval(self) -> int:
218        """
219        Return the (dynamic) poll interval for the player.
220
221        Only used if 'needs_poll' is set to True.
222        This should return the interval in seconds.
223        """
224        return self._attr_poll_interval
225
226    @property
227    def hidden_by_default(self) -> bool:
228        """Return if the player should be hidden in the UI by default."""
229        return self._attr_hidden_by_default
230
231    @property
232    def expose_to_ha_by_default(self) -> bool:
233        """Return if the player should be exposed to Home Assistant by default."""
234        return self._attr_expose_to_ha_by_default
235
236    @property
237    def enabled_by_default(self) -> bool:
238        """Return if the player should be enabled by default."""
239        return self._attr_enabled_by_default
240
241    @property
242    def static_group_members(self) -> list[str]:
243        """
244        Return the static group members for a player group.
245
246        For PlayerType.GROUP return the player_ids of members that must/can not be removed by
247        the user. For all other player types return an empty list.
248        """
249        return self._attr_static_group_members
250
251    @property
252    def powered(self) -> bool | None:
253        """
254        Return if the player is powered on.
255
256        If the player does not support PlayerFeature.POWER,
257        or the state is (currently) unknown, this property may return None.
258        """
259        return self._attr_powered
260
261    @property
262    def volume_level(self) -> int | None:
263        """
264        Return the current volume level (0..100) of the player.
265
266        If the player does not support PlayerFeature.VOLUME_SET,
267        or the state is (currently) unknown, this property may return None.
268        """
269        return self._attr_volume_level
270
271    @property
272    def volume_muted(self) -> bool | None:
273        """
274        Return the current mute state of the player.
275
276        If the player does not support PlayerFeature.VOLUME_MUTE,
277        or the state is (currently) unknown, this property may return None.
278        """
279        return self._attr_volume_muted
280
281    @property
282    def active_source(self) -> str | None:
283        """
284        Return the (id of) the active source of the player.
285
286        Only required if the player supports PlayerFeature.SELECT_SOURCE.
287
288        Set to None if the player is not currently playing a source or
289        the player_id if the player is currently playing a MA queue.
290        """
291        return self._attr_active_source
292
293    @property
294    def group_members(self) -> list[str]:
295        """
296        Return the group members of the player.
297
298        If there are other players synced/grouped with this player,
299        this should return the id's of players synced to this player,
300        and this should include the player's own id (as first item in the list).
301
302        If there are currently no group members, this should return an empty list.
303        """
304        return self._attr_group_members
305
306    @property
307    def can_group_with(self) -> set[str]:
308        """
309        Return the id's of players this player can group with.
310
311        This should return set of player_id's this player can group/sync with
312        or just the provider's instance_id if all players can group with each other.
313        """
314        return self._attr_can_group_with
315
316    @cached_property
317    def synced_to(self) -> str | None:
318        """Return the id of the player this player is synced to (sync leader)."""
319        # default implementation, feel free to override if your
320        # provider has a more efficient way to determine this
321        if self.group_members and self.group_members[0] != self.player_id:
322            return self.group_members[0]
323        for player in self.mass.players.all_players(
324            return_unavailable=False, return_protocol_players=True
325        ):
326            if player.type == PlayerType.GROUP:
327                continue
328            if self.player_id in player.group_members and player.player_id != self.player_id:
329                return player.player_id
330        return None
331
332    @property
333    def current_media(self) -> PlayerMedia | None:
334        """Return the current media being played by the player."""
335        return self._attr_current_media
336
337    @property
338    def source_list(self) -> list[PlayerSource]:
339        """Return list of available (native) sources for this player."""
340        return self._attr_source_list
341
342    @property
343    def active_sound_mode(self) -> str | None:
344        """Return active sound mode of this player."""
345        return self._attr_active_sound_mode
346
347    @cached_property
348    def sound_mode_list(self) -> UniqueList[PlayerSoundMode]:
349        """Return available PlayerSoundModes for Player."""
350        return UniqueList(self._attr_sound_mode_list)
351
352    @cached_property
353    def options(self) -> UniqueList[PlayerOption]:
354        """Return all PlayerOptions for Player."""
355        return UniqueList(self._attr_options)
356
357    async def power(self, powered: bool) -> None:
358        """
359        Handle POWER command on the player.
360
361        Will only be called if the PlayerFeature.POWER is supported.
362
363        :param powered: bool if player should be powered on or off.
364        """
365        raise NotImplementedError("power needs to be implemented when PlayerFeature.POWER is set")
366
367    async def volume_set(self, volume_level: int) -> None:
368        """
369        Handle VOLUME_SET command on the player.
370
371        Will only be called if the PlayerFeature.VOLUME_SET is supported.
372
373        :param volume_level: volume level (0..100) to set on the player.
374        """
375        raise NotImplementedError(
376            "volume_set needs to be implemented when PlayerFeature.VOLUME_SET is set"
377        )
378
379    async def volume_mute(self, muted: bool) -> None:
380        """
381        Handle VOLUME MUTE command on the player.
382
383        Will only be called if the PlayerFeature.VOLUME_MUTE is supported.
384
385        :param muted: bool if player should be muted.
386        """
387        raise NotImplementedError(
388            "volume_mute needs to be implemented when PlayerFeature.VOLUME_MUTE is set"
389        )
390
391    async def play(self) -> None:
392        """Handle PLAY command on the player."""
393        raise NotImplementedError("play needs to be implemented")
394
395    async def stop(self) -> None:
396        """
397        Handle STOP command on the player.
398
399        Will be called to stop the stream/playback if the player has play_media support.
400        """
401        raise NotImplementedError(
402            "stop needs to be implemented when PlayerFeature.PLAY_MEDIA is set"
403        )
404
405    async def pause(self) -> None:
406        """
407        Handle PAUSE command on the player.
408
409        Will only be called if the player reports PlayerFeature.PAUSE is supported.
410        """
411        raise NotImplementedError("pause needs to be implemented when PlayerFeature.PAUSE is set")
412
413    async def next_track(self) -> None:
414        """
415        Handle NEXT_TRACK command on the player.
416
417        Will only be called if the player reports PlayerFeature.NEXT_PREVIOUS
418        is supported and the player's currently selected source supports it.
419        """
420        raise NotImplementedError(
421            "next_track needs to be implemented when PlayerFeature.NEXT_PREVIOUS is set"
422        )
423
424    async def previous_track(self) -> None:
425        """
426        Handle PREVIOUS_TRACK command on the player.
427
428        Will only be called if the player reports PlayerFeature.NEXT_PREVIOUS
429        is supported and the player's currently selected source supports it.
430        """
431        raise NotImplementedError(
432            "previous_track needs to be implemented when PlayerFeature.NEXT_PREVIOUS is set"
433        )
434
435    async def seek(self, position: int) -> None:
436        """
437        Handle SEEK command on the player.
438
439        Seek to a specific position in the current track.
440        Will only be called if the player reports PlayerFeature.SEEK is
441        supported and the player is NOT currently playing a MA queue.
442
443        :param position: The position to seek to, in seconds.
444        """
445        raise NotImplementedError("seek needs to be implemented when PlayerFeature.SEEK is set")
446
447    async def play_media(
448        self,
449        media: PlayerMedia,
450    ) -> None:
451        """
452        Handle PLAY MEDIA command on given player.
453
454        This is called by the Player controller to start playing Media on the player,
455        which can be a MA queue item/stream or a native source.
456        The provider's own implementation should work out how to handle this request.
457
458        :param media: Details of the item that needs to be played on the player.
459        """
460        raise NotImplementedError(
461            "play_media needs to be implemented when PlayerFeature.PLAY_MEDIA is set"
462        )
463
464    async def on_protocol_playback(
465        self,
466        output_protocol: OutputProtocol,
467    ) -> None:
468        """
469        Handle callback when playback starts on a protocol output.
470
471        Called by the Player Controller after play_media is executed on a protocol player.
472        Allows the native player implementation to perform special logic when protocol
473        playback starts.
474
475        Optional - providers can override to implement protocol-specific logic.
476
477        :param output_protocol: The OutputProtocol object containing protocol details.
478        """
479        return  # Optional callback - no-op by default
480
481    async def enqueue_next_media(self, media: PlayerMedia) -> None:
482        """
483        Handle enqueuing of the next (queue) item on the player.
484
485        Called when player reports it started buffering a queue item
486        and when the queue items updated.
487
488        A PlayerProvider implementation is in itself responsible for handling this
489        so that the queue items keep playing until its empty or the player stopped.
490
491        Will only be called if the player reports PlayerFeature.ENQUEUE is
492        supported and the player is currently playing a MA queue.
493
494        This will NOT be called if the end of the queue is reached (and repeat disabled).
495        This will NOT be called if the player is using flow mode to playback the queue.
496
497         :param media: Details of the item that needs to be enqueued on the player.
498        """
499        raise NotImplementedError(
500            "enqueue_next_media needs to be implemented when PlayerFeature.ENQUEUE is set"
501        )
502
503    async def play_announcement(
504        self, announcement: PlayerMedia, volume_level: int | None = None
505    ) -> None:
506        """
507        Handle (native) playback of an announcement on the player.
508
509        Will only be called if the PlayerFeature.PLAY_ANNOUNCEMENT is supported.
510
511        :param announcement: Details of the announcement that needs to be played on the player.
512        :param volume_level: The volume level to play the announcement at (0..100).
513            If not set, the player should use the current volume level.
514        """
515        raise NotImplementedError(
516            "play_announcement needs to be implemented when PlayerFeature.PLAY_ANNOUNCEMENT is set"
517        )
518
519    async def select_source(self, source: str) -> None:
520        """
521        Handle SELECT SOURCE command on the player.
522
523        Will only be called if the PlayerFeature.SELECT_SOURCE is supported.
524
525        :param source: The source(id) to select, as defined in the source_list.
526        """
527        raise NotImplementedError(
528            "select_source needs to be implemented when PlayerFeature.SELECT_SOURCE is set"
529        )
530
531    async def select_sound_mode(self, sound_mode: str) -> None:
532        """
533        Handle SELECT SOUND MODE command on the player.
534
535        Will only be called if the PlayerFeature.SELECT_SOUND_MODE is supported.
536
537        :param source: The sound_mode(id) to select, as defined in the sound_mode_list.
538        """
539        raise NotImplementedError(
540            "select_sound_mode needs to be implemented when PlayerFeature.SELECT_SOUND_MODE is set"
541        )
542
543    async def set_option(self, option_key: str, option_value: PlayerOptionValueType) -> None:
544        """
545        Handle SET_OPTION command on the player.
546
547        Will only be called if the PlayerFeature.OPTIONS is supported.
548
549        :param option_key: The option_key of the PlayerOption
550        :param option_value: The new value of the PlayerOption
551        """
552        raise NotImplementedError(
553            "set_option needs to be implemented when PlayerFeature.Option is set"
554        )
555
556    async def set_members(
557        self,
558        player_ids_to_add: list[str] | None = None,
559        player_ids_to_remove: list[str] | None = None,
560    ) -> None:
561        """
562        Handle SET_MEMBERS command on the player.
563
564        Group or ungroup the given child player(s) to/from this player.
565        Will only be called if the PlayerFeature.SET_MEMBERS is supported.
566
567        :param player_ids_to_add: List of player_id's to add to the group.
568        :param player_ids_to_remove: List of player_id's to remove from the group.
569        """
570        raise NotImplementedError(
571            "set_members needs to be implemented when PlayerFeature.SET_MEMBERS is set"
572        )
573
574    async def poll(self) -> None:
575        """
576        Poll player for state updates.
577
578        This is called by the Player Manager;
579        if the 'needs_poll' property is True.
580        """
581        raise NotImplementedError("poll needs to be implemented when needs_poll is True")
582
583    async def get_config_entries(
584        self,
585        action: str | None = None,
586        values: dict[str, ConfigValueType] | None = None,
587    ) -> list[ConfigEntry]:
588        """
589        Return all (provider/player specific) Config Entries for the player.
590
591        action: [optional] action key called from config entries UI.
592        values: the (intermediate) raw values for config entries sent with the action.
593        """
594        # Return any (player/provider specific) config entries for a player.
595        # To override the default config entries, simply define an entry with the same key
596        # and it will be used instead of the default one.
597        return []
598
599    async def on_config_updated(self) -> None:
600        """
601        Handle logic when the player is loaded or updated.
602
603        Override this method in your player implementation if you need
604        to perform any additional setup logic after the player is registered and
605        the self.config was loaded, and whenever the config changes.
606        """
607        return
608
609    async def on_unload(self) -> None:
610        """Handle logic when the player is unloaded from the Player controller."""
611        for callback in self._on_unload_callbacks:
612            try:
613                callback()
614            except Exception as err:
615                self.logger.error(
616                    "Error calling on_unload callback for player %s: %s",
617                    self.player_id,
618                    err,
619                )
620
621    async def group_with(self, target_player_id: str) -> None:
622        """
623        Handle GROUP_WITH command on the player.
624
625        Group this player to the given syncleader/target.
626        Will only be called if the PlayerFeature.SET_MEMBERS is supported.
627
628        :param target_player: player_id of the target player / sync leader.
629        """
630        # convenience helper method
631        # no need to implement unless your player/provider has an optimized way to execute this
632        # default implementation will simply call set_members
633        # to add the target player to the group.
634        target_player = self.mass.players.get_player(target_player_id, raise_unavailable=True)
635        assert target_player  # for type checking
636        await target_player.set_members(player_ids_to_add=[self.player_id])
637
638    async def ungroup(self) -> None:
639        """
640        Handle UNGROUP command on the player.
641
642        Remove the player from any (sync)groups it currently is grouped to.
643        If this player is the sync leader (or group player),
644        all child's will be ungrouped and the group dissolved.
645
646        Will only be called if the PlayerFeature.SET_MEMBERS is supported.
647        """
648        # convenience helper method
649        # no need to implement unless your player/provider has an optimized way to execute this
650        # default implementation will simply call set_members
651        if self.synced_to:
652            if parent_player := self.mass.players.get_player(self.synced_to):
653                # if this player is synced to another player, remove self from that group
654                await parent_player.set_members(player_ids_to_remove=[self.player_id])
655        elif self.group_members:
656            await self.set_members(player_ids_to_remove=self.group_members)
657
658    def _on_player_media_updated(self) -> None:  # noqa: B027
659        """Handle callback when the current media of the player is updated."""
660        # optional callback for players that want to be informed when the final
661        # current media is updated (after applying group/sync membership logic).
662        # for instance to update any display information on the physical player.
663
664    # DO NOT OVERWRITE BELOW !
665    # These properties and methods are either managed by core logic or they
666    # are used to perform a very specific function. Overwriting these may
667    # produce undesirable effects.
668
669    @property
670    @final
671    def player_id(self) -> str:
672        """Return the id of the player."""
673        return self._player_id
674
675    @property
676    @final
677    def provider(self) -> PlayerProvider:
678        """Return the provider of the player."""
679        return self._provider
680
681    @property
682    @final
683    def provider_id(self) -> str:
684        """Return the provider (instance) id of the player."""
685        return self._provider.instance_id
686
687    @property
688    @final
689    def config(self) -> PlayerConfig:
690        """Return the config of the player."""
691        return self._config
692
693    @property
694    @final
695    def extra_attributes(self) -> dict[str, EXTRA_ATTRIBUTES_TYPES]:
696        """
697        Return the extra attributes of the player.
698
699        This is a dict that can be used to pass any extra (serializable)
700        attributes over the API, to be consumed by the UI (or another APi client, such as HA).
701        This is not persisted and not used or validated by the core logic.
702        """
703        return self._extra_attributes
704
705    @property
706    @final
707    def extra_data(self) -> dict[str, Any]:
708        """
709        Return the extra data of the player.
710
711        This is a dict that can be used to store any extra data
712        that is not part of the player state or config.
713        This is not persisted and not exposed on the API.
714        """
715        return self._extra_data
716
717    @cached_property
718    @final
719    def display_name(self) -> str:
720        """Return the (FINAL) display name of the player."""
721        if custom_name := self._config.name:
722            # always prefer the custom name over the default name
723            return custom_name
724        return self.name or self._config.default_name or self.player_id
725
726    @cached_property
727    @final
728    def enabled(self) -> bool:
729        """Return if the player is enabled."""
730        return self._config.enabled
731
732    @property
733    @final
734    def initialized(self) -> asyncio.Event:
735        """
736        Return if the player is initialized.
737
738        Used by player controller to indicate initial registration completed.
739        """
740        return self.__initialized
741
742    @property
743    def corrected_elapsed_time(self) -> float | None:
744        """Return the corrected/realtime elapsed time."""
745        if self.elapsed_time is None or self.elapsed_time_last_updated is None:
746            return None
747        if self.playback_state == PlaybackState.PLAYING:
748            return self.elapsed_time + (time.time() - self.elapsed_time_last_updated)
749        return self.elapsed_time
750
751    @cached_property
752    @final
753    def icon(self) -> str:
754        """Return the player icon."""
755        return cast("str", self._config.get_value(CONF_ENTRY_PLAYER_ICON.key))
756
757    @cached_property
758    @final
759    def power_control(self) -> str:
760        """Return the power control type."""
761        if conf := self.mass.config.get_raw_player_config_value(self.player_id, CONF_POWER_CONTROL):
762            return str(conf)
763        # not explicitly set, use native if supported
764        if PlayerFeature.POWER in self.supported_features:
765            return PLAYER_CONTROL_NATIVE
766        # note that we do not try to use protocol players for power control,
767        # as this is very unlikely to be provided by a generic protocol and if it does,
768        # it will be handled automatically on stream start/stop.
769        return PLAYER_CONTROL_NONE
770
771    @cached_property
772    @final
773    def volume_control(self) -> str:
774        """Return the volume control type."""
775        if conf := self.mass.config.get_raw_player_config_value(
776            self.player_id, CONF_VOLUME_CONTROL
777        ):
778            return str(conf)
779        # not explicitly set, use native if supported
780        if PlayerFeature.VOLUME_SET in self.supported_features:
781            return PLAYER_CONTROL_NATIVE
782        # check for protocol player with volume support, and use that if found
783        if protocol_player := self._get_protocol_player_for_feature(PlayerFeature.VOLUME_SET):
784            return protocol_player.player_id
785        return PLAYER_CONTROL_NONE
786
787    @cached_property
788    @final
789    def mute_control(self) -> str:
790        """Return the mute control type."""
791        if conf := self.mass.config.get_raw_player_config_value(self.player_id, CONF_MUTE_CONTROL):
792            return str(conf)
793        # not explicitly set, use native if supported
794        if PlayerFeature.VOLUME_MUTE in self.supported_features:
795            return PLAYER_CONTROL_NATIVE
796        # check for protocol player with volume mute support, and use that if found
797        if protocol_player := self._get_protocol_player_for_feature(PlayerFeature.VOLUME_MUTE):
798            return protocol_player.player_id
799        return PLAYER_CONTROL_NONE
800
801    @cached_property
802    @final
803    def group_volume(self) -> int:
804        """
805        Return the group volume level.
806
807        If this player is a group player or syncgroup, this will return the average volume
808        level of all (powered on) child players in the group.
809        If the player is not a group player or syncgroup, this will return the volume level
810        of the player itself (if set), or 0 if not set.
811        """
812        if len(self.state.group_members) == 0:
813            # player is not a group or syncgroup
814            return self.state.volume_level or 0
815        # calculate group volume from all (turned on) players
816        group_volume = 0
817        active_players = 0
818        for child_player in self.mass.players.iter_group_members(
819            self, only_powered=True, exclude_self=self.type != PlayerType.PLAYER
820        ):
821            if (child_volume := child_player.state.volume_level) is None:
822                continue
823            group_volume += child_volume
824            active_players += 1
825        if active_players:
826            group_volume = int(group_volume / active_players)
827        return group_volume
828
829    @cached_property
830    @final
831    def hide_in_ui(self) -> bool:
832        """
833        Return the hide player in UI options.
834
835        This is a convenience property based on the config entry.
836        """
837        return bool(self._config.get_value(CONF_HIDE_IN_UI, self.hidden_by_default))
838
839    @cached_property
840    @final
841    def expose_to_ha(self) -> bool:
842        """
843        Return if the player should be exposed to Home Assistant.
844
845        This is a convenience property that returns True if the player is set to be exposed
846        to Home Assistant, based on the config entry.
847        """
848        return bool(self._config.get_value(CONF_EXPOSE_PLAYER_TO_HA, self.expose_to_ha_by_default))
849
850    @property
851    @final
852    def mass_queue_active(self) -> bool:
853        """
854        Return if the/a Music Assistant Queue is currently active for this player.
855
856        This is a convenience property that returns True if the
857        player currently has a Music Assistant Queue as active source.
858        """
859        return bool(self.mass.players.get_active_queue(self))
860
861    @cached_property
862    @final
863    def flow_mode(self) -> bool:
864        """
865        Return if the player needs flow mode.
866
867        Will use 'requires_flow_mode' unless overridden by flow_mode config.
868        Considers the active output protocol's flow_mode if a protocol is active.
869        """
870        # If an output protocol is active (and not native), use the protocol player's flow_mode
871        # The protocol player will handle its own config check
872        if (
873            self.__attr_active_output_protocol
874            and self.__attr_active_output_protocol != "native"
875            and (
876                protocol_player := self.mass.players.get_player(self.__attr_active_output_protocol)
877            )
878        ):
879            return protocol_player.flow_mode
880        # Check native player's config override
881        if bool(self._config.get_value(CONF_FLOW_MODE)) is True:
882            # flow mode explicitly enabled in config
883            return True
884        return self.requires_flow_mode
885
886    @property
887    @final
888    def supports_enqueue(self) -> bool:
889        """
890        Return if the player supports enqueueing tracks.
891
892        This considers the active output protocol's capabilities if one is active.
893        If a protocol player is active, checks that protocol's ENQUEUE feature.
894        Otherwise checks the native player's ENQUEUE feature.
895        """
896        return self._check_feature_with_active_protocol(PlayerFeature.ENQUEUE)
897
898    @property
899    @final
900    def state(self) -> PlayerState:
901        """Return the current (and FINAL) PlayerState of the player."""
902        return self._state
903
904    # Protocol-related properties and helpers
905
906    @cached_property
907    @final
908    def is_native_player(self) -> bool:
909        """Return True if this player is a native player."""
910        is_universal_player = self.provider.domain == "universal_player"
911        has_play_media = PlayerFeature.PLAY_MEDIA in self.supported_features
912        return self.type != PlayerType.PROTOCOL and not is_universal_player and has_play_media
913
914    @cached_property
915    @final
916    def output_protocols(self) -> list[OutputProtocol]:
917        """
918        Return all output options for this player.
919
920        Includes:
921        - Native playback (if player supports PLAY_MEDIA and is not a protocol/universal player)
922        - Active protocol players from linked_output_protocols
923        - Disabled protocols from cached linked_protocol_player_ids in config
924
925        Each entry has an available flag indicating current availability.
926        """
927        result: list[OutputProtocol] = []
928
929        # Add native playback option if applicable
930        if self.is_native_player:
931            result.append(
932                OutputProtocol(
933                    output_protocol_id="native",
934                    name=self.provider.name,
935                    protocol_domain=self.provider.domain,
936                    priority=0,  # Native is always highest priority
937                    available=self.available,
938                    is_native=True,
939                )
940            )
941
942        # Add active protocol players
943        active_ids: set[str] = set()
944        for linked in self.__attr_linked_protocols:
945            active_ids.add(linked.output_protocol_id)
946            # Check if the protocol player is actually available
947            protocol_player = self.mass.players.get_player(linked.output_protocol_id)
948            is_available = protocol_player.available if protocol_player else False
949            if protocol_player and not is_available:
950                self.logger.debug(
951                    "Protocol player %s (%s) is unavailable for %s",
952                    linked.output_protocol_id,
953                    linked.protocol_domain,
954                    self.display_name,
955                )
956            # Use provider name if available, else domain title
957            if protocol_player:
958                name = protocol_player.provider.name
959            else:
960                name = linked.protocol_domain.title() if linked.protocol_domain else "Unknown"
961            result.append(
962                OutputProtocol(
963                    output_protocol_id=linked.output_protocol_id,
964                    name=name,
965                    protocol_domain=linked.protocol_domain,
966                    priority=linked.priority,
967                    available=is_available,
968                )
969            )
970
971        # Add disabled protocols from cache
972        cached_protocol_ids: list[str] = self.mass.config.get(
973            f"{CONF_PLAYERS}/{self.player_id}/values/{CONF_LINKED_PROTOCOL_PLAYER_IDS}",
974            [],
975        )
976        for protocol_id in cached_protocol_ids:
977            if protocol_id in active_ids:
978                continue  # Already included above
979            # Get stored config to determine protocol domain
980            if raw_conf := self.mass.config.get(f"{CONF_PLAYERS}/{protocol_id}"):
981                provider_id = raw_conf.get("provider", "")
982                protocol_domain = provider_id.split("--")[0] if provider_id else "unknown"
983                priority = PROTOCOL_PRIORITY.get(protocol_domain, 100)
984                result.append(
985                    OutputProtocol(
986                        output_protocol_id=protocol_id,
987                        name=protocol_domain.title(),
988                        protocol_domain=protocol_domain,
989                        priority=priority,
990                        available=False,  # Disabled protocols are not available
991                    )
992                )
993
994        # Sort by priority (lower = more preferred)
995        result.sort(key=lambda o: o.priority)
996        return result
997
998    @property
999    @final
1000    def linked_output_protocols(self) -> list[OutputProtocol]:
1001        """Return the list of actively linked output protocol players."""
1002        return self.__attr_linked_protocols
1003
1004    @property
1005    @final
1006    def protocol_parent_id(self) -> str | None:
1007        """Return the parent player_id if this is a protocol player linked to a native player."""
1008        return self.__attr_protocol_parent_id
1009
1010    @property
1011    @final
1012    def active_output_protocol(self) -> str | None:
1013        """Return the currently active output protocol ID."""
1014        return self.__attr_active_output_protocol
1015
1016    @final
1017    def set_active_output_protocol(self, protocol_id: str | None) -> None:
1018        """
1019        Set the currently active output protocol ID.
1020
1021        :param protocol_id: The protocol player_id to set as active, "native" for native playback,
1022            or None to clear the active protocol.
1023        """
1024        if self.__attr_active_output_protocol == protocol_id:
1025            return  # No change
1026        if protocol_id == self.player_id:
1027            protocol_id = "native"  # Normalize to "native" for native player
1028        if protocol_id:
1029            protocol_name = protocol_id
1030            if protocol_id == "native":
1031                protocol_name = "Native"
1032            elif protocol_player := self.mass.players.get_player(protocol_id):
1033                protocol_name = protocol_player.provider.name
1034            self.logger.info(
1035                "Setting active output protocol on %s to %s",
1036                self.display_name,
1037                protocol_name,
1038            )
1039        else:
1040            self.logger.info(
1041                "Clearing active output protocol on %s",
1042                self.display_name,
1043            )
1044        self.__attr_active_output_protocol = protocol_id
1045        self.update_state()
1046
1047    @final
1048    def set_linked_output_protocols(self, protocols: list[OutputProtocol]) -> None:
1049        """
1050        Set the actively linked output protocol players.
1051
1052        :param protocols: List of OutputProtocol objects representing active protocol players.
1053        """
1054        self.__attr_linked_protocols = protocols
1055        self.mass.players.trigger_player_update(self.player_id)
1056
1057    @final
1058    def set_protocol_parent_id(self, parent_id: str | None) -> None:
1059        """
1060        Set the parent player_id for protocol players.
1061
1062        :param parent_id: The player_id of the parent player, or None to clear.
1063        """
1064        self.__attr_protocol_parent_id = parent_id
1065        self.mass.players.trigger_player_update(self.player_id)
1066
1067    @final
1068    def get_linked_protocol(self, protocol_domain: str) -> OutputProtocol | None:
1069        """Get a linked protocol by domain with current availability."""
1070        for linked in self.__attr_linked_protocols:
1071            if linked.protocol_domain == protocol_domain:
1072                protocol_player = self.mass.players.get_player(linked.output_protocol_id)
1073                current_available = protocol_player.available if protocol_player else False
1074                return OutputProtocol(
1075                    output_protocol_id=linked.output_protocol_id,
1076                    name=protocol_player.provider.name
1077                    if protocol_player
1078                    else linked.protocol_domain.title(),
1079                    protocol_domain=linked.protocol_domain,
1080                    priority=linked.priority,
1081                    available=current_available,
1082                    is_native=False,
1083                )
1084        return None
1085
1086    @final
1087    def get_output_protocol_by_domain(self, protocol_domain: str) -> OutputProtocol | None:
1088        """
1089        Get an output protocol by domain, including native protocol.
1090
1091        Unlike get_linked_protocol, this also checks if the player's native protocol
1092        matches the requested domain.
1093
1094        :param protocol_domain: The protocol domain to search for (e.g., "airplay", "sonos").
1095        """
1096        for output_protocol in self.output_protocols:
1097            if output_protocol.protocol_domain == protocol_domain:
1098                return output_protocol
1099        return None
1100
1101    @final
1102    def get_protocol_player(self, player_id: str) -> Player | None:
1103        """Get the protocol Player for a given player_id."""
1104        if player_id == "native":
1105            return self if PlayerFeature.PLAY_MEDIA in self.supported_features else None
1106        return self.mass.players.get_player(player_id)
1107
1108    @final
1109    def get_preferred_protocol_player(self) -> Player | None:
1110        """Get the best available protocol player by priority."""
1111        for linked in sorted(self.__attr_linked_protocols, key=lambda x: x.priority):
1112            if protocol_player := self.mass.players.get_player(linked.output_protocol_id):
1113                if protocol_player.available:
1114                    return protocol_player
1115        return None
1116
1117    @final
1118    def update_state(self, force_update: bool = False, signal_event: bool = True) -> None:
1119        """
1120        Update the PlayerState from the current state of the player.
1121
1122        This method should be called to update the player's state
1123        and signal any changes to the PlayerController.
1124
1125        :param force_update: If True, a state update event will be
1126        pushed even if the state has not actually changed.
1127        :param signal_event: If True, signal the state update event to the PlayerController.
1128        """
1129        self.mass.verify_event_loop_thread("player.update_state")
1130        # clear the dict for the cached properties
1131        self._cache.clear()
1132        # calculate the new state
1133        prev_media_checksum = self._get_player_media_checksum()
1134        changed_values = self.__calculate_player_state()
1135        if prev_media_checksum != self._get_player_media_checksum():
1136            # current media changed, call the media updated callback
1137            self._on_player_media_updated()
1138        # ignore some values that are not relevant for the state
1139        changed_values.pop("elapsed_time_last_updated", None)
1140        changed_values.pop("extra_attributes.seq_no", None)
1141        changed_values.pop("extra_attributes.last_poll", None)
1142        changed_values.pop("current_media.elapsed_time_last_updated", None)
1143        # persist the default name if it changed
1144        if self.name and self.config.default_name != self.name:
1145            self.mass.config.set_player_default_name(self.player_id, self.name)
1146        # persist the player type if it changed
1147        if self.type != self._config.player_type:
1148            self.mass.config.set_player_type(self.player_id, self.type)
1149        # return early if nothing changed (unless force_update is True)
1150        if len(changed_values) == 0 and not force_update:
1151            return
1152
1153        # signal the state update to the PlayerController
1154        if signal_event:
1155            self.mass.players.signal_player_state_update(self, changed_values)
1156
1157    @final
1158    def set_current_media(  # noqa: PLR0913
1159        self,
1160        uri: str,
1161        media_type: MediaType = MediaType.UNKNOWN,
1162        title: str | None = None,
1163        artist: str | None = None,
1164        album: str | None = None,
1165        image_url: str | None = None,
1166        duration: int | None = None,
1167        source_id: str | None = None,
1168        queue_item_id: str | None = None,
1169        custom_data: dict[str, Any] | None = None,
1170        clear_all: bool = False,
1171    ) -> None:
1172        """
1173        Set current_media helper.
1174
1175        Assumes use of '_attr_current_media'.
1176        """
1177        if self._attr_current_media is None or clear_all:
1178            self._attr_current_media = PlayerMedia(
1179                uri=uri,
1180                media_type=media_type,
1181            )
1182        self._attr_current_media.uri = uri
1183        if media_type != MediaType.UNKNOWN:
1184            self._attr_current_media.media_type = media_type
1185        if title:
1186            self._attr_current_media.title = title
1187        if artist:
1188            self._attr_current_media.artist = artist
1189        if album:
1190            self._attr_current_media.album = album
1191        if image_url:
1192            self._attr_current_media.image_url = image_url
1193        if duration:
1194            self._attr_current_media.duration = duration
1195        if source_id:
1196            self._attr_current_media.source_id = source_id
1197        if queue_item_id:
1198            self._attr_current_media.queue_item_id = queue_item_id
1199        if custom_data:
1200            self._attr_current_media.custom_data = custom_data
1201
1202    @final
1203    def set_config(self, config: PlayerConfig) -> None:
1204        """
1205        Set/update the player config.
1206
1207        May only be called by the PlayerController.
1208        """
1209        # TODO: validate that caller is the PlayerController ?
1210        self._config = config
1211
1212    @final
1213    def set_initialized(self) -> None:
1214        """Set the player as initialized."""
1215        self.__initialized.set()
1216
1217    @final
1218    def to_dict(self) -> dict[str, Any]:
1219        """Return the (serializable) dict representation of the Player."""
1220        return self.state.to_dict()
1221
1222    @final
1223    def supports_feature(self, feature: PlayerFeature) -> bool:
1224        """Return True if this player supports the given feature."""
1225        return feature in self.supported_features
1226
1227    @final
1228    def check_feature(self, feature: PlayerFeature) -> None:
1229        """Check if this player supports the given feature."""
1230        if not self.supports_feature(feature):
1231            raise UnsupportedFeaturedException(
1232                f"Player {self.display_name} does not support feature {feature.name}"
1233            )
1234
1235    @final
1236    def _get_player_media_checksum(self) -> str:
1237        """Return a checksum for the current media."""
1238        if not (media := self.state.current_media):
1239            return ""
1240        return (
1241            f"{media.uri}|{media.title}|{media.source_id}|{media.queue_item_id}|"
1242            f"{media.image_url}|{media.duration}|{media.elapsed_time}"
1243        )
1244
1245    @final
1246    def _check_feature_with_active_protocol(
1247        self, feature: PlayerFeature, active_only: bool = False
1248    ) -> bool:
1249        """
1250        Check if a feature is supported considering the active output protocol.
1251
1252        If an active output protocol is set (and not native), checks that protocol
1253        player's features. Otherwise checks the native player's features.
1254
1255        :param feature: The PlayerFeature to check.
1256        :return: True if the feature is supported by the active protocol or native player.
1257        """
1258        # If active output protocol is set and not native, check protocol player's features
1259        if (
1260            self.__attr_active_output_protocol
1261            and self.__attr_active_output_protocol != "native"
1262            and (
1263                protocol_player := self.mass.players.get_player(self.__attr_active_output_protocol)
1264            )
1265        ):
1266            return feature in protocol_player.supported_features
1267        # Otherwise check native player's features
1268        return feature in self.supported_features
1269
1270    @final
1271    def _get_protocol_player_for_feature(
1272        self,
1273        feature: PlayerFeature,
1274    ) -> Player | None:
1275        """Get player(protocol) which has the given PlayerFeature."""
1276        # prefer native player
1277        if feature in self.supported_features:
1278            return self
1279        # Otherwise, use the first available linked protocol
1280        for linked in self.linked_output_protocols:
1281            if (
1282                (protocol_player := self.mass.players.get_player(linked.output_protocol_id))
1283                and protocol_player.available
1284                and feature in protocol_player.supported_features
1285            ):
1286                return protocol_player
1287
1288        return None
1289
1290    @final
1291    def __calculate_player_state(
1292        self,
1293    ) -> dict[str, tuple[Any, Any]]:
1294        """
1295        Calculate the (current) and FINAL PlayerState.
1296
1297        This method is called when we're updating the player,
1298        and we compare the current state with the previous state to determine
1299        if we need to signal a state change to API consumers.
1300
1301        Returns a dict with the state attributes that have changed.
1302        """
1303        playback_state, elapsed_time, elapsed_time_last_updated = self.__final_playback_state
1304        prev_state = deepcopy(self._state)
1305        self._state = PlayerState(
1306            player_id=self.player_id,
1307            provider=self.provider_id,
1308            type=self.type,
1309            available=self.enabled and self.available,
1310            device_info=self.device_info,
1311            supported_features=self.__final_supported_features,
1312            playback_state=playback_state,
1313            elapsed_time=elapsed_time,
1314            elapsed_time_last_updated=elapsed_time_last_updated,
1315            powered=self.__final_power_state,
1316            volume_level=self.__final_volume_level,
1317            volume_muted=self.__final_volume_muted_state,
1318            group_members=UniqueList(self.__final_group_members),
1319            static_group_members=UniqueList(self.static_group_members),
1320            can_group_with=self.__final_can_group_with,
1321            synced_to=self.__final_synced_to,
1322            active_source=self.__final_active_source,
1323            source_list=self.__final_source_list,
1324            active_group=self.__final_active_group,
1325            current_media=self.__final_current_media,
1326            active_sound_mode=self.active_sound_mode,
1327            sound_mode_list=self.sound_mode_list,
1328            options=self.options,
1329            name=self.display_name,
1330            enabled=self.enabled,
1331            hide_in_ui=self.hide_in_ui,
1332            expose_to_ha=self.expose_to_ha,
1333            icon=self.icon,
1334            group_volume=self.group_volume,
1335            extra_attributes=self.extra_attributes,
1336            power_control=self.power_control,
1337            volume_control=self.volume_control,
1338            mute_control=self.mute_control,
1339            output_protocols=self.output_protocols,
1340            active_output_protocol=self.__attr_active_output_protocol,
1341        )
1342
1343        # track stop called state
1344        if (
1345            prev_state.playback_state == PlaybackState.IDLE
1346            and self._state.playback_state != PlaybackState.IDLE
1347        ):
1348            self.__stop_called = False
1349        elif (
1350            prev_state.playback_state != PlaybackState.IDLE
1351            and self._state.playback_state == PlaybackState.IDLE
1352        ):
1353            self.__stop_called = True
1354            # when we're going to idle,
1355            # we want to reset the active mass source after a short delay
1356            # this is done using a timer which gets reset if the player starts playing again
1357            # before the timer is up, using the task_id
1358            self.mass.call_later(
1359                2, self.set_active_mass_source, None, task_id=f"set_mass_source_{self.player_id}"
1360            )
1361
1362        return get_changed_dataclass_values(
1363            prev_state,
1364            self._state,
1365            recursive=True,
1366        )
1367
1368    @cached_property
1369    @final
1370    def __final_playback_state(self) -> tuple[PlaybackState, float | None, float | None]:
1371        """
1372        Return the FINAL playback state based on the playercontrol which may have been set-up.
1373
1374        Returns a tuple of (playback_state, elapsed_time, elapsed_time_last_updated).
1375        """
1376        # If an output protocol is active (and not native), use the protocol player's state
1377        if (
1378            self.__attr_active_output_protocol
1379            and self.__attr_active_output_protocol != "native"
1380            and (
1381                protocol_player := self.mass.players.get_player(self.__attr_active_output_protocol)
1382            )
1383            and protocol_player.playback_state != PlaybackState.IDLE
1384        ):
1385            return (
1386                protocol_player.state.playback_state,
1387                protocol_player.state.elapsed_time,
1388                protocol_player.state.elapsed_time_last_updated,
1389            )
1390        # If we're synced, use the syncleader state for playback state and elapsed time
1391        # NOTE: Don't do this for the active group player,
1392        # because the group player relies on the sync leader for state info.
1393        parent_id = self.__final_synced_to
1394        if parent_id and (parent_player := self.mass.players.get_player(parent_id)):
1395            return (
1396                parent_player.state.playback_state,
1397                parent_player.state.elapsed_time,
1398                parent_player.state.elapsed_time_last_updated,
1399            )
1400        return (self.playback_state, self.elapsed_time, self.elapsed_time_last_updated)
1401
1402    @cached_property
1403    @final
1404    def __final_power_state(self) -> bool | None:
1405        """Return the FINAL power state based on the playercontrol which may have been set-up."""
1406        power_control = self.power_control
1407        if power_control == PLAYER_CONTROL_FAKE:
1408            return bool(self.extra_data.get(ATTR_FAKE_POWER, False))
1409        if power_control == PLAYER_CONTROL_NATIVE:
1410            return self.powered
1411        if power_control == PLAYER_CONTROL_NONE:
1412            return None
1413        # handle player control for power if set
1414        if control := self.mass.players.get_player_control(power_control):
1415            return control.power_state
1416        return None
1417
1418    @cached_property
1419    @final
1420    def __final_volume_level(self) -> int | None:
1421        """Return the FINAL volume level based on the playercontrol which may have been set-up."""
1422        volume_control = self.volume_control
1423        if volume_control == PLAYER_CONTROL_FAKE:
1424            return int(self.extra_data.get(ATTR_FAKE_VOLUME, 0))
1425        if volume_control == PLAYER_CONTROL_NATIVE:
1426            return self.volume_level
1427        if volume_control == PLAYER_CONTROL_NONE:
1428            return None
1429        # handle protocol player as volume control
1430        if control := self.mass.players.get_player(volume_control):
1431            return control.volume_level
1432        # handle player control for volume if set
1433        if player_control := self.mass.players.get_player_control(volume_control):
1434            return player_control.volume_level
1435        return None
1436
1437    @cached_property
1438    @final
1439    def __final_volume_muted_state(self) -> bool | None:
1440        """Return the FINAL mute state based on any playercontrol which may have been set-up."""
1441        mute_control = self.mute_control
1442        if mute_control == PLAYER_CONTROL_FAKE:
1443            return bool(self.extra_data.get(ATTR_FAKE_MUTE, False))
1444        if mute_control == PLAYER_CONTROL_NATIVE:
1445            return self.volume_muted
1446        if mute_control == PLAYER_CONTROL_NONE:
1447            return None
1448        # handle protocol player as mute control
1449        if control := self.mass.players.get_player(mute_control):
1450            return control.volume_muted
1451        # handle player control for mute if set
1452        if player_control := self.mass.players.get_player_control(mute_control):
1453            return player_control.volume_muted
1454        return None
1455
1456    @cached_property
1457    @final
1458    def __final_active_group(self) -> str | None:
1459        """
1460        Return the player id of any playergroup that is currently active for this player.
1461
1462        This will return the id of the groupplayer if any groups are active.
1463        If no groups are currently active, this will return None.
1464        """
1465        if self.type == PlayerType.PROTOCOL:
1466            # protocol players should not have an active group,
1467            # they follow the group state of their parent player
1468            return None
1469        for group_player in self.mass.players.all_players(
1470            return_unavailable=False, return_disabled=False
1471        ):
1472            if group_player.type != PlayerType.GROUP:
1473                continue
1474            if group_player.player_id == self.player_id:
1475                continue
1476            if group_player.playback_state not in (PlaybackState.PLAYING, PlaybackState.PAUSED):
1477                continue
1478            if self.player_id in group_player.group_members:
1479                return group_player.player_id
1480        return None
1481
1482    @cached_property
1483    @final
1484    def __final_current_media(self) -> PlayerMedia | None:
1485        """Return the FINAL current media for the player."""
1486        if self.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS):
1487            # if an announcement is in progress, return announcement details
1488            return PlayerMedia(
1489                uri="announcement",
1490                media_type=MediaType.ANNOUNCEMENT,
1491                title="ANNOUNCEMENT",
1492            )
1493
1494        # if the player is grouped/synced, use the current_media of the group/parent player
1495        if parent_player_id := (self.__final_active_group or self.__final_synced_to):
1496            if parent_player_id != self.player_id and (
1497                parent_player := self.mass.players.get_player(parent_player_id)
1498            ):
1499                return parent_player.state.current_media
1500            return None  # if parent player not found, return None for current media
1501        # if this is a protocol player, use the current_media of the parent player
1502        if self.type == PlayerType.PROTOCOL and self.__attr_protocol_parent_id:
1503            if parent_player := self.mass.players.get_player(self.__attr_protocol_parent_id):
1504                return parent_player.state.current_media
1505        # if a pluginsource is currently active, return those details
1506        active_source = self.__final_active_source
1507        if (
1508            active_source
1509            and (source := self.mass.players.get_plugin_source(active_source))
1510            and source.metadata
1511        ):
1512            return PlayerMedia(
1513                uri=source.metadata.uri or source.id,
1514                media_type=MediaType.PLUGIN_SOURCE,
1515                title=source.metadata.title,
1516                artist=source.metadata.artist,
1517                album=source.metadata.album,
1518                image_url=source.metadata.image_url,
1519                duration=source.metadata.duration,
1520                source_id=source.id,
1521                elapsed_time=source.metadata.elapsed_time,
1522                elapsed_time_last_updated=source.metadata.elapsed_time_last_updated,
1523            )
1524        # if MA queue is active, return those details
1525        active_queue: PlayerQueue | None = None
1526        if not active_queue and active_source:
1527            active_queue = self.mass.player_queues.get(active_source)
1528        if not active_queue and self.active_source is None:
1529            active_queue = self.mass.player_queues.get(self.player_id)
1530        if active_queue and (current_item := active_queue.current_item):
1531            item_image_url = (
1532                # the image format needs to be 500x500 jpeg for maximum compatibility with players
1533                self.mass.metadata.get_image_url(current_item.image, size=500, image_format="jpeg")
1534                if current_item.image
1535                else None
1536            )
1537            if current_item.streamdetails and (
1538                stream_metadata := current_item.streamdetails.stream_metadata
1539            ):
1540                # handle stream metadata in streamdetails (e.g. for radio stream)
1541                return PlayerMedia(
1542                    uri=current_item.uri,
1543                    media_type=current_item.media_type,
1544                    title=stream_metadata.title or current_item.name,
1545                    artist=stream_metadata.artist,
1546                    album=stream_metadata.album or stream_metadata.description or current_item.name,
1547                    image_url=(stream_metadata.image_url or item_image_url),
1548                    duration=stream_metadata.duration or current_item.duration,
1549                    source_id=active_queue.queue_id,
1550                    queue_item_id=current_item.queue_item_id,
1551                    elapsed_time=stream_metadata.elapsed_time or int(active_queue.elapsed_time),
1552                    elapsed_time_last_updated=stream_metadata.elapsed_time_last_updated
1553                    or active_queue.elapsed_time_last_updated,
1554                )
1555            if media_item := current_item.media_item:
1556                # normal media item
1557                # we use getattr here to avoid issues with different media item types
1558                version = getattr(media_item, "version", None)
1559                album = getattr(media_item, "album", None)
1560                podcast = getattr(media_item, "podcast", None)
1561                metadata = getattr(media_item, "metadata", None)
1562                description = getattr(metadata, "description", None) if metadata else None
1563                return PlayerMedia(
1564                    uri=str(media_item.uri),
1565                    media_type=media_item.media_type,
1566                    title=f"{media_item.name} ({version})" if version else media_item.name,
1567                    artist=getattr(media_item, "artist_str", None),
1568                    album=album.name if album else podcast.name if podcast else description,
1569                    # the image format needs to be 500x500 jpeg for maximum player compatibility
1570                    image_url=self.mass.metadata.get_image_url(
1571                        current_item.media_item.image, size=500, image_format="jpeg"
1572                    )
1573                    or item_image_url
1574                    if current_item.media_item.image
1575                    else item_image_url,
1576                    duration=media_item.duration,
1577                    source_id=active_queue.queue_id,
1578                    queue_item_id=current_item.queue_item_id,
1579                    elapsed_time=int(active_queue.elapsed_time),
1580                    elapsed_time_last_updated=active_queue.elapsed_time_last_updated,
1581                )
1582
1583            # fallback to basic current item details
1584            return PlayerMedia(
1585                uri=current_item.uri,
1586                media_type=current_item.media_type,
1587                title=current_item.name,
1588                image_url=item_image_url,
1589                duration=current_item.duration,
1590                source_id=active_queue.queue_id,
1591                queue_item_id=current_item.queue_item_id,
1592                elapsed_time=int(active_queue.elapsed_time),
1593                elapsed_time_last_updated=active_queue.elapsed_time_last_updated,
1594            )
1595        if active_queue:
1596            # queue is active but no current item
1597            return None
1598        # return native current media if no group/queue is active
1599        if self.current_media:
1600            return PlayerMedia(
1601                uri=self.current_media.uri,
1602                media_type=self.current_media.media_type,
1603                title=self.current_media.title,
1604                artist=self.current_media.artist,
1605                album=self.current_media.album,
1606                image_url=self.current_media.image_url,
1607                duration=self.current_media.duration,
1608                source_id=self.current_media.source_id or active_source,
1609                queue_item_id=self.current_media.queue_item_id,
1610                elapsed_time=self.current_media.elapsed_time or int(self.elapsed_time)
1611                if self.elapsed_time
1612                else None,
1613                elapsed_time_last_updated=self.current_media.elapsed_time_last_updated
1614                or self.elapsed_time_last_updated,
1615            )
1616        return None
1617
1618    @cached_property
1619    @final
1620    def __final_source_list(self) -> UniqueList[PlayerSource]:
1621        """Return the FINAL source list for the player."""
1622        sources = UniqueList(self.source_list)
1623        if self.type == PlayerType.PROTOCOL:
1624            return sources
1625        # always ensure the Music Assistant Queue is in the source list
1626        mass_source = next((x for x in sources if x.id == self.player_id), None)
1627        if mass_source is None:
1628            # if the MA queue is not in the source list, add it
1629            mass_source = PlayerSource(
1630                id=self.player_id,
1631                name="Music Assistant Queue",
1632                passive=False,
1633                # TODO: Do we want to dynamically set these based on the queue state ?
1634                can_play_pause=True,
1635                can_seek=True,
1636                can_next_previous=True,
1637            )
1638            sources.append(mass_source)
1639        # append all/any plugin sources (convert to PlayerSource to avoid deepcopy issues)
1640        for plugin_source in self.mass.players.get_plugin_sources():
1641            if hasattr(plugin_source, "as_player_source"):
1642                sources.append(plugin_source.as_player_source())
1643            else:
1644                sources.append(plugin_source)
1645        return sources
1646
1647    @cached_property
1648    @final
1649    def __final_group_members(self) -> list[str]:
1650        """Return the FINAL group members of this player."""
1651        if self.__final_synced_to:
1652            # If player is synced to another player, it has no group members itself
1653            return []
1654
1655        # Start by translating native group_members to visible player IDs
1656        # This handles cases where a native player (e.g., native AirPlay) has grouped
1657        # protocol players (e.g., Sonos AirPlay protocol players) that need translation
1658        members: list[str] = []
1659        if self.type == PlayerType.PROTOCOL:
1660            # protocol players use their own group members without translation
1661            members.extend(self.group_members)
1662        else:
1663            translated_members = self._translate_protocol_ids_to_visible(set(self.group_members))
1664            for member in translated_members:
1665                if member.player_id not in members:
1666                    members.append(member.player_id)
1667
1668        # If there's an active linked protocol, include its group members (translated)
1669        if self.__attr_active_output_protocol and self.__attr_active_output_protocol != "native":
1670            if protocol_player := self.mass.players.get_player(self.__attr_active_output_protocol):
1671                # Translate protocol player IDs to visible player IDs
1672                protocol_members = self._translate_protocol_ids_to_visible(
1673                    set(protocol_player.group_members)
1674                )
1675                for member in protocol_members:
1676                    if member.player_id not in members:
1677                        members.append(member.player_id)
1678
1679        if self.type != PlayerType.GROUP:
1680            # Ensure the player_id is first in the group_members list
1681            if len(members) > 0 and members[0] != self.player_id:
1682                members = [self.player_id, *[m for m in members if m != self.player_id]]
1683            # If the only member is self, return empty list
1684            if members == [self.player_id]:
1685                return []
1686        return members
1687
1688    @cached_property
1689    @final
1690    def __final_synced_to(self) -> str | None:
1691        """
1692        Return the FINAL synced_to state.
1693
1694        This checks both native sync state and protocol player sync state,
1695        translating protocol player IDs to visible player IDs.
1696        """
1697        # First check the native synced_to from the property
1698        if native_synced_to := self.synced_to:
1699            return native_synced_to
1700
1701        for linked in self.__attr_linked_protocols:
1702            if not (protocol_player := self.mass.players.get_player(linked.output_protocol_id)):
1703                continue
1704            if protocol_player.synced_to:
1705                # Protocol player is synced, translate to visible player
1706                if proto_sync_parent := self.mass.players.get_player(protocol_player.synced_to):
1707                    if proto_sync_parent.type != PlayerType.PROTOCOL:
1708                        # Sync parent is already a visible player (e.g., native AirPlay player)
1709                        return proto_sync_parent.player_id
1710                    if proto_sync_parent.protocol_parent_id and (
1711                        parent := self.mass.players.get_player(proto_sync_parent.protocol_parent_id)
1712                    ):
1713                        # Sync parent is a protocol player, return its visible parent
1714                        return parent.player_id
1715
1716        return None
1717
1718    @cached_property
1719    @final
1720    def __final_supported_features(self) -> set[PlayerFeature]:
1721        """Return the FINAL supported features based supported output protocol(s)."""
1722        base_features = self.supported_features.copy()
1723        if self.__attr_active_output_protocol and self.__attr_active_output_protocol != "native":
1724            # Active linked protocol: add from that specific protocol
1725            if protocol_player := self.mass.players.get_player(self.__attr_active_output_protocol):
1726                for feature in protocol_player.supported_features:
1727                    if feature in ACTIVE_PROTOCOL_FEATURES:
1728                        base_features.add(feature)
1729        # Append (allowed features) from all linked protocols
1730        for linked in self.__attr_linked_protocols:
1731            if protocol_player := self.mass.players.get_player(linked.output_protocol_id):
1732                for feature in protocol_player.supported_features:
1733                    if feature in PROTOCOL_FEATURES:
1734                        base_features.add(feature)
1735        return base_features
1736
1737    @cached_property
1738    @final
1739    def __final_can_group_with(self) -> set[str]:
1740        """
1741        Return the FINAL set of player id's this player can group with.
1742
1743        This is a convenience property which calculates the final can_group_with set
1744        based on any linked protocol players and current player/grouped state.
1745
1746        If player is synced to a native parent: return empty set (already grouped).
1747        If player is synced to a protocol: can still group with other players.
1748        If no active linked protocol: return can_group_with from all active output protocols.
1749        If active linked protocol: return native can_group_with + active protocol's.
1750
1751        All protocol player IDs are translated to their visible parent player IDs.
1752        """
1753
1754        def _should_include_player(player: Player) -> bool:
1755            """Check if a player should be included in the can-group-with set."""
1756            if not player.available:
1757                return False
1758            if player.player_id == self.player_id:
1759                return False  # Don't include self
1760            # Don't include (playing) players that have group members (they are group leaders)
1761            if (  # noqa: SIM103
1762                player.state.playback_state in (PlaybackState.PLAYING, PlaybackState.PAUSED)
1763                and player.group_members
1764            ):
1765                return False
1766            return True
1767
1768        if self.__final_synced_to:
1769            # player is already synced/grouped, cannot group with others
1770            return set()
1771
1772        expanded_can_group_with = self._expand_can_group_with()
1773        # Scenario 1: Player is a protocol player - just return the (expanded) result
1774        if self.type == PlayerType.PROTOCOL:
1775            return {x.player_id for x in expanded_can_group_with}
1776
1777        result: set[str] = set()
1778        # always start with the native can_group_with options (expanded from provider instance IDs)
1779        # NOTE we need to translate protocol player IDs to visible player IDs here as well,
1780        # to cover cases where a native player (e.g., native AirPlay) has grouped protocol players
1781        # (e.g., Sonos AirPlay protocol players)
1782        for player in expanded_can_group_with:
1783            if player.type == PlayerType.PROTOCOL:
1784                if not player.protocol_parent_id:
1785                    continue
1786                parent_player = self.mass.players.get_player(player.protocol_parent_id)
1787                if not parent_player or not _should_include_player(parent_player):
1788                    continue
1789                result.add(parent_player.player_id)
1790            elif _should_include_player(player):
1791                result.add(player.player_id)
1792
1793        # Scenario 2: External source is active - don't include protocol-based grouping
1794        # When an external source (e.g., Spotify Connect, TV) is active, grouping via
1795        # protocols (AirPlay, Sendspin, etc.) wouldn't work - only native grouping is available.
1796        if self._has_external_source_active():
1797            return result
1798
1799        # Translate can_group_with from active linked protocol(s) and add to result
1800        for linked in self.__attr_linked_protocols:
1801            if protocol_player := self.mass.players.get_player(linked.output_protocol_id):
1802                for player in self._translate_protocol_ids_to_visible(
1803                    protocol_player.state.can_group_with
1804                ):
1805                    if not _should_include_player(player):
1806                        continue
1807                    result.add(player.player_id)
1808        return result
1809
1810    @cached_property
1811    @final
1812    def __final_active_source(self) -> str | None:
1813        """
1814        Calculate the final active source based on any group memberships, source plugins etc.
1815
1816        Note: When an output protocol is active, the source remains the parent player's
1817        source since protocol players don't have their own queue/source - they only
1818        handle the actual streaming/playback.
1819        """
1820        # if the player is grouped/synced, use the active source of the group/parent player
1821        if parent_player_id := (self.__final_synced_to or self.__final_active_group):
1822            if parent_player := self.mass.players.get_player(parent_player_id):
1823                return parent_player.state.active_source
1824            return None  # should not happen but just in case
1825        if self.type == PlayerType.PROTOCOL:
1826            if self.protocol_parent_id and (
1827                parent_player := self.mass.players.get_player(self.protocol_parent_id)
1828            ):
1829                # if this is a protocol player, use the active source of the parent player
1830                return parent_player.state.active_source
1831            # fallback to None here if parent player not found,
1832            # protocol players should not have an active source themselves
1833            return None
1834        # if a plugin source is active that belongs to this player, return that
1835        for plugin_source in self.mass.players.get_plugin_sources():
1836            if plugin_source.in_use_by == self.player_id:
1837                return plugin_source.id
1838        output_protocol_domain: str | None = None
1839        if self.active_output_protocol and self.active_output_protocol != "native":
1840            if protocol_player := self.mass.players.get_player(self.active_output_protocol):
1841                output_protocol_domain = protocol_player.provider.domain
1842        # active source as reported by the player itself
1843        if (
1844            self.active_source
1845            # try to catch cases where player reports an active source
1846            # that is actually from an active output protocol (e.g. AirPlay)
1847            and self.active_source.lower() != output_protocol_domain
1848        ):
1849            return self.active_source
1850        # return the (last) known MA source - fallback to player's own queue source if none
1851        return self.__active_mass_source or self.player_id
1852
1853    @final
1854    def _translate_protocol_ids_to_visible(self, player_ids: set[str]) -> set[Player]:
1855        """
1856        Translate protocol player IDs to their visible parent players.
1857
1858        Protocol players are hidden and users interact with visible players
1859        (native or universal). This method translates protocol player IDs
1860        back to the visible (parent) players.
1861
1862        :param player_ids: Set of player IDs.
1863        :return: Set of visible players.
1864        """
1865        result: set[Player] = set()
1866        if not player_ids:
1867            return result
1868        for player_id in player_ids:
1869            target_player = self.mass.players.get_player(player_id)
1870            if not target_player:
1871                continue
1872            if target_player.type != PlayerType.PROTOCOL:
1873                # Non-protocol player is already visible - include directly
1874                result.add(target_player)
1875                continue
1876            # This is a protocol player - find its visible parent
1877            if not target_player.protocol_parent_id:
1878                continue
1879            parent_player = self.mass.players.get_player(target_player.protocol_parent_id)
1880            if not parent_player:
1881                continue
1882            result.add(parent_player)
1883        return result
1884
1885    @final
1886    def _has_external_source_active(self) -> bool:
1887        """
1888        Check if an external (non-MA-managed) source is currently active.
1889
1890        External sources include things like Spotify Connect, TV input, etc.
1891        When an external source is active, protocol-based grouping is not available.
1892
1893        :return: True if an external source is active, False otherwise.
1894        """
1895        active_source = self.__final_active_source
1896        if active_source is None:
1897            return False
1898
1899        # Player's own ID means MA queue is (or was) active
1900        if active_source == self.player_id:
1901            return False
1902
1903        # Check if it's a known queue ID
1904        if self.mass.player_queues.get(active_source):
1905            return False
1906
1907        # Check if it's a plugin source - if not, it's an external source
1908        return not any(
1909            plugin_source.id == active_source
1910            for plugin_source in self.mass.players.get_plugin_sources()
1911        )
1912
1913    @final
1914    def _expand_can_group_with(self) -> set[Player]:
1915        """
1916        Expand the 'can-group-with' to include all players from provider instance IDs.
1917
1918        This method expands any provider instance IDs (e.g., "airplay", "chromecast")
1919        in the group members to all (available) players of that provider
1920
1921        :return: Set of available players in the can-group-with.
1922        """
1923        result = set()
1924
1925        for member_id in self.can_group_with:
1926            if player := self.mass.players.get_player(member_id):
1927                result.add(player)
1928                continue  # already a player ID
1929            # Check if member_id is a provider instance ID
1930            if provider := self.mass.get_provider(member_id):
1931                for player in self.mass.players.all_players(
1932                    return_unavailable=False,  # Only include available players
1933                    provider_filter=provider.instance_id,
1934                    return_protocol_players=True,
1935                ):
1936                    result.add(player)
1937        return result
1938
1939    # The id of the (last) active mass source.
1940    # This is to keep track of the last active MA source for the player,
1941    # so we can restore it when needed (e.g. after switching to a plugin source).
1942    __active_mass_source: str | None = None
1943
1944    @final
1945    def set_active_mass_source(self, value: str | None) -> None:
1946        """
1947        Set the id of the (last) active mass source.
1948
1949        This is to keep track of the last active MA source for the player,
1950        so we can restore it when needed (e.g. after switching to a plugin source).
1951        """
1952        self.mass.cancel_timer(f"set_mass_source_{self.player_id}")
1953        self.__active_mass_source = value
1954        self.update_state()
1955
1956    __stop_called: bool = False
1957
1958    @final
1959    def mark_stop_called(self) -> None:
1960        """Mark that the STOP command was called on the player."""
1961        self.__stop_called = True
1962
1963    @property
1964    @final
1965    def stop_called(self) -> bool:
1966        """
1967        Return True if the STOP command was called on the player.
1968
1969        This is used to differentiate between a user-initiated stop
1970        and a natural end of playback (e.g. end of track/queue).
1971        mainly for debugging/logging purposes by the streams controller.
1972        """
1973        return self.__stop_called
1974
1975    def __hash__(self) -> int:
1976        """Return a hash of the Player."""
1977        return hash(self.player_id)
1978
1979    def __str__(self) -> str:
1980        """Return a string representation of the Player."""
1981        return f"Player {self.name} ({self.player_id})"
1982
1983    def __repr__(self) -> str:
1984        """Return a string representation of the Player."""
1985        return f"<Player name={self.name} id={self.player_id} available={self.available}>"
1986
1987    def __eq__(self, other: object) -> bool:
1988        """Check equality of two Player objects."""
1989        if not isinstance(other, Player):
1990            return False
1991        return self.player_id == other.player_id
1992
1993    def __ne__(self, other: object) -> bool:
1994        """Check inequality of two Player objects."""
1995        return not self.__eq__(other)
1996
1997
1998__all__ = [
1999    # explicitly re-export the models we imported from the models package,
2000    # for convenience reasons
2001    "EXTRA_ATTRIBUTES_TYPES",
2002    "DeviceInfo",
2003    "Player",
2004    "PlayerMedia",
2005    "PlayerSource",
2006    "PlayerState",
2007]
2008