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