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        if self.__attr_active_output_protocol == protocol_id:
1014            return  # No change
1015        if protocol_id == self.player_id:
1016            protocol_id = "native"  # Normalize to "native" for native player
1017        if protocol_id:
1018            protocol_name = protocol_id
1019            if protocol_id == "native":
1020                protocol_name = "Native"
1021            elif protocol_player := self.mass.players.get_player(protocol_id):
1022                protocol_name = protocol_player.provider.name
1023            self.logger.info(
1024                "Setting active output protocol on %s to %s",
1025                self.display_name,
1026                protocol_name,
1027            )
1028        else:
1029            self.logger.info(
1030                "Clearing active output protocol on %s",
1031                self.display_name,
1032            )
1033        self.__attr_active_output_protocol = protocol_id
1034        self.update_state()
1035
1036    @final
1037    def set_linked_output_protocols(self, protocols: list[OutputProtocol]) -> None:
1038        """
1039        Set the actively linked output protocol players.
1040
1041        :param protocols: List of OutputProtocol objects representing active protocol players.
1042        """
1043        self.__attr_linked_protocols = protocols
1044        self.mass.players.trigger_player_update(self.player_id)
1045
1046    @final
1047    def set_protocol_parent_id(self, parent_id: str | None) -> None:
1048        """
1049        Set the parent player_id for protocol players.
1050
1051        :param parent_id: The player_id of the parent player, or None to clear.
1052        """
1053        self.__attr_protocol_parent_id = parent_id
1054        self.mass.players.trigger_player_update(self.player_id)
1055
1056    @final
1057    def get_linked_protocol(self, protocol_domain: str) -> OutputProtocol | None:
1058        """Get a linked protocol by domain with current availability."""
1059        for linked in self.__attr_linked_protocols:
1060            if linked.protocol_domain == protocol_domain:
1061                protocol_player = self.mass.players.get_player(linked.output_protocol_id)
1062                current_available = protocol_player.available if protocol_player else False
1063                return OutputProtocol(
1064                    output_protocol_id=linked.output_protocol_id,
1065                    name=protocol_player.provider.name
1066                    if protocol_player
1067                    else linked.protocol_domain.title(),
1068                    protocol_domain=linked.protocol_domain,
1069                    priority=linked.priority,
1070                    available=current_available,
1071                    is_native=False,
1072                )
1073        return None
1074
1075    @final
1076    def get_output_protocol_by_domain(self, protocol_domain: str) -> OutputProtocol | None:
1077        """
1078        Get an output protocol by domain, including native protocol.
1079
1080        Unlike get_linked_protocol, this also checks if the player's native protocol
1081        matches the requested domain.
1082
1083        :param protocol_domain: The protocol domain to search for (e.g., "airplay", "sonos").
1084        """
1085        for output_protocol in self.output_protocols:
1086            if output_protocol.protocol_domain == protocol_domain:
1087                return output_protocol
1088        return None
1089
1090    @final
1091    def get_protocol_player(self, player_id: str) -> Player | None:
1092        """Get the protocol Player for a given player_id."""
1093        if player_id == "native":
1094            return self if PlayerFeature.PLAY_MEDIA in self.supported_features else None
1095        return self.mass.players.get_player(player_id)
1096
1097    @final
1098    def get_preferred_protocol_player(self) -> Player | None:
1099        """Get the best available protocol player by priority."""
1100        for linked in sorted(self.__attr_linked_protocols, key=lambda x: x.priority):
1101            if protocol_player := self.mass.players.get_player(linked.output_protocol_id):
1102                if protocol_player.available:
1103                    return protocol_player
1104        return None
1105
1106    @final
1107    def update_state(self, force_update: bool = False, signal_event: bool = True) -> None:
1108        """
1109        Update the PlayerState from the current state of the player.
1110
1111        This method should be called to update the player's state
1112        and signal any changes to the PlayerController.
1113
1114        :param force_update: If True, a state update event will be
1115        pushed even if the state has not actually changed.
1116        :param signal_event: If True, signal the state update event to the PlayerController.
1117        """
1118        self.mass.verify_event_loop_thread("player.update_state")
1119        # clear the dict for the cached properties
1120        self._cache.clear()
1121        # calculate the new state
1122        prev_media_checksum = self._get_player_media_checksum()
1123        changed_values = self.__calculate_player_state()
1124        if prev_media_checksum != self._get_player_media_checksum():
1125            # current media changed, call the media updated callback
1126            self._on_player_media_updated()
1127        # ignore some values that are not relevant for the state
1128        changed_values.pop("elapsed_time_last_updated", None)
1129        changed_values.pop("extra_attributes.seq_no", None)
1130        changed_values.pop("extra_attributes.last_poll", None)
1131        changed_values.pop("current_media.elapsed_time_last_updated", None)
1132        # persist the default name if it changed
1133        if self.name and self.config.default_name != self.name:
1134            self.mass.config.set_player_default_name(self.player_id, self.name)
1135        # persist the player type if it changed
1136        if self.type != self._config.player_type:
1137            self.mass.config.set_player_type(self.player_id, self.type)
1138        # return early if nothing changed (unless force_update is True)
1139        if len(changed_values) == 0 and not force_update:
1140            return
1141
1142        # signal the state update to the PlayerController
1143        if signal_event:
1144            self.mass.players.signal_player_state_update(self, changed_values)
1145
1146    @final
1147    def set_current_media(  # noqa: PLR0913
1148        self,
1149        uri: str,
1150        media_type: MediaType = MediaType.UNKNOWN,
1151        title: str | None = None,
1152        artist: str | None = None,
1153        album: str | None = None,
1154        image_url: str | None = None,
1155        duration: int | None = None,
1156        source_id: str | None = None,
1157        queue_item_id: str | None = None,
1158        custom_data: dict[str, Any] | None = None,
1159        clear_all: bool = False,
1160    ) -> None:
1161        """
1162        Set current_media helper.
1163
1164        Assumes use of '_attr_current_media'.
1165        """
1166        if self._attr_current_media is None or clear_all:
1167            self._attr_current_media = PlayerMedia(
1168                uri=uri,
1169                media_type=media_type,
1170            )
1171        self._attr_current_media.uri = uri
1172        if media_type != MediaType.UNKNOWN:
1173            self._attr_current_media.media_type = media_type
1174        if title:
1175            self._attr_current_media.title = title
1176        if artist:
1177            self._attr_current_media.artist = artist
1178        if album:
1179            self._attr_current_media.album = album
1180        if image_url:
1181            self._attr_current_media.image_url = image_url
1182        if duration:
1183            self._attr_current_media.duration = duration
1184        if source_id:
1185            self._attr_current_media.source_id = source_id
1186        if queue_item_id:
1187            self._attr_current_media.queue_item_id = queue_item_id
1188        if custom_data:
1189            self._attr_current_media.custom_data = custom_data
1190
1191    @final
1192    def set_config(self, config: PlayerConfig) -> None:
1193        """
1194        Set/update the player config.
1195
1196        May only be called by the PlayerController.
1197        """
1198        # TODO: validate that caller is the PlayerController ?
1199        self._config = config
1200
1201    @final
1202    def set_initialized(self) -> None:
1203        """Set the player as initialized."""
1204        self.__initialized.set()
1205
1206    @final
1207    def to_dict(self) -> dict[str, Any]:
1208        """Return the (serializable) dict representation of the Player."""
1209        return self.state.to_dict()
1210
1211    @final
1212    def supports_feature(self, feature: PlayerFeature) -> bool:
1213        """Return True if this player supports the given feature."""
1214        return feature in self.supported_features
1215
1216    @final
1217    def check_feature(self, feature: PlayerFeature) -> None:
1218        """Check if this player supports the given feature."""
1219        if not self.supports_feature(feature):
1220            raise UnsupportedFeaturedException(
1221                f"Player {self.display_name} does not support feature {feature.name}"
1222            )
1223
1224    @final
1225    def _get_player_media_checksum(self) -> str:
1226        """Return a checksum for the current media."""
1227        if not (media := self.state.current_media):
1228            return ""
1229        return (
1230            f"{media.uri}|{media.title}|{media.source_id}|{media.queue_item_id}|"
1231            f"{media.image_url}|{media.duration}|{media.elapsed_time}"
1232        )
1233
1234    @final
1235    def _check_feature_with_active_protocol(
1236        self, feature: PlayerFeature, active_only: bool = False
1237    ) -> bool:
1238        """
1239        Check if a feature is supported considering the active output protocol.
1240
1241        If an active output protocol is set (and not native), checks that protocol
1242        player's features. Otherwise checks the native player's features.
1243
1244        :param feature: The PlayerFeature to check.
1245        :return: True if the feature is supported by the active protocol or native player.
1246        """
1247        # If active output protocol is set and not native, check protocol player's features
1248        if (
1249            self.__attr_active_output_protocol
1250            and self.__attr_active_output_protocol != "native"
1251            and (
1252                protocol_player := self.mass.players.get_player(self.__attr_active_output_protocol)
1253            )
1254        ):
1255            return feature in protocol_player.supported_features
1256        # Otherwise check native player's features
1257        return feature in self.supported_features
1258
1259    @final
1260    def _get_protocol_player_for_feature(
1261        self,
1262        feature: PlayerFeature,
1263    ) -> Player | None:
1264        """Get player(protocol) which has the given PlayerFeature."""
1265        # prefer native player
1266        if feature in self.supported_features:
1267            return self
1268        # Otherwise, use the first available linked protocol
1269        for linked in self.linked_output_protocols:
1270            if (
1271                (protocol_player := self.mass.players.get_player(linked.output_protocol_id))
1272                and protocol_player.available
1273                and feature in protocol_player.supported_features
1274            ):
1275                return protocol_player
1276
1277        return None
1278
1279    @final
1280    def __calculate_player_state(
1281        self,
1282    ) -> dict[str, tuple[Any, Any]]:
1283        """
1284        Calculate the (current) and FINAL PlayerState.
1285
1286        This method is called when we're updating the player,
1287        and we compare the current state with the previous state to determine
1288        if we need to signal a state change to API consumers.
1289
1290        Returns a dict with the state attributes that have changed.
1291        """
1292        playback_state, elapsed_time, elapsed_time_last_updated = self.__final_playback_state
1293        prev_state = deepcopy(self._state)
1294        self._state = PlayerState(
1295            player_id=self.player_id,
1296            provider=self.provider_id,
1297            type=self.type,
1298            available=self.enabled and self.available,
1299            device_info=self.device_info,
1300            supported_features=self.__final_supported_features,
1301            playback_state=playback_state,
1302            elapsed_time=elapsed_time,
1303            elapsed_time_last_updated=elapsed_time_last_updated,
1304            powered=self.__final_power_state,
1305            volume_level=self.__final_volume_level,
1306            volume_muted=self.__final_volume_muted_state,
1307            group_members=UniqueList(self.__final_group_members),
1308            static_group_members=UniqueList(self.static_group_members),
1309            can_group_with=self.__final_can_group_with,
1310            synced_to=self.__final_synced_to,
1311            active_source=self.__final_active_source,
1312            source_list=self.__final_source_list,
1313            active_group=self.__final_active_group,
1314            current_media=self.__final_current_media,
1315            active_sound_mode=self.active_sound_mode,
1316            sound_mode_list=self.sound_mode_list,
1317            options=self.options,
1318            name=self.display_name,
1319            enabled=self.enabled,
1320            hide_in_ui=self.hide_in_ui,
1321            expose_to_ha=self.expose_to_ha,
1322            icon=self.icon,
1323            group_volume=self.group_volume,
1324            extra_attributes=self.extra_attributes,
1325            power_control=self.power_control,
1326            volume_control=self.volume_control,
1327            mute_control=self.mute_control,
1328            output_protocols=self.output_protocols,
1329            active_output_protocol=self.__attr_active_output_protocol,
1330        )
1331
1332        # track stop called state
1333        if (
1334            prev_state.playback_state == PlaybackState.IDLE
1335            and self._state.playback_state != PlaybackState.IDLE
1336        ):
1337            self.__stop_called = False
1338        elif (
1339            prev_state.playback_state != PlaybackState.IDLE
1340            and self._state.playback_state == PlaybackState.IDLE
1341        ):
1342            self.__stop_called = True
1343            # when we're going to idle,
1344            # we want to reset the active mass source after a short delay
1345            # this is done using a timer which gets reset if the player starts playing again
1346            # before the timer is up, using the task_id
1347            self.mass.call_later(
1348                2, self.set_active_mass_source, None, task_id=f"set_mass_source_{self.player_id}"
1349            )
1350
1351        return get_changed_dataclass_values(
1352            prev_state,
1353            self._state,
1354            recursive=True,
1355        )
1356
1357    @cached_property
1358    @final
1359    def __final_playback_state(self) -> tuple[PlaybackState, float | None, float | None]:
1360        """
1361        Return the FINAL playback state based on the playercontrol which may have been set-up.
1362
1363        Returns a tuple of (playback_state, elapsed_time, elapsed_time_last_updated).
1364        """
1365        # If an output protocol is active (and not native), use the protocol player's state
1366        if (
1367            self.__attr_active_output_protocol
1368            and self.__attr_active_output_protocol != "native"
1369            and (
1370                protocol_player := self.mass.players.get_player(self.__attr_active_output_protocol)
1371            )
1372            and protocol_player.playback_state != PlaybackState.IDLE
1373        ):
1374            return (
1375                protocol_player.state.playback_state,
1376                protocol_player.state.elapsed_time,
1377                protocol_player.state.elapsed_time_last_updated,
1378            )
1379        # If we're synced, use the syncleader state for playback state and elapsed time
1380        # NOTE: Don't do this for the active group player,
1381        # because the group player relies on the sync leader for state info.
1382        parent_id = self.__final_synced_to
1383        if parent_id and (parent_player := self.mass.players.get_player(parent_id)):
1384            return (
1385                parent_player.state.playback_state,
1386                parent_player.state.elapsed_time,
1387                parent_player.state.elapsed_time_last_updated,
1388            )
1389        return (self.playback_state, self.elapsed_time, self.elapsed_time_last_updated)
1390
1391    @cached_property
1392    @final
1393    def __final_power_state(self) -> bool | None:
1394        """Return the FINAL power state based on the playercontrol which may have been set-up."""
1395        power_control = self.power_control
1396        if power_control == PLAYER_CONTROL_FAKE:
1397            return bool(self.extra_data.get(ATTR_FAKE_POWER, False))
1398        if power_control == PLAYER_CONTROL_NATIVE:
1399            return self.powered
1400        if power_control == PLAYER_CONTROL_NONE:
1401            return None
1402        # handle player control for power if set
1403        if control := self.mass.players.get_player_control(power_control):
1404            return control.power_state
1405        return None
1406
1407    @cached_property
1408    @final
1409    def __final_volume_level(self) -> int | None:
1410        """Return the FINAL volume level based on the playercontrol which may have been set-up."""
1411        volume_control = self.volume_control
1412        if volume_control == PLAYER_CONTROL_FAKE:
1413            return int(self.extra_data.get(ATTR_FAKE_VOLUME, 0))
1414        if volume_control == PLAYER_CONTROL_NATIVE:
1415            return self.volume_level
1416        if volume_control == PLAYER_CONTROL_NONE:
1417            return None
1418        # handle protocol player as volume control
1419        if control := self.mass.players.get_player(volume_control):
1420            return control.volume_level
1421        # handle player control for volume if set
1422        if player_control := self.mass.players.get_player_control(volume_control):
1423            return player_control.volume_level
1424        return None
1425
1426    @cached_property
1427    @final
1428    def __final_volume_muted_state(self) -> bool | None:
1429        """Return the FINAL mute state based on any playercontrol which may have been set-up."""
1430        mute_control = self.mute_control
1431        if mute_control == PLAYER_CONTROL_FAKE:
1432            return bool(self.extra_data.get(ATTR_FAKE_MUTE, False))
1433        if mute_control == PLAYER_CONTROL_NATIVE:
1434            return self.volume_muted
1435        if mute_control == PLAYER_CONTROL_NONE:
1436            return None
1437        # handle protocol player as mute control
1438        if control := self.mass.players.get_player(mute_control):
1439            return control.volume_muted
1440        # handle player control for mute if set
1441        if player_control := self.mass.players.get_player_control(mute_control):
1442            return player_control.volume_muted
1443        return None
1444
1445    @cached_property
1446    @final
1447    def __final_active_group(self) -> str | None:
1448        """
1449        Return the player id of any playergroup that is currently active for this player.
1450
1451        This will return the id of the groupplayer if any groups are active.
1452        If no groups are currently active, this will return None.
1453        """
1454        if self.type == PlayerType.PROTOCOL:
1455            # protocol players should not have an active group,
1456            # they follow the group state of their parent player
1457            return None
1458        for group_player in self.mass.players.all_players(
1459            return_unavailable=False, return_disabled=False
1460        ):
1461            if group_player.type != PlayerType.GROUP:
1462                continue
1463            if group_player.player_id == self.player_id:
1464                continue
1465            if group_player.playback_state not in (PlaybackState.PLAYING, PlaybackState.PAUSED):
1466                continue
1467            if self.player_id in group_player.group_members:
1468                return group_player.player_id
1469        return None
1470
1471    @cached_property
1472    @final
1473    def __final_current_media(self) -> PlayerMedia | None:
1474        """Return the FINAL current media for the player."""
1475        if self.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS):
1476            # if an announcement is in progress, return announcement details
1477            return PlayerMedia(
1478                uri="announcement",
1479                media_type=MediaType.ANNOUNCEMENT,
1480                title="ANNOUNCEMENT",
1481            )
1482
1483        # if the player is grouped/synced, use the current_media of the group/parent player
1484        if parent_player_id := (self.__final_active_group or self.__final_synced_to):
1485            if parent_player_id != self.player_id and (
1486                parent_player := self.mass.players.get_player(parent_player_id)
1487            ):
1488                return parent_player.state.current_media
1489            return None  # if parent player not found, return None for current media
1490        # if this is a protocol player, use the current_media of the parent player
1491        if self.type == PlayerType.PROTOCOL and self.__attr_protocol_parent_id:
1492            if parent_player := self.mass.players.get_player(self.__attr_protocol_parent_id):
1493                return parent_player.state.current_media
1494        # if a pluginsource is currently active, return those details
1495        active_source = self.__final_active_source
1496        if (
1497            active_source
1498            and (source := self.mass.players.get_plugin_source(active_source))
1499            and source.metadata
1500        ):
1501            return PlayerMedia(
1502                uri=source.metadata.uri or source.id,
1503                media_type=MediaType.PLUGIN_SOURCE,
1504                title=source.metadata.title,
1505                artist=source.metadata.artist,
1506                album=source.metadata.album,
1507                image_url=source.metadata.image_url,
1508                duration=source.metadata.duration,
1509                source_id=source.id,
1510                elapsed_time=source.metadata.elapsed_time,
1511                elapsed_time_last_updated=source.metadata.elapsed_time_last_updated,
1512            )
1513        # if MA queue is active, return those details
1514        active_queue: PlayerQueue | None = None
1515        if not active_queue and active_source:
1516            active_queue = self.mass.player_queues.get(active_source)
1517        if not active_queue and self.active_source is None:
1518            active_queue = self.mass.player_queues.get(self.player_id)
1519        if active_queue and (current_item := active_queue.current_item):
1520            item_image_url = (
1521                # the image format needs to be 500x500 jpeg for maximum compatibility with players
1522                self.mass.metadata.get_image_url(current_item.image, size=500, image_format="jpeg")
1523                if current_item.image
1524                else None
1525            )
1526            if current_item.streamdetails and (
1527                stream_metadata := current_item.streamdetails.stream_metadata
1528            ):
1529                # handle stream metadata in streamdetails (e.g. for radio stream)
1530                return PlayerMedia(
1531                    uri=current_item.uri,
1532                    media_type=current_item.media_type,
1533                    title=stream_metadata.title or current_item.name,
1534                    artist=stream_metadata.artist,
1535                    album=stream_metadata.album or stream_metadata.description or current_item.name,
1536                    image_url=(stream_metadata.image_url or item_image_url),
1537                    duration=stream_metadata.duration or current_item.duration,
1538                    source_id=active_queue.queue_id,
1539                    queue_item_id=current_item.queue_item_id,
1540                    elapsed_time=stream_metadata.elapsed_time or int(active_queue.elapsed_time),
1541                    elapsed_time_last_updated=stream_metadata.elapsed_time_last_updated
1542                    or active_queue.elapsed_time_last_updated,
1543                )
1544            if media_item := current_item.media_item:
1545                # normal media item
1546                # we use getattr here to avoid issues with different media item types
1547                version = getattr(media_item, "version", None)
1548                album = getattr(media_item, "album", None)
1549                podcast = getattr(media_item, "podcast", None)
1550                metadata = getattr(media_item, "metadata", None)
1551                description = getattr(metadata, "description", None) if metadata else None
1552                return PlayerMedia(
1553                    uri=str(media_item.uri),
1554                    media_type=media_item.media_type,
1555                    title=f"{media_item.name} ({version})" if version else media_item.name,
1556                    artist=getattr(media_item, "artist_str", None),
1557                    album=album.name if album else podcast.name if podcast else description,
1558                    # the image format needs to be 500x500 jpeg for maximum player compatibility
1559                    image_url=self.mass.metadata.get_image_url(
1560                        current_item.media_item.image, size=500, image_format="jpeg"
1561                    )
1562                    or item_image_url
1563                    if current_item.media_item.image
1564                    else item_image_url,
1565                    duration=media_item.duration,
1566                    source_id=active_queue.queue_id,
1567                    queue_item_id=current_item.queue_item_id,
1568                    elapsed_time=int(active_queue.elapsed_time),
1569                    elapsed_time_last_updated=active_queue.elapsed_time_last_updated,
1570                )
1571
1572            # fallback to basic current item details
1573            return PlayerMedia(
1574                uri=current_item.uri,
1575                media_type=current_item.media_type,
1576                title=current_item.name,
1577                image_url=item_image_url,
1578                duration=current_item.duration,
1579                source_id=active_queue.queue_id,
1580                queue_item_id=current_item.queue_item_id,
1581                elapsed_time=int(active_queue.elapsed_time),
1582                elapsed_time_last_updated=active_queue.elapsed_time_last_updated,
1583            )
1584        if active_queue:
1585            # queue is active but no current item
1586            return None
1587        # return native current media if no group/queue is active
1588        if self.current_media:
1589            return PlayerMedia(
1590                uri=self.current_media.uri,
1591                media_type=self.current_media.media_type,
1592                title=self.current_media.title,
1593                artist=self.current_media.artist,
1594                album=self.current_media.album,
1595                image_url=self.current_media.image_url,
1596                duration=self.current_media.duration,
1597                source_id=self.current_media.source_id or active_source,
1598                queue_item_id=self.current_media.queue_item_id,
1599                elapsed_time=self.current_media.elapsed_time or int(self.elapsed_time)
1600                if self.elapsed_time
1601                else None,
1602                elapsed_time_last_updated=self.current_media.elapsed_time_last_updated
1603                or self.elapsed_time_last_updated,
1604            )
1605        return None
1606
1607    @cached_property
1608    @final
1609    def __final_source_list(self) -> UniqueList[PlayerSource]:
1610        """Return the FINAL source list for the player."""
1611        sources = UniqueList(self.source_list)
1612        if self.type == PlayerType.PROTOCOL:
1613            return sources
1614        # always ensure the Music Assistant Queue is in the source list
1615        mass_source = next((x for x in sources if x.id == self.player_id), None)
1616        if mass_source is None:
1617            # if the MA queue is not in the source list, add it
1618            mass_source = PlayerSource(
1619                id=self.player_id,
1620                name="Music Assistant Queue",
1621                passive=False,
1622                # TODO: Do we want to dynamically set these based on the queue state ?
1623                can_play_pause=True,
1624                can_seek=True,
1625                can_next_previous=True,
1626            )
1627            sources.append(mass_source)
1628        # append all/any plugin sources (convert to PlayerSource to avoid deepcopy issues)
1629        for plugin_source in self.mass.players.get_plugin_sources():
1630            if hasattr(plugin_source, "as_player_source"):
1631                sources.append(plugin_source.as_player_source())
1632            else:
1633                sources.append(plugin_source)
1634        return sources
1635
1636    @cached_property
1637    @final
1638    def __final_group_members(self) -> list[str]:
1639        """Return the FINAL group members of this player."""
1640        if self.__final_synced_to:
1641            # If player is synced to another player, it has no group members itself
1642            return []
1643
1644        # Start by translating native group_members to visible player IDs
1645        # This handles cases where a native player (e.g., native AirPlay) has grouped
1646        # protocol players (e.g., Sonos AirPlay protocol players) that need translation
1647        members: list[str] = []
1648        if self.type == PlayerType.PROTOCOL:
1649            # protocol players use their own group members without translation
1650            members.extend(self.group_members)
1651        else:
1652            translated_members = self._translate_protocol_ids_to_visible(set(self.group_members))
1653            for member in translated_members:
1654                if member.player_id not in members:
1655                    members.append(member.player_id)
1656
1657        # If there's an active linked protocol, include its group members (translated)
1658        if self.__attr_active_output_protocol and self.__attr_active_output_protocol != "native":
1659            if protocol_player := self.mass.players.get_player(self.__attr_active_output_protocol):
1660                # Translate protocol player IDs to visible player IDs
1661                protocol_members = self._translate_protocol_ids_to_visible(
1662                    set(protocol_player.group_members)
1663                )
1664                for member in protocol_members:
1665                    if member.player_id not in members:
1666                        members.append(member.player_id)
1667
1668        if self.type != PlayerType.GROUP:
1669            # Ensure the player_id is first in the group_members list
1670            if len(members) > 0 and members[0] != self.player_id:
1671                members = [self.player_id, *[m for m in members if m != self.player_id]]
1672            # If the only member is self, return empty list
1673            if members == [self.player_id]:
1674                return []
1675        return members
1676
1677    @cached_property
1678    @final
1679    def __final_synced_to(self) -> str | None:
1680        """
1681        Return the FINAL synced_to state.
1682
1683        This checks both native sync state and protocol player sync state,
1684        translating protocol player IDs to visible player IDs.
1685        """
1686        # First check the native synced_to from the property
1687        if native_synced_to := self.synced_to:
1688            return native_synced_to
1689
1690        for linked in self.__attr_linked_protocols:
1691            if not (protocol_player := self.mass.players.get_player(linked.output_protocol_id)):
1692                continue
1693            if protocol_player.synced_to:
1694                # Protocol player is synced, translate to visible player
1695                if proto_sync_parent := self.mass.players.get_player(protocol_player.synced_to):
1696                    if proto_sync_parent.type != PlayerType.PROTOCOL:
1697                        # Sync parent is already a visible player (e.g., native AirPlay player)
1698                        return proto_sync_parent.player_id
1699                    if proto_sync_parent.protocol_parent_id and (
1700                        parent := self.mass.players.get_player(proto_sync_parent.protocol_parent_id)
1701                    ):
1702                        # Sync parent is a protocol player, return its visible parent
1703                        return parent.player_id
1704
1705        return None
1706
1707    @cached_property
1708    @final
1709    def __final_supported_features(self) -> set[PlayerFeature]:
1710        """Return the FINAL supported features based supported output protocol(s)."""
1711        base_features = self.supported_features.copy()
1712        if self.__attr_active_output_protocol and self.__attr_active_output_protocol != "native":
1713            # Active linked protocol: add from that specific protocol
1714            if protocol_player := self.mass.players.get_player(self.__attr_active_output_protocol):
1715                for feature in protocol_player.supported_features:
1716                    if feature in ACTIVE_PROTOCOL_FEATURES:
1717                        base_features.add(feature)
1718        # Append (allowed features) from all linked protocols
1719        for linked in self.__attr_linked_protocols:
1720            if protocol_player := self.mass.players.get_player(linked.output_protocol_id):
1721                for feature in protocol_player.supported_features:
1722                    if feature in PROTOCOL_FEATURES:
1723                        base_features.add(feature)
1724        return base_features
1725
1726    @cached_property
1727    @final
1728    def __final_can_group_with(self) -> set[str]:
1729        """
1730        Return the FINAL set of player id's this player can group with.
1731
1732        This is a convenience property which calculates the final can_group_with set
1733        based on any linked protocol players and current player/grouped state.
1734
1735        If player is synced to a native parent: return empty set (already grouped).
1736        If player is synced to a protocol: can still group with other players.
1737        If no active linked protocol: return can_group_with from all active output protocols.
1738        If active linked protocol: return native can_group_with + active protocol's.
1739
1740        All protocol player IDs are translated to their visible parent player IDs.
1741        """
1742
1743        def _should_include_player(player: Player) -> bool:
1744            """Check if a player should be included in the can-group-with set."""
1745            if not player.available:
1746                return False
1747            if player.player_id == self.player_id:
1748                return False  # Don't include self
1749            # Don't include (playing) players that have group members (they are group leaders)
1750            if (  # noqa: SIM103
1751                player.state.playback_state in (PlaybackState.PLAYING, PlaybackState.PAUSED)
1752                and player.group_members
1753            ):
1754                return False
1755            return True
1756
1757        if self.__final_synced_to:
1758            # player is already synced/grouped, cannot group with others
1759            return set()
1760
1761        expanded_can_group_with = self._expand_can_group_with()
1762        # Scenario 1: Player is a protocol player - just return the (expanded) result
1763        if self.type == PlayerType.PROTOCOL:
1764            return {x.player_id for x in expanded_can_group_with}
1765
1766        result: set[str] = set()
1767        # always start with the native can_group_with options (expanded from provider instance IDs)
1768        # NOTE we need to translate protocol player IDs to visible player IDs here as well,
1769        # to cover cases where a native player (e.g., native AirPlay) has grouped protocol players
1770        # (e.g., Sonos AirPlay protocol players)
1771        for player in expanded_can_group_with:
1772            if player.type == PlayerType.PROTOCOL:
1773                if not player.protocol_parent_id:
1774                    continue
1775                parent_player = self.mass.players.get_player(player.protocol_parent_id)
1776                if not parent_player or not _should_include_player(parent_player):
1777                    continue
1778                result.add(parent_player.player_id)
1779            elif _should_include_player(player):
1780                result.add(player.player_id)
1781
1782        # Scenario 2: External source is active - don't include protocol-based grouping
1783        # When an external source (e.g., Spotify Connect, TV) is active, grouping via
1784        # protocols (AirPlay, Sendspin, etc.) wouldn't work - only native grouping is available.
1785        if self._has_external_source_active():
1786            return result
1787
1788        # Translate can_group_with from active linked protocol(s) and add to result
1789        for linked in self.__attr_linked_protocols:
1790            if protocol_player := self.mass.players.get_player(linked.output_protocol_id):
1791                for player in self._translate_protocol_ids_to_visible(
1792                    protocol_player.state.can_group_with
1793                ):
1794                    if not _should_include_player(player):
1795                        continue
1796                    result.add(player.player_id)
1797        return result
1798
1799    @cached_property
1800    @final
1801    def __final_active_source(self) -> str | None:
1802        """
1803        Calculate the final active source based on any group memberships, source plugins etc.
1804
1805        Note: When an output protocol is active, the source remains the parent player's
1806        source since protocol players don't have their own queue/source - they only
1807        handle the actual streaming/playback.
1808        """
1809        # if the player is grouped/synced, use the active source of the group/parent player
1810        if parent_player_id := (self.__final_synced_to or self.__final_active_group):
1811            if parent_player := self.mass.players.get_player(parent_player_id):
1812                return parent_player.state.active_source
1813            return None  # should not happen but just in case
1814        if self.type == PlayerType.PROTOCOL:
1815            if self.protocol_parent_id and (
1816                parent_player := self.mass.players.get_player(self.protocol_parent_id)
1817            ):
1818                # if this is a protocol player, use the active source of the parent player
1819                return parent_player.state.active_source
1820            # fallback to None here if parent player not found,
1821            # protocol players should not have an active source themselves
1822            return None
1823        # if a plugin source is active that belongs to this player, return that
1824        for plugin_source in self.mass.players.get_plugin_sources():
1825            if plugin_source.in_use_by == self.player_id:
1826                return plugin_source.id
1827        output_protocol_domain: str | None = None
1828        if self.active_output_protocol and self.active_output_protocol != "native":
1829            if protocol_player := self.mass.players.get_player(self.active_output_protocol):
1830                output_protocol_domain = protocol_player.provider.domain
1831        # active source as reported by the player itself
1832        if (
1833            self.active_source
1834            # try to catch cases where player reports an active source
1835            # that is actually from an active output protocol (e.g. AirPlay)
1836            and self.active_source.lower() != output_protocol_domain
1837        ):
1838            return self.active_source
1839        # return the (last) known MA source - fallback to player's own queue source if none
1840        return self.__active_mass_source or self.player_id
1841
1842    @final
1843    def _translate_protocol_ids_to_visible(self, player_ids: set[str]) -> set[Player]:
1844        """
1845        Translate protocol player IDs to their visible parent players.
1846
1847        Protocol players are hidden and users interact with visible players
1848        (native or universal). This method translates protocol player IDs
1849        back to the visible (parent) players.
1850
1851        :param player_ids: Set of player IDs.
1852        :return: Set of visible players.
1853        """
1854        result: set[Player] = set()
1855        if not player_ids:
1856            return result
1857        for player_id in player_ids:
1858            target_player = self.mass.players.get_player(player_id)
1859            if not target_player:
1860                continue
1861            if target_player.type != PlayerType.PROTOCOL:
1862                # Non-protocol player is already visible - include directly
1863                result.add(target_player)
1864                continue
1865            # This is a protocol player - find its visible parent
1866            if not target_player.protocol_parent_id:
1867                continue
1868            parent_player = self.mass.players.get_player(target_player.protocol_parent_id)
1869            if not parent_player:
1870                continue
1871            result.add(parent_player)
1872        return result
1873
1874    @final
1875    def _has_external_source_active(self) -> bool:
1876        """
1877        Check if an external (non-MA-managed) source is currently active.
1878
1879        External sources include things like Spotify Connect, TV input, etc.
1880        When an external source is active, protocol-based grouping is not available.
1881
1882        :return: True if an external source is active, False otherwise.
1883        """
1884        active_source = self.__final_active_source
1885        if active_source is None:
1886            return False
1887
1888        # Player's own ID means MA queue is (or was) active
1889        if active_source == self.player_id:
1890            return False
1891
1892        # Check if it's a known queue ID
1893        if self.mass.player_queues.get(active_source):
1894            return False
1895
1896        # Check if it's a plugin source - if not, it's an external source
1897        return not any(
1898            plugin_source.id == active_source
1899            for plugin_source in self.mass.players.get_plugin_sources()
1900        )
1901
1902    @final
1903    def _expand_can_group_with(self) -> set[Player]:
1904        """
1905        Expand the 'can-group-with' to include all players from provider instance IDs.
1906
1907        This method expands any provider instance IDs (e.g., "airplay", "chromecast")
1908        in the group members to all (available) players of that provider
1909
1910        :return: Set of available players in the can-group-with.
1911        """
1912        result = set()
1913
1914        for member_id in self.can_group_with:
1915            if player := self.mass.players.get_player(member_id):
1916                result.add(player)
1917                continue  # already a player ID
1918            # Check if member_id is a provider instance ID
1919            if provider := self.mass.get_provider(member_id):
1920                for player in self.mass.players.all_players(
1921                    return_unavailable=False,  # Only include available players
1922                    provider_filter=provider.instance_id,
1923                    return_protocol_players=True,
1924                ):
1925                    result.add(player)
1926        return result
1927
1928    # The id of the (last) active mass source.
1929    # This is to keep track of the last active MA source for the player,
1930    # so we can restore it when needed (e.g. after switching to a plugin source).
1931    __active_mass_source: str | None = None
1932
1933    @final
1934    def set_active_mass_source(self, value: str | None) -> None:
1935        """
1936        Set the id of the (last) active mass source.
1937
1938        This is to keep track of the last active MA source for the player,
1939        so we can restore it when needed (e.g. after switching to a plugin source).
1940        """
1941        self.mass.cancel_timer(f"set_mass_source_{self.player_id}")
1942        self.__active_mass_source = value
1943        self.update_state()
1944
1945    __stop_called: bool = False
1946
1947    @final
1948    def mark_stop_called(self) -> None:
1949        """Mark that the STOP command was called on the player."""
1950        self.__stop_called = True
1951
1952    @property
1953    @final
1954    def stop_called(self) -> bool:
1955        """
1956        Return True if the STOP command was called on the player.
1957
1958        This is used to differentiate between a user-initiated stop
1959        and a natural end of playback (e.g. end of track/queue).
1960        mainly for debugging/logging purposes by the streams controller.
1961        """
1962        return self.__stop_called
1963
1964    def __hash__(self) -> int:
1965        """Return a hash of the Player."""
1966        return hash(self.player_id)
1967
1968    def __str__(self) -> str:
1969        """Return a string representation of the Player."""
1970        return f"Player {self.name} ({self.player_id})"
1971
1972    def __repr__(self) -> str:
1973        """Return a string representation of the Player."""
1974        return f"<Player name={self.name} id={self.player_id} available={self.available}>"
1975
1976    def __eq__(self, other: object) -> bool:
1977        """Check equality of two Player objects."""
1978        if not isinstance(other, Player):
1979            return False
1980        return self.player_id == other.player_id
1981
1982    def __ne__(self, other: object) -> bool:
1983        """Check inequality of two Player objects."""
1984        return not self.__eq__(other)
1985
1986
1987__all__ = [
1988    # explicitly re-export the models we imported from the models package,
1989    # for convenience reasons
1990    "EXTRA_ATTRIBUTES_TYPES",
1991    "DeviceInfo",
1992    "Player",
1993    "PlayerMedia",
1994    "PlayerSource",
1995    "PlayerState",
1996]
1997