music-assistant-server

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