/
/
/
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 the serverside Player object is not the same as the clientside Player object,
7which is a dataclass in the models package containing the player state.
8"""
9
10from __future__ import annotations
11
12import time
13from abc import ABC, abstractmethod
14from collections.abc import Callable
15from copy import deepcopy
16from typing import TYPE_CHECKING, Any, cast, final
17
18from music_assistant_models.constants import (
19 EXTRA_ATTRIBUTES_TYPES,
20 PLAYER_CONTROL_FAKE,
21 PLAYER_CONTROL_NATIVE,
22 PLAYER_CONTROL_NONE,
23)
24from music_assistant_models.enums import MediaType, PlaybackState, PlayerFeature, PlayerType
25from music_assistant_models.errors import UnsupportedFeaturedException
26from music_assistant_models.player import DeviceInfo, PlayerMedia, PlayerSource
27from music_assistant_models.player import Player as PlayerState
28from music_assistant_models.unique_list import UniqueList
29from propcache import under_cached_property as cached_property
30
31from music_assistant.constants import (
32 ATTR_ANNOUNCEMENT_IN_PROGRESS,
33 ATTR_FAKE_MUTE,
34 ATTR_FAKE_POWER,
35 ATTR_FAKE_VOLUME,
36 CONF_ENTRY_PLAYER_ICON,
37 CONF_EXPOSE_PLAYER_TO_HA,
38 CONF_FLOW_MODE,
39 CONF_HIDE_IN_UI,
40 CONF_MUTE_CONTROL,
41 CONF_POWER_CONTROL,
42 CONF_SMART_FADES_MODE,
43 CONF_VOLUME_CONTROL,
44)
45from music_assistant.helpers.util import get_changed_dataclass_values
46
47if TYPE_CHECKING:
48 from music_assistant_models.config_entries import ConfigEntry, ConfigValueType, PlayerConfig
49
50 from .player_provider import PlayerProvider
51
52
53class Player(ABC):
54 """
55 Base representation of a Player within the Music Assistant Server.
56
57 Player Provider implementations should inherit from this base model.
58 """
59
60 _attr_type: PlayerType = PlayerType.PLAYER
61 _attr_supported_features: set[PlayerFeature]
62 _attr_group_members: list[str]
63 _attr_static_group_members: list[str]
64 _attr_device_info: DeviceInfo
65 _attr_can_group_with: set[str]
66 _attr_source_list: list[PlayerSource]
67 _attr_available: bool = True
68 _attr_name: str | None = None
69 _attr_powered: bool | None = None
70 _attr_playback_state: PlaybackState = PlaybackState.IDLE
71 _attr_volume_level: int | None = None
72 _attr_volume_muted: bool | None = None
73 _attr_elapsed_time: float | None = None
74 _attr_elapsed_time_last_updated: float | None = None
75 _attr_active_source: str | None = None
76 _attr_current_media: PlayerMedia | None = None
77 _attr_needs_poll: bool = False
78 _attr_poll_interval: int = 30
79 _attr_hidden_by_default: bool = False
80 _attr_expose_to_ha_by_default: bool = True
81 _attr_enabled_by_default: bool = True
82
83 def __init__(self, provider: PlayerProvider, player_id: str) -> None:
84 """Initialize the Player."""
85 # set mass as public variable
86 self.mass = provider.mass
87 self.logger = provider.logger
88 # initialize mutable attributes
89 self._attr_supported_features = set()
90 self._attr_group_members = []
91 self._attr_static_group_members = []
92 self._attr_device_info = DeviceInfo()
93 self._attr_can_group_with = set()
94 self._attr_source_list = []
95 # do not override/overwrite these private attributes below!
96 self._cache: dict[str, Any] = {} # storage dict for cached properties
97 self._player_id = player_id
98 self._provider = provider
99 self.mass.config.create_default_player_config(
100 player_id, self.provider_id, self.type, self.name, self.enabled_by_default
101 )
102 self._config = self.mass.config.get_base_player_config(player_id, self.provider_id)
103 self._extra_data: dict[str, Any] = {}
104 self._extra_attributes: dict[str, Any] = {}
105 self._on_unload_callbacks: list[Callable[[], None]] = []
106 self.__active_mass_source = player_id
107 # The PlayerState is the (snapshotted) final state of the player
108 # after applying any config overrides and other transformations,
109 # such as the display name and player controls.
110 # the state is updated when calling 'update_state' and is what is sent over the API.
111 self._state = PlayerState(
112 player_id=self.player_id,
113 provider=self.provider_id,
114 type=self.type,
115 name=self.display_name,
116 available=self.available,
117 device_info=self.device_info,
118 supported_features=self.supported_features,
119 playback_state=self.playback_state,
120 )
121
122 @property
123 def type(self) -> PlayerType:
124 """Return the type of the player."""
125 return self._attr_type
126
127 @property
128 def available(self) -> bool:
129 """Return if the player is available."""
130 return self._attr_available
131
132 @property
133 def name(self) -> str | None:
134 """Return the name of the player."""
135 return self._attr_name
136
137 @property
138 def supported_features(self) -> set[PlayerFeature]:
139 """Return the supported features of the player."""
140 return self._attr_supported_features
141
142 @property
143 def playback_state(self) -> PlaybackState:
144 """Return the current playback state of the player."""
145 return self._attr_playback_state
146
147 @property
148 def requires_flow_mode(self) -> bool:
149 """
150 Return if the player needs flow mode.
151
152 Will by default be set to True if the player does not support PlayerFeature.ENQUEUE
153 or has crossfade enabled without gapless support.
154 """
155 if PlayerFeature.ENQUEUE not in self.supported_features:
156 # without enqueue support, flow mode is required
157 return True
158 return (
159 # player has crossfade enabled without gapless support - flow mode is required
160 PlayerFeature.GAPLESS_PLAYBACK not in self.supported_features
161 and str(self._config.get_value(CONF_SMART_FADES_MODE)) != "disabled"
162 )
163
164 @property
165 def device_info(self) -> DeviceInfo:
166 """Return the device info of the player."""
167 return self._attr_device_info
168
169 @property
170 def elapsed_time(self) -> float | None:
171 """Return the elapsed time in (fractional) seconds of the current track (if any)."""
172 return self._attr_elapsed_time
173
174 @property
175 def elapsed_time_last_updated(self) -> float | None:
176 """
177 Return when the elapsed time was last updated.
178
179 return: The (UTC) timestamp when the elapsed time was last updated,
180 or None if it was never updated (or unknown).
181 """
182 return self._attr_elapsed_time_last_updated
183
184 @property
185 def group_members(self) -> list[str]:
186 """
187 Return the group members of the player.
188
189 If there are other players synced/grouped with this player,
190 this should return the id's of players synced to this player,
191 and this should include the player's own id (as first item in the list).
192
193 If there are currently no group members, this should return an empty list.
194 """
195 if self.type == PlayerType.PLAYER and (
196 len(self._attr_group_members) >= 1 and self.player_id not in self._attr_group_members
197 ):
198 # always ensure the player_id is in the group_members list for players
199 return [self.player_id, *self._attr_group_members]
200 if self._attr_group_members == [self.player_id]:
201 return []
202 return self._attr_group_members
203
204 @property
205 def static_group_members(self) -> list[str]:
206 """
207 Return the static group members for a player group.
208
209 For PlayerType.GROUP return the player_ids of members that must not be removed by
210 the user.
211 For all other player types return an empty list.
212 """
213 return self._attr_static_group_members
214
215 @property
216 def can_group_with(self) -> set[str]:
217 """
218 Return the id's of players this player can group with.
219
220 This should return set of player_id's this player can group/sync with
221 or just the provider's instance_id if all players can group with each other.
222 """
223 return self._attr_can_group_with
224
225 @property
226 def needs_poll(self) -> bool:
227 """Return if the player needs to be polled for state updates."""
228 return self._attr_needs_poll
229
230 @property
231 def poll_interval(self) -> int:
232 """
233 Return the (dynamic) poll interval for the player.
234
235 Only used if 'needs_poll' is set to True.
236 This should return the interval in seconds.
237 """
238 return self._attr_poll_interval
239
240 @property
241 def hidden_by_default(self) -> bool:
242 """Return if the player should be hidden in the UI by default."""
243 return self._attr_hidden_by_default
244
245 @property
246 def expose_to_ha_by_default(self) -> bool:
247 """Return if the player should be exposed to Home Assistant by default."""
248 return self._attr_expose_to_ha_by_default
249
250 @property
251 def enabled_by_default(self) -> bool:
252 """Return if the player should be enabled by default."""
253 return self._attr_enabled_by_default
254
255 @property
256 def _powered(self) -> bool | None:
257 """
258 Return if the player is powered on.
259
260 If the player does not support PlayerFeature.POWER,
261 or the state is (currently) unknown, this property may return None.
262
263 Note that this is NOT the final power state of the player,
264 as it may be overridden by a playercontrol.
265 Hence it's marked as a private property.
266 The final power state can be retrieved by using the 'powered' property.
267 """
268 return self._attr_powered
269
270 @property
271 def _volume_level(self) -> int | None:
272 """
273 Return the current volume level (0..100) of the player.
274
275 If the player does not support PlayerFeature.VOLUME_SET,
276 or the state is (currently) unknown, this property may return None.
277
278 Note that this is NOT the final volume level state of the player,
279 as it may be overridden by a playercontrol.
280 Hence it's marked as a private property.
281 The final volume level state can be retrieved by using the 'volume_level' property.
282 """
283 return self._attr_volume_level
284
285 @property
286 def _volume_muted(self) -> bool | None:
287 """
288 Return the current mute state of the player.
289
290 If the player does not support PlayerFeature.VOLUME_MUTE,
291 or the state is (currently) unknown, this property may return None.
292
293 Note that this is NOT the final muted state of the player,
294 as it may be overridden by a playercontrol.
295 Hence it's marked as a private property.
296 The final muted state can be retrieved by using the 'volume_muted' property.
297 """
298 return self._attr_volume_muted
299
300 @property
301 def _active_source(self) -> str | None:
302 """
303 Return the (id of) the active source of the player.
304
305 Only required if the player supports PlayerFeature.SELECT_SOURCE.
306
307 Set to None if the player is not currently playing a source or
308 the player_id if the player is currently playing a MA queue.
309
310 Note that this is NOT the final active source of the player,
311 as it may be overridden by a active group/sync membership.
312 Hence it's marked as a private property.
313 The final active source can be retrieved by using the 'active_source' property.
314 """
315 return self._attr_active_source
316
317 @property
318 def _current_media(self) -> PlayerMedia | None:
319 """
320 Return the current media being played by the player.
321
322 Note that this is NOT the final current media of the player,
323 as it may be overridden by a active group/sync membership.
324 Hence it's marked as a private property.
325 The final current media can be retrieved by using the 'current_media' property.
326 """
327 return self._attr_current_media
328
329 @property
330 def _source_list(self) -> list[PlayerSource]:
331 """
332 Return list of available (native) sources for this player.
333
334 Note that this is NOT the final source list of the player,
335 as we inject the MA queue source if the player is currently playing a MA queue.
336 Hence it's marked as a private property.
337 The final source list can be retrieved by using the 'source_list' property.
338 """
339 return self._attr_source_list
340
341 async def power(self, powered: bool) -> None:
342 """
343 Handle POWER command on the player.
344
345 Will only be called if the PlayerFeature.POWER is supported.
346
347 :param powered: bool if player should be powered on or off.
348 """
349 raise NotImplementedError("power needs to be implemented when PlayerFeature.POWER is set")
350
351 async def volume_set(self, volume_level: int) -> None:
352 """
353 Handle VOLUME_SET command on the player.
354
355 Will only be called if the PlayerFeature.VOLUME_SET is supported.
356
357 :param volume_level: volume level (0..100) to set on the player.
358 """
359 raise NotImplementedError(
360 "volume_set needs to be implemented when PlayerFeature.VOLUME_SET is set"
361 )
362
363 async def volume_mute(self, muted: bool) -> None:
364 """
365 Handle VOLUME MUTE command on the player.
366
367 Will only be called if the PlayerFeature.VOLUME_MUTE is supported.
368
369 :param muted: bool if player should be muted.
370 """
371 raise NotImplementedError(
372 "volume_mute needs to be implemented when PlayerFeature.VOLUME_MUTE is set"
373 )
374
375 async def play(self) -> None:
376 """Handle PLAY command on the player."""
377 raise NotImplementedError("play needs to be implemented")
378
379 @abstractmethod
380 async def stop(self) -> None:
381 """
382 Handle STOP command on the player.
383
384 Will only be called if the player reports PlayerFeature.PAUSE is supported or
385 player supports resuming of stopped playback.
386 """
387 raise NotImplementedError("stop needs to be implemented")
388
389 async def pause(self) -> None:
390 """
391 Handle PAUSE command on the player.
392
393 Will only be called if the player reports PlayerFeature.PAUSE is supported.
394 """
395 raise NotImplementedError("pause needs to be implemented when PlayerFeature.PAUSE is set")
396
397 async def next_track(self) -> None:
398 """
399 Handle NEXT_TRACK command on the player.
400
401 Will only be called if the player reports PlayerFeature.NEXT_PREVIOUS
402 is supported and the player is not currently playing a MA queue.
403 """
404 raise NotImplementedError(
405 "next_track needs to be implemented when PlayerFeature.NEXT_PREVIOUS is set"
406 )
407
408 async def previous_track(self) -> None:
409 """
410 Handle PREVIOUS_TRACK command on the player.
411
412 Will only be called if the player reports PlayerFeature.NEXT_PREVIOUS
413 is supported and the player is not currently playing a MA queue.
414 """
415 raise NotImplementedError(
416 "previous_track needs to be implemented when PlayerFeature.NEXT_PREVIOUS is set"
417 )
418
419 async def seek(self, position: int) -> None:
420 """
421 Handle SEEK command on the player.
422
423 Seek to a specific position in the current track.
424 Will only be called if the player reports PlayerFeature.SEEK is
425 supported and the player is NOT currently playing a MA queue.
426
427 :param position: The position to seek to, in seconds.
428 """
429 raise NotImplementedError("seek needs to be implemented when PlayerFeature.SEEK is set")
430
431 @abstractmethod
432 async def play_media(
433 self,
434 media: PlayerMedia,
435 ) -> None:
436 """
437 Handle PLAY MEDIA command on given player.
438
439 This is called by the Player controller to start playing Media on the player,
440 which can be a MA queue item/stream or a native source.
441 The provider's own implementation should work out how to handle this request.
442
443 :param media: Details of the item that needs to be played on the player.
444 """
445 raise NotImplementedError("play_media needs to be implemented")
446
447 async def enqueue_next_media(self, media: PlayerMedia) -> None:
448 """
449 Handle enqueuing of the next (queue) item on the player.
450
451 Called when player reports it started buffering a queue item
452 and when the queue items updated.
453
454 A PlayerProvider implementation is in itself responsible for handling this
455 so that the queue items keep playing until its empty or the player stopped.
456
457 Will only be called if the player reports PlayerFeature.ENQUEUE is
458 supported and the player is currently playing a MA queue.
459
460 This will NOT be called if the end of the queue is reached (and repeat disabled).
461 This will NOT be called if the player is using flow mode to playback the queue.
462
463 :param media: Details of the item that needs to be enqueued on the player.
464 """
465 raise NotImplementedError(
466 "enqueue_next_media needs to be implemented when PlayerFeature.ENQUEUE is set"
467 )
468
469 async def play_announcement(
470 self, announcement: PlayerMedia, volume_level: int | None = None
471 ) -> None:
472 """
473 Handle (native) playback of an announcement on the player.
474
475 Will only be called if the PlayerFeature.PLAY_ANNOUNCEMENT is supported.
476
477 :param announcement: Details of the announcement that needs to be played on the player.
478 :param volume_level: The volume level to play the announcement at (0..100).
479 If not set, the player should use the current volume level.
480 """
481 raise NotImplementedError(
482 "play_announcement needs to be implemented when PlayerFeature.PLAY_ANNOUNCEMENT is set"
483 )
484
485 async def select_source(self, source: str) -> None:
486 """
487 Handle SELECT SOURCE command on the player.
488
489 Will only be called if the PlayerFeature.SELECT_SOURCE is supported.
490
491 :param source: The source(id) to select, as defined in the source_list.
492 """
493 raise NotImplementedError(
494 "select_source needs to be implemented when PlayerFeature.SELECT_SOURCE is set"
495 )
496
497 async def set_members(
498 self,
499 player_ids_to_add: list[str] | None = None,
500 player_ids_to_remove: list[str] | None = None,
501 ) -> None:
502 """
503 Handle SET_MEMBERS command on the player.
504
505 Group or ungroup the given child player(s) to/from this player.
506 Will only be called if the PlayerFeature.SET_MEMBERS is supported.
507
508 :param player_ids_to_add: List of player_id's to add to the group.
509 :param player_ids_to_remove: List of player_id's to remove from the group.
510 """
511 raise NotImplementedError(
512 "set_members needs to be implemented when PlayerFeature.SET_MEMBERS is set"
513 )
514
515 async def poll(self) -> None:
516 """
517 Poll player for state updates.
518
519 This is called by the Player Manager;
520 if the 'needs_poll' property is True.
521 """
522 raise NotImplementedError("poll needs to be implemented when needs_poll is True")
523
524 async def get_config_entries(
525 self,
526 action: str | None = None,
527 values: dict[str, ConfigValueType] | None = None,
528 ) -> list[ConfigEntry]:
529 """
530 Return all (provider/player specific) Config Entries for the player.
531
532 action: [optional] action key called from config entries UI.
533 values: the (intermediate) raw values for config entries sent with the action.
534 """
535 # Return any (player/provider specific) config entries for a player.
536 # To override the default config entries, simply define an entry with the same key
537 # and it will be used instead of the default one.
538 return []
539
540 async def on_config_updated(self) -> None:
541 """
542 Handle logic when the player is loaded or updated.
543
544 Override this method in your player implementation if you need
545 to perform any additional setup logic after the player is registered and
546 the self.config was loaded, and whenever the config changes.
547 """
548 return
549
550 async def on_unload(self) -> None:
551 """Handle logic when the player is unloaded from the Player controller."""
552 for callback in self._on_unload_callbacks:
553 try:
554 callback()
555 except Exception as err:
556 self.logger.error(
557 "Error calling on_unload callback for player %s: %s",
558 self.player_id,
559 err,
560 )
561
562 async def group_with(self, target_player_id: str) -> None:
563 """
564 Handle GROUP_WITH command on the player.
565
566 Group this player to the given syncleader/target.
567 Will only be called if the PlayerFeature.SET_MEMBERS is supported.
568
569 :param target_player: player_id of the target player / sync leader.
570 """
571 # convenience helper method
572 # no need to implement unless your player/provider has an optimized way to execute this
573 # default implementation will simply call set_members
574 # to add the target player to the group.
575 target_player = self.mass.players.get(target_player_id, raise_unavailable=True)
576 assert target_player # for type checking
577 await target_player.set_members(player_ids_to_add=[self.player_id])
578
579 async def ungroup(self) -> None:
580 """
581 Handle UNGROUP command on the player.
582
583 Remove the player from any (sync)groups it currently is grouped to.
584 If this player is the sync leader (or group player),
585 all child's will be ungrouped and the group dissolved.
586
587 Will only be called if the PlayerFeature.SET_MEMBERS is supported.
588 """
589 # convenience helper method
590 # no need to implement unless your player/provider has an optimized way to execute this
591 # default implementation will simply call set_members
592 if self.synced_to:
593 if parent_player := self.mass.players.get(self.synced_to):
594 # if this player is synced to another player, remove self from that group
595 await parent_player.set_members(player_ids_to_remove=[self.player_id])
596 elif self.group_members:
597 await self.set_members(player_ids_to_remove=self.group_members)
598
599 @property
600 def synced_to(self) -> str | None:
601 """
602 Return the id of the player this player is synced to (sync leader).
603
604 If this player is not synced to another player (or is the sync leader itself),
605 this should return None.
606 If it is part of a (permanent) group, this should also return None.
607 """
608 # default implementation: feel free to override
609 for player in self.mass.players.all():
610 if player.player_id == self.player_id:
611 # skip self
612 continue
613 if player.type == PlayerType.PLAYER and self.player_id in player.group_members:
614 # this player is synced to another player, but not part of a (permanent) group
615 return player.player_id
616 return None
617
618 def _on_player_media_updated(self) -> None: # noqa: B027
619 """Handle callback when the current media of the player is updated."""
620 # optional callback for players that want to be informed when the final
621 # current media is updated (after applying group/sync membership logic).
622 # for instance to update any display information on the physical player.
623
624 # DO NOT OVERWRITE BELOW !
625 # These properties and methods are either managed by core logic or they
626 # are used to perform a very specific function. Overwriting these may
627 # produce undesirable effects.
628
629 @property
630 @final
631 def player_id(self) -> str:
632 """Return the id of the player."""
633 return self._player_id
634
635 @property
636 @final
637 def provider(self) -> PlayerProvider:
638 """Return the provider of the player."""
639 return self._provider
640
641 @property
642 @final
643 def provider_id(self) -> str:
644 """Return the provider (instance) id of the player."""
645 return self._provider.instance_id
646
647 @property
648 @final
649 def config(self) -> PlayerConfig:
650 """Return the config of the player."""
651 return self._config
652
653 @property
654 @final
655 def extra_attributes(self) -> dict[str, EXTRA_ATTRIBUTES_TYPES]:
656 """
657 Return the extra attributes of the player.
658
659 This is a dict that can be used to pass any extra (serializable)
660 attributes over the API, to be consumed by the UI (or another APi client, such as HA).
661 This is not persisted and not used or validated by the core logic.
662 """
663 return self._extra_attributes
664
665 @property
666 @final
667 def extra_data(self) -> dict[str, Any]:
668 """
669 Return the extra data of the player.
670
671 This is a dict that can be used to store any extra data
672 that is not part of the player state or config.
673 This is not persisted and not exposed on the API.
674 """
675 return self._extra_data
676
677 @cached_property
678 @final
679 def display_name(self) -> str:
680 """Return the display name of the player."""
681 if custom_name := self._config.name:
682 # always prefer the custom name over the default name
683 return custom_name
684 return self.name or self._config.default_name or self.player_id
685
686 @property
687 @final
688 def powered(self) -> bool | None:
689 """
690 Return the FINAL power state of the player.
691
692 This is a convenience property which calculates the final power state
693 based on the playercontrol which may have been set-up.
694 """
695 power_control = self.power_control
696 if power_control == PLAYER_CONTROL_FAKE:
697 return bool(self.extra_data.get(ATTR_FAKE_POWER, False))
698 if power_control == PLAYER_CONTROL_NATIVE:
699 return self._powered
700 if power_control == PLAYER_CONTROL_NONE:
701 return None
702 if control := self.mass.players.get_player_control(power_control):
703 return control.power_state
704 return None
705
706 @property
707 @final
708 def volume_level(self) -> int | None:
709 """
710 Return the FINAL volume level of the player.
711
712 This is a convenience property which calculates the final volume level
713 based on the playercontrol which may have been set-up.
714 """
715 volume_control = self.volume_control
716 if volume_control == PLAYER_CONTROL_FAKE:
717 return int(self.extra_data.get(ATTR_FAKE_VOLUME, 0))
718 if volume_control == PLAYER_CONTROL_NATIVE:
719 return self._volume_level
720 if volume_control == PLAYER_CONTROL_NONE:
721 return None
722 if control := self.mass.players.get_player_control(volume_control):
723 return control.volume_level
724 return None
725
726 @property
727 @final
728 def volume_muted(self) -> bool | None:
729 """
730 Return the FINAL mute state of the player.
731
732 This is a convenience property which calculates the final mute state
733 based on the playercontrol which may have been set-up.
734 """
735 mute_control = self.mute_control
736 if mute_control == PLAYER_CONTROL_FAKE:
737 return bool(self.extra_data.get(ATTR_FAKE_MUTE, False))
738 if mute_control == PLAYER_CONTROL_NATIVE:
739 return self._volume_muted
740 if mute_control == PLAYER_CONTROL_NONE:
741 return None
742 if control := self.mass.players.get_player_control(mute_control):
743 return control.volume_muted
744 return None
745
746 @property
747 @final
748 def active_source(self) -> str | None:
749 """
750 Return the FINAL active source of the player.
751
752 This is a convenience property which calculates the final active source
753 based on any group memberships or source plugins that can be active.
754 """
755 # if the player is grouped/synced, use the active source of the group/parent player
756 if parent_player_id := (self.active_group or self.synced_to):
757 if parent_player_id != self.player_id and (
758 parent_player := self.mass.players.get(parent_player_id)
759 ):
760 return parent_player.active_source
761 for plugin_source in self.mass.players.get_plugin_sources():
762 if plugin_source.in_use_by == self.player_id:
763 return plugin_source.id
764 if (
765 self.playback_state in (PlaybackState.PLAYING, PlaybackState.PAUSED)
766 and self._active_source
767 ):
768 # active source as reported by the player itself
769 # but only if playing/paused, otherwise we always prefer the MA source
770 return self._active_source
771 # return the (last) known MA source
772 return self.__active_mass_source
773
774 @cached_property
775 @final
776 def source_list(self) -> UniqueList[PlayerSource]:
777 """
778 Return the FINAL source list of the player.
779
780 This is a convenience property with the calculated final source list
781 based on any group memberships or source plugins that can be active.
782 """
783 return self.__attr_source_list or UniqueList()
784
785 @cached_property
786 @final
787 def enabled(self) -> bool:
788 """Return if the player is enabled."""
789 return self._config.enabled
790
791 @property
792 def corrected_elapsed_time(self) -> float | None:
793 """Return the corrected/realtime elapsed time."""
794 if self.elapsed_time is None or self.elapsed_time_last_updated is None:
795 return None
796 if self.playback_state == PlaybackState.PLAYING:
797 return self.elapsed_time + (time.time() - self.elapsed_time_last_updated)
798 return self.elapsed_time
799
800 @property
801 @final
802 def active_groups(self) -> list[str]:
803 """
804 Return the player ids of all playergroups that are currently active for this player.
805
806 This will return the ids of the groupplayers if any groups are active.
807 If no groups are currently active, this will return an empty list.
808 """
809 return self.__attr_active_groups or []
810
811 @property
812 @final
813 def active_group(self) -> str | None:
814 """
815 Return the player id of the (first) playergroup that is currently active for this player.
816
817 This will return the id of the groupplayer if a group is active.
818 If no group is currently active, this will return None.
819 """
820 active_groups = self.active_groups
821 return active_groups[0] if active_groups else None
822
823 @property
824 @final
825 def current_media(self) -> PlayerMedia | None:
826 """
827 Return the current media being played by the player.
828
829 This is a convenience property with the calculates current media
830 based on any group memberships or source plugins that can be active.
831 """
832 return self.__attr_current_media
833
834 @cached_property
835 @final
836 def icon(self) -> str:
837 """Return the player icon."""
838 return cast("str", self._config.get_value(CONF_ENTRY_PLAYER_ICON.key))
839
840 @cached_property
841 @final
842 def power_control(self) -> str:
843 """Return the power control type."""
844 if conf := self._config.get_value(CONF_POWER_CONTROL):
845 return str(conf)
846 return PLAYER_CONTROL_NONE
847
848 @cached_property
849 @final
850 def volume_control(self) -> str:
851 """Return the volume control type."""
852 if conf := self._config.get_value(CONF_VOLUME_CONTROL):
853 return str(conf)
854 return PLAYER_CONTROL_NONE
855
856 @cached_property
857 @final
858 def mute_control(self) -> str:
859 """Return the mute control type."""
860 if conf := self._config.get_value(CONF_MUTE_CONTROL):
861 return str(conf)
862 return PLAYER_CONTROL_NONE
863
864 @property
865 @final
866 def group_volume(self) -> int:
867 """
868 Return the group volume level.
869
870 If this player is a group player or syncgroup, this will return the average volume
871 level of all (powered on) child players in the group.
872 If the player is not a group player or syncgroup, this will return the volume level
873 of the player itself (if set), or 0 if not set.
874 """
875 if len(self.group_members) == 0:
876 # player is not a group or syncgroup
877 return self.volume_level or 0
878 # calculate group volume from all (turned on) players
879 group_volume = 0
880 active_players = 0
881 for child_player in self.mass.players.iter_group_members(
882 self, only_powered=True, exclude_self=self.type != PlayerType.PLAYER
883 ):
884 if (child_volume := child_player.volume_level) is None:
885 continue
886 group_volume += child_volume
887 active_players += 1
888 if active_players:
889 group_volume = int(group_volume / active_players)
890 return group_volume
891
892 @cached_property
893 @final
894 def hide_in_ui(self) -> bool:
895 """
896 Return the hide player in UI options.
897
898 This is a convenience property based on the config entry.
899 """
900 return bool(self._config.get_value(CONF_HIDE_IN_UI, self.hidden_by_default))
901
902 @cached_property
903 @final
904 def expose_to_ha(self) -> bool:
905 """
906 Return if the player should be exposed to Home Assistant.
907
908 This is a convenience property that returns True if the player is set to be exposed
909 to Home Assistant, based on the config entry.
910 """
911 return bool(self._config.get_value(CONF_EXPOSE_PLAYER_TO_HA))
912
913 @property
914 @final
915 def mass_queue_active(self) -> bool:
916 """
917 Return if the/a Music Assistant Queue is currently active for this player.
918
919 This is a convenience property that returns True if the
920 player currently has a Music Assistant Queue as active source.
921 """
922 return bool(self.mass.players.get_active_queue(self))
923
924 @property
925 @final
926 def flow_mode(self) -> bool:
927 """
928 Return if the player needs flow mode.
929
930 Will use 'requires_flow_mode' unless overridden by flow_mode config.
931 """
932 if bool(self._config.get_value(CONF_FLOW_MODE)) is True:
933 # flow mode explicitly enabled in config
934 return True
935 return self.requires_flow_mode
936
937 @property
938 @final
939 def state(self) -> PlayerState:
940 """Return the current PlayerState of the player."""
941 return self._state
942
943 @final
944 def update_state(self, force_update: bool = False, signal_event: bool = True) -> None:
945 """
946 Update the PlayerState with the current state of the player.
947
948 This method should be called to update the player's state
949 and signal any changes to the PlayerController.
950
951 :param force_update: If True, a state update event will be
952 pushed even if the state has not actually changed.
953 :param signal_event: If True, signal the state update event to the PlayerController.
954 """
955 self.mass.verify_event_loop_thread("player.update_state")
956 # clear the dict for the cached properties
957 self._cache.clear()
958 # calculate the new state
959 prev_media_checksum = self._get_player_media_checksum()
960 changed_values = self.__calculate_state()
961 if prev_media_checksum != self._get_player_media_checksum():
962 # current media changed, call the media updated callback
963 self._on_player_media_updated()
964 # ignore some values that are not relevant for the state
965 changed_values.pop("elapsed_time_last_updated", None)
966 changed_values.pop("extra_attributes.seq_no", None)
967 changed_values.pop("extra_attributes.last_poll", None)
968 changed_values.pop("current_media.elapsed_time_last_updated", None)
969 # persist the default name if it changed
970 if self.name and self.config.default_name != self.name:
971 self.mass.config.set_player_default_name(self.player_id, self.name)
972 # persist the player type if it changed
973 if self.type != self._config.player_type:
974 self.mass.config.set_player_type(self.player_id, self.type)
975 # return early if nothing changed (unless force_update is True)
976 if len(changed_values) == 0 and not force_update:
977 return
978 # signal the state update to the PlayerController
979 if signal_event:
980 self.mass.players.signal_player_state_update(self, changed_values)
981
982 @final
983 def set_current_media( # noqa: PLR0913
984 self,
985 uri: str,
986 media_type: MediaType = MediaType.UNKNOWN,
987 title: str | None = None,
988 artist: str | None = None,
989 album: str | None = None,
990 image_url: str | None = None,
991 duration: int | None = None,
992 source_id: str | None = None,
993 queue_item_id: str | None = None,
994 custom_data: dict[str, Any] | None = None,
995 clear_all: bool = False,
996 ) -> None:
997 """
998 Set current_media helper.
999
1000 Assumes use of '_attr_current_media'.
1001 """
1002 if self._attr_current_media is None or clear_all:
1003 self._attr_current_media = PlayerMedia(
1004 uri=uri,
1005 media_type=media_type,
1006 )
1007 self._attr_current_media.uri = uri
1008 if media_type != MediaType.UNKNOWN:
1009 self._attr_current_media.media_type = media_type
1010 if title:
1011 self._attr_current_media.title = title
1012 if artist:
1013 self._attr_current_media.artist = artist
1014 if album:
1015 self._attr_current_media.album = album
1016 if image_url:
1017 self._attr_current_media.image_url = image_url
1018 if duration:
1019 self._attr_current_media.duration = duration
1020 if source_id:
1021 self._attr_current_media.source_id = source_id
1022 if queue_item_id:
1023 self._attr_current_media.queue_item_id = queue_item_id
1024 if custom_data:
1025 self._attr_current_media.custom_data = custom_data
1026
1027 @final
1028 def set_config(self, config: PlayerConfig) -> None:
1029 """
1030 Set/update the player config.
1031
1032 May only be called by the PlayerController.
1033 """
1034 # TODO: validate that caller is the PlayerController ?
1035 self._config = config
1036
1037 @final
1038 def to_dict(self) -> dict[str, Any]:
1039 """Return the (serializable) dict representation of the Player."""
1040 return self.state.to_dict()
1041
1042 @final
1043 def supports_feature(self, feature: PlayerFeature) -> bool:
1044 """Return True if this player supports the given feature."""
1045 return feature in self.supported_features
1046
1047 @final
1048 def check_feature(self, feature: PlayerFeature) -> None:
1049 """Check if this player supports the given feature."""
1050 if not self.supports_feature(feature):
1051 raise UnsupportedFeaturedException(
1052 f"Player {self.display_name} does not support feature {feature.name}"
1053 )
1054
1055 def _get_player_media_checksum(self) -> str:
1056 """Return a checksum for the current media."""
1057 if not (media := self.current_media):
1058 return ""
1059 return (
1060 f"{media.uri}|{media.title}|{media.source_id}|{media.queue_item_id}|"
1061 f"{media.image_url}|{media.duration}|{media.elapsed_time}"
1062 )
1063
1064 def __calculate_state(
1065 self,
1066 ) -> dict[str, tuple[Any, Any]]:
1067 """
1068 Calculate the (current) PlayerState.
1069
1070 This method is called when we're updating the player,
1071 and we compare the current state with the previous state to determine
1072 if we need to signal a state change to API consumers.
1073
1074 Returns a dict with the state attributes that have changed.
1075 """
1076 self.__attr_active_groups = self.__calculate_active_groups()
1077 self.__attr_current_media = self.__calculate_current_media()
1078 self.__attr_source_list = self.__calculate_source_list()
1079 prev_state = deepcopy(self._state)
1080 self._state = PlayerState(
1081 player_id=self.player_id,
1082 provider=self.provider_id,
1083 type=self.type,
1084 available=self.enabled and self.available,
1085 device_info=self.device_info,
1086 supported_features=self.supported_features,
1087 playback_state=self.playback_state,
1088 elapsed_time=self.elapsed_time,
1089 elapsed_time_last_updated=self.elapsed_time_last_updated,
1090 powered=self.powered,
1091 volume_level=self.volume_level,
1092 volume_muted=self.volume_muted,
1093 group_members=UniqueList(self.group_members),
1094 static_group_members=UniqueList(self.static_group_members),
1095 can_group_with=self.can_group_with,
1096 synced_to=self.synced_to,
1097 active_source=self.active_source,
1098 source_list=self.source_list,
1099 active_group=self.active_group,
1100 current_media=self.current_media,
1101 name=self.display_name,
1102 enabled=self.enabled,
1103 hide_in_ui=self.hide_in_ui,
1104 expose_to_ha=self.expose_to_ha,
1105 icon=self.icon,
1106 group_volume=self.group_volume,
1107 extra_attributes=self.extra_attributes,
1108 power_control=self.power_control,
1109 volume_control=self.volume_control,
1110 mute_control=self.mute_control,
1111 )
1112
1113 # correct group_members if needed
1114 if self._state.group_members == [self.player_id]:
1115 self._state.group_members.clear()
1116 elif (
1117 self._state.group_members
1118 and self.player_id not in self._state.group_members
1119 and self.type == PlayerType.PLAYER
1120 ):
1121 self._state.group_members.set([self.player_id, *self._state.group_members])
1122
1123 # track stop called state
1124 if (
1125 prev_state.playback_state == PlaybackState.IDLE
1126 and self._state.playback_state != PlaybackState.IDLE
1127 ):
1128 self.__stop_called = False
1129 elif (
1130 prev_state.playback_state != PlaybackState.IDLE
1131 and self._state.playback_state == PlaybackState.IDLE
1132 ):
1133 self.__stop_called = True
1134
1135 # Auto correct player state if player is synced (or group child)
1136 # This is because some players/providers do not accurately update this info
1137 # for the sync child's.
1138 if self._state.synced_to and (sync_leader := self.mass.players.get(self._state.synced_to)):
1139 self._state.playback_state = sync_leader.playback_state
1140 self._state.elapsed_time = sync_leader.elapsed_time
1141 self._state.elapsed_time_last_updated = sync_leader.elapsed_time_last_updated
1142
1143 return get_changed_dataclass_values(
1144 prev_state,
1145 self._state,
1146 recursive=True,
1147 )
1148
1149 __attr_active_groups: list[str] | None = None
1150
1151 def __calculate_active_groups(self) -> list[str]:
1152 """Calculate the active groups for the player."""
1153 active_groups = []
1154 for player in self.mass.players.all(return_unavailable=False, return_disabled=False):
1155 if player.type != PlayerType.GROUP:
1156 continue
1157 if player.player_id == self.player_id:
1158 continue
1159 if not (player.powered or player.playback_state == PlaybackState.PLAYING):
1160 continue
1161 if self.player_id in player.group_members:
1162 active_groups.append(player.player_id)
1163 return active_groups
1164
1165 __attr_current_media: PlayerMedia | None = None
1166
1167 def __calculate_current_media(self) -> PlayerMedia | None:
1168 """Calculate the current media for the player."""
1169 if self.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS):
1170 # if an announcement is in progress, return announcement details
1171 return PlayerMedia(
1172 uri="announcement",
1173 media_type=MediaType.ANNOUNCEMENT,
1174 title="ANNOUNCEMENT",
1175 )
1176 # if the player is grouped/synced, use the current_media of the group/parent player
1177 if parent_player_id := (self.active_group or self.synced_to):
1178 if parent_player_id != self.player_id and (
1179 parent_player := self.mass.players.get(parent_player_id)
1180 ):
1181 return parent_player.current_media
1182 # if a pluginsource is currently active, return those details
1183 if (
1184 self.active_source
1185 and (source := self.mass.players.get_plugin_source(self.active_source))
1186 and source.metadata
1187 ):
1188 return PlayerMedia(
1189 uri=source.metadata.uri or source.id,
1190 media_type=MediaType.PLUGIN_SOURCE,
1191 title=source.metadata.title,
1192 artist=source.metadata.artist,
1193 album=source.metadata.album,
1194 image_url=source.metadata.image_url,
1195 duration=source.metadata.duration,
1196 source_id=source.id,
1197 elapsed_time=source.metadata.elapsed_time,
1198 elapsed_time_last_updated=source.metadata.elapsed_time_last_updated,
1199 )
1200 # if MA queue is active, return those details
1201 active_queue = None
1202 if self._current_media and self._current_media.source_id:
1203 active_queue = self.mass.player_queues.get(self._current_media.source_id)
1204 if not active_queue and self.active_source:
1205 active_queue = self.mass.player_queues.get(self.active_source)
1206 if not active_queue and self._active_source is None:
1207 active_queue = self.mass.player_queues.get(self.player_id)
1208
1209 if active_queue and (current_item := active_queue.current_item):
1210 item_image_url = (
1211 # the image format needs to be 500x500 jpeg for maximum compatibility with players
1212 self.mass.metadata.get_image_url(current_item.image, size=500, image_format="jpeg")
1213 if current_item.image
1214 else None
1215 )
1216 if current_item.streamdetails and (
1217 stream_metadata := current_item.streamdetails.stream_metadata
1218 ):
1219 # handle stream metadata in streamdetails (e.g. for radio stream)
1220 return PlayerMedia(
1221 uri=current_item.uri,
1222 media_type=current_item.media_type,
1223 title=stream_metadata.title or current_item.name,
1224 artist=stream_metadata.artist,
1225 album=stream_metadata.album or stream_metadata.description or current_item.name,
1226 image_url=(stream_metadata.image_url or item_image_url),
1227 duration=stream_metadata.duration or current_item.duration,
1228 source_id=active_queue.queue_id,
1229 queue_item_id=current_item.queue_item_id,
1230 elapsed_time=stream_metadata.elapsed_time or int(active_queue.elapsed_time),
1231 elapsed_time_last_updated=stream_metadata.elapsed_time_last_updated
1232 or active_queue.elapsed_time_last_updated,
1233 )
1234 if media_item := current_item.media_item:
1235 # normal media item
1236 # we use getattr here to avoid issues with different media item types
1237 version = getattr(media_item, "version", None)
1238 album = getattr(media_item, "album", None)
1239 podcast = getattr(media_item, "podcast", None)
1240 metadata = getattr(media_item, "metadata", None)
1241 description = getattr(metadata, "description", None) if metadata else None
1242 return PlayerMedia(
1243 uri=str(media_item.uri),
1244 media_type=media_item.media_type,
1245 title=f"{media_item.name} ({version})" if version else media_item.name,
1246 artist=getattr(media_item, "artist_str", None),
1247 album=album.name if album else podcast.name if podcast else description,
1248 # the image format needs to be 500x500 jpeg for maximum player compatibility
1249 image_url=self.mass.metadata.get_image_url(
1250 current_item.media_item.image, size=500, image_format="jpeg"
1251 )
1252 or item_image_url
1253 if current_item.media_item.image
1254 else item_image_url,
1255 duration=media_item.duration,
1256 source_id=active_queue.queue_id,
1257 queue_item_id=current_item.queue_item_id,
1258 elapsed_time=int(active_queue.elapsed_time),
1259 elapsed_time_last_updated=active_queue.elapsed_time_last_updated,
1260 )
1261
1262 # fallback to basic current item details
1263 return PlayerMedia(
1264 uri=current_item.uri,
1265 media_type=current_item.media_type,
1266 title=current_item.name,
1267 image_url=item_image_url,
1268 duration=current_item.duration,
1269 source_id=active_queue.queue_id,
1270 queue_item_id=current_item.queue_item_id,
1271 elapsed_time=int(active_queue.elapsed_time),
1272 elapsed_time_last_updated=active_queue.elapsed_time_last_updated,
1273 )
1274 if active_queue:
1275 # queue is active but no current item
1276 return None
1277 # return native current media if no group/queue is active
1278 if self._current_media:
1279 return PlayerMedia(
1280 uri=self._current_media.uri,
1281 media_type=self._current_media.media_type,
1282 title=self._current_media.title,
1283 artist=self._current_media.artist,
1284 album=self._current_media.album,
1285 image_url=self._current_media.image_url,
1286 duration=self._current_media.duration,
1287 source_id=self._current_media.source_id or self._active_source,
1288 queue_item_id=self._current_media.queue_item_id,
1289 elapsed_time=self._current_media.elapsed_time or int(self.elapsed_time)
1290 if self.elapsed_time
1291 else None,
1292 elapsed_time_last_updated=self._current_media.elapsed_time_last_updated
1293 or self.elapsed_time_last_updated,
1294 )
1295 return None
1296
1297 __attr_source_list: UniqueList[PlayerSource] | None = None
1298
1299 def __calculate_source_list(self) -> UniqueList[PlayerSource]:
1300 """Calculate the source list for the player."""
1301 sources = UniqueList(self._source_list)
1302 # always ensure the Music Assistant Queue is in the source list
1303 mass_source = next((x for x in sources if x.id == self.player_id), None)
1304 if mass_source is None:
1305 # if the MA queue is not in the source list, add it
1306 mass_source = PlayerSource(
1307 id=self.player_id,
1308 name="Music Assistant Queue",
1309 passive=False,
1310 # TODO: Do we want to dynamically set these based on the queue state ?
1311 can_play_pause=True,
1312 can_seek=True,
1313 can_next_previous=True,
1314 )
1315 sources.append(mass_source)
1316 # append all/any plugin sources (convert to PlayerSource to avoid deepcopy issues)
1317 for plugin_source in self.mass.players.get_plugin_sources():
1318 if hasattr(plugin_source, "as_player_source"):
1319 sources.append(plugin_source.as_player_source())
1320 else:
1321 sources.append(plugin_source)
1322 return sources
1323
1324 # The id of the (last) active mass source.
1325 # This is to keep track of the last active MA source for the player,
1326 # so we can restore it when needed (e.g. after switching to a plugin source).
1327 __active_mass_source: str = ""
1328
1329 def set_active_mass_source(self, value: str) -> None:
1330 """
1331 Set the id of the (last) active mass source.
1332
1333 This is to keep track of the last active MA source for the player,
1334 so we can restore it when needed (e.g. after switching to a plugin source).
1335 """
1336 self.__active_mass_source = value
1337 self.update_state()
1338
1339 __stop_called: bool = False
1340
1341 def mark_stop_called(self) -> None:
1342 """Mark that the STOP command was called on the player."""
1343 self.__stop_called = True
1344
1345 @property
1346 def stop_called(self) -> bool:
1347 """
1348 Return True if the STOP command was called on the player.
1349
1350 This is used to differentiate between a user-initiated stop
1351 and a natural end of playback (e.g. end of track/queue).
1352 mainly for debugging/logging purposes by the streams controller.
1353 """
1354 return self.__stop_called
1355
1356 def __hash__(self) -> int:
1357 """Return a hash of the Player."""
1358 return hash(self.player_id)
1359
1360 def __str__(self) -> str:
1361 """Return a string representation of the Player."""
1362 return f"Player {self.name} ({self.player_id})"
1363
1364 def __repr__(self) -> str:
1365 """Return a string representation of the Player."""
1366 return f"<Player name={self.name} id={self.player_id} available={self.available}>"
1367
1368 def __eq__(self, other: object) -> bool:
1369 """Check equality of two Player objects."""
1370 if not isinstance(other, Player):
1371 return False
1372 return self.player_id == other.player_id
1373
1374 def __ne__(self, other: object) -> bool:
1375 """Check inequality of two Player objects."""
1376 return not self.__eq__(other)
1377
1378
1379__all__ = [
1380 # explicitly re-export the models we imported from the models package,
1381 # for convenience reasons
1382 "EXTRA_ATTRIBUTES_TYPES",
1383 "DeviceInfo",
1384 "Player",
1385 "PlayerMedia",
1386 "PlayerSource",
1387 "PlayerState",
1388]
1389
1390
1391class GroupPlayer(Player):
1392 """Helper class for a (generic) group player."""
1393
1394 _attr_type: PlayerType = PlayerType.GROUP
1395
1396 @cached_property
1397 def synced_to(self) -> str | None:
1398 """Return the id of the player this player is synced to (sync leader)."""
1399 # default implementation: groups can't be synced
1400 return None
1401
1402 async def volume_set(self, volume_level: int) -> None:
1403 """
1404 Handle VOLUME_SET command on the player.
1405
1406 :param volume_level: volume level (0..100) to set on the player.
1407 """
1408 # Default implementation:
1409 # This will set the (relative) volume level on all child players.
1410 # free to override if you want to handle this differently.
1411 await self.mass.players.set_group_volume(self, volume_level)
1412