music-assistant-server

1"""Demo Player implementation."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7from music_assistant_models.config_entries import ConfigEntry, ConfigValueType
8from music_assistant_models.enums import ConfigEntryType, PlaybackState, PlayerFeature
9from music_assistant_models.player import 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        }
32        self._set_attributes()
33
34    async def on_config_updated(self) -> None:
35        """Handle logic when the player is loaded or updated."""
36        # OPTIONAL
37        # This method is optional and should be implemented if you need to handle
38        # any initialization logic after the config was initially loaded or updated.
39        # This is called after the player is registered and self.config was loaded.
40        # And also when the config was updated.
41        # You don't need to call update_state() here.
42
43    @property
44    def needs_poll(self) -> bool:
45        """Return if the player needs to be polled for state updates."""
46        # MANDATORY
47        # this should return True if the player needs to be polled for state updates,
48        # If you player does not need to be polled, you can return False.
49        return True
50
51    @property
52    def poll_interval(self) -> int:
53        """Return the interval in seconds to poll the player for state updates."""
54        # OPTIONAL
55        # used in conjunction with the needs_poll property.
56        # this should return the interval in seconds to poll the player for state updates.
57        return 5 if self._attr_playback_state == PlaybackState.PLAYING else 30
58
59    @property
60    def source_list(self) -> list[PlayerSource]:
61        """Return list of available (native) sources for this player."""
62        # OPTIONAL - required only if you specified PlayerFeature.SELECT_SOURCE
63        # this is an optional property that you can implement if your
64        # player supports (external) source control (aux, HDMI, etc.).
65        # If your player does not support sources, you can leave this out completely.
66        return [
67            PlayerSource(
68                id="line_in",
69                name="Line-In",
70                passive=False,
71                can_play_pause=False,
72                can_next_previous=False,
73                can_seek=False,
74            ),
75            PlayerSource(
76                id="spotify_connect",
77                name="Spotify",
78                # by specifying passive=True, we indicate that this source
79                # is not actively selectable by the user from the UI.
80                passive=True,
81                can_play_pause=True,
82                can_next_previous=True,
83                can_seek=True,
84            ),
85        ]
86
87    async def get_config_entries(
88        self,
89        action: str | None = None,
90        values: dict[str, ConfigValueType] | None = None,
91    ) -> list[ConfigEntry]:
92        """Return all (provider/player specific) Config Entries for the player."""
93        # OPTIONAL
94        # this method is optional and should be implemented if you need player specific
95        # configuration entries. If you do not need player specific configuration entries,
96        # you can leave this method out completely.
97        # note that the config controller will always add a set of default config entries
98        # if you want, you can override those by specifying the same key as a default entry.
99        return [
100            # example of a player specific config entry
101            # you can also override a default entry by specifying the same key
102            # as a default entry, but with a different type or default value.
103            ConfigEntry(
104                key="demo_player_setting",
105                type=ConfigEntryType.STRING,
106                label="Demo Player Setting",
107                required=False,
108                default_value="default_value",
109                description="This is a demo player setting.",
110            ),
111        ]
112
113    async def power(self, powered: bool) -> None:
114        """Handle POWER command on the player."""
115        # OPTIONAL - required only if you specified PlayerFeature.POWER
116        # this method should send a power on/off command to the given player.
117        logger = self.provider.logger.getChild(self.player_id)
118        if powered:
119            # In this demo implementation we just set the power state to ON
120            # and optimistically update the state.
121            # In a real implementation you would read the actual value from the player
122            # either from a callback or by polling the player.
123            logger.info("Received POWER ON command on player %s", self.display_name)
124            self._attr_powered = True
125        else:
126            # In this demo implementation we just set the power state to OFF
127            # and optimistically update the state.
128            # In a real implementation you would read the actual value from the player
129            # either from a callback or by polling the player.
130            logger.info("Received POWER OFF command on player %s", self.display_name)
131            self._attr_powered = False
132        # update the player state in the player manager
133        self.update_state()
134
135    async def volume_set(self, volume_level: int) -> None:
136        """Handle VOLUME_SET command on the player."""
137        # OPTIONAL - required only if you specified PlayerFeature.VOLUME_SET
138        # this method should send a volume set command to the given player.
139
140        # In this demo implementation we just set the volume level
141        # and optimistically update the state.
142        # In a real implementation you would send a command to the actual player and
143        # get the actual value from the player either from a callback or by polling the player.
144        logger = self.provider.logger.getChild(self.player_id)
145        logger.info(
146            "Received VOLUME_SET command on player %s with level %s",
147            self.display_name,
148            volume_level,
149        )
150        self._attr_volume_level = volume_level  # volume level is between 0 and 100
151        # update the player state in the player manager
152        self.update_state()
153
154    async def volume_mute(self, muted: bool) -> None:
155        """Handle VOLUME MUTE command on the player."""
156        # OPTIONAL - required only if you specified PlayerFeature.VOLUME_MUTE
157        # this method should send a volume mute command to the given player.
158        logger = self.provider.logger.getChild(self.player_id)
159        logger.info(
160            "Received VOLUME_MUTE command on player %s with muted %s", self.display_name, muted
161        )
162        self._attr_volume_muted = muted
163        self.update_state()
164
165    async def play(self) -> None:
166        """Play command."""
167        # MANDATORY
168        # this method is mandatory and should be implemented.
169        # this method should send a play/resume command to the given player.
170        # normally this is the point where you would resume playback
171        # on your actual player device.
172
173        # In this demo implementation we just set the playback state to PLAYING
174        # and optimistically set the playback state to PLAYING.
175        # In a real implementation you actually send a command to the player
176        # wait for the player to report a new state before updating the playback state.
177        logger = self.provider.logger.getChild(self.player_id)
178        logger.info("Received PLAY command on player %s", self.display_name)
179        self._attr_playback_state = PlaybackState.PLAYING
180        self.update_state()
181
182    async def stop(self) -> None:
183        """Stop command."""
184        # MANDATORY
185        # this method is mandatory and should be implemented.
186        # this method should send a stop command to the given player.
187        # normally this is the point where you would stop playback
188        # on your actual player device.
189
190        # In this demo implementation we just set the playback state to IDLE
191        # and optimistically set the playback state to IDLE.
192        # In a real implementation you actually send a command to the player
193        # wait for the player to report a new state before updating the playback state.
194        logger = self.provider.logger.getChild(self.player_id)
195        logger.info("Received STOP command on player %s", self.display_name)
196        self._attr_playback_state = PlaybackState.IDLE
197        self._attr_active_source = None
198        self._attr_current_media = None
199        self.update_state()
200
201    async def pause(self) -> None:
202        """Pause command."""
203        # OPTIONAL - required only if you specified PlayerFeature.PAUSE
204        # this method should send a pause command to the given player.
205
206        # In this demo implementation we just set the playback state to PAUSED
207        # and optimistically set the playback state to PAUSED.
208        # In a real implementation you actually send a command to the player
209        # wait for the player to report a new state before updating the playback state.
210        logger = self.provider.logger.getChild(self.player_id)
211        logger.info("Received PAUSE command on player %s", self.display_name)
212        self._attr_playback_state = PlaybackState.PAUSED
213        self.update_state()
214
215    async def next_track(self) -> None:
216        """Next command."""
217        # OPTIONAL - required only if you specified PlayerFeature.NEXT_PREVIOUS
218        # this method should send a next track command to the given player.
219        # Note that this is only needed/used if the player is playing a 3rd party
220        # stream (e.g. Spotify, YouTube, etc.) and the player supports skipping to the next track.
221        # When the player is playing MA content, this is already handled in the Queue controller.
222
223    async def previous_track(self) -> None:
224        """Previous command."""
225        # OPTIONAL - required only if you specified PlayerFeature.NEXT_PREVIOUS
226        # this method should send a previous track command to the given player.
227        # Note that this is only needed/used if the player is playing a 3rd party
228        # stream (e.g. Spotify, YouTube, etc.) and the player supports skipping to the next track.
229        # When the player is playing MA content, this is already handled in the Queue controller.
230
231    async def seek(self, position: int) -> None:
232        """SEEK command on the player."""
233        # OPTIONAL - required only if you specified PlayerFeature.SEEK
234        # this method should send a seek command to the given player.
235        # the position is the position in seconds to seek to in the current playing item.
236
237    async def play_media(self, media: PlayerMedia) -> None:
238        """Play media command."""
239        # MANDATORY
240        # This method is mandatory and should be implemented.
241        # This method should handle the play_media command for the given player.
242        # It will be called when media needs to be played on the player.
243        # The media object contains all the details needed to play the item.
244
245        # In 99% of the cases this will be called by the Queue controller to play
246        # a single item from the queue on the player and the uri within the media
247        # object will then contain the URL to play that single queue item.
248
249        # If your player provider does not support enqueuing of items,
250        # the queue controller will simply call this play_media method for
251        # each item in the queue to play them one by one.
252
253        # In order to support true gapless and/or enqueuing, we offer the option of
254        # 'flow_mode' playback. In that case the queue controller will stitch together
255        # all songs in the playbook queue into a single stream and send that to the player.
256        # In that case the URI (and metadata) received here is that of the 'flow mode' stream.
257
258        # Examples of player providers that use flow mode for playback by default are AirPlay,
259        # SnapCast and Fully Kiosk.
260
261        # Examples of player providers that optionally use 'flow mode' are Google Cast and
262        # Home Assistant. They provide a config entry to enable flow mode playback.
263
264        # Examples of player providers that natively support enqueuing of items are Sonos,
265        # Slimproto and Google Cast.
266
267        # In this demo implementation we just optimistically set the state.
268        # In a real implementation you actually send a command to the player
269        # wait for the player to report a new state before updating the playback state.
270        url = await self.provider.mass.streams.resolve_stream_url(self.player_id, media)
271        logger = self.provider.logger.getChild(self.player_id)
272        logger.info("Received PLAY_MEDIA command on player %s with url %s", self.display_name, url)
273        self._attr_current_media = media
274        self._attr_playback_state = PlaybackState.PLAYING
275        self.update_state()
276
277    async def enqueue_next_media(self, media: PlayerMedia) -> None:
278        """Handle enqueuing of the next (queue) item on the player."""
279        # OPTIONAL - required only if you specified PlayerFeature.ENQUEUE
280        # This method is optional and should be implemented if you want to support
281        # enqueuing of the next item on the player.
282        # This will be called when the player reports it started buffering a queue item
283        # and when the queue items updated.
284        # A PlayerProvider implementation is in itself responsible for handling this
285        # so that the queue items keep playing until its empty or the player stopped.
286
287    async def play_announcement(
288        self, announcement: PlayerMedia, volume_level: int | None = None
289    ) -> None:
290        """Handle (native) playback of an announcement on the player."""
291        # OPTIONAL - required only if you specified PlayerFeature.PLAY_ANNOUNCEMENT
292        # This method is optional and should be implemented if the player supports
293        # NATIVE playback of announcements (with ducking etc.).
294        # The announcement object contains all the details needed to play the announcement.
295        # The volume_level is optional and can be used to set the volume level for the announcement.
296        # If you do not use the announcement playerfeature, the default behavior is to play the
297        # announcement as a regular media item using the play_media method and the MA player manager
298        # will take care of setting the volume level for the announcement and resuming etc.
299
300    async def select_source(self, source: str) -> None:
301        """Handle SELECT SOURCE command on the player."""
302        # OPTIONAL - required only if you specified PlayerFeature.SELECT_SOURCE
303        # This method is optional and should be implemented if the player supports
304        # selecting a source (e.g. HDMI, AUX, etc.) on the player.
305        # The source is the source ID to select on the player.
306        # available sources are specified in the Player.source_list property
307
308    async def set_members(
309        self,
310        player_ids_to_add: list[str] | None = None,
311        player_ids_to_remove: list[str] | None = None,
312    ) -> None:
313        """Handle SET_MEMBERS command on the player."""
314        # OPTIONAL - required only if you specified PlayerFeature.SET_MEMBERS
315        # This method is optional and should be implemented if the player supports
316        # syncing/grouping with other players.
317
318    async def poll(self) -> None:
319        """Poll player for state updates."""
320        # OPTIONAL - This is called by the Player Manager if the 'needs_poll' property is True.
321        self._set_attributes()
322        self.update_state()
323
324    async def on_unload(self) -> None:
325        """Handle logic when the player is unloaded from the Player controller."""
326        # OPTIONAL
327        # this method is optional and should be implemented if you need to handle
328        # any logic when the player is unloaded from the Player controller.
329        # This is called when the player is removed from the Player controller.
330        self.logger.info("Player %s unloaded", self.name)
331
332    def _set_attributes(self) -> None:
333        """Update/set (dynamic) properties."""
334        self._attr_powered = True
335        self._attr_volume_muted = False
336        self._attr_volume_level = 50
337