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