/
/
/
1"""
2Adaptor for converting BBC Sounds objects to Music Assistant media items.
3
4Many Sounds API endpoints return containers of "PlayableObjects" which can be a
5range of different types. The auntie-sounds library detects these differing
6types and provides a sensible set of objects to work with, e.g. RadioShow.
7
8This adaptor maps those objects to the most sensible type for MA.
9"""
10
11from abc import ABC, abstractmethod
12from dataclasses import dataclass
13from datetime import datetime, tzinfo
14from typing import TYPE_CHECKING, Any, ClassVar, cast
15
16from music_assistant_models.enums import ContentType, ImageType, MediaType, StreamType
17from music_assistant_models.errors import MusicAssistantError
18from music_assistant_models.media_items import (
19 AudioFormat,
20 BrowseFolder,
21 MediaItemChapter,
22 MediaItemImage,
23 MediaItemMetadata,
24 ProviderMapping,
25 Radio,
26 RecommendationFolder,
27 Track,
28)
29from music_assistant_models.media_items import Podcast as MAPodcast
30from music_assistant_models.media_items import PodcastEpisode as MAPodcastEpisode
31from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
32from music_assistant_models.unique_list import UniqueList
33from sounds.models import (
34 Category,
35 Collection,
36 LiveStation,
37 MenuItem,
38 Playlist,
39 Podcast,
40 PodcastEpisode,
41 RadioClip,
42 RadioSeries,
43 RadioShow,
44 RecommendedMenuItem,
45 Schedule,
46 SoundsTypes,
47 Station,
48 StationSearchResult,
49)
50
51import music_assistant.helpers.datetime as dt
52from music_assistant.helpers.datetime import LOCAL_TIMEZONE
53from music_assistant.providers.bbc_sounds.constants import _Constants
54
55if TYPE_CHECKING:
56 from music_assistant.providers.bbc_sounds import BBCSoundsProvider
57
58
59def _date_convertor(
60 timestamp: str | datetime,
61 date_format: str,
62 timezone: tzinfo | None = LOCAL_TIMEZONE,
63) -> str:
64 if isinstance(timestamp, str):
65 timestamp = dt.from_iso_string(timestamp)
66 else:
67 timestamp = timestamp.astimezone(timezone)
68 return timestamp.strftime(date_format)
69
70
71def _to_time(timestamp: str | datetime) -> str:
72 return _date_convertor(timestamp, "%H:%M")
73
74
75def _to_date_and_time(timestamp: str | datetime) -> str:
76 return _date_convertor(timestamp, "%a %d %B %H:%M")
77
78
79def _to_date(timestamp: str | datetime) -> str:
80 return _date_convertor(timestamp, "%d/%m/%y")
81
82
83class ConversionError(MusicAssistantError):
84 """Raised when object conversion fails."""
85
86
87class ImageProvider:
88 """Handles image URL resolution and MediaItemImage creation."""
89
90 # TODO: keeping this in for demo purposes
91 ICON_BASE_URL = (
92 "https://cdn.jsdelivr.net/gh/kieranhogg/auntie-sounds@main/src/sounds/icons/solid"
93 )
94
95 ICON_MAPPING: ClassVar[dict[str, str]] = {
96 "listen_live": "listen_live",
97 "continue_listening": "continue",
98 "editorial_collection": "editorial",
99 "local_rail": "my_location",
100 "single_item_promo": "featured",
101 "collections": "collections",
102 "categories": "categories",
103 "recommendations": "my_sounds",
104 "unmissable_speech": "speech",
105 "podcasts": "speech",
106 "unmissable_music": "music",
107 "music": "music",
108 "explore": "categories",
109 "stations": "latest",
110 "news": "news",
111 }
112
113 @classmethod
114 def get_icon_url(cls, icon_id: str) -> str | None:
115 """Get icon URL for a given icon ID."""
116 if icon_id is not None:
117 if icon_id in cls.ICON_MAPPING:
118 return f"{cls.ICON_BASE_URL}/{cls.ICON_MAPPING[icon_id]}.png"
119 if "latest_playables_for_curation" in icon_id:
120 return f"{cls.ICON_BASE_URL}/news.png"
121 return None
122
123 @classmethod
124 def create_image(
125 cls, url: str, provider: str, image_type: ImageType = ImageType.THUMB
126 ) -> MediaItemImage:
127 """Create a MediaItemImage from a URL."""
128 return MediaItemImage(
129 path=url,
130 provider=provider,
131 type=image_type,
132 remotely_accessible=True,
133 )
134
135 @classmethod
136 def create_metadata_with_image(
137 cls,
138 url: str | None,
139 provider: str,
140 description: str | None = None,
141 chapters: list[MediaItemChapter] | None = None,
142 ) -> MediaItemMetadata:
143 """Create metadata with optional image and description."""
144 metadata = MediaItemMetadata()
145 if url:
146 metadata.add_image(cls.create_image(url, provider))
147 if description:
148 metadata.description = description
149 if chapters:
150 metadata.chapters = chapters
151 return metadata
152
153
154@dataclass
155class Context:
156 """Context information for object conversion."""
157
158 provider: BBCSoundsProvider
159 provider_domain: str
160 path_parts: list[str] | None = None
161 force_type: (
162 type[
163 Track
164 | LiveStation
165 | Radio
166 | MAPodcast
167 | MAPodcastEpisode
168 | BrowseFolder
169 | RecommendationFolder
170 | RecommendedMenuItem
171 ]
172 | None
173 ) = None
174
175
176class BaseConverter(ABC):
177 """Base model."""
178
179 def __init__(self, context: Context):
180 """Create a new instance."""
181 self.context = context
182 self.logger = self.context.provider.logger
183
184 @abstractmethod
185 def can_convert(self, source_obj: Any) -> bool:
186 """Check if this converter can handle the source object."""
187
188 @abstractmethod
189 async def get_stream_details(self, source_obj: Any) -> StreamDetails | None:
190 """Convert the source object to a stream."""
191
192 @abstractmethod
193 async def convert(
194 self, source_obj: Any
195 ) -> (
196 Track
197 | LiveStation
198 | Radio
199 | MAPodcast
200 | MAPodcastEpisode
201 | BrowseFolder
202 | RecommendationFolder
203 | RecommendedMenuItem
204 ):
205 """Convert the source object to target type."""
206
207 def _create_provider_mapping(self, item_id: str) -> ProviderMapping:
208 """Create provider mapping for the item."""
209 return self.context.provider._get_provider_mapping(item_id)
210
211 def _get_attr(self, obj: Any, attr_path: str, default: Any = None) -> Any:
212 """
213 Get (optionally-nested) attribute from object.
214
215 Supports e.g. _get_attr(object, "thing.other_thing")
216 """
217 # TODO: I'm fairly sure there is existing code/libs for this?
218 try:
219 current = obj
220 for part in attr_path.split("."):
221 if hasattr(current, part):
222 current = getattr(current, part)
223 elif isinstance(current, dict) and part in current:
224 current = current[part]
225 else:
226 return default
227 return current
228 except AttributeError, KeyError, TypeError:
229 return default
230
231 def _get_synopsis(self, obj: Any) -> str | None:
232 """Return the fullest synopsis the given object carries, if any."""
233 for length in ("long", "medium", "short"):
234 if synopsis := self._get_attr(obj, f"synopses.{length}"):
235 return str(synopsis)
236 return None
237
238
239class StationConverter(BaseConverter):
240 """Converts Station-related objects."""
241
242 ConvertableTypes = Station | LiveStation | StationSearchResult
243
244 def can_convert(self, source_obj: Any) -> bool:
245 """Check if this converter can convert to a Station object."""
246 return isinstance(source_obj, self.ConvertableTypes)
247
248 async def get_stream_details(self, source_obj: Any) -> StreamDetails | None:
249 """Convert the source object to a stream."""
250 if not isinstance(source_obj, self.ConvertableTypes):
251 return None
252 # TODO: can't seek this stream
253 station = await self.convert(source_obj)
254 if not station or not source_obj.stream:
255 return None
256 show_time = self._get_attr(source_obj, "titles.secondary")
257 show_title = self._get_attr(source_obj, "titles.primary")
258 programme_name = f"{show_time} ⢠{show_title}"
259 stream_details = None
260 if station and source_obj.stream:
261 stream_metadata = StreamMetadata(
262 title=programme_name,
263 )
264
265 if station.image is not None:
266 stream_metadata.image_url = station.image.path
267
268 stream_details = StreamDetails(
269 stream_metadata=stream_metadata,
270 media_type=MediaType.RADIO,
271 stream_type=StreamType.HLS
272 if self.context.provider.stream_format == _Constants.HLS
273 else StreamType.HTTP,
274 path=str(source_obj.stream),
275 item_id=station.item_id,
276 provider=station.provider,
277 audio_format=AudioFormat(
278 content_type=ContentType.try_parse(str(source_obj.stream))
279 ),
280 data={
281 "provider": self.context.provider_domain,
282 "station": station.item_id,
283 },
284 )
285 return stream_details
286
287 async def convert(self, source_obj: Any) -> Radio:
288 """Convert the source object to target type."""
289 if isinstance(source_obj, Station):
290 return self._convert_station(source_obj)
291 if isinstance(source_obj, LiveStation):
292 return self._convert_live_station(source_obj)
293 if isinstance(source_obj, StationSearchResult):
294 return self._convert_station_search_result(source_obj)
295 self.logger.error(f"Failed to convert station {type(source_obj)}: {source_obj}")
296 raise ConversionError(f"Failed to convert station {type(source_obj)}: {source_obj}")
297
298 def _convert_station(self, station: Station) -> Radio:
299 """Convert Station object."""
300 image_url = self._get_attr(station, "image_url")
301
302 radio = Radio(
303 item_id=station.id,
304 # Add BBC prefix back to station to help identify station within MA
305 name=f"BBC {self._get_attr(station, 'title', 'Unknown')}",
306 provider=self.context.provider_domain,
307 metadata=ImageProvider.create_metadata_with_image(
308 image_url, self.context.provider_domain
309 ),
310 provider_mappings={self._create_provider_mapping(station.id)},
311 )
312 if station.stream:
313 radio.uri = station.stream.uri
314 return radio
315
316 def _convert_live_station(self, station: LiveStation) -> Radio:
317 """Convert LiveStation object."""
318 name = self._get_attr(station, "network.short_title", "Unknown")
319 image_url = self._get_attr(station, "network.logo_url")
320
321 return Radio(
322 item_id=station.id,
323 name=f"BBC {name}",
324 provider=self.context.provider_domain,
325 metadata=ImageProvider.create_metadata_with_image(
326 image_url, self.context.provider_domain
327 ),
328 provider_mappings={self._create_provider_mapping(station.id)},
329 )
330
331 def _convert_station_search_result(self, station: StationSearchResult) -> Radio:
332 """Convert StationSearchResult object."""
333 return Radio(
334 item_id=station.service_id,
335 name=f"BBC {station.station_name}",
336 provider=self.context.provider_domain,
337 metadata=ImageProvider.create_metadata_with_image(
338 station.station_image_url, self.context.provider_domain
339 ),
340 provider_mappings={self._create_provider_mapping(station.service_id)},
341 )
342
343
344class PodcastConverter(BaseConverter):
345 """Converts podcast-related objects."""
346
347 ConvertableTypes = Podcast | PodcastEpisode | RadioShow | RadioClip | RadioSeries
348 OutputTypes = MAPodcast | MAPodcastEpisode | Track
349 SCHEDULE_ITEM_FORMAT = "{start} {show_name} ⢠{show_title} ({date})"
350 SCHEDULE_ITEM_DEFAULT_FORMAT = "{show_name} ⢠{show_title}"
351 PODCAST_EPISODE_DEFAULT_FORMAT = "{episode_title} ({date})"
352 PODCAST_EPISODE_DETAILED_FORMAT = "{episode_title} ⢠{detail} ({date})"
353
354 def _format_show_title(self, show: RadioShow) -> str:
355 if show is None:
356 return "Unknown show"
357 if show.start and show.titles:
358 return self.SCHEDULE_ITEM_FORMAT.format(
359 start=_to_time(show.start),
360 show_name=show.titles["primary"],
361 show_title=show.titles["secondary"],
362 date=_to_date(show.start),
363 )
364 if show.titles:
365 # TODO: when getting a schedule listing, we have a broadcast time
366 # when we fetch the streaming details later we lose that from the new API call
367 title = self.SCHEDULE_ITEM_DEFAULT_FORMAT.format(
368 show_name=show.titles["primary"],
369 show_title=show.titles["secondary"],
370 )
371 date = show.release.get("date") if show.release else None
372 if date and isinstance(date, (str, datetime)):
373 title += f" ({_to_date(date)})"
374 return title
375 return "Unknown"
376
377 def _format_podcast_episode_title(self, episode: PodcastEpisode) -> str:
378 # Similar to show, but not quite: we expect to see this in the context of a podcast detail
379 # page
380 if episode is None:
381 return "Unknown episode"
382
383 if episode.release:
384 date = episode.release.get("date")
385 elif episode.availability:
386 date = episode.availability.get("from")
387 else:
388 date = None
389 if isinstance(date, (str, datetime)) and episode.titles:
390 datestamp = _to_date(date)
391 title = self.PODCAST_EPISODE_DEFAULT_FORMAT.format(
392 episode_title=episode.titles.get("secondary"),
393 date=datestamp,
394 )
395 else:
396 title = str(episode.titles.get("secondary")) if episode.titles else "Unknown episode"
397 return title
398
399 def can_convert(self, source_obj: Any) -> bool:
400 """Check if this converter can convert to a Podcast object."""
401 if self.context.force_type:
402 return issubclass(self.context.force_type, self.OutputTypes)
403 return isinstance(source_obj, self.ConvertableTypes)
404
405 async def get_stream_details(self, source_obj: Any) -> StreamDetails | None:
406 """Convert the source object to a stream."""
407 if isinstance(source_obj, (Podcast, RadioSeries)):
408 return None
409 stream_details = None
410 episode = await self.convert(source_obj)
411 if episode and isinstance(episode, MAPodcastEpisode) and source_obj.stream:
412 stream_details = StreamDetails(
413 stream_metadata=StreamMetadata(
414 title=episode.name,
415 uri=source_obj.stream,
416 ),
417 media_type=MediaType.PODCAST_EPISODE,
418 stream_type=StreamType.HLS
419 if self.context.provider.stream_format == _Constants.HLS
420 else StreamType.HTTP,
421 path=source_obj.stream,
422 item_id=source_obj.id,
423 provider=self.context.provider_domain,
424 audio_format=AudioFormat(content_type=ContentType.try_parse(source_obj.stream)),
425 allow_seek=True,
426 can_seek=True,
427 duration=(episode.duration or None),
428 seek_position=(int(episode.position) if episode.position else 0),
429 seconds_streamed=(int(episode.position) if episode.position else 0),
430 )
431 elif episode and isinstance(episode, Track) and source_obj.stream:
432 # Try to work out the best network/series name to display
433 if source_obj.network and source_obj.network.id == "bbc_webonly":
434 title = "BBC News"
435 elif source_obj.network:
436 title = f"BBC {source_obj.network.short_title}"
437 elif source_obj.container:
438 title = source_obj.container.title
439 elif source_obj.titles:
440 title = self._get_attr(source_obj, "titles.primary")
441 elif episode.metadata and episode.metadata.description:
442 title = episode.metadata.description
443
444 if not title:
445 title = ""
446
447 metadata = StreamMetadata(title=title, uri=source_obj.stream)
448 if episode.metadata.images:
449 metadata.image_url = episode.metadata.images[0].path
450
451 stream_details = StreamDetails(
452 stream_metadata=metadata,
453 media_type=MediaType.TRACK,
454 stream_type=StreamType.HLS
455 if self.context.provider.stream_format == _Constants.HLS
456 else StreamType.HTTP,
457 path=source_obj.stream,
458 item_id=episode.item_id,
459 provider=self.context.provider_domain,
460 audio_format=AudioFormat(content_type=ContentType.try_parse(source_obj.stream)),
461 can_seek=True,
462 duration=episode.duration,
463 )
464 return stream_details
465
466 async def convert(self, source_obj: Any) -> OutputTypes:
467 """Convert podcast objects."""
468 if isinstance(source_obj, (Podcast, RadioSeries)) or self.context.force_type is Podcast:
469 return await self._convert_podcast(source_obj)
470 if isinstance(source_obj, PodcastEpisode):
471 return await self._convert_podcast_episode(source_obj)
472 if isinstance(source_obj, RadioShow):
473 return await self._convert_radio_show(source_obj)
474 if isinstance(source_obj, RadioClip) or self.context.force_type is Track:
475 return await self._convert_radio_clip(source_obj)
476 self.logger.error(f"Failed to convert podcast object {type(source_obj)}: {source_obj}")
477 raise ConversionError(f"Browse conversion failed: {source_obj}")
478
479 async def _convert_podcast(self, podcast: Podcast | RadioSeries) -> MAPodcast:
480 name = self._get_attr(podcast, "titles.primary") or self._get_attr(podcast, "title")
481 description = self._get_synopsis(podcast)
482 image_url = self._get_attr(podcast, "image_url") or self._get_attr(
483 podcast, "sub_items.image_url"
484 )
485
486 return MAPodcast(
487 item_id=podcast.id,
488 name=name,
489 provider=self.context.provider_domain,
490 metadata=ImageProvider.create_metadata_with_image(
491 image_url, self.context.provider_domain, description
492 ),
493 provider_mappings={self._create_provider_mapping(podcast.item_id)},
494 )
495
496 async def _convert_podcast_episode(self, episode: PodcastEpisode) -> MAPodcastEpisode:
497 duration = self._get_attr(episode, "duration.value")
498 progress_ms = self._get_attr(episode, "progress.value")
499 resume_position = (progress_ms * 1000) if progress_ms else None
500 description = self._get_synopsis(episode)
501
502 # Handle parent podcast
503 podcast = None
504 if hasattr(episode, "container") and episode.container:
505 podcast = await PodcastConverter(self.context).convert(
506 cast("Podcast", episode.container)
507 )
508
509 if not podcast or not isinstance(podcast, MAPodcast):
510 raise ConversionError(f"No podcast for episode {episode}")
511 if not episode or not episode.pid:
512 raise ConversionError(f"No podcast episode for {episode}")
513
514 return MAPodcastEpisode(
515 item_id=episode.pid,
516 name=self._format_podcast_episode_title(episode),
517 provider=self.context.provider_domain,
518 duration=duration,
519 position=0,
520 resume_position_ms=resume_position,
521 metadata=ImageProvider.create_metadata_with_image(
522 episode.image_url,
523 self.context.provider_domain,
524 description,
525 ),
526 podcast=podcast,
527 provider_mappings={self._create_provider_mapping(episode.pid)},
528 uri=episode.stream,
529 )
530
531 async def _convert_radio_show(self, show: RadioShow) -> MAPodcastEpisode | Track:
532 duration = self._get_attr(show, "duration.value")
533 progress_ms = self._get_attr(show, "progress.value")
534 resume_position = (progress_ms * 1000) if progress_ms else None
535
536 if not show or not show.pid:
537 raise ConversionError(f"No radio show for {show}")
538
539 # Determine if this should be an episode or track based on duration/context
540 # TODO: picked a sensible default but need to investigate if this makes sense
541 # Track example: latest BBC News, PodcastEpisode: latest episode of a radio show
542 if (
543 self.context.force_type == Track
544 or (
545 not self.context.force_type
546 and duration
547 and duration < _Constants.TRACK_DURATION_THRESHOLD
548 )
549 or (not hasattr(show, "container") or not show.container)
550 ):
551 return Track(
552 item_id=show.pid,
553 name=self._format_show_title(show),
554 provider=self.context.provider_domain,
555 duration=duration,
556 metadata=ImageProvider.create_metadata_with_image(
557 url=show.image_url,
558 provider=self.context.provider_domain,
559 description=self._get_synopsis(show),
560 ),
561 provider_mappings={self._create_provider_mapping(show.pid)},
562 )
563 # Handle as episode
564 podcast = None
565 if hasattr(show, "container") and show.container:
566 podcast = await PodcastConverter(self.context).convert(cast("Podcast", show.container))
567
568 if not podcast or not isinstance(podcast, MAPodcast):
569 raise ConversionError(f"No podcast for episode for {show}")
570
571 return MAPodcastEpisode(
572 item_id=show.pid,
573 name=self._format_show_title(show),
574 provider=self.context.provider_domain,
575 duration=duration,
576 resume_position_ms=resume_position,
577 metadata=ImageProvider.create_metadata_with_image(
578 url=show.image_url,
579 provider=self.context.provider_domain,
580 description=self._get_synopsis(show),
581 ),
582 podcast=podcast,
583 provider_mappings={self._create_provider_mapping(show.pid)},
584 position=1,
585 )
586
587 async def _convert_radio_clip(self, clip: RadioClip) -> Track | MAPodcastEpisode:
588 duration = self._get_attr(clip, "duration.value")
589 description = self._get_synopsis(clip)
590
591 if not clip or not clip.pid:
592 raise ConversionError(f"No clip for {clip}")
593
594 if self.context.force_type is MAPodcastEpisode:
595 podcast = None
596 if hasattr(clip, "container") and clip.container:
597 podcast = await PodcastConverter(self.context).convert(
598 cast("Podcast", clip.container)
599 )
600
601 if not podcast or not isinstance(podcast, MAPodcast):
602 raise ConversionError(f"No podcast for episode for {clip}")
603 return MAPodcastEpisode(
604 item_id=clip.pid,
605 name=self._get_attr(clip, "titles.entity_title", "Unknown title"),
606 provider=self.context.provider_domain,
607 duration=duration,
608 metadata=ImageProvider.create_metadata_with_image(
609 clip.image_url, self.context.provider_domain, description
610 ),
611 provider_mappings={self._create_provider_mapping(clip.pid)},
612 podcast=podcast,
613 position=0,
614 )
615 return Track(
616 item_id=clip.pid,
617 name=self._get_attr(clip, "titles.entity_title", "Unknown Track"),
618 provider=self.context.provider_domain,
619 duration=duration,
620 metadata=ImageProvider.create_metadata_with_image(
621 clip.image_url, self.context.provider_domain, description
622 ),
623 provider_mappings={self._create_provider_mapping(clip.pid)},
624 )
625
626
627class BrowseConverter(BaseConverter):
628 """Converts browsable objects like menus, categories, collections."""
629
630 ConvertableTypes = MenuItem | Category | Collection | Schedule | RecommendedMenuItem | Playlist
631 OutputTypes = BrowseFolder | RecommendationFolder
632
633 def can_convert(self, source_obj: Any) -> bool:
634 """Check if this converter can convert to a Browsable object."""
635 can_convert = False
636 if self.context.force_type:
637 can_convert = issubclass(self.context.force_type, self.OutputTypes)
638 else:
639 can_convert = isinstance(source_obj, self.ConvertableTypes)
640 return can_convert
641
642 async def get_stream_details(self, source_obj: Any) -> StreamDetails | None:
643 """Convert the source object to a stream."""
644 return None
645
646 async def convert(self, source_obj: Any) -> OutputTypes:
647 """Convert browsable objects."""
648 if isinstance(source_obj, MenuItem) and self.context.force_type is not RecommendationFolder:
649 return self._convert_menu_item(source_obj)
650 if isinstance(source_obj, (Category, Collection, Playlist)):
651 return self._convert_category_or_collection(source_obj)
652 if isinstance(source_obj, Schedule):
653 return self._convert_schedule(source_obj)
654 if isinstance(source_obj, RecommendedMenuItem):
655 return await self._convert_recommended_item(source_obj)
656 self.logger.error(f"Failed to convert browse object {type(source_obj)}: {source_obj}")
657 raise ConversionError(f"Browse conversion failed: {source_obj}")
658
659 def _convert_menu_item(self, item: MenuItem) -> BrowseFolder | RecommendationFolder:
660 """Convert MenuItem to BrowseFolder or RecommendationFolder."""
661 if not item or not item.title:
662 raise ConversionError(f"No menu item {item}")
663 if not item.image_url:
664 image_url = ImageProvider.get_icon_url(item.item_id)
665 image = (
666 ImageProvider.create_image(image_url, self.context.provider_domain)
667 if image_url
668 else None
669 )
670 elif item.image_url:
671 image = ImageProvider.create_image(item.image_url, self.context.provider_domain)
672
673 path = self._build_path(item.item_id)
674
675 return_type = BrowseFolder
676
677 if self.context.force_type is RecommendationFolder:
678 return_type = RecommendationFolder
679
680 return return_type(
681 item_id=item.item_id,
682 name=item.title,
683 provider=self.context.provider_domain,
684 path=path,
685 image=image,
686 )
687
688 def _convert_category_or_collection(
689 self, item: Category | Collection | Playlist
690 ) -> BrowseFolder:
691 """Convert Category, Collection or Playlist to BrowseFolder."""
692 if isinstance(item, Playlist):
693 if not isinstance(self.context.path_parts, list):
694 raise ConversionError("Path not provided for Playlist item")
695 path = "/".join([*self.context.path_parts, item.item_id])
696 else:
697 path_prefix = "categories" if isinstance(item, Category) else "collections"
698 path = f"{self.context.provider_domain}://{path_prefix}/{item.item_id}"
699
700 return BrowseFolder(
701 item_id=item.item_id,
702 name=self._get_attr(item, "titles.primary", "Untitled folder"),
703 provider=self.context.provider_domain,
704 path=path,
705 image=(
706 ImageProvider.create_image(item.image_url, self.context.provider_domain)
707 if item.image_url
708 else None
709 ),
710 )
711
712 def _convert_schedule(self, schedule: Schedule) -> BrowseFolder:
713 """Convert Schedule to BrowseFolder."""
714 return BrowseFolder(
715 item_id="schedule",
716 name="Schedule",
717 translation_key="schedule",
718 provider=self.context.provider_domain,
719 path=self._build_path("schedule"),
720 )
721
722 async def _convert_recommended_item(self, item: RecommendedMenuItem) -> RecommendationFolder:
723 """Convert RecommendedMenuItem to RecommendationFolder."""
724 if not item or not item.sub_items or not item.title:
725 raise ConversionError(f"Incorrect format for item {item}")
726
727 # TODO this is messy
728 new_adaptor = Adaptor(provider=self.context.provider)
729 items: list[Track | Radio | MAPodcast | MAPodcastEpisode | BrowseFolder] = []
730 for sub_item in item.sub_items:
731 new_item = await new_adaptor.new_object(sub_item)
732 if (
733 new_item is not None
734 and not isinstance(new_item, RecommendationFolder)
735 and not isinstance(new_item, RecommendedMenuItem)
736 ):
737 items.append(new_item)
738
739 return RecommendationFolder(
740 item_id=item.item_id,
741 name=item.title,
742 provider=self.context.provider_domain,
743 items=UniqueList(items),
744 )
745
746 def _build_path(self, item_id: str) -> str:
747 """Build path for browse items."""
748 if self.context.path_parts:
749 return "/".join([*self.context.path_parts, item_id])
750 return f"{self.context.provider_domain}://{item_id}"
751
752
753class Adaptor:
754 """An adaptor object to convert Sounds API objects into MA ones."""
755
756 def __init__(self, provider: BBCSoundsProvider):
757 """Create new adaptor."""
758 self.provider = provider
759 self.logger = self.provider.logger
760 self._converters: list[BaseConverter] = []
761
762 def _create_context(
763 self,
764 path_parts: list[str] | None = None,
765 force_type: (
766 type[
767 Track
768 | Radio
769 | MAPodcast
770 | MAPodcastEpisode
771 | BrowseFolder
772 | RecommendationFolder
773 | RecommendedMenuItem
774 ]
775 | None
776 ) = None,
777 ) -> Context:
778 return Context(
779 provider=self.provider,
780 provider_domain=self.provider.domain,
781 path_parts=path_parts,
782 force_type=force_type,
783 )
784
785 async def new_streamable_object(
786 self,
787 source_obj: SoundsTypes,
788 force_type: type[Track | Radio | MAPodcastEpisode] | None = None,
789 path_parts: list[str] | None = None,
790 ) -> StreamDetails | None:
791 """
792 Convert an auntie-sounds object to appropriate Music Assistant object.
793
794 Args:
795 source_obj: The source object from Sounds API via auntie-sounds
796 force_type: Force conversion to specific type if the expected target type is known
797 path_parts: Path parts for browse items to construct the object's path
798
799 Returns:
800 Converted Music Assistant media item or None if no converter found
801 """
802 if source_obj is None:
803 return None
804
805 context = self._create_context(path_parts, force_type)
806
807 converters = [
808 StationConverter(context),
809 PodcastConverter(context),
810 BrowseConverter(context),
811 ]
812
813 for converter in converters:
814 if converter.can_convert(source_obj):
815 try:
816 stream_details = await converter.get_stream_details(source_obj)
817 except AttributeError as e:
818 self.logger.error(f"Error converting object: {e!s}")
819 return None
820 self.provider.logger.debug(
821 f"Successfully converted {type(source_obj).__name__}"
822 f" to {type(stream_details).__name__}"
823 )
824 return stream_details
825 self.provider.logger.warning(
826 f"No stream converter found for type {type(source_obj).__name__}"
827 )
828 return None
829
830 async def new_object(
831 self,
832 source_obj: SoundsTypes,
833 force_type: (
834 type[
835 Track
836 | Radio
837 | MAPodcast
838 | MAPodcastEpisode
839 | BrowseFolder
840 | RecommendationFolder
841 | RecommendedMenuItem
842 ]
843 | None
844 ) = None,
845 path_parts: list[str] | None = None,
846 ) -> (
847 Track
848 | Radio
849 | MAPodcast
850 | MAPodcastEpisode
851 | BrowseFolder
852 | RecommendationFolder
853 | RecommendedMenuItem
854 | None
855 ):
856 """
857 Convert an auntie-sounds object to appropriate Music Assistant object.
858
859 Args:
860 source_obj: The source object from Sounds API via auntie-sounds
861 force_type: Force conversion to specific type if the expected target type is known
862 path_parts: Path parts for browse items to construct the object's path
863
864 Returns:
865 Converted Music Assistant media item or None if no converter found
866 """
867 if source_obj is None:
868 return None
869
870 context = self._create_context(path_parts, force_type)
871
872 converters = [
873 StationConverter(context),
874 PodcastConverter(context),
875 BrowseConverter(context),
876 ]
877 for converter in converters:
878 self.logger.debug(f"Checking if converter {converter} can convert {type(source_obj)}")
879 if converter.can_convert(source_obj):
880 try:
881 result = await converter.convert(source_obj)
882 except AttributeError as e:
883 self.logger.error(f"Error converting object: {e!s}")
884 return None
885 if context.force_type:
886 assert type(result) is context.force_type, (
887 f"Forced type to {context.force_type} but received {type(result)} "
888 f"using {type(converter)}"
889 )
890 self.provider.logger.debug(
891 f"Successfully converted {type(source_obj).__name__}"
892 f" to {type(result).__name__} {result}"
893 )
894 return result
895 self.logger.debug(f"Converter {converter} could not convert {type(source_obj)}")
896
897 self.logger.warning(f"No converter found for type {type(source_obj).__name__}")
898 self.logger.debug(str(source_obj))
899 return None
900