music-assistant-server

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