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