/
/
/
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._report_skipped_item(MediaType.ALBUM, album, "album_id", 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._report_skipped_item(MediaType.ARTIST, artist, "artist_id", 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 as error:
157 self._report_skipped_item(MediaType.TRACK, track, "track_id", error)
158 continue
159 except (KeyError, TypeError, InvalidDataError) as error:
160 self._report_skipped_item(MediaType.TRACK, track, "track_id", error)
161 continue
162
163 def _get_artist_item_mapping(self, artist_id: str, artist_obj: dict[str, Any]) -> ItemMapping:
164 if (not artist_id and artist_obj["name"] == "Various Artists") or artist_id == "0":
165 artist_id = VARIOUS_ARTISTS_MBID
166 return self._get_item_mapping(MediaType.ARTIST, artist_id, str(artist_obj.get("name")))
167
168 def _get_item_mapping(self, media_type: MediaType, key: str, name: str) -> ItemMapping:
169 return ItemMapping(
170 media_type=media_type,
171 item_id=key,
172 provider=self.instance_id,
173 name=name,
174 )
175
176 async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
177 """Retrieve playlists from iBroadcast."""
178 for playlist in (await self._client.get_playlists()).values():
179 # Skip the auto generated playlist
180 if playlist["type"] != "recently-played" and playlist["type"] != "thumbsup":
181 yield await self._parse_playlist(playlist)
182
183 @use_cache(3600 * 24 * 7) # Cache for 7 days
184 async def get_playlist(self, prov_playlist_id: str) -> Playlist:
185 """Get full playlist details by id."""
186 playlist_obj = await self._client.get_playlist(int(prov_playlist_id))
187 try:
188 playlist = await self._parse_playlist(playlist_obj)
189 except (KeyError, TypeError, InvalidDataError, IndexError) as error:
190 self.logger.debug("Parse playlist failed: %s", playlist_obj, exc_info=error)
191 raise MediaNotFoundError(f"Playlist {prov_playlist_id} could not be parsed") from error
192 return playlist
193
194 @use_cache(3600, allow_expired_cache=True) # Cache for 1 hour
195 async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
196 """Get playlist tracks."""
197 tracks: list[Track] = []
198 if page > 0:
199 return tracks
200 playlist_obj = await self._client.get_playlist(int(prov_playlist_id))
201 if "tracks" not in playlist_obj:
202 return tracks
203 return await self._get_tracks(playlist_obj["tracks"], True)
204
205 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
206 """Return the content details for the given track when it will be streamed."""
207 # How to buildup a stream url:
208 # [streaming_server]/[url]?Expires=[now]&Signature=[user token]&file_id=[file ID]
209 # &user_id=[user ID]&platform=[your app name]&version=[your app version]
210 # See https://devguide.ibroadcast.com/?p=streaming-server
211 url = await self._client.get_full_stream_url(int(item_id), "music-assistant")
212
213 # Replace the bitrate path segment with "orig" to request the original upload format.
214 # By default the bitrate is always 128kbps.
215 parsed = urlparse(url)
216 path_parts = parsed.path.split("/")
217 if len(path_parts) > 1 and path_parts[1].isdigit():
218 path_parts[1] = "orig"
219 url = urlunparse(parsed._replace(path="/".join(path_parts)))
220
221 return StreamDetails(
222 provider=self.instance_id,
223 item_id=item_id,
224 audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
225 stream_type=StreamType.HTTP,
226 path=url,
227 can_seek=True,
228 allow_seek=True,
229 )
230
231 async def _get_tracks(self, track_ids: list[int], is_playlist: bool = False) -> list[Track]:
232 """Retrieve a list of tracks based on provided track IDs."""
233 tracks = []
234 for index, track_id in enumerate(track_ids, 1):
235 track_obj = await self._client.get_track(track_id)
236 if track_obj is not None:
237 track = await self._parse_track(track_obj)
238 if is_playlist:
239 track.position = index
240 tracks.append(track)
241 return tracks
242
243 async def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
244 """Parse a iBroadcast user response to Artist model object."""
245 artist_id = artist_obj["artist_id"]
246 artist = Artist(
247 item_id=artist_id,
248 name=artist_obj["name"],
249 provider=self.instance_id,
250 provider_mappings={
251 ProviderMapping(
252 item_id=artist_id,
253 provider_domain=self.domain,
254 provider_instance=self.instance_id,
255 url=f"https://media.ibroadcast.com/?view=container&container_id={artist_id}&type=artists",
256 )
257 },
258 )
259 # Artwork
260 if "artwork_id" in artist_obj:
261 artist.metadata.images = UniqueList(
262 [
263 MediaItemImage(
264 type=ImageType.THUMB,
265 path=await self._client.get_artist_artwork_url(artist_id),
266 provider=self.instance_id,
267 remotely_accessible=True,
268 )
269 ]
270 )
271 return artist
272
273 async def _parse_album(self, album_obj: dict[str, Any]) -> Album:
274 """Parse ibroadcast album object to generic layout."""
275 album_id = album_obj["album_id"]
276 name, version = parse_title_and_version(album_obj["name"])
277 album = Album(
278 item_id=album_id,
279 provider=self.instance_id,
280 name=name,
281 year=album_obj["year"],
282 version=version,
283 provider_mappings={
284 ProviderMapping(
285 item_id=album_id,
286 provider_domain=self.domain,
287 provider_instance=self.instance_id,
288 audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
289 url=f"https://media.ibroadcast.com/?view=container&container_id={album_id}&type=albums",
290 )
291 },
292 )
293 if album_obj["artist_id"] == 0:
294 artist = Artist(
295 item_id=VARIOUS_ARTISTS_MBID,
296 name=VARIOUS_ARTISTS_NAME,
297 provider=self.instance_id,
298 provider_mappings={
299 ProviderMapping(
300 item_id=VARIOUS_ARTISTS_MBID,
301 provider_domain=self.domain,
302 provider_instance=self.instance_id,
303 )
304 },
305 )
306 album.artists.append(artist)
307 else:
308 artist_mapping = self._get_item_mapping(
309 MediaType.ARTIST,
310 album_obj["artist_id"],
311 (await self._client.get_artist(album_obj["artist_id"]))["name"]
312 if await self._client.get_artist(album_obj["artist_id"])
313 else UNKNOWN_ARTIST,
314 )
315 album.artists.append(artist_mapping)
316
317 if "rating" in album_obj and album_obj["rating"] == 5:
318 album.favorite = True
319 # iBroadcast doesn't seem to know album type - try inference
320 album.album_type = infer_album_type(name, version)
321
322 # There is only an artwork in the tracks, lets get the first track one
323 artwork_url = await self._client.get_album_artwork_url(album_id)
324 if artwork_url:
325 album.metadata.images = UniqueList([self._get_artwork_object(artwork_url)])
326 return album
327
328 def _get_artwork_object(self, url: str) -> MediaItemImage:
329 return MediaItemImage(
330 type=ImageType.THUMB,
331 path=url,
332 provider=self.instance_id,
333 remotely_accessible=True,
334 )
335
336 async def _parse_track(self, track_obj: dict[str, Any]) -> Track:
337 """Parse an iBroadcast track object to a Track model object."""
338 track = Track(
339 item_id=track_obj["track_id"],
340 provider=self.instance_id,
341 name=track_obj["title"],
342 provider_mappings={
343 ProviderMapping(
344 item_id=track_obj["track_id"],
345 provider_domain=self.domain,
346 provider_instance=self.instance_id,
347 available=not track_obj["trashed"],
348 audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
349 )
350 },
351 )
352 if track_obj["album_id"]:
353 album = await self._client.get_album(track_obj["album_id"])
354
355 if "rating" in track_obj and track_obj["rating"] == 5:
356 track.favorite = True
357 if "length" in track_obj and str(track_obj["length"]).isdigit():
358 track.duration = track_obj["length"]
359 # use the disc number if available
360 if album and album["disc"] > 0:
361 track.disc_number = album["disc"]
362 track.track_number = int(track_obj["track"])
363 # otherwise, track number might look like 201, meaning, disc 2, track 1
364 elif track_obj["track"] > 99:
365 track.disc_number = int(str(track_obj["track"])[:1])
366 track.track_number = int(str(track_obj["track"])[1:])
367 # or just the track number and no disc number
368 else:
369 track.track_number = int(track_obj["track"])
370 # Track artists
371 if "artist_id" in track_obj:
372 artist_id = track_obj["artist_id"]
373 track.artists = UniqueList(
374 [self._get_artist_item_mapping(artist_id, await self._client.get_artist(artist_id))]
375 )
376 # additional artists structure: 'artists_additional': [[artist id, phrase, type]]
377 track.artists.extend(
378 [
379 self._get_artist_item_mapping(
380 additional_artist[0],
381 await self._client.get_artist(additional_artist[0]),
382 )
383 for additional_artist in track_obj["artists_additional"]
384 if additional_artist[0]
385 ]
386 )
387 # guard that track has valid artists
388 if not track.artists:
389 msg = "Track is missing artists"
390 raise InvalidDataError(msg)
391
392 # Artwork
393 track.metadata.images = UniqueList(
394 [
395 self._get_artwork_object(
396 await self._client.get_track_artwork_url(track_obj["track_id"])
397 )
398 ]
399 )
400 # Genre
401 genres: set[str] = set()
402 if track_obj["genre"]:
403 genres.add(track_obj["genre"])
404 if track_obj["genres_additional"]:
405 genres.add(track_obj["genres_additional"])
406 track.metadata.genres = genres
407 # album info
408 if album:
409 track.album = self._get_item_mapping(
410 MediaType.ALBUM, track_obj["album_id"], album["name"]
411 )
412 return track
413
414 async def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
415 """Parse an iBroadcast Playlist response to a Playlist object."""
416 playlist_id = str(playlist_obj["playlist_id"])
417 playlist = Playlist(
418 item_id=playlist_id,
419 provider=self.instance_id,
420 name=playlist_obj["name"],
421 provider_mappings={
422 ProviderMapping(
423 item_id=playlist_id,
424 provider_domain=self.domain,
425 provider_instance=self.instance_id,
426 )
427 },
428 )
429 # Can be supported in future, the API has options available
430 playlist.is_editable = False
431 playlist.metadata.images = UniqueList(
432 [
433 self._get_artwork_object(
434 await self._client.get_playlist_artwork_url(int(playlist_id))
435 )
436 ]
437 )
438 if "description" in playlist_obj:
439 playlist.metadata.description = playlist_obj["description"]
440 return playlist
441
442 def _report_skipped_item(
443 self, media_type: MediaType, item_obj: dict[str, Any], id_key: str, err: Exception
444 ) -> None:
445 """
446 Report a library item that was dropped while listing the library.
447
448 :param media_type: Media type of the skipped item.
449 :param item_obj: Raw api object of the skipped item.
450 :param id_key: Key under which the raw object holds the item id, which iBroadcast
451 returns as a number while the library stores it as text.
452 :param err: The error that made the item unusable.
453 """
454 item_id = item_obj.get(id_key)
455 self.report_skipped_sync_item(
456 media_type, str(item_id) if item_id is not None else None, err
457 )
458