/
/
/
1"""
2Universal Player implementation.
3
4A virtual player for devices that have no native (vendor-specific) provider in
5Music Assistant but support one or more generic streaming protocols such as
6AirPlay, Sendspin, Chromecast, or DLNA.
7
8The Universal Player is automatically created when a protocol player with
9PlayerType.PROTOCOL is registered, providing a unified interface while delegating
10actual playback to the underlying protocol player(s).
11"""
12
13from __future__ import annotations
14
15from typing import TYPE_CHECKING
16
17from music_assistant_models.enums import PlayerFeature
18
19from music_assistant.constants import CONF_PREFERRED_OUTPUT_PROTOCOL
20from music_assistant.models.player import DeviceInfo, Player
21
22if TYPE_CHECKING:
23 from .provider import UniversalPlayerProvider
24
25
26class UniversalPlayer(Player):
27 """
28 Universal Player implementation.
29
30 A virtual player for devices without native Music Assistant support that use
31 generic streaming protocols. It does NOT have PLAY_MEDIA capability on its own.
32 Playback is always delegated to one of the linked protocol players via the protocol
33 linking system.
34 """
35
36 def __init__(
37 self,
38 provider: UniversalPlayerProvider,
39 player_id: str,
40 name: str,
41 device_info: DeviceInfo,
42 protocol_player_ids: list[str],
43 ) -> None:
44 """
45 Initialize UniversalPlayer instance.
46
47 :param provider: The UniversalPlayerProvider instance.
48 :param player_id: Unique player ID (typically based on MAC address).
49 :param name: Display name for the player.
50 :param device_info: Device information aggregated from protocol players.
51 :param protocol_player_ids: List of protocol player IDs to link.
52 """
53 super().__init__(provider, player_id)
54 self._protocol_player_ids = protocol_player_ids
55 # Set player attributes
56 self._attr_name = name
57 self._attr_device_info = device_info
58 # Start as unavailable - will be updated when protocol players are linked
59 self._attr_available = False
60 # a universal player does not have any features on its own,
61 # it delegates to protocol players
62 self._attr_supported_features = set()
63
64 def _get_control_target(
65 self, required_feature: PlayerFeature, require_active: bool = False
66 ) -> Player | None:
67 """Get the best player to send control commands to.
68
69 Prefers the active output protocol, otherwise uses the first available
70 protocol player that supports the needed feature.
71 """
72 # If we have an active protocol, use that
73 if (
74 self.active_output_protocol
75 and self.active_output_protocol != "native"
76 and (protocol_player := self.mass.players.get_player(self.active_output_protocol))
77 and required_feature in protocol_player.supported_features
78 ):
79 return protocol_player
80
81 # If require_active is set, and no active protocol found, return None
82 if require_active:
83 return None
84
85 # Otherwise, use the first available linked protocol
86 for protocol_player_id in self._protocol_player_ids:
87 if (
88 (protocol_player := self.mass.players.get_player(protocol_player_id))
89 and protocol_player.available
90 and required_feature in protocol_player.supported_features
91 ):
92 return protocol_player
93
94 return None
95
96 def update_from_protocol_players(self) -> None:
97 """
98 Update state from linked protocol players.
99
100 Called to sync state like volume, availability from protocol players.
101 """
102 # Aggregate availability - available if any protocol is available
103 self._attr_available = any(
104 (p := self.mass.players.get_player(pid)) and p.available
105 for pid in self._protocol_player_ids
106 )
107 # Get volume from best control target
108 if target := self._get_control_target(PlayerFeature.VOLUME_SET):
109 if target.volume_level is not None:
110 self._attr_volume_level = target.volume_level
111 if target := self._get_control_target(PlayerFeature.VOLUME_MUTE):
112 if target.volume_muted is not None:
113 self._attr_volume_muted = target.volume_muted
114
115 self.update_state()
116
117 def add_protocol_player(self, protocol_player_id: str) -> None:
118 """Add a protocol player to this universal player."""
119 if protocol_player_id not in self._protocol_player_ids:
120 self._protocol_player_ids.append(protocol_player_id)
121
122 def remove_protocol_player(self, protocol_player_id: str) -> None:
123 """Remove a protocol player from this universal player."""
124 if protocol_player_id in self._protocol_player_ids:
125 self._protocol_player_ids.remove(protocol_player_id)
126
127 def _get_preferred_protocol_player(self) -> Player | None:
128 """
129 Get the preferred protocol player for this universal player.
130
131 Selection priority:
132 1. Active output protocol (if set and available)
133 2. User's preferred output protocol (from settings), fallback to highest
134 priority if preferred is not available
135 """
136 # 1. Active output protocol takes precedence
137 if (
138 self.active_output_protocol
139 and self.active_output_protocol != "native"
140 and (protocol_player := self.mass.players.get_player(self.active_output_protocol))
141 and protocol_player.available
142 ):
143 return protocol_player
144
145 # 2. User's preferred output protocol (with fallback to highest priority)
146 preferred = self.mass.config.get_raw_player_config_value(
147 self.player_id, CONF_PREFERRED_OUTPUT_PROTOCOL
148 )
149 if preferred and (protocol_player := self.mass.players.get_player(str(preferred))):
150 if protocol_player.available:
151 return protocol_player
152
153 # Fallback: if user's preferred protocol is not available,
154 # use the highest priority available protocol
155 for protocol in sorted(self.linked_output_protocols, key=lambda x: x.priority):
156 if protocol_player := self.mass.players.get_player(protocol.output_protocol_id):
157 if protocol_player.available:
158 return protocol_player
159
160 return None
161