/
/
/
1"""Test/Demo provider that creates a collection of fake media items."""
2
3from __future__ import annotations
4
5import random
6from collections.abc import AsyncGenerator
7from datetime import UTC, datetime
8from typing import TYPE_CHECKING
9
10from music_assistant_models.config_entries import ConfigEntry
11from music_assistant_models.enums import (
12 ConfigEntryType,
13 ContentType,
14 ImageType,
15 MediaType,
16 ProviderFeature,
17 StreamType,
18)
19from music_assistant_models.media_items import (
20 Album,
21 Artist,
22 Audiobook,
23 AudioFormat,
24 ItemMapping,
25 MediaItemChapter,
26 MediaItemImage,
27 MediaItemMetadata,
28 Podcast,
29 PodcastEpisode,
30 ProviderMapping,
31 Track,
32 UniqueList,
33)
34from music_assistant_models.streamdetails import StreamDetails
35
36from music_assistant.constants import (
37 DEFAULT_GENRES,
38 MASS_LOGO,
39 SILENCE_FILE_LONG,
40 VARIOUS_ARTISTS_FANART,
41)
42from music_assistant.models.music_provider import MusicProvider
43
44if TYPE_CHECKING:
45 from music_assistant_models.config_entries import ProviderConfig
46 from music_assistant_models.provider import ProviderManifest
47
48 from music_assistant.mass import MusicAssistant
49 from music_assistant.models import ProviderInstanceType
50
51
52DEFAULT_THUMB = MediaItemImage(
53 type=ImageType.THUMB,
54 path=MASS_LOGO,
55 provider="builtin",
56 remotely_accessible=False,
57)
58
59DEFAULT_FANART = MediaItemImage(
60 type=ImageType.FANART,
61 path=VARIOUS_ARTISTS_FANART,
62 provider="builtin",
63 remotely_accessible=False,
64)
65
66CONF_KEY_NUM_ARTISTS = "num_artists"
67CONF_KEY_NUM_ALBUMS = "num_albums"
68CONF_KEY_NUM_TRACKS = "num_tracks"
69CONF_KEY_NUM_PODCASTS = "num_podcasts"
70CONF_KEY_NUM_AUDIOBOOKS = "num_audiobooks"
71
72SUPPORTED_FEATURES = {
73 ProviderFeature.BROWSE,
74 ProviderFeature.LIBRARY_ARTISTS,
75 ProviderFeature.LIBRARY_ALBUMS,
76 ProviderFeature.LIBRARY_TRACKS,
77 ProviderFeature.LIBRARY_PODCASTS,
78 ProviderFeature.LIBRARY_AUDIOBOOKS,
79 ProviderFeature.SIMILAR_TRACKS,
80}
81
82
83async def setup(
84 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
85) -> ProviderInstanceType:
86 """Initialize provider(instance) with given configuration."""
87 return TestProvider(mass, manifest, config, SUPPORTED_FEATURES)
88
89
90class TestProvider(MusicProvider):
91 """Test/Demo provider that creates a collection of fake media items."""
92
93 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
94 """Return Config entries to configure this provider."""
95 return (
96 ConfigEntry(
97 key=CONF_KEY_NUM_ARTISTS,
98 type=ConfigEntryType.INTEGER,
99 label="Number of (test) artists",
100 description="Number of test artists to generate",
101 default_value=5,
102 required=False,
103 ),
104 ConfigEntry(
105 key=CONF_KEY_NUM_ALBUMS,
106 type=ConfigEntryType.INTEGER,
107 label="Number of (test) albums per artist",
108 description="Number of test albums to generate per artist",
109 default_value=5,
110 required=False,
111 ),
112 ConfigEntry(
113 key=CONF_KEY_NUM_TRACKS,
114 type=ConfigEntryType.INTEGER,
115 label="Number of (test) tracks per album",
116 description="Number of test tracks to generate per artist-album",
117 default_value=20,
118 required=False,
119 ),
120 ConfigEntry(
121 key=CONF_KEY_NUM_PODCASTS,
122 type=ConfigEntryType.INTEGER,
123 label="Number of (test) podcasts",
124 description="Number of test podcasts to generate",
125 default_value=5,
126 required=False,
127 ),
128 ConfigEntry(
129 key=CONF_KEY_NUM_AUDIOBOOKS,
130 type=ConfigEntryType.INTEGER,
131 label="Number of (test) audiobooks",
132 description="Number of test audiobooks to generate",
133 default_value=5,
134 required=False,
135 ),
136 )
137
138 @property
139 def is_streaming_provider(self) -> bool:
140 """Return True if the provider is a streaming provider."""
141 return False
142
143 async def get_library_genres(self) -> AsyncGenerator[str]:
144 """Retrieve library genres from the provider."""
145 for genre in DEFAULT_GENRES:
146 yield genre
147
148 async def get_item_genre_names(self, media_type: MediaType, item_id: str) -> set[str]:
149 """Return genre names for a single item."""
150 if media_type == MediaType.ARTIST:
151 seed = item_id
152 elif media_type == MediaType.ALBUM:
153 seed = item_id.split("_", 2)[0]
154 elif media_type == MediaType.TRACK:
155 seed = item_id.split("_", 3)[0]
156 elif media_type == MediaType.PODCAST:
157 seed = item_id
158 elif media_type == MediaType.PODCAST_EPISODE:
159 seed = item_id.split("_", 2)[0]
160 elif media_type == MediaType.AUDIOBOOK:
161 seed = item_id
162 else:
163 return set()
164 return {random.Random(seed).choice(DEFAULT_GENRES)}
165
166 async def get_track(self, prov_track_id: str) -> Track:
167 """Get full track details by id."""
168 artist_idx, album_idx, track_idx = prov_track_id.split("_", 3)
169 genre = random.Random(artist_idx).choice(DEFAULT_GENRES)
170 return Track(
171 item_id=prov_track_id,
172 provider=self.instance_id,
173 name=f"{genre} Test Track {artist_idx} - {album_idx} - {track_idx}",
174 duration=60,
175 artists=UniqueList([await self.get_artist(artist_idx)]),
176 album=await self.get_album(f"{artist_idx}_{album_idx}"),
177 provider_mappings={
178 ProviderMapping(
179 item_id=prov_track_id,
180 provider_domain=self.domain,
181 provider_instance=self.instance_id,
182 ),
183 },
184 metadata=MediaItemMetadata(
185 images=UniqueList([DEFAULT_THUMB]),
186 genres={genre},
187 release_date=datetime(2021, 6, 15, tzinfo=UTC),
188 ),
189 disc_number=1,
190 track_number=int(track_idx),
191 )
192
193 async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
194 """Return a deterministic set of similar tracks (the catalogue neighbours of the seed)."""
195 num_artists = self.config.get_value(CONF_KEY_NUM_ARTISTS) or 5
196 num_albums = self.config.get_value(CONF_KEY_NUM_ALBUMS) or 5
197 num_tracks = self.config.get_value(CONF_KEY_NUM_TRACKS) or 20
198 assert isinstance(num_artists, int)
199 assert isinstance(num_albums, int)
200 assert isinstance(num_tracks, int)
201 total = num_artists * num_albums * num_tracks
202 artist_idx, album_idx, track_idx = (int(part) for part in prov_track_id.split("_", 3))
203 seed_ordinal = (artist_idx * num_albums + album_idx) * num_tracks + track_idx
204 # walk the catalogue starting just after the seed so the result is stable and seed-excluded
205 similar: list[Track] = []
206 for step in range(1, total):
207 if len(similar) >= limit:
208 break
209 ordinal = (seed_ordinal + step) % total
210 artist, remainder = divmod(ordinal, num_albums * num_tracks)
211 album, track = divmod(remainder, num_tracks)
212 similar.append(await self.get_track(f"{artist}_{album}_{track}"))
213 return similar
214
215 async def get_artist(self, prov_artist_id: str) -> Artist:
216 """Get full artist details by id."""
217 genre = random.Random(prov_artist_id).choice(DEFAULT_GENRES)
218 return Artist(
219 item_id=prov_artist_id,
220 provider=self.instance_id,
221 name=f"{genre} Test Artist {prov_artist_id}",
222 metadata=MediaItemMetadata(
223 images=UniqueList([DEFAULT_THUMB, DEFAULT_FANART]),
224 genres={genre},
225 ),
226 provider_mappings={
227 ProviderMapping(
228 item_id=prov_artist_id,
229 provider_domain=self.domain,
230 provider_instance=self.instance_id,
231 )
232 },
233 )
234
235 async def get_album(self, prov_album_id: str) -> Album:
236 """Get full artist details by id."""
237 artist_idx, album_idx = prov_album_id.split("_", 2)
238 genre = random.Random(artist_idx).choice(DEFAULT_GENRES)
239 return Album(
240 item_id=prov_album_id,
241 provider=self.instance_id,
242 name=f"{genre} Test Album {album_idx}",
243 artists=UniqueList([await self.get_artist(artist_idx)]),
244 provider_mappings={
245 ProviderMapping(
246 item_id=prov_album_id,
247 provider_domain=self.domain,
248 provider_instance=self.instance_id,
249 )
250 },
251 metadata=MediaItemMetadata(images=UniqueList([DEFAULT_THUMB]), genres={genre}),
252 )
253
254 async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
255 """Get all tracks for the given album id."""
256 num_tracks = self.config.get_value(CONF_KEY_NUM_TRACKS) or 20
257 assert isinstance(num_tracks, int)
258 artist_idx, album_idx = prov_album_id.split("_", 2)
259 return [
260 await self.get_track(f"{artist_idx}_{album_idx}_{track_idx}")
261 for track_idx in range(num_tracks)
262 ]
263
264 async def get_podcast(self, prov_podcast_id: str) -> Podcast:
265 """Get full podcast details by id."""
266 genre = random.Random(prov_podcast_id).choice(DEFAULT_GENRES)
267 return Podcast(
268 item_id=prov_podcast_id,
269 provider=self.instance_id,
270 name=f"{genre} Test Podcast {prov_podcast_id}",
271 metadata=MediaItemMetadata(images=UniqueList([DEFAULT_THUMB]), genres={genre}),
272 provider_mappings={
273 ProviderMapping(
274 item_id=prov_podcast_id,
275 provider_domain=self.domain,
276 provider_instance=self.instance_id,
277 )
278 },
279 publisher="Test Publisher",
280 )
281
282 async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
283 """Get full audiobook details by id."""
284 genre = random.Random(prov_audiobook_id).choice(DEFAULT_GENRES)
285 return Audiobook(
286 item_id=prov_audiobook_id,
287 provider=self.instance_id,
288 name=f"{genre} Test Audiobook {prov_audiobook_id}",
289 metadata=MediaItemMetadata(
290 images=UniqueList([DEFAULT_THUMB]),
291 description="This is a description for Test Audiobook",
292 chapters=[
293 MediaItemChapter(position=1, name="Chapter 1", start=10, end=20),
294 MediaItemChapter(position=2, name="Chapter 2", start=20, end=40),
295 MediaItemChapter(position=2, name="Chapter 3", start=40),
296 ],
297 genres={genre},
298 ),
299 provider_mappings={
300 ProviderMapping(
301 item_id=prov_audiobook_id,
302 provider_domain=self.domain,
303 provider_instance=self.instance_id,
304 )
305 },
306 publisher="Test Publisher",
307 authors=UniqueList(["AudioBook Author"]),
308 narrators=UniqueList(["AudioBook Narrator"]),
309 duration=60,
310 )
311
312 async def get_library_artists(self) -> AsyncGenerator[Artist]:
313 """Retrieve library artists from the provider."""
314 num_artists = self.config.get_value(CONF_KEY_NUM_ARTISTS)
315 assert isinstance(num_artists, int)
316 for artist_idx in range(num_artists):
317 yield await self.get_artist(str(artist_idx))
318
319 async def get_library_albums(self) -> AsyncGenerator[Album]:
320 """Retrieve library albums from the provider."""
321 num_artists = self.config.get_value(CONF_KEY_NUM_ARTISTS) or 5
322 assert isinstance(num_artists, int)
323 num_albums = self.config.get_value(CONF_KEY_NUM_ALBUMS)
324 assert isinstance(num_albums, int)
325 for artist_idx in range(num_artists):
326 for album_idx in range(num_albums):
327 album_item_id = f"{artist_idx}_{album_idx}"
328 yield await self.get_album(album_item_id)
329
330 async def get_library_tracks(self) -> AsyncGenerator[Track]:
331 """Retrieve library tracks from the provider."""
332 num_artists = self.config.get_value(CONF_KEY_NUM_ARTISTS) or 5
333 assert isinstance(num_artists, int)
334 num_albums = self.config.get_value(CONF_KEY_NUM_ALBUMS) or 5
335 assert isinstance(num_albums, int)
336 num_tracks = self.config.get_value(CONF_KEY_NUM_TRACKS)
337 assert isinstance(num_tracks, int)
338 for artist_idx in range(num_artists):
339 for album_idx in range(num_albums):
340 for track_idx in range(num_tracks):
341 track_item_id = f"{artist_idx}_{album_idx}_{track_idx}"
342 yield await self.get_track(track_item_id)
343
344 async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
345 """Retrieve library tracks from the provider."""
346 num_podcasts = self.config.get_value(CONF_KEY_NUM_PODCASTS)
347 assert isinstance(num_podcasts, int)
348 for podcast_idx in range(num_podcasts):
349 yield await self.get_podcast(str(podcast_idx))
350
351 async def get_library_audiobooks(self) -> AsyncGenerator[Audiobook]:
352 """Retrieve library audiobooks from the provider."""
353 num_audiobooks = self.config.get_value(CONF_KEY_NUM_AUDIOBOOKS)
354 assert isinstance(num_audiobooks, int)
355 for audiobook_idx in range(num_audiobooks):
356 yield await self.get_audiobook(str(audiobook_idx))
357
358 async def get_podcast_episodes(
359 self,
360 prov_podcast_id: str,
361 ) -> AsyncGenerator[PodcastEpisode]:
362 """Get all PodcastEpisodes for given podcast id."""
363 num_episodes = 25
364 for episode_idx in range(num_episodes):
365 yield await self.get_podcast_episode(f"{prov_podcast_id}_{episode_idx}")
366
367 async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
368 """Get (full) podcast episode details by id."""
369 podcast_id, episode_idx = prov_episode_id.split("_", 2)
370 genre = random.Random(podcast_id).choice(DEFAULT_GENRES)
371 return PodcastEpisode(
372 item_id=prov_episode_id,
373 provider=self.instance_id,
374 name=f"{genre} Test PodcastEpisode {podcast_id}-{episode_idx}",
375 duration=60,
376 podcast=ItemMapping(
377 item_id=podcast_id,
378 provider=self.instance_id,
379 name=f"Test Podcast {podcast_id}",
380 media_type=MediaType.PODCAST,
381 image=DEFAULT_THUMB,
382 ),
383 provider_mappings={
384 ProviderMapping(
385 item_id=prov_episode_id,
386 provider_domain=self.domain,
387 provider_instance=self.instance_id,
388 )
389 },
390 metadata=MediaItemMetadata(
391 description="This is a description for "
392 f"Test PodcastEpisode {episode_idx} of Test Podcast {podcast_id}",
393 genres={genre},
394 ),
395 position=int(episode_idx),
396 )
397
398 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
399 """Get streamdetails for a track/radio."""
400 return StreamDetails(
401 provider=self.instance_id,
402 item_id=item_id,
403 audio_format=AudioFormat(
404 content_type=ContentType.OGG,
405 sample_rate=48000,
406 bit_depth=16,
407 channels=2,
408 ),
409 media_type=media_type,
410 stream_type=StreamType.HTTP,
411 path=SILENCE_FILE_LONG,
412 can_seek=True,
413 allow_seek=True,
414 )
415