/
/
/
1"""Demo Player implementation."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7from music_assistant_models.config_entries import ConfigEntry
8from music_assistant_models.enums import ConfigEntryType, PlaybackState, PlayerFeature
9from music_assistant_models.player import PlayerOptionValueType, PlayerSource
10
11from music_assistant.models.player import Player, PlayerMedia
12
13if TYPE_CHECKING:
14 from .provider import DemoPlayerprovider
15
16
17class DemoPlayer(Player):
18 """DemoPlayer in Music Assistant."""
19
20 def __init__(self, provider: DemoPlayerprovider, player_id: str) -> None:
21 """Initialize the Player."""
22 super().__init__(provider, player_id)
23 # init some static variables
24 self._attr_name = f"Demo Player {player_id}"
25 self._attr_supported_features = {
26 PlayerFeature.PLAY_MEDIA,
27 PlayerFeature.POWER,
28 PlayerFeature.VOLUME_SET,
29 PlayerFeature.VOLUME_MUTE,
30 PlayerFeature.PLAY_ANNOUNCEMENT,
31 PlayerFeature.SET_MEMBERS,
32 }
33 # demo players can be (sync) grouped with other players of this provider
34 self._attr_can_group_with = {provider.instance_id}
35 self._set_attributes()
36
37 async def on_config_updated(self) -> None:
38 """Handle logic when the PlayerConfig is first loaded or updated."""
39 # OPTIONAL
40 # This method is optional and should be implemented if you need to handle
41 # any initialization logic after the config was initially loaded or updated.
42 # This is called after the player is registered and self.config was loaded.
43 # And also when the config was updated.
44 # You don't need to call update_state() here.
45
46 @property
47 def needs_poll(self) -> bool:
48 """Return if the player needs to be polled for state updates."""
49 # MANDATORY
50 # this should return True if the player needs to be polled for state updates,
51 # If you player does not need to be polled, you can return False.
52 return True
53
54 @property
55 def poll_interval(self) -> int:
56 """Return the interval in seconds to poll the player for state updates."""
57 # OPTIONAL
58 # used in conjunction with the needs_poll property.
59 # this should return the interval in seconds to poll the player for state updates.
60 return 5 if self._attr_playback_state == PlaybackState.PLAYING else 30
61
62 @property
63 def source_list(self) -> list[PlayerSource]:
64 """Return list of available (native) sources for this player."""
65 # OPTIONAL - required only if you specified PlayerFeature.SELECT_SOURCE
66 # this is an optional property that you can implement if your
67 # player supports (external) source control (aux, HDMI, etc.).
68 # If your player does not support sources, you can leave this out completely.
69 return [
70 PlayerSource(
71 id="line_in",
72 name="Line-In",
73 passive=False,
74 can_play_pause=False,
75 can_next_previous=False,
76 can_seek=False,
77 ),
78 PlayerSource(
79 id="spotify_connect",
80 name="Spotify",
81 # by specifying passive=True, we indicate that this source
82 # is not actively selectable by the user from the UI.
83 passive=True,
84 can_play_pause=True,
85 can_next_previous=True,
86 can_seek=True,
87 ),
88 ]
89
90 async def get_config_entries(self) -> list[ConfigEntry]:
91 """Return all (provider/player specific) Config Entries for the player."""
92 # OPTIONAL
93 # this method is optional and should be implemented if you need player specific
94 # configuration entries. If you do not need player specific configuration entries,
95 # you can leave this method out completely.
96 # note that the config controller will always add a set of default config entries
97 # if you want, you can override those by specifying the same key as a default entry.
98 return [
99 # example of a player specific config entry
100 # you can also override a default entry by specifying the same key
101 # as a default entry, but with a different type or default value.
102 ConfigEntry(
103 key="demo_player_setting",
104 type=ConfigEntryType.STRING,
105 label="Demo Player Setting",
106 required=False,
107 default_value="default_value",
108 description="This is a demo player setting.",
109 ),
110 ]
111
112 async def power(self, powered: bool) -> None:
113 """Handle POWER command on the player."""
114 # OPTIONAL - required only if you specified PlayerFeature.POWER
115 # this method should send a power on/off command to the given player.
116 logger = self.provider.logger.getChild(self.player_id)
117 if powered:
118 # In this demo implementation we just set the power state to ON
119 # and optimistically update the state.
120 # In a real implementation you would read the actual value from the player
121 # either from a callback or by polling the player.
122 logger.info("Received POWER ON command on player %s", self.display_name)
123 self._attr_powered = True
124 else:
125 # In this demo implementation we just set the power state to OFF
126 # and optimistically update the state.
127 # In a real implementation you would read the actual value from the player
128 # either from a callback or by polling the player.
129 logger.info("Received POWER OFF command on player %s", self.display_name)
130 self._attr_powered = False
131 # update the player state in the player manager
132 self.update_state()
133
134 async def volume_set(self, volume_level: int) -> None:
135 """Handle VOLUME_SET command on the player."""
136 # OPTIONAL - required only if you specified PlayerFeature.VOLUME_SET
137 # this method should send a volume set command to the given player.
138
139 # In this demo implementation we just set the volume level
140 # and optimistically update the state.
141 # In a real implementation you would send a command to the actual player and
142 # get the actual value from the player either from a callback or by polling the player.
143 logger = self.provider.logger.getChild(self.player_id)
144 logger.info(
145 "Received VOLUME_SET command on player %s with level %s",
146 self.display_name,
147 volume_level,
148 )
149 self._attr_volume_level = volume_level # volume level is between 0 and 100
150 # update the player state in the player manager
151 self.update_state()
152
153 async def volume_mute(self, muted: bool) -> None:
154 """Handle VOLUME MUTE command on the player."""
155 # OPTIONAL - required only if you specified PlayerFeature.VOLUME_MUTE
156 # this method should send a volume mute command to the given player.
157 logger = self.provider.logger.getChild(self.player_id)
158 logger.info(
159 "Received VOLUME_MUTE command on player %s with muted %s", self.display_name, muted
160 )
161 self._attr_volume_muted = muted
162 self.update_state()
163
164 async def play(self) -> None:
165 """Play command."""
166 # MANDATORY
167 # this method is mandatory and should be implemented.
168 # this method should send a play/resume command to the given player.
169 # normally this is the point where you would resume playback
170 # on your actual player device.
171
172 # In this demo implementation we just set the playback state to PLAYING
173 # and optimistically set the playback state to PLAYING.
174 # In a real implementation you actually send a command to the player
175 # wait for the player to report a new state before updating the playback state.
176 logger = self.provider.logger.getChild(self.player_id)
177 logger.info("Received PLAY command on player %s", self.display_name)
178 self._attr_playback_state = PlaybackState.PLAYING
179 self.update_state()
180
181 async def stop(self) -> None:
182 """Stop command."""
183 # MANDATORY
184 # this method is mandatory and should be implemented.
185 # this method should send a stop command to the given player.
186 # normally this is the point where you would stop playback
187 # on your actual player device.
188
189 # In this demo implementation we just set the playback state to IDLE
190 # and optimistically set the playback state to IDLE.
191 # In a real implementation you actually send a command to the player
192 # wait for the player to report a new state before updating the playback state.
193 logger = self.provider.logger.getChild(self.player_id)
194 logger.info("Received STOP command on player %s", self.display_name)
195 self._attr_playback_state = PlaybackState.IDLE
196 self._attr_active_source = None
197 self._attr_current_media = None
198 self.update_state()
199
200 async def pause(self) -> None:
201 """Pause command."""
202 # OPTIONAL - required only if you specified PlayerFeature.PAUSE
203 # this method should send a pause command to the given player.
204
205 # In this demo implementation we just set the playback state to PAUSED
206 # and optimistically set the playback state to PAUSED.
207 # In a real implementation you actually send a command to the player
208 # wait for the player to report a new state before updating the playback state.
209 logger = self.provider.logger.getChild(self.player_id)
210 logger.info("Received PAUSE command on player %s", self.display_name)
211 self._attr_playback_state = PlaybackState.PAUSED
212 self.update_state()
213
214 async def next_track(self) -> None:
215 """Next command."""
216 # OPTIONAL - required only if you specified PlayerFeature.NEXT_PREVIOUS
217 # this method should send a next track command to the given player.
218 # Note that this is only needed/used if the player is playing a 3rd party
219 # stream (e.g. Spotify, YouTube, etc.) and the player supports skipping to the next track.
220 # When the player is playing MA content, this is already handled in the Queue controller.
221
222 async def previous_track(self) -> None:
223 """Previous command."""
224 # OPTIONAL - required only if you specified PlayerFeature.NEXT_PREVIOUS
225 # this method should send a previous track command to the given player.
226 # Note that this is only needed/used if the player is playing a 3rd party
227 # stream (e.g. Spotify, YouTube, etc.) and the player supports skipping to the next track.
228 # When the player is playing MA content, this is already handled in the Queue controller.
229
230 async def seek(self, position: int) -> None:
231 """SEEK command on the player."""
232 # OPTIONAL - required only if you specified PlayerFeature.SEEK
233 # this method should send a seek command to the given player.
234 # the position is the position in seconds to seek to in the current playing item.
235
236 async def play_media(self, media: PlayerMedia) -> None:
237 """Play media command."""
238 # MANDATORY
239 # This method is mandatory and should be implemented.
240 # This method should handle the play_media command for the given player.
241 # It will be called when media needs to be played on the player.
242 # The media object contains all the details needed to play the item.
243
244 # In 99% of the cases this will be called by the Queue controller to play
245 # a single item from the queue on the player and the uri within the media
246 # object will then contain the URL to play that single queue item.
247
248 # If your player provider does not support enqueuing of items,
249 # the queue controller will simply call this play_media method for
250 # each item in the queue to play them one by one.
251
252 # In order to support true gapless and/or enqueuing, we offer the option of
253 # 'flow_mode' playback. In that case the queue controller will stitch together
254 # all songs in the playbook queue into a single stream and send that to the player.
255 # In that case the URI (and metadata) received here is that of the 'flow mode' stream.
256
257 # Examples of player providers that use flow mode for playback by default are AirPlay,
258 # SnapCast and Fully Kiosk.
259
260 # Examples of player providers that optionally use 'flow mode' are Google Cast and
261 # Home Assistant. They provide a config entry to enable flow mode playback.
262
263 # Examples of player providers that natively support enqueuing of items are Sonos,
264 # Slimproto and Google Cast.
265
266 # In this demo implementation we just optimistically set the state.
267 # In a real implementation you actually send a command to the player
268 # wait for the player to report a new state before updating the playback state.
269 url = await self.provider.mass.streams.resolve_stream_url(self.player_id, media)
270 logger = self.provider.logger.getChild(self.player_id)
271 logger.info("Received PLAY_MEDIA command on player %s with url %s", self.display_name, url)
272 self._attr_current_media = media
273 self._attr_playback_state = PlaybackState.PLAYING
274 self.update_state()
275
276 async def enqueue_next_media(self, media: PlayerMedia) -> None:
277 """Handle enqueuing of the next (queue) item on the player."""
278 # OPTIONAL - required only if you specified PlayerFeature.ENQUEUE
279 # This method is optional and should be implemented if you want to support
280 # enqueuing of the next item on the player.
281 # This will be called when the player reports it started buffering a queue item
282 # and when the queue items updated.
283 # A PlayerProvider implementation is in itself responsible for handling this
284 # so that the queue items keep playing until its empty or the player stopped.
285
286 async def play_announcement(
287 self, announcement: PlayerMedia, volume_level: int | None = None
288 ) -> None:
289 """Handle (native) playback of an announcement on the player."""
290 # OPTIONAL - required only if you specified PlayerFeature.PLAY_ANNOUNCEMENT
291 # This method is optional and should be implemented if the player supports
292 # NATIVE playback of announcements (with ducking etc.).
293 # The announcement object contains all the details needed to play the announcement.
294 # The volume_level is optional and can be used to set the volume level for the announcement.
295 # If you do not use the announcement playerfeature, the default behavior is to play the
296 # announcement as a regular media item using the play_media method and the MA player manager
297 # will take care of setting the volume level for the announcement and resuming etc.
298
299 async def select_source(self, source: str) -> None:
300 """Handle SELECT SOURCE command on the player."""
301 # OPTIONAL - required only if you specified PlayerFeature.SELECT_SOURCE
302 # This method is optional and should be implemented if the player supports
303 # selecting a source (e.g. HDMI, AUX, etc.) on the player.
304 # The source is the source ID to select on the player.
305 # available sources are specified in the Player.source_list property
306
307 async def select_sound_mode(self, sound_mode: str) -> None:
308 """Handle SELECT SOUND MODE command on the player."""
309 # OPTIONAL - required only if you specified PlayerFeature.SELECT_SOUND_MODE.
310 # This method is optional and should be implemented if the player supports
311 # selecting a native sound mode (e.g. stereo, 5.1, classic...).
312 # The sound_mode is the sound mode's id, and available sound modes
313 # are specified in the Player.sound_mode_list property.
314
315 async def set_members(
316 self,
317 player_ids_to_add: list[str] | None = None,
318 player_ids_to_remove: list[str] | None = None,
319 ) -> None:
320 """Handle SET_MEMBERS command on the player."""
321 # OPTIONAL - required only if you specified PlayerFeature.SET_MEMBERS
322 # This method is optional and should be implemented if the player supports
323 # syncing/grouping with other players.
324 # This demo keeps a simple in-memory member list: the leader tracks its
325 # group_members and each member's synced_to is then derived automatically by
326 # the base Player from the leader's list. A real provider would also issue the
327 # native sync command to the device here.
328 members = dict.fromkeys(self._attr_group_members)
329 for member_id in player_ids_to_add or []:
330 members[member_id] = None
331 for member_id in player_ids_to_remove or []:
332 members.pop(member_id, None)
333 other_member_ids = [pid for pid in members if pid != self.player_id]
334 # the leader is always the first member while the group is non-empty
335 self._attr_group_members = [self.player_id, *other_member_ids] if other_member_ids else []
336 self.update_state()
337 # refresh affected members so their derived synced_to / can_group_with recompute
338 for member_id in [*(player_ids_to_add or []), *(player_ids_to_remove or [])]:
339 if (member := self.mass.players.get_player(member_id)) is not None:
340 member.update_state()
341
342 async def set_option(self, option_key: str, option_value: PlayerOptionValueType) -> None:
343 """Handle SET_OPTION command on the player."""
344 # OPTIONAL - required only if you specified PlayerFeature.OPTIONS.
345 # PlayerOptions are native settings of the player adjustable on the fly,
346 # e.g. a treble value, toggable settings etc. In the ui they are accessible
347 # on the player menu, or via
348 # player settings -> players -> <select player> -> player options.
349 #
350 # The options are mapped to respective Home Assistant entities, if the integration
351 # is used. The mapping for non read-only option is:
352 # PlayerOptionType.BOOLEAN (w and w/o options) -> SwitchEntity
353 # PlayerOptionType.FLOAT or PlayerOptionType.INTEGER (w/o options) -> NumberEntity
354 # PlayerOptionType.STRING (w/o options) -> TextEntity
355 # PlayerOptionType.FLOAT/ INTEGER/ STRING with options -> SelectEntity.
356 # Read-only options will be mapped to SensorEntities once a provider using them exists.
357 #
358 # The translation key of the respective option must be present in the HA integration,
359 # otherwise, the option is ignored. Currently available keys are listed in the following
360 # dictionaries right at the top of the respective entity type:
361 #
362 # PLAYER_OPTIONS_SWITCH in
363 # https://github.com/home-assistant/core/blob/dev/homeassistant/components/music_assistant/switch.py
364 # PLAYER_OPTIONS_SELECT in
365 # https://github.com/home-assistant/core/blob/dev/homeassistant/components/music_assistant/select.py
366 # PLAYER_OPTIONS_NUMBER in
367 # https://github.com/home-assistant/core/blob/dev/homeassistant/components/music_assistant/number.py
368 # PLAYER_OPTIONS_TEXT in
369 # https://github.com/home-assistant/core/blob/dev/homeassistant/components/music_assistant/text.py
370 #
371 # Before creating a PR to HA core to support new translation keys, complete the review process in MA.
372 # We are also happy to create the HA PR for you after the review process.
373 # The MusicCast provider can serve as an example for player options.
374
375 async def poll(self) -> None:
376 """Poll player for state updates."""
377 # OPTIONAL - This is called by the Player Manager if the 'needs_poll' property is True.
378 self._set_attributes()
379 self.update_state()
380
381 async def on_unload(self) -> None:
382 """Handle logic when the player is unloaded from the Player controller."""
383 # OPTIONAL
384 # this method is optional and should be implemented if you need to handle
385 # any logic when the player is unloaded from the Player controller.
386 # This is called when the player is removed from the Player controller.
387 self.logger.info("Player %s unloaded", self.name)
388
389 def _set_attributes(self) -> None:
390 """Update/set (dynamic) properties."""
391 self._attr_powered = True
392 self._attr_volume_muted = False
393 self._attr_volume_level = 50
394