music-assistant-server

1"""
2DEMO/TEMPLATE Plugin Provider for Music Assistant.
3
4This is an empty plugin provider with no actual implementation.
5Its meant to get started developing a new plugin provider for Music Assistant.
6
7Use it as a reference to discover what methods exists and what they should return.
8Also it is good to look at existing plugin providers to get a better understanding.
9
10In general, a plugin provider does not have any mandatory implementation details.
11It provides additional functionality to Music Assistant and most often it will
12interact with the existing core controllers and event logic. For example a Scrobble plugin.
13
14If your plugin needs to communicate with external services or devices, you need to
15use a dedicated (async) library for that. You can add these dependencies to the
16manifest.json file in the requirements section,
17which is a list of (versioned!) python modules (pip syntax) that should be installed
18when the provider is selected by the user.
19
20To add a new plugin provider to Music Assistant, you need to create a new folder
21in the providers folder with the name of your provider (e.g. 'my_plugin_provider').
22In that folder you should create (at least) a __init__.py file and a manifest.json file.
23
24Optional is an icon.svg file that will be used as the icon for the provider in the UI,
25but we also support that you specify a material design icon in the manifest.json file.
26
27IMPORTANT NOTE:
28We strongly recommend developing on either macOS or Linux and start your development
29environment by running the setup.sh scripts in the scripts folder of the repository.
30This will create a virtual environment and install all dependencies needed for development.
31See also our general DEVELOPMENT.md guide in the repository for more information.
32
33"""
34
35from __future__ import annotations
36
37from collections.abc import AsyncGenerator, Sequence
38from typing import TYPE_CHECKING
39
40from music_assistant_models.enums import (
41    ContentType,
42    EventType,
43    MediaType,
44    ProviderFeature,
45    StreamType,
46)
47from music_assistant_models.errors import MediaNotFoundError
48from music_assistant_models.media_items import (
49    AudioSource,
50    Playlist,
51    ProviderMapping,
52    SearchResults,
53    UniqueList,
54)
55from music_assistant_models.media_items.audio_format import AudioFormat
56from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
57
58from music_assistant.models.plugin import PluginProvider, SourceControlValue
59
60if TYPE_CHECKING:
61    from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
62    from music_assistant_models.enums import SourceControl
63    from music_assistant_models.event import MassEvent
64    from music_assistant_models.media_items import (
65        BrowseFolder,
66        ItemMapping,
67        MediaItemType,
68        RecommendationFolder,
69        Track,
70    )
71    from music_assistant_models.provider import ProviderManifest
72
73    from music_assistant.mass import MusicAssistant
74    from music_assistant.models import ProviderInstanceType
75
76
77# stable id for the single AudioSource this demo provider exposes;
78# combined with the provider instance_id this forms the persistent browse/play uri
79# (e.g. `<instance_id>://audio_source/main`) listed under "Live Inputs". AudioSource
80# items are not favoritable / library-backed in MA core today.
81AUDIO_SOURCE_ID = "main"
82
83SUPPORTED_FEATURES = {
84    # MANDATORY
85    # this constant should contain a set of provider-level features
86    # that your provider supports or an empty set if none.
87    # see the ProviderFeature enum for all available features
88    # at time of writing the only plugin-specific feature is the
89    # 'AUDIO_SOURCE' feature which indicates that this provider can
90    # provide a (single) audio source to Music Assistant, such as a live stream.
91    # we add this feature here to demonstrate the concept.
92    ProviderFeature.AUDIO_SOURCE
93}
94
95
96async def setup(
97    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
98) -> ProviderInstanceType:
99    """Initialize provider(instance) with given configuration."""
100    # setup is called when the user wants to setup a new provider instance.
101    # you are free to do any preflight checks here and but you must return
102    # an instance of the provider.
103    return MyDemoPluginprovider(mass, manifest, config, SUPPORTED_FEATURES)
104
105
106class MyDemoPluginprovider(PluginProvider):
107    """
108    Example/demo Plugin provider.
109
110    Note that this is always subclassed from PluginProvider,
111    which in turn is a subclass of the generic Provider model.
112
113    The base implementation already takes care of some convenience methods,
114    such as the mass object and the logger. Take a look at the base class
115    for more information on what is available.
116
117    Just like with any other subclass, make sure that if you override
118    any of the default methods (such as __init__), you call the super() method.
119    In most cases its not needed to override any of the builtin methods and you only
120    implement the abc methods with your actual implementation.
121    """
122
123    # tracks which queue currently owns the exclusive AudioSource. Set in
124    # on_source_selected (NOT in get_stream_details — that path also runs from
125    # queue preload, where claiming would block a later cross-queue handoff).
126    _in_use_by_queue: str | None = None
127    # tracks the active stream_session_id for the current stream request.
128    # Paired with _in_use_by_queue: same-queue reconnects refresh this token
129    # without changing _in_use_by_queue, so stream loops and generator
130    # finallys must guard their lock release on both still matching.
131    _active_session_id: str | None = None
132
133    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
134        """
135        Return the (options) config entries for this (existing) provider instance.
136
137        Return an empty tuple when the provider has no options. Interactive setup
138        input (if any) is collected by a ``setup_flow.py`` module; one-shot buttons
139        are declared here as ``ConfigEntryType.ACTION`` entries and handled in
140        ``handle_config_action``.
141        """
142        return ()
143
144    async def loaded_in_mass(self) -> None:
145        """Call after the provider has been loaded."""
146        # OPTIONAL
147        # this is an optional method that you can implement if
148        # relevant or leave out completely if not needed.
149        # it will be called after the provider has been fully loaded into Music Assistant.
150        # you can use this for instance to trigger custom (non-mdns) discovery of plugins
151        # or any other logic that needs to run after the provider is fully loaded.
152
153        # as reference we will subscribe here to an event on the MA eventbus
154        # this is just an example and you can remove this if not needed.
155        async def handle_event(event: MassEvent) -> None:
156            if event.event == EventType.MEDIA_ITEM_PLAYED:
157                # example implementation of handling a media item played event
158                self.logger.info("Media item played event received: %s", event.data)
159
160        self.mass.subscribe(handle_event, EventType.MEDIA_ITEM_PLAYED)
161
162    async def unload(self, is_removed: bool = False) -> None:
163        """
164        Handle unload/close of the provider.
165
166        Called when provider is deregistered (e.g. MA exiting or config reloading).
167        is_removed will be set to True when the provider is removed from the configuration.
168        """
169        # OPTIONAL
170        # this is an optional method that you can implement if
171        # relevant or leave out completely if not needed.
172        # it will be called when the provider is unloaded from Music Assistant.
173        # this means also when the provider is getting reloaded
174
175    async def get_audio_sources(self) -> list[AudioSource]:
176        """Return the AudioSources this plugin currently exposes."""
177        # OPTIONAL
178        # Will only be called if ProviderFeature.AUDIO_SOURCE is declared.
179        #
180        # Return one or more AudioSource MediaItems describing the live
181        # inputs this plugin offers. AudioSources show up under the global
182        # "Live Inputs" browse node and can be played on any player via
183        # the standard play_media flow. Capability flags (can_play_pause,
184        # can_seek, can_next_previous) drive which control buttons the UI
185        # surfaces and which commands the server proxies through to
186        # on_source_control.
187        #
188        # Most plugins expose a single source; for those, build it once in
189        # __init__ and return the cached instance here. Plugins that expose
190        # multiple sources (e.g. a hardware bridge with multiple inputs) can
191        # rebuild the list at call time.
192        return [
193            AudioSource(
194                item_id=AUDIO_SOURCE_ID,
195                provider=self.instance_id,
196                name=self.name,
197                provider_mappings={
198                    ProviderMapping(
199                        item_id=AUDIO_SOURCE_ID,
200                        provider_domain=self.domain,
201                        provider_instance=self.instance_id,
202                        audio_format=AudioFormat(
203                            content_type=ContentType.PCM_S16LE,
204                            sample_rate=44100,
205                            bit_depth=16,
206                            channels=2,
207                        ),
208                    )
209                },
210                can_play_pause=False,
211                can_seek=False,
212                can_next_previous=False,
213                exclusive=True,
214                allow_external_trigger=False,
215            )
216        ]
217
218    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
219        """Return StreamDetails for streaming an item owned by this plugin."""
220        # OPTIONAL
221        # Called for any playable item this plugin exposes; media_type tells you
222        # which kind. This demo only has an AudioSource.
223        #
224        # MUST be side-effect-free — no exclusivity claim, no busy raise.
225        # MA calls this from both the streaming path AND from queue preload, so
226        # mutating provider state here would let a preload accidentally claim
227        # the source and block a subsequent cross-queue handoff. Ownership is
228        # claimed in on_source_selected (which fires only on the actual stream
229        # request, not on preload) — see the example below.
230        #
231        # Return a StreamDetails with stream_type=CUSTOM when audio comes from
232        # an async generator (get_audio_stream below), or stream_type=NAMED_PIPE
233        # plus a path when audio comes from a named pipe / file. stream_metadata
234        # carries the initial track info; update it at runtime via
235        # mass.streams.update_stream_metadata(queue_id, ...) — same mechanism
236        # ICY radio metadata uses.
237        #
238        # Silence-during-pause contract:
239        # - stream_type=CUSTOM: the server wraps your audio generator with a
240        #   silence-keepalive, so you can just stop yielding bytes while the
241        #   upstream device is paused. The wrapper inserts silence frames at
242        #   the declared PCM format and the player stays connected.
243        # - stream_type=NAMED_PIPE: the underlying process MUST keep writing
244        #   silence to the pipe during pause (most popular binaries like
245        #   shairport-sync and librespot do this in passthrough mode). If
246        #   the producer actually stops writing, ffmpeg will block and the
247        #   player will eventually disconnect.
248        if item_id != AUDIO_SOURCE_ID:
249            raise MediaNotFoundError(f"Unknown AudioSource: {item_id}")
250        return StreamDetails(
251            provider=self.instance_id,
252            item_id=item_id,
253            audio_format=AudioFormat(
254                content_type=ContentType.PCM_S16LE,
255                sample_rate=44100,
256                bit_depth=16,
257                channels=2,
258            ),
259            media_type=MediaType.AUDIO_SOURCE,
260            stream_type=StreamType.CUSTOM,
261            stream_metadata=StreamMetadata(title=self.name),
262        )
263
264    async def get_audio_stream(
265        self, streamdetails: StreamDetails, seek_position: int = 0
266    ) -> AsyncGenerator[bytes]:
267        """Yield raw audio bytes for the given streamdetails."""
268        # OPTIONAL
269        # Will only be called when get_stream_details returned
270        # stream_type=StreamType.CUSTOM. Yield bytes in the PCM format declared
271        # by streamdetails.audio_format. Release any per-stream resources in a
272        # try/finally — the consumer closes the generator when playback ends.
273        #
274        # Lock release pattern: snapshot BOTH the queue id AND the active
275        # session id at stream start, then guard the finally release on both
276        # still matching. A queue_id-only guard is unsafe for same-queue
277        # reconnects: when the player drops + reopens the same URL,
278        # on_source_selected fires again with a fresh stream_session_id (but
279        # the same queue id), and the prior generator's teardown would
280        # otherwise clear the lock that now belongs to the new session.
281        consumer_queue = self._in_use_by_queue
282        captured_session_id = self._active_session_id
283        # 100ms of silence at 44.1kHz/16bit/stereo PCM — matches the audio_format
284        # declared in get_stream_details. Replace with your actual byte source.
285        pcm_chunk = b"\x00" * (44100 * 2 * 2 // 10)
286        try:
287            for _ in range(100):
288                yield pcm_chunk
289        finally:
290            if (
291                self._in_use_by_queue == consumer_queue
292                and self._active_session_id == captured_session_id
293            ):
294                self._in_use_by_queue = None
295
296    async def on_source_control(
297        self,
298        source_id: str,
299        action: SourceControl,
300        value: SourceControlValue = None,
301    ) -> None:
302        """Handle a playback control command for the active AudioSource."""
303        # OPTIONAL
304        # Called when the AudioSource is the active queue item and the user
305        # invokes a control whose capability flag (can_play_pause, can_seek,
306        # can_next_previous) is True. value carries SEEK position in seconds
307        # and is unused for other actions.
308        # Plugins that advertise no controls do not need to override this.
309        # Volume sync (e.g. mirroring the upstream app's volume slider) lives
310        # on the separate on_volume_change hook below.
311        raise NotImplementedError
312
313    async def on_volume_change(self, source_id: str, volume: int) -> None:
314        """React to a volume change on the player streaming this AudioSource."""
315        # OPTIONAL
316        # Implement when the plugin wants to sync the upstream device's volume
317        # display with MA (e.g. Spotify Connect mirroring the Spotify app slider).
318        # Fired only on the direct queue owner — group volume changes fire once
319        # at the group level, not per child.
320
321    async def on_source_selected(
322        self, source_id: str, player_id: str, queue_id: str, stream_session_id: str
323    ) -> None:
324        """React to an AudioSource being selected for playback on a player."""
325        # OPTIONAL — fires only on the actual stream request (not on queue
326        # preload). This is the single point where exclusive AudioSources
327        # claim ownership: doing it here (instead of in get_stream_details)
328        # keeps the preload path side-effect-free so a player-switch handoff
329        # is not blocked at queue prep time.
330        #
331        # Plugins MUST store stream_session_id so the matching
332        # on_source_unselected callback can reject stale teardowns from
333        # superseded same-queue requests. Typical pattern for exclusive
334        # sources:
335        #
336        #     if self._active_player_id and self._active_player_id != player_id:
337        #         await self.mass.players.cmd_stop(self._active_player_id)
338        #     self._in_use_by_queue = queue_id  # claim (overwrites prior queue's)
339        #     self._active_session_id = stream_session_id
340        #     self._active_player_id = player_id
341        #
342        # If allow_player_switch is False and the requesting player is not the
343        # configured target, redirect via mass.player_queues.play_media(target,
344        # uri) and then RAISE so the original (disallowed) request does not
345        # continue into get_stream_details and stream to the wrong player.
346
347    async def on_source_unselected(
348        self, source_id: str, queue_id: str, stream_session_id: str
349    ) -> None:
350        """React to MA tearing down this AudioSource's stream from a queue."""
351        # OPTIONAL — fires in the queue-item stream handler's finally block, so
352        # it runs regardless of how streaming ended (normal completion, client
353        # disconnect, exception). Lets NAMED_PIPE plugins release ownership
354        # without depending on an external session event.
355        #
356        # Guard with a stream_session_id match — NOT just a queue_id match —
357        # so a stale callback from a superseded same-queue request (player
358        # drops + reopens the same URL before the prior finally fires) cannot
359        # clear the live claim of the new stream:
360        #
361        #     if self._active_session_id != stream_session_id:
362        #         return
363        #     self._active_session_id = None
364        #     if self._in_use_by_queue == queue_id:
365        #         self._in_use_by_queue = None
366
367    async def search(
368        self,
369        search_query: str,
370        media_types: list[MediaType],
371        limit: int = 5,
372    ) -> SearchResults:
373        """Perform a search against this plugin's content."""
374        # OPTIONAL
375        # Will only be called if ProviderFeature.SEARCH is declared.
376        # Return a SearchResults with items belonging to this plugin (typically
377        # Playlist items whose provider_mappings point back to this plugin).
378        # The global search controller interleaves the result with results
379        # from other providers.
380        return SearchResults()
381
382    async def get_similar_tracks(self, track: Track, limit: int = 25) -> list[Track]:
383        """Retrieve a list of similar tracks for the given track."""
384        # OPTIONAL
385        # Will only be called if ProviderFeature.SIMILAR_TRACKS is declared.
386        # Results should be Track objects with provider_mappings pointing to
387        # existing music providers so MA's playback path resolves normally.
388        return []
389
390    async def get_recommendations(self) -> list[RecommendationFolder]:
391        """Get this plugin's available recommendation rows, without items."""
392        # OPTIONAL
393        # Will only be called if ProviderFeature.RECOMMENDATIONS is declared.
394        # Return the recommendation rows this plugin offers as
395        # RecommendationFolder objects WITHOUT items (leave the default empty
396        # items list). This must be fast: hardcoded or locally cached row
397        # descriptors only, no backend calls — MA uses it to instantly render
398        # the row shells and the recommendations settings page.
399        # Keep each row's item_id stable across calls and restarts: the user's
400        # per-row preferences (enabled/hidden, order) are keyed on it.
401        return []
402
403    async def get_recommendation_items(
404        self, item_id: str
405    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
406        """Get the items for a single recommendation row."""
407        # OPTIONAL
408        # Will only be called if ProviderFeature.RECOMMENDATIONS is declared.
409        # Called separately (possibly in parallel) for each enabled row
410        # returned by get_recommendations; any (slow) backend fetching belongs
411        # here, not in get_recommendations. Rows may contain Playlist items
412        # pointing back to this plugin via their provider_mappings; users can
413        # add such a Playlist to their library through the standard
414        # add-to-library flow.
415        # Return an empty UniqueList for an unknown item_id.
416        return UniqueList()
417
418    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
419        """Browse this plugin's contents."""
420        # OPTIONAL
421        # Will only be called if ProviderFeature.BROWSE is declared.
422        # Plugins surfacing playlists should yield Playlist items here. Each
423        # Playlist MUST have at least one ProviderMapping pointing back to
424        # this plugin's instance_id/domain so MA can resolve get_playlist /
425        # get_playlist_tracks against the correct provider when the user adds
426        # it to their library.
427        return []
428
429    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
430        """Return details of a single playlist owned by this plugin."""
431        # OPTIONAL
432        # Implement when surfacing playlists via browse / recommendations so
433        # MA can refresh metadata once a user adds it to their library.
434        return Playlist(
435            item_id=prov_playlist_id,
436            provider=self.instance_id,
437            name="Example Playlist",
438            provider_mappings={
439                ProviderMapping(
440                    item_id=prov_playlist_id,
441                    provider_domain=self.domain,
442                    provider_instance=self.instance_id,
443                )
444            },
445        )
446
447    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
448        """Return a page of tracks for a playlist owned by this plugin."""
449        # OPTIONAL
450        # Implement when surfacing playlists. Tracks SHOULD carry
451        # ProviderMappings pointing to real music providers so MA's playback
452        # path resolves normally.
453        return []
454