music-assistant-server

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
232class StationConverter(BaseConverter):
233    """Converts Station-related objects."""
234
235    ConvertableTypes = Station | LiveStation | StationSearchResult
236
237    def can_convert(self, source_obj: Any) -> bool:
238        """Check if this converter can convert to a Station object."""
239        return isinstance(source_obj, self.ConvertableTypes)
240
241    async def get_stream_details(self, source_obj: Any) -> StreamDetails | None:
242        """Convert the source object to a stream."""
243        if not isinstance(source_obj, self.ConvertableTypes):
244            return None
245        # TODO: can't seek this stream
246        station = await self.convert(source_obj)
247        if not station or not source_obj.stream:
248            return None
249        show_time = self._get_attr(source_obj, "titles.secondary")
250        show_title = self._get_attr(source_obj, "titles.primary")
251        programme_name = f"{show_time} • {show_title}"
252        stream_details = None
253        if station and source_obj.stream:
254            stream_metadata = StreamMetadata(
255                title=programme_name,
256            )
257
258            if station.image is not None:
259                stream_metadata.image_url = station.image.path
260
261            stream_details = StreamDetails(
262                stream_metadata=stream_metadata,
263                media_type=MediaType.RADIO,
264                stream_type=StreamType.HLS
265                if self.context.provider.stream_format == _Constants.HLS
266                else StreamType.HTTP,
267                path=str(source_obj.stream),
268                item_id=station.item_id,
269                provider=station.provider,
270                audio_format=AudioFormat(
271                    content_type=ContentType.try_parse(str(source_obj.stream))
272                ),
273                data={
274                    "provider": self.context.provider_domain,
275                    "station": station.item_id,
276                },
277            )
278        return stream_details
279
280    async def convert(self, source_obj: Any) -> Radio:
281        """Convert the source object to target type."""
282        if isinstance(source_obj, Station):
283            return self._convert_station(source_obj)
284        if isinstance(source_obj, LiveStation):
285            return self._convert_live_station(source_obj)
286        if isinstance(source_obj, StationSearchResult):
287            return self._convert_station_search_result(source_obj)
288        self.logger.error(f"Failed to convert station {type(source_obj)}: {source_obj}")
289        raise ConversionError(f"Failed to convert station {type(source_obj)}: {source_obj}")
290
291    def _convert_station(self, station: Station) -> Radio:
292        """Convert Station object."""
293        image_url = self._get_attr(station, "image_url")
294
295        radio = Radio(
296            item_id=station.id,
297            # Add BBC prefix back to station to help identify station within MA
298            name=f"BBC {self._get_attr(station, 'title', 'Unknown')}",
299            provider=self.context.provider_domain,
300            metadata=ImageProvider.create_metadata_with_image(
301                image_url, self.context.provider_domain
302            ),
303            provider_mappings={self._create_provider_mapping(station.id)},
304        )
305        if station.stream:
306            radio.uri = station.stream.uri
307        return radio
308
309    def _convert_live_station(self, station: LiveStation) -> Radio:
310        """Convert LiveStation object."""
311        name = self._get_attr(station, "network.short_title", "Unknown")
312        image_url = self._get_attr(station, "network.logo_url")
313
314        return Radio(
315            item_id=station.id,
316            name=f"BBC {name}",
317            provider=self.context.provider_domain,
318            metadata=ImageProvider.create_metadata_with_image(
319                image_url, self.context.provider_domain
320            ),
321            provider_mappings={self._create_provider_mapping(station.id)},
322        )
323
324    def _convert_station_search_result(self, station: StationSearchResult) -> Radio:
325        """Convert StationSearchResult object."""
326        return Radio(
327            item_id=station.service_id,
328            name=f"BBC {station.station_name}",
329            provider=self.context.provider_domain,
330            metadata=ImageProvider.create_metadata_with_image(
331                station.station_image_url, self.context.provider_domain
332            ),
333            provider_mappings={self._create_provider_mapping(station.service_id)},
334        )
335
336
337class PodcastConverter(BaseConverter):
338    """Converts podcast-related objects."""
339
340    ConvertableTypes = Podcast | PodcastEpisode | RadioShow | RadioClip | RadioSeries
341    OutputTypes = MAPodcast | MAPodcastEpisode | Track
342    SCHEDULE_ITEM_FORMAT = "{start} {show_name} • {show_title} ({date})"
343    SCHEDULE_ITEM_DEFAULT_FORMAT = "{show_name} • {show_title}"
344    PODCAST_EPISODE_DEFAULT_FORMAT = "{episode_title} ({date})"
345    PODCAST_EPISODE_DETAILED_FORMAT = "{episode_title} • {detail} ({date})"
346
347    def _format_show_title(self, show: RadioShow) -> str:
348        if show is None:
349            return "Unknown show"
350        if show.start and show.titles:
351            return self.SCHEDULE_ITEM_FORMAT.format(
352                start=_to_time(show.start),
353                show_name=show.titles["primary"],
354                show_title=show.titles["secondary"],
355                date=_to_date(show.start),
356            )
357        if show.titles:
358            # TODO: when getting a schedule listing, we have a broadcast time
359            # when we fetch the streaming details later we lose that from the new API call
360            title = self.SCHEDULE_ITEM_DEFAULT_FORMAT.format(
361                show_name=show.titles["primary"],
362                show_title=show.titles["secondary"],
363            )
364            date = show.release.get("date") if show.release else None
365            if date and isinstance(date, (str, datetime)):
366                title += f" ({_to_date(date)})"
367            return title
368        return "Unknown"
369
370    def _format_podcast_episode_title(self, episode: PodcastEpisode) -> str:
371        # Similar to show, but not quite: we expect to see this in the context of a podcast detail
372        # page
373        if episode is None:
374            return "Unknown episode"
375
376        if episode.release:
377            date = episode.release.get("date")
378        elif episode.availability:
379            date = episode.availability.get("from")
380        else:
381            date = None
382        if isinstance(date, (str, datetime)) and episode.titles:
383            datestamp = _to_date(date)
384            title = self.PODCAST_EPISODE_DEFAULT_FORMAT.format(
385                episode_title=episode.titles.get("secondary"),
386                date=datestamp,
387            )
388        else:
389            title = str(episode.titles.get("secondary")) if episode.titles else "Unknown episode"
390        return title
391
392    def can_convert(self, source_obj: Any) -> bool:
393        """Check if this converter can convert to a Podcast object."""
394        if self.context.force_type:
395            return issubclass(self.context.force_type, self.OutputTypes)
396        return isinstance(source_obj, self.ConvertableTypes)
397
398    async def get_stream_details(self, source_obj: Any) -> StreamDetails | None:
399        """Convert the source object to a stream."""
400        if isinstance(source_obj, (Podcast, RadioSeries)):
401            return None
402        stream_details = None
403        episode = await self.convert(source_obj)
404        if (
405            episode
406            and isinstance(episode, MAPodcastEpisode)
407            and (episode.metadata.description or episode.name)
408            and source_obj.stream
409        ):
410            stream_details = StreamDetails(
411                stream_metadata=StreamMetadata(
412                    title=episode.metadata.description or episode.name,
413                    uri=source_obj.stream,
414                ),
415                media_type=MediaType.PODCAST_EPISODE,
416                stream_type=StreamType.HLS
417                if self.context.provider.stream_format == _Constants.HLS
418                else StreamType.HTTP,
419                path=source_obj.stream,
420                item_id=source_obj.id,
421                provider=self.context.provider_domain,
422                audio_format=AudioFormat(content_type=ContentType.try_parse(source_obj.stream)),
423                allow_seek=True,
424                can_seek=True,
425                duration=(episode.duration or None),
426                seek_position=(int(episode.position) if episode.position else 0),
427                seconds_streamed=(int(episode.position) if episode.position else 0),
428            )
429        elif episode and isinstance(episode, Track) and source_obj.stream:
430            # Try to work out the best network/series name to display
431            if source_obj.network and source_obj.network.id == "bbc_webonly":
432                title = "BBC News"
433            elif source_obj.network:
434                title = f"BBC {source_obj.network.short_title}"
435            elif source_obj.container:
436                title = source_obj.container.title
437            elif episode.metadata and episode.metadata.description:
438                title = episode.metadata.description
439            elif source_obj.titles:
440                title = source_obj.titles["primary"]
441
442            if not title:
443                title = ""
444
445            metadata = StreamMetadata(title=title, uri=source_obj.stream)
446            if episode.metadata.images:
447                metadata.image_url = episode.metadata.images[0].path
448
449            stream_details = StreamDetails(
450                stream_metadata=metadata,
451                media_type=MediaType.TRACK,
452                stream_type=StreamType.HLS
453                if self.context.provider.stream_format == _Constants.HLS
454                else StreamType.HTTP,
455                path=source_obj.stream,
456                item_id=episode.item_id,
457                provider=self.context.provider_domain,
458                audio_format=AudioFormat(content_type=ContentType.try_parse(source_obj.stream)),
459                can_seek=True,
460                duration=episode.duration,
461            )
462        return stream_details
463
464    async def convert(self, source_obj: Any) -> OutputTypes:
465        """Convert podcast objects."""
466        if isinstance(source_obj, (Podcast, RadioSeries)) or self.context.force_type is Podcast:
467            return await self._convert_podcast(source_obj)
468        if isinstance(source_obj, PodcastEpisode):
469            return await self._convert_podcast_episode(source_obj)
470        if isinstance(source_obj, RadioShow):
471            return await self._convert_radio_show(source_obj)
472        if isinstance(source_obj, RadioClip) or self.context.force_type is Track:
473            return await self._convert_radio_clip(source_obj)
474        self.logger.error(f"Failed to convert podcast object {type(source_obj)}: {source_obj}")
475        raise ConversionError(f"Browse conversion failed: {source_obj}")
476
477    async def _convert_podcast(self, podcast: Podcast | RadioSeries) -> MAPodcast:
478        name = self._get_attr(podcast, "titles.primary") or self._get_attr(podcast, "title")
479        description = self._get_attr(podcast, "synopses.long") or self._get_attr(
480            podcast, "synopses.short"
481        )
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_attr(episode, "synopses.short")
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=show.synopses.get("long") if show.synopses else None,
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                show.image_url, self.context.provider_domain
579            ),
580            podcast=podcast,
581            provider_mappings={self._create_provider_mapping(show.pid)},
582            position=1,
583        )
584
585    async def _convert_radio_clip(self, clip: RadioClip) -> Track | MAPodcastEpisode:
586        duration = self._get_attr(clip, "duration.value")
587        description = self._get_attr(clip, "network.short_title")
588
589        if not clip or not clip.pid:
590            raise ConversionError(f"No clip for {clip}")
591
592        if self.context.force_type is MAPodcastEpisode:
593            podcast = None
594            if hasattr(clip, "container") and clip.container:
595                podcast = await PodcastConverter(self.context).convert(
596                    cast("Podcast", clip.container)
597                )
598
599            if not podcast or not isinstance(podcast, MAPodcast):
600                raise ConversionError(f"No podcast for episode for {clip}")
601            return MAPodcastEpisode(
602                item_id=clip.pid,
603                name=self._get_attr(clip, "titles.entity_title", "Unknown title"),
604                provider=self.context.provider_domain,
605                duration=duration,
606                metadata=ImageProvider.create_metadata_with_image(
607                    clip.image_url, self.context.provider_domain, description
608                ),
609                provider_mappings={self._create_provider_mapping(clip.pid)},
610                podcast=podcast,
611                position=0,
612            )
613        return Track(
614            item_id=clip.pid,
615            name=self._get_attr(clip, "titles.entity_title", "Unknown Track"),
616            provider=self.context.provider_domain,
617            duration=duration,
618            metadata=ImageProvider.create_metadata_with_image(
619                clip.image_url, self.context.provider_domain, description
620            ),
621            provider_mappings={self._create_provider_mapping(clip.pid)},
622        )
623
624
625class BrowseConverter(BaseConverter):
626    """Converts browsable objects like menus, categories, collections."""
627
628    ConvertableTypes = MenuItem | Category | Collection | Schedule | RecommendedMenuItem | Playlist
629    OutputTypes = BrowseFolder | RecommendationFolder
630
631    def can_convert(self, source_obj: Any) -> bool:
632        """Check if this converter can convert to a Browsable object."""
633        can_convert = False
634        if self.context.force_type:
635            can_convert = issubclass(self.context.force_type, self.OutputTypes)
636        else:
637            can_convert = isinstance(source_obj, self.ConvertableTypes)
638        return can_convert
639
640    async def get_stream_details(self, source_obj: Any) -> StreamDetails | None:
641        """Convert the source object to a stream."""
642        return None
643
644    async def convert(self, source_obj: Any) -> OutputTypes:
645        """Convert browsable objects."""
646        if isinstance(source_obj, MenuItem) and self.context.force_type is not RecommendationFolder:
647            return self._convert_menu_item(source_obj)
648        if isinstance(source_obj, (Category, Collection, Playlist)):
649            return self._convert_category_or_collection(source_obj)
650        if isinstance(source_obj, Schedule):
651            return self._convert_schedule(source_obj)
652        if isinstance(source_obj, RecommendedMenuItem):
653            return await self._convert_recommended_item(source_obj)
654        self.logger.error(f"Failed to convert browse object {type(source_obj)}: {source_obj}")
655        raise ConversionError(f"Browse conversion failed: {source_obj}")
656
657    def _convert_menu_item(self, item: MenuItem) -> BrowseFolder | RecommendationFolder:
658        """Convert MenuItem to BrowseFolder or RecommendationFolder."""
659        if not item or not item.title:
660            raise ConversionError(f"No menu item {item}")
661        if not item.image_url:
662            image_url = ImageProvider.get_icon_url(item.item_id)
663            image = (
664                ImageProvider.create_image(image_url, self.context.provider_domain)
665                if image_url
666                else None
667            )
668        elif item.image_url:
669            image = ImageProvider.create_image(item.image_url, self.context.provider_domain)
670
671        path = self._build_path(item.item_id)
672
673        return_type = BrowseFolder
674
675        if self.context.force_type is RecommendationFolder:
676            return_type = RecommendationFolder
677
678        return return_type(
679            item_id=item.item_id,
680            name=item.title,
681            provider=self.context.provider_domain,
682            path=path,
683            image=image,
684        )
685
686    def _convert_category_or_collection(
687        self, item: Category | Collection | Playlist
688    ) -> BrowseFolder:
689        """Convert Category, Collection or Playlist to BrowseFolder."""
690        if isinstance(item, Playlist):
691            if not isinstance(self.context.path_parts, list):
692                raise ConversionError("Path not provided for Playlist item")
693            path = "/".join([*self.context.path_parts, item.item_id])
694        else:
695            path_prefix = "categories" if isinstance(item, Category) else "collections"
696            path = f"{self.context.provider_domain}://{path_prefix}/{item.item_id}"
697
698        return BrowseFolder(
699            item_id=item.item_id,
700            name=self._get_attr(item, "titles.primary", "Untitled folder"),
701            provider=self.context.provider_domain,
702            path=path,
703            image=(
704                ImageProvider.create_image(item.image_url, self.context.provider_domain)
705                if item.image_url
706                else None
707            ),
708        )
709
710    def _convert_schedule(self, schedule: Schedule) -> BrowseFolder:
711        """Convert Schedule to BrowseFolder."""
712        return BrowseFolder(
713            item_id="schedule",
714            name="Schedule",
715            translation_key="schedule",
716            provider=self.context.provider_domain,
717            path=self._build_path("schedule"),
718        )
719
720    async def _convert_recommended_item(self, item: RecommendedMenuItem) -> RecommendationFolder:
721        """Convert RecommendedMenuItem to RecommendationFolder."""
722        if not item or not item.sub_items or not item.title:
723            raise ConversionError(f"Incorrect format for item {item}")
724
725        # TODO this is messy
726        new_adaptor = Adaptor(provider=self.context.provider)
727        items: list[Track | Radio | MAPodcast | MAPodcastEpisode | BrowseFolder] = []
728        for sub_item in item.sub_items:
729            new_item = await new_adaptor.new_object(sub_item)
730            if (
731                new_item is not None
732                and not isinstance(new_item, RecommendationFolder)
733                and not isinstance(new_item, RecommendedMenuItem)
734            ):
735                items.append(new_item)
736
737        return RecommendationFolder(
738            item_id=item.item_id,
739            name=item.title,
740            provider=self.context.provider_domain,
741            items=UniqueList(items),
742        )
743
744    def _build_path(self, item_id: str) -> str:
745        """Build path for browse items."""
746        if self.context.path_parts:
747            return "/".join([*self.context.path_parts, item_id])
748        return f"{self.context.provider_domain}://{item_id}"
749
750
751class Adaptor:
752    """An adaptor object to convert Sounds API objects into MA ones."""
753
754    def __init__(self, provider: BBCSoundsProvider):
755        """Create new adaptor."""
756        self.provider = provider
757        self.logger = self.provider.logger
758        self._converters: list[BaseConverter] = []
759
760    def _create_context(
761        self,
762        path_parts: list[str] | None = None,
763        force_type: (
764            type[
765                Track
766                | Radio
767                | MAPodcast
768                | MAPodcastEpisode
769                | BrowseFolder
770                | RecommendationFolder
771                | RecommendedMenuItem
772            ]
773            | None
774        ) = None,
775    ) -> Context:
776        return Context(
777            provider=self.provider,
778            provider_domain=self.provider.domain,
779            path_parts=path_parts,
780            force_type=force_type,
781        )
782
783    async def new_streamable_object(
784        self,
785        source_obj: SoundsTypes,
786        force_type: type[Track | Radio | MAPodcastEpisode] | None = None,
787        path_parts: list[str] | None = None,
788    ) -> StreamDetails | None:
789        """
790        Convert an auntie-sounds object to appropriate Music Assistant object.
791
792        Args:
793            source_obj: The source object from Sounds API via auntie-sounds
794            force_type: Force conversion to specific type if the expected target type is known
795            path_parts: Path parts for browse items to construct the object's path
796
797        Returns:
798            Converted Music Assistant media item or None if no converter found
799        """
800        if source_obj is None:
801            return None
802
803        context = self._create_context(path_parts, force_type)
804
805        converters = [
806            StationConverter(context),
807            PodcastConverter(context),
808            BrowseConverter(context),
809        ]
810
811        for converter in converters:
812            if converter.can_convert(source_obj):
813                try:
814                    stream_details = await converter.get_stream_details(source_obj)
815                except AttributeError as e:
816                    self.logger.error(f"Error converting object: {e!s}")
817                    return None
818                self.provider.logger.debug(
819                    f"Successfully converted {type(source_obj).__name__}"
820                    f" to {type(stream_details).__name__}"
821                )
822                return stream_details
823        self.provider.logger.warning(
824            f"No stream converter found for type {type(source_obj).__name__}"
825        )
826        return None
827
828    async def new_object(
829        self,
830        source_obj: SoundsTypes,
831        force_type: (
832            type[
833                Track
834                | Radio
835                | MAPodcast
836                | MAPodcastEpisode
837                | BrowseFolder
838                | RecommendationFolder
839                | RecommendedMenuItem
840            ]
841            | None
842        ) = None,
843        path_parts: list[str] | None = None,
844    ) -> (
845        Track
846        | Radio
847        | MAPodcast
848        | MAPodcastEpisode
849        | BrowseFolder
850        | RecommendationFolder
851        | RecommendedMenuItem
852        | None
853    ):
854        """
855        Convert an auntie-sounds object to appropriate Music Assistant object.
856
857        Args:
858            source_obj: The source object from Sounds API via auntie-sounds
859            force_type: Force conversion to specific type if the expected target type is known
860            path_parts: Path parts for browse items to construct the object's path
861
862        Returns:
863            Converted Music Assistant media item or None if no converter found
864        """
865        if source_obj is None:
866            return None
867
868        context = self._create_context(path_parts, force_type)
869
870        converters = [
871            StationConverter(context),
872            PodcastConverter(context),
873            BrowseConverter(context),
874        ]
875        for converter in converters:
876            self.logger.debug(f"Checking if converter {converter} can convert {type(source_obj)}")
877            if converter.can_convert(source_obj):
878                try:
879                    result = await converter.convert(source_obj)
880                except AttributeError as e:
881                    self.logger.error(f"Error converting object: {e!s}")
882                    return None
883                if context.force_type:
884                    assert type(result) is context.force_type, (
885                        f"Forced type to {context.force_type} but received {type(result)} "
886                        f"using {type(converter)}"
887                    )
888                self.provider.logger.debug(
889                    f"Successfully converted {type(source_obj).__name__}"
890                    f" to {type(result).__name__} {result}"
891                )
892                return result
893            self.logger.debug(f"Converter {converter} could not convert {type(source_obj)}")
894
895        self.logger.warning(f"No converter found for type {type(source_obj).__name__}")
896        self.logger.debug(str(source_obj))
897        return None
898