music-assistant-server

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