/
/
/
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_player: str | None = None
127 # tracks the active stream_session_id for the current stream request.
128 # Paired with _in_use_by_player: same-queue reconnects refresh this token
129 # without changing _in_use_by_player, 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.players.update_source_metadata(player_id, ...).
236 #
237 # Silence-during-pause contract:
238 # - stream_type=CUSTOM: the server wraps your audio generator with a
239 # silence-keepalive, so you can just stop yielding bytes while the
240 # upstream device is paused. The wrapper inserts silence frames at
241 # the declared PCM format and the player stays connected.
242 # - stream_type=NAMED_PIPE: the underlying process MUST keep writing
243 # silence to the pipe during pause (most popular binaries like
244 # shairport-sync and librespot do this in passthrough mode). If
245 # the producer actually stops writing, ffmpeg will block and the
246 # player will eventually disconnect.
247 if item_id != AUDIO_SOURCE_ID:
248 raise MediaNotFoundError(f"Unknown AudioSource: {item_id}")
249 return StreamDetails(
250 provider=self.instance_id,
251 item_id=item_id,
252 audio_format=AudioFormat(
253 content_type=ContentType.PCM_S16LE,
254 sample_rate=44100,
255 bit_depth=16,
256 channels=2,
257 ),
258 media_type=MediaType.AUDIO_SOURCE,
259 stream_type=StreamType.CUSTOM,
260 stream_metadata=StreamMetadata(title=self.name),
261 )
262
263 async def get_audio_stream(
264 self, streamdetails: StreamDetails, seek_position: int = 0
265 ) -> AsyncGenerator[bytes]:
266 """Yield raw audio bytes for the given streamdetails."""
267 # OPTIONAL
268 # Will only be called when get_stream_details returned
269 # stream_type=StreamType.CUSTOM. Yield bytes in the PCM format declared
270 # by streamdetails.audio_format. Release any per-stream resources in a
271 # try/finally â the consumer closes the generator when playback ends.
272 #
273 # Lock release pattern: snapshot BOTH the queue id AND the active
274 # session id at stream start, then guard the finally release on both
275 # still matching. A queue_id-only guard is unsafe for same-queue
276 # reconnects: when the player drops + reopens the same URL,
277 # on_source_selected fires again with a fresh stream_session_id (but
278 # the same queue id), and the prior generator's teardown would
279 # otherwise clear the lock that now belongs to the new session.
280 consumer_queue = self._in_use_by_player
281 captured_session_id = self._active_session_id
282 # 100ms of silence at 44.1kHz/16bit/stereo PCM â matches the audio_format
283 # declared in get_stream_details. Replace with your actual byte source.
284 pcm_chunk = b"\x00" * (44100 * 2 * 2 // 10)
285 try:
286 for _ in range(100):
287 yield pcm_chunk
288 finally:
289 if (
290 self._in_use_by_player == consumer_queue
291 and self._active_session_id == captured_session_id
292 ):
293 self._in_use_by_player = None
294
295 async def on_source_control(
296 self,
297 source_id: str,
298 action: SourceControl,
299 value: SourceControlValue = None,
300 ) -> None:
301 """Handle a playback control command for the active AudioSource."""
302 # OPTIONAL
303 # Called when the AudioSource is the active queue item and the user
304 # invokes a control whose capability flag (can_play_pause, can_seek,
305 # can_next_previous) is True. value carries SEEK position in seconds
306 # and is unused for other actions.
307 # Plugins that advertise no controls do not need to override this.
308 # Volume sync (e.g. mirroring the upstream app's volume slider) lives
309 # on the separate on_volume_change hook below.
310 raise NotImplementedError
311
312 async def on_volume_change(self, source_id: str, volume: int) -> None:
313 """React to a volume change on the player streaming this AudioSource."""
314 # OPTIONAL
315 # Implement when the plugin wants to sync the upstream device's volume
316 # display with MA (e.g. Spotify Connect mirroring the Spotify app slider).
317 # Fired only on the direct queue owner â group volume changes fire once
318 # at the group level, not per child.
319
320 async def on_source_selected(
321 self, source_id: str, player_id: str, owner_player_id: str, stream_session_id: str
322 ) -> None:
323 """React to an AudioSource being selected for playback on a player."""
324 # OPTIONAL â fires only on the actual stream request (not on queue
325 # preload). This is the single point where exclusive AudioSources
326 # claim ownership: doing it here (instead of in get_stream_details)
327 # keeps the preload path side-effect-free so a player-switch handoff
328 # is not blocked at queue prep time.
329 #
330 # Plugins MUST store stream_session_id so the matching
331 # on_source_unselected callback can reject stale teardowns from
332 # superseded same-queue requests. Typical pattern for exclusive
333 # sources:
334 #
335 # if self._active_player_id and self._active_player_id != player_id:
336 # await self.mass.players.cmd_stop(self._active_player_id)
337 # self._in_use_by_player = owner_player_id # claim (overwrites prior queue's)
338 # self._active_session_id = stream_session_id
339 # self._active_player_id = player_id
340 #
341 # If allow_player_switch is False and the requesting player is not the
342 # configured target, redirect via mass.player_queues.play_media(target,
343 # uri) and then RAISE so the original (disallowed) request does not
344 # continue into get_stream_details and stream to the wrong player.
345
346 async def on_source_unselected(
347 self, source_id: str, owner_player_id: str, stream_session_id: str
348 ) -> None:
349 """React to MA tearing down this AudioSource's stream from a queue."""
350 # OPTIONAL â fires in the queue-item stream handler's finally block, so
351 # it runs regardless of how streaming ended (normal completion, client
352 # disconnect, exception). Lets NAMED_PIPE plugins release ownership
353 # without depending on an external session event.
354 #
355 # Guard with a stream_session_id match â NOT just a owner_player_id match â
356 # so a stale callback from a superseded same-queue request (player
357 # drops + reopens the same URL before the prior finally fires) cannot
358 # clear the live claim of the new stream:
359 #
360 # if self._active_session_id != stream_session_id:
361 # return
362 # self._active_session_id = None
363 # if self._in_use_by_player == owner_player_id:
364 # self._in_use_by_player = None
365
366 async def search(
367 self,
368 search_query: str,
369 media_types: list[MediaType],
370 limit: int = 5,
371 ) -> SearchResults:
372 """Perform a search against this plugin's content."""
373 # OPTIONAL
374 # Will only be called if ProviderFeature.SEARCH is declared.
375 # Return a SearchResults with items belonging to this plugin (typically
376 # Playlist items whose provider_mappings point back to this plugin).
377 # The global search controller interleaves the result with results
378 # from other providers.
379 return SearchResults()
380
381 async def get_similar_tracks(self, track: Track, limit: int = 25) -> list[Track]:
382 """Retrieve a list of similar tracks for the given track."""
383 # OPTIONAL
384 # Will only be called if ProviderFeature.SIMILAR_TRACKS is declared.
385 # Results should be Track objects with provider_mappings pointing to
386 # existing music providers so MA's playback path resolves normally.
387 return []
388
389 async def get_recommendations(self) -> list[RecommendationFolder]:
390 """Get this plugin's available recommendation rows, without items."""
391 # OPTIONAL
392 # Will only be called if ProviderFeature.RECOMMENDATIONS is declared.
393 # Return the recommendation rows this plugin offers as
394 # RecommendationFolder objects WITHOUT items (leave the default empty
395 # items list). This must be fast: hardcoded or locally cached row
396 # descriptors only, no backend calls â MA uses it to instantly render
397 # the row shells and the recommendations settings page.
398 # Keep each row's item_id stable across calls and restarts: the user's
399 # per-row preferences (enabled/hidden, order) are keyed on it.
400 return []
401
402 async def get_recommendation_items(
403 self, item_id: str
404 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
405 """Get the items for a single recommendation row."""
406 # OPTIONAL
407 # Will only be called if ProviderFeature.RECOMMENDATIONS is declared.
408 # Called separately (possibly in parallel) for each enabled row
409 # returned by get_recommendations; any (slow) backend fetching belongs
410 # here, not in get_recommendations. Rows may contain Playlist items
411 # pointing back to this plugin via their provider_mappings; users can
412 # add such a Playlist to their library through the standard
413 # add-to-library flow.
414 # Return an empty UniqueList for an unknown item_id.
415 return UniqueList()
416
417 async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
418 """Browse this plugin's contents."""
419 # OPTIONAL
420 # Will only be called if ProviderFeature.BROWSE is declared.
421 # Plugins surfacing playlists should yield Playlist items here. Each
422 # Playlist MUST have at least one ProviderMapping pointing back to
423 # this plugin's instance_id/domain so MA can resolve get_playlist /
424 # get_playlist_tracks against the correct provider when the user adds
425 # it to their library.
426 return []
427
428 async def get_playlist(self, prov_playlist_id: str) -> Playlist:
429 """Return details of a single playlist owned by this plugin."""
430 # OPTIONAL
431 # Implement when surfacing playlists via browse / recommendations so
432 # MA can refresh metadata once a user adds it to their library.
433 return Playlist(
434 item_id=prov_playlist_id,
435 provider=self.instance_id,
436 name="Example Playlist",
437 provider_mappings={
438 ProviderMapping(
439 item_id=prov_playlist_id,
440 provider_domain=self.domain,
441 provider_instance=self.instance_id,
442 )
443 },
444 )
445
446 async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
447 """Return a page of tracks for a playlist owned by this plugin."""
448 # OPTIONAL
449 # Implement when surfacing playlists. Tracks SHOULD carry
450 # ProviderMappings pointing to real music providers so MA's playback
451 # path resolves normally.
452 return []
453