music-assistant-server

1"""iBroadcast support for MusicAssistant."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any
6from urllib.parse import urlparse, urlunparse
7
8from ibroadcastaio import IBroadcastClient
9from music_assistant_models.enums import (
10    ContentType,
11    ImageType,
12    MediaType,
13    ProviderFeature,
14    StreamType,
15)
16from music_assistant_models.errors import InvalidDataError, LoginFailed, MediaNotFoundError
17from music_assistant_models.media_items import (
18    Album,
19    Artist,
20    AudioFormat,
21    ItemMapping,
22    MediaItemImage,
23    Playlist,
24    ProviderMapping,
25    Track,
26    UniqueList,
27)
28from music_assistant_models.streamdetails import StreamDetails
29
30from music_assistant.constants import (
31    CONF_PASSWORD,
32    CONF_USERNAME,
33    UNKNOWN_ARTIST,
34    VARIOUS_ARTISTS_MBID,
35    VARIOUS_ARTISTS_NAME,
36)
37from music_assistant.controllers.cache import use_cache
38from music_assistant.helpers.util import infer_album_type, parse_title_and_version
39from music_assistant.models.music_provider import MusicProvider
40
41SUPPORTED_FEATURES = {
42    ProviderFeature.LIBRARY_ARTISTS,
43    ProviderFeature.LIBRARY_TRACKS,
44    ProviderFeature.LIBRARY_ALBUMS,
45    ProviderFeature.LIBRARY_PLAYLISTS,
46    ProviderFeature.BROWSE,
47    ProviderFeature.ARTIST_ALBUMS,
48}
49
50
51if TYPE_CHECKING:
52    from collections.abc import AsyncGenerator
53
54    from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
55    from music_assistant_models.provider import ProviderManifest
56
57    from music_assistant.mass import MusicAssistant
58    from music_assistant.models import ProviderInstanceType
59
60
61async def setup(
62    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
63) -> ProviderInstanceType:
64    """Initialize provider(instance) with given configuration."""
65    return IBroadcastProvider(mass, manifest, config, SUPPORTED_FEATURES)
66
67
68class IBroadcastProvider(MusicProvider):
69    """Provider for iBroadcast."""
70
71    _user_id: str
72    _client: IBroadcastClient
73
74    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
75        """Return Config entries to setup this provider."""
76        return ()
77
78    async def handle_async_init(self) -> None:
79        """Set up the iBroadcast provider."""
80        username = self.get_setup_value(CONF_USERNAME)
81        password = self.get_setup_value(CONF_PASSWORD)
82        if not username or not password:
83            msg = "Invalid login credentials"
84            raise LoginFailed(msg)
85        self._client = IBroadcastClient(self.mass.http_session)
86        status = await self._client.login(username, password)
87        self._user_id = status["user"]["id"]
88
89        # temporary call to refresh library until ibroadcast provides a detailed api
90        await self._client.refresh_library()
91
92    async def get_library_albums(self) -> AsyncGenerator[Album]:
93        """Retrieve library albums from ibroadcast."""
94        for album in (await self._client.get_albums()).values():
95            try:
96                yield await self._parse_album(album)
97            except (KeyError, TypeError, InvalidDataError, IndexError) as error:
98                self.logger.debug("Parse album failed: %s", album, exc_info=error)
99                continue
100
101    @use_cache(3600 * 24 * 7)  # Cache for 7 days
102    async def get_album(self, prov_album_id: str) -> Album:
103        """Get full album details by id."""
104        album_obj = await self._client.get_album(int(prov_album_id))
105        return await self._parse_album(album_obj)
106
107    async def get_library_artists(self) -> AsyncGenerator[Artist]:
108        """Retrieve all library artists from iBroadcast."""
109        for artist in (await self._client.get_artists()).values():
110            try:
111                yield await self._parse_artist(artist)
112            except (KeyError, TypeError, InvalidDataError, IndexError) as error:
113                self.logger.debug("Parse artist failed: %s", artist, exc_info=error)
114                continue
115
116    @use_cache(3600 * 24 * 7, allow_expired_cache=True)  # Cache for 7 days
117    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
118        """Get a list of albums for the given artist."""
119        albums_objs = [
120            album
121            for album in (await self._client.get_albums()).values()
122            if album["artist_id"] == int(prov_artist_id)
123        ]
124        albums = []
125        for album in albums_objs:
126            try:
127                albums.append(await self._parse_album(album))
128            except (KeyError, TypeError, InvalidDataError, IndexError) as error:
129                self.logger.debug("Parse album failed: %s", album, exc_info=error)
130                continue
131        return albums
132
133    @use_cache(3600 * 24 * 7, allow_expired_cache=True)  # Cache for 7 days
134    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
135        """Get album tracks for given album id."""
136        album = await self._client.get_album(int(prov_album_id))
137        return await self._get_tracks(album["tracks"])
138
139    @use_cache(3600 * 24 * 7)  # Cache for 7 days
140    async def get_track(self, prov_track_id: str) -> Track:
141        """Get full track details by id."""
142        track_obj = await self._client.get_track(int(prov_track_id))
143        return await self._parse_track(track_obj)
144
145    @use_cache(3600 * 24 * 7)  # Cache for 7 days
146    async def get_artist(self, prov_artist_id: str) -> Artist:
147        """Get full artist details by id."""
148        artist_obj = await self._client.get_artist(int(prov_artist_id))
149        return await self._parse_artist(artist_obj)
150
151    async def get_library_tracks(self) -> AsyncGenerator[Track]:
152        """Retrieve library tracks from iBroadcast."""
153        for track in (await self._client.get_tracks()).values():
154            try:
155                yield await self._parse_track(track)
156            except IndexError:
157                continue
158            except (KeyError, TypeError, InvalidDataError) as error:
159                self.logger.debug("Parse track failed: %s", track, exc_info=error)
160                continue
161
162    def _get_artist_item_mapping(self, artist_id: str, artist_obj: dict[str, Any]) -> ItemMapping:
163        if (not artist_id and artist_obj["name"] == "Various Artists") or artist_id == "0":
164            artist_id = VARIOUS_ARTISTS_MBID
165        return self._get_item_mapping(MediaType.ARTIST, artist_id, str(artist_obj.get("name")))
166
167    def _get_item_mapping(self, media_type: MediaType, key: str, name: str) -> ItemMapping:
168        return ItemMapping(
169            media_type=media_type,
170            item_id=key,
171            provider=self.instance_id,
172            name=name,
173        )
174
175    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
176        """Retrieve playlists from iBroadcast."""
177        for playlist in (await self._client.get_playlists()).values():
178            # Skip the auto generated playlist
179            if playlist["type"] != "recently-played" and playlist["type"] != "thumbsup":
180                yield await self._parse_playlist(playlist)
181
182    @use_cache(3600 * 24 * 7)  # Cache for 7 days
183    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
184        """Get full playlist details by id."""
185        playlist_obj = await self._client.get_playlist(int(prov_playlist_id))
186        try:
187            playlist = await self._parse_playlist(playlist_obj)
188        except (KeyError, TypeError, InvalidDataError, IndexError) as error:
189            self.logger.debug("Parse playlist failed: %s", playlist_obj, exc_info=error)
190            raise MediaNotFoundError(f"Playlist {prov_playlist_id} could not be parsed") from error
191        return playlist
192
193    @use_cache(3600, allow_expired_cache=True)  # Cache for 1 hour
194    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
195        """Get playlist tracks."""
196        tracks: list[Track] = []
197        if page > 0:
198            return tracks
199        playlist_obj = await self._client.get_playlist(int(prov_playlist_id))
200        if "tracks" not in playlist_obj:
201            return tracks
202        return await self._get_tracks(playlist_obj["tracks"], True)
203
204    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
205        """Return the content details for the given track when it will be streamed."""
206        # How to buildup a stream url:
207        # [streaming_server]/[url]?Expires=[now]&Signature=[user token]&file_id=[file ID]
208        # &user_id=[user ID]&platform=[your app name]&version=[your app version]
209        # See https://devguide.ibroadcast.com/?p=streaming-server
210        url = await self._client.get_full_stream_url(int(item_id), "music-assistant")
211
212        # Replace the bitrate path segment with "orig" to request the original upload format.
213        # By default the bitrate is always 128kbps.
214        parsed = urlparse(url)
215        path_parts = parsed.path.split("/")
216        if len(path_parts) > 1 and path_parts[1].isdigit():
217            path_parts[1] = "orig"
218            url = urlunparse(parsed._replace(path="/".join(path_parts)))
219
220        return StreamDetails(
221            provider=self.instance_id,
222            item_id=item_id,
223            audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
224            stream_type=StreamType.HTTP,
225            path=url,
226            can_seek=True,
227            allow_seek=True,
228        )
229
230    async def _get_tracks(self, track_ids: list[int], is_playlist: bool = False) -> list[Track]:
231        """Retrieve a list of tracks based on provided track IDs."""
232        tracks = []
233        for index, track_id in enumerate(track_ids, 1):
234            track_obj = await self._client.get_track(track_id)
235            if track_obj is not None:
236                track = await self._parse_track(track_obj)
237                if is_playlist:
238                    track.position = index
239                tracks.append(track)
240        return tracks
241
242    async def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
243        """Parse a iBroadcast user response to Artist model object."""
244        artist_id = artist_obj["artist_id"]
245        artist = Artist(
246            item_id=artist_id,
247            name=artist_obj["name"],
248            provider=self.instance_id,
249            provider_mappings={
250                ProviderMapping(
251                    item_id=artist_id,
252                    provider_domain=self.domain,
253                    provider_instance=self.instance_id,
254                    url=f"https://media.ibroadcast.com/?view=container&container_id={artist_id}&type=artists",
255                )
256            },
257        )
258        # Artwork
259        if "artwork_id" in artist_obj:
260            artist.metadata.images = UniqueList(
261                [
262                    MediaItemImage(
263                        type=ImageType.THUMB,
264                        path=await self._client.get_artist_artwork_url(artist_id),
265                        provider=self.instance_id,
266                        remotely_accessible=True,
267                    )
268                ]
269            )
270        return artist
271
272    async def _parse_album(self, album_obj: dict[str, Any]) -> Album:
273        """Parse ibroadcast album object to generic layout."""
274        album_id = album_obj["album_id"]
275        name, version = parse_title_and_version(album_obj["name"])
276        album = Album(
277            item_id=album_id,
278            provider=self.instance_id,
279            name=name,
280            year=album_obj["year"],
281            version=version,
282            provider_mappings={
283                ProviderMapping(
284                    item_id=album_id,
285                    provider_domain=self.domain,
286                    provider_instance=self.instance_id,
287                    audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
288                    url=f"https://media.ibroadcast.com/?view=container&container_id={album_id}&type=albums",
289                )
290            },
291        )
292        if album_obj["artist_id"] == 0:
293            artist = Artist(
294                item_id=VARIOUS_ARTISTS_MBID,
295                name=VARIOUS_ARTISTS_NAME,
296                provider=self.instance_id,
297                provider_mappings={
298                    ProviderMapping(
299                        item_id=VARIOUS_ARTISTS_MBID,
300                        provider_domain=self.domain,
301                        provider_instance=self.instance_id,
302                    )
303                },
304            )
305            album.artists.append(artist)
306        else:
307            artist_mapping = self._get_item_mapping(
308                MediaType.ARTIST,
309                album_obj["artist_id"],
310                (await self._client.get_artist(album_obj["artist_id"]))["name"]
311                if await self._client.get_artist(album_obj["artist_id"])
312                else UNKNOWN_ARTIST,
313            )
314            album.artists.append(artist_mapping)
315
316        if "rating" in album_obj and album_obj["rating"] == 5:
317            album.favorite = True
318        # iBroadcast doesn't seem to know album type - try inference
319        album.album_type = infer_album_type(name, version)
320
321        # There is only an artwork in the tracks, lets get the first track one
322        artwork_url = await self._client.get_album_artwork_url(album_id)
323        if artwork_url:
324            album.metadata.images = UniqueList([self._get_artwork_object(artwork_url)])
325        return album
326
327    def _get_artwork_object(self, url: str) -> MediaItemImage:
328        return MediaItemImage(
329            type=ImageType.THUMB,
330            path=url,
331            provider=self.instance_id,
332            remotely_accessible=True,
333        )
334
335    async def _parse_track(self, track_obj: dict[str, Any]) -> Track:
336        """Parse an iBroadcast track object to a Track model object."""
337        track = Track(
338            item_id=track_obj["track_id"],
339            provider=self.instance_id,
340            name=track_obj["title"],
341            provider_mappings={
342                ProviderMapping(
343                    item_id=track_obj["track_id"],
344                    provider_domain=self.domain,
345                    provider_instance=self.instance_id,
346                    available=not track_obj["trashed"],
347                    audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
348                )
349            },
350        )
351        if track_obj["album_id"]:
352            album = await self._client.get_album(track_obj["album_id"])
353
354        if "rating" in track_obj and track_obj["rating"] == 5:
355            track.favorite = True
356        if "length" in track_obj and str(track_obj["length"]).isdigit():
357            track.duration = track_obj["length"]
358        # use the disc number if available
359        if album and album["disc"] > 0:
360            track.disc_number = album["disc"]
361            track.track_number = int(track_obj["track"])
362        # otherwise, track number might look like 201, meaning, disc 2, track 1
363        elif track_obj["track"] > 99:
364            track.disc_number = int(str(track_obj["track"])[:1])
365            track.track_number = int(str(track_obj["track"])[1:])
366        # or just the track number and no disc number
367        else:
368            track.track_number = int(track_obj["track"])
369        # Track artists
370        if "artist_id" in track_obj:
371            artist_id = track_obj["artist_id"]
372            track.artists = UniqueList(
373                [self._get_artist_item_mapping(artist_id, await self._client.get_artist(artist_id))]
374            )
375            # additional artists structure: 'artists_additional': [[artist id, phrase, type]]
376            track.artists.extend(
377                [
378                    self._get_artist_item_mapping(
379                        additional_artist[0],
380                        await self._client.get_artist(additional_artist[0]),
381                    )
382                    for additional_artist in track_obj["artists_additional"]
383                    if additional_artist[0]
384                ]
385            )
386            # guard that track has valid artists
387            if not track.artists:
388                msg = "Track is missing artists"
389                raise InvalidDataError(msg)
390
391        # Artwork
392        track.metadata.images = UniqueList(
393            [
394                self._get_artwork_object(
395                    await self._client.get_track_artwork_url(track_obj["track_id"])
396                )
397            ]
398        )
399        # Genre
400        genres: set[str] = set()
401        if track_obj["genre"]:
402            genres.add(track_obj["genre"])
403        if track_obj["genres_additional"]:
404            genres.add(track_obj["genres_additional"])
405        track.metadata.genres = genres
406        # album info
407        if album:
408            track.album = self._get_item_mapping(
409                MediaType.ALBUM, track_obj["album_id"], album["name"]
410            )
411        return track
412
413    async def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
414        """Parse an iBroadcast Playlist response to a Playlist object."""
415        playlist_id = str(playlist_obj["playlist_id"])
416        playlist = Playlist(
417            item_id=playlist_id,
418            provider=self.instance_id,
419            name=playlist_obj["name"],
420            provider_mappings={
421                ProviderMapping(
422                    item_id=playlist_id,
423                    provider_domain=self.domain,
424                    provider_instance=self.instance_id,
425                )
426            },
427        )
428        # Can be supported in future, the API has options available
429        playlist.is_editable = False
430        playlist.metadata.images = UniqueList(
431            [
432                self._get_artwork_object(
433                    await self._client.get_playlist_artwork_url(int(playlist_id))
434                )
435            ]
436        )
437        if "description" in playlist_obj:
438            playlist.metadata.description = playlist_obj["description"]
439        return playlist
440