music-assistant-server

1"""RadioBrowser musicprovider support for MusicAssistant."""
2
3from __future__ import annotations
4
5from collections.abc import Sequence
6from typing import TYPE_CHECKING
7
8from music_assistant_models.enums import (
9    ContentType,
10    ImageType,
11    LinkType,
12    MediaType,
13    ProviderFeature,
14    StreamType,
15)
16from music_assistant_models.errors import MediaNotFoundError, ProviderUnavailableError
17from music_assistant_models.media_items import (
18    AudioFormat,
19    BrowseFolder,
20    MediaItemImage,
21    MediaItemLink,
22    MediaItemType,
23    ProviderMapping,
24    Radio,
25    SearchResults,
26    UniqueList,
27)
28from music_assistant_models.streamdetails import StreamDetails
29from radios import FilterBy, Order, RadioBrowser, RadioBrowserError, Station
30
31from music_assistant.controllers.cache import use_cache
32from music_assistant.models.music_provider import MusicProvider
33
34SUPPORTED_FEATURES = {
35    ProviderFeature.SEARCH,
36    ProviderFeature.BROWSE,
37}
38
39if TYPE_CHECKING:
40    from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
41    from music_assistant_models.provider import ProviderManifest
42
43    from music_assistant.mass import MusicAssistant
44    from music_assistant.models import ProviderInstanceType
45
46
47async def setup(
48    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
49) -> ProviderInstanceType:
50    """Initialize provider(instance) with given configuration."""
51    return RadioBrowserProvider(mass, manifest, config, SUPPORTED_FEATURES)
52
53
54class RadioBrowserProvider(MusicProvider):
55    """Provider implementation for RadioBrowser."""
56
57    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
58        """Return Config entries to configure this provider."""
59        return ()
60
61    async def handle_async_init(self) -> None:
62        """Handle async initialization of the provider."""
63        self.radios = RadioBrowser(
64            session=self.mass.http_session, user_agent=f"MusicAssistant/{self.mass.version}"
65        )
66        try:
67            await self.radios.stats()
68        except RadioBrowserError as err:
69            raise ProviderUnavailableError(f"RadioBrowser API unavailable: {err}") from err
70
71    @use_cache(3600 * 24 * 14)  # Cache for 14 days
72    async def search(
73        self, search_query: str, media_types: list[MediaType], limit: int = 10
74    ) -> SearchResults:
75        """Perform search on musicprovider."""
76        result = SearchResults()
77        if MediaType.RADIO not in media_types:
78            return result
79
80        try:
81            searchresult = await self.radios.search(name=search_query, limit=limit)
82            result.radio = [await self._parse_radio(item) for item in searchresult]
83        except RadioBrowserError as err:
84            self.logger.warning("RadioBrowser search failed for query '%s': %s", search_query, err)
85
86        return result
87
88    async def browse(self, path: str) -> Sequence[MediaItemType | BrowseFolder]:
89        """Browse this provider's items."""
90        path_parts = [] if "://" not in path else path.split("://")[1].split("/")
91
92        subpath = path_parts[0] if len(path_parts) > 0 else ""
93        subsubpath = path_parts[1] if len(path_parts) > 1 else ""
94        subsubsubpath = path_parts[2] if len(path_parts) > 2 else ""
95
96        if not subpath:
97            return [
98                BrowseFolder(
99                    item_id="popularity",
100                    provider=self.domain,
101                    path=path + "popularity",
102                    name="",
103                    translation_key="radiobrowser_by_popularity",
104                ),
105                BrowseFolder(
106                    item_id="category",
107                    provider=self.domain,
108                    path=path + "category",
109                    name="",
110                    translation_key="radiobrowser_by_category",
111                ),
112            ]
113
114        if subpath == "popularity":
115            if not subsubpath:
116                return [
117                    BrowseFolder(
118                        item_id="popular",
119                        provider=self.domain,
120                        path=path + "/popular",
121                        name="",
122                        translation_key="radiobrowser_by_clicks",
123                    ),
124                    BrowseFolder(
125                        item_id="votes",
126                        provider=self.domain,
127                        path=path + "/votes",
128                        name="",
129                        translation_key="radiobrowser_by_votes",
130                    ),
131                ]
132
133            if subsubpath == "popular":
134                return await self.get_by_popularity()
135
136            if subsubpath == "votes":
137                return await self.get_by_votes()
138
139        if subpath == "category":
140            if not subsubpath:
141                return [
142                    BrowseFolder(
143                        item_id="country",
144                        provider=self.domain,
145                        path=path + "/country",
146                        name="",
147                        translation_key="radiobrowser_by_country",
148                    ),
149                    BrowseFolder(
150                        item_id="language",
151                        provider=self.domain,
152                        path=path + "/language",
153                        name="",
154                        translation_key="radiobrowser_by_language",
155                    ),
156                    BrowseFolder(
157                        item_id="tag",
158                        provider=self.domain,
159                        path=path + "/tag",
160                        name="",
161                        translation_key="radiobrowser_by_tag",
162                    ),
163                ]
164
165            if subsubpath == "country":
166                if subsubsubpath:
167                    return await self.get_by_country(subsubsubpath)
168                return await self.get_country_folders(path)
169
170            if subsubpath == "language":
171                if subsubsubpath:
172                    return await self.get_by_language(subsubsubpath)
173                return await self.get_language_folders(path)
174
175            if subsubpath == "tag":
176                if subsubsubpath:
177                    return await self.get_by_tag(subsubsubpath)
178                return await self.get_tag_folders(path)
179
180        return []
181
182    @use_cache(3600 * 6)  # Cache for 6 hours
183    async def get_by_popularity(self) -> Sequence[Radio]:
184        """Get radio stations by popularity."""
185        try:
186            stations = await self.radios.stations(
187                hide_broken=True,
188                limit=1000,
189                order=Order.CLICK_COUNT,
190                reverse=True,
191            )
192            return [await self._parse_radio(station) for station in stations]
193        except RadioBrowserError as err:
194            raise ProviderUnavailableError(f"Failed to fetch popular stations: {err}") from err
195
196    @use_cache(3600 * 6)  # Cache for 6 hours
197    async def get_by_votes(self) -> Sequence[Radio]:
198        """Get radio stations by votes."""
199        try:
200            stations = await self.radios.stations(
201                hide_broken=True,
202                limit=1000,
203                order=Order.VOTES,
204                reverse=True,
205            )
206            return [await self._parse_radio(station) for station in stations]
207        except RadioBrowserError as err:
208            raise ProviderUnavailableError(f"Failed to fetch stations by votes: {err}") from err
209
210    @use_cache(3600 * 24 * 7)  # Cache for 7 days
211    async def get_country_folders(self, base_path: str) -> list[BrowseFolder]:
212        """Get a list of country names as BrowseFolder."""
213        try:
214            countries = await self.radios.countries(order=Order.NAME, hide_broken=True, limit=1000)
215        except RadioBrowserError as err:
216            raise ProviderUnavailableError(f"Failed to fetch countries: {err}") from err
217
218        items: list[BrowseFolder] = []
219        for country in countries:
220            folder = BrowseFolder(
221                item_id=country.code.lower(),
222                provider=self.domain,
223                path=base_path + "/" + country.code.lower(),
224                name=country.name,
225            )
226            if country.favicon and country.favicon.strip():
227                folder.image = MediaItemImage(
228                    type=ImageType.THUMB,
229                    path=country.favicon,
230                    provider=self.instance_id,
231                    remotely_accessible=True,
232                )
233            items.append(folder)
234        return items
235
236    @use_cache(3600 * 24 * 7)  # Cache for 7 days
237    async def get_language_folders(self, base_path: str) -> list[BrowseFolder]:
238        """Get a list of language names as BrowseFolder."""
239        try:
240            languages = await self.radios.languages(
241                order=Order.STATION_COUNT, reverse=True, hide_broken=True, limit=1000
242            )
243        except RadioBrowserError as err:
244            raise ProviderUnavailableError(f"Failed to fetch languages: {err}") from err
245
246        return [
247            BrowseFolder(
248                item_id=language.name,
249                provider=self.domain,
250                path=base_path + "/" + language.name,
251                name=language.name,
252            )
253            for language in languages
254        ]
255
256    @use_cache(3600 * 24 * 7)  # Cache for 7 days
257    async def get_tag_folders(self, base_path: str) -> list[BrowseFolder]:
258        """Get a list of tag names as BrowseFolder."""
259        try:
260            tags = await self.radios.tags(
261                hide_broken=True,
262                order=Order.STATION_COUNT,
263                reverse=True,
264                limit=100,
265            )
266        except RadioBrowserError as err:
267            raise ProviderUnavailableError(f"Failed to fetch tags: {err}") from err
268
269        tags.sort(key=lambda tag: tag.name)
270        return [
271            BrowseFolder(
272                item_id=tag.name,
273                provider=self.domain,
274                path=base_path + "/" + tag.name,
275                name=tag.name.title(),
276            )
277            for tag in tags
278        ]
279
280    @use_cache(3600 * 24)  # Cache for 1 day
281    async def get_by_country(self, country_code: str) -> list[Radio]:
282        """Get radio stations by country."""
283        try:
284            stations = await self.radios.stations(
285                filter_by=FilterBy.COUNTRY_CODE_EXACT,
286                filter_term=country_code,
287                hide_broken=True,
288                limit=1000,
289                order=Order.CLICK_COUNT,
290                reverse=True,
291            )
292            return [await self._parse_radio(station) for station in stations]
293        except RadioBrowserError as err:
294            raise ProviderUnavailableError(
295                f"Failed to fetch stations for country {country_code}: {err}"
296            ) from err
297
298    @use_cache(3600 * 24)  # Cache for 1 day
299    async def get_by_language(self, language: str) -> list[Radio]:
300        """Get radio stations by language."""
301        try:
302            stations = await self.radios.stations(
303                filter_by=FilterBy.LANGUAGE_EXACT,
304                filter_term=language,
305                hide_broken=True,
306                limit=1000,
307                order=Order.CLICK_COUNT,
308                reverse=True,
309            )
310            return [await self._parse_radio(station) for station in stations]
311        except RadioBrowserError as err:
312            raise ProviderUnavailableError(
313                f"Failed to fetch stations for language {language}: {err}"
314            ) from err
315
316    @use_cache(3600 * 24)  # Cache for 1 day
317    async def get_by_tag(self, tag: str) -> list[Radio]:
318        """Get radio stations by tag."""
319        try:
320            stations = await self.radios.stations(
321                filter_by=FilterBy.TAG_EXACT,
322                filter_term=tag,
323                hide_broken=True,
324                limit=1000,
325                order=Order.CLICK_COUNT,
326                reverse=True,
327            )
328            return [await self._parse_radio(station) for station in stations]
329        except RadioBrowserError as err:
330            raise ProviderUnavailableError(
331                f"Failed to fetch stations for tag {tag}: {err}"
332            ) from err
333
334    @use_cache(3600 * 24 * 14)  # Cache for 14 days
335    async def get_radio(self, prov_radio_id: str) -> Radio:
336        """Get radio station details."""
337        try:
338            radio = await self.radios.station(uuid=prov_radio_id)
339            if not radio:
340                raise MediaNotFoundError(f"Radio station {prov_radio_id} not found")
341            return await self._parse_radio(radio)
342        except RadioBrowserError as err:
343            raise ProviderUnavailableError(
344                f"Failed to fetch radio station {prov_radio_id}: {err}"
345            ) from err
346
347    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
348        """Get streamdetails for a radio station."""
349        try:
350            stream = await self.radios.station(uuid=item_id)
351            if not stream:
352                raise MediaNotFoundError(f"Radio station {item_id} not found")
353
354            await self.radios.station_click(uuid=item_id)
355
356            stream_url = stream.url_resolved or stream.url
357            if not stream_url:
358                raise MediaNotFoundError(f"Radio station {item_id} has no stream URL")
359
360            return StreamDetails(
361                provider=self.domain,
362                item_id=item_id,
363                audio_format=AudioFormat(
364                    content_type=ContentType.try_parse(stream.codec),
365                    bit_rate=stream.bitrate or None,
366                ),
367                media_type=MediaType.RADIO,
368                stream_type=StreamType.HTTP,
369                path=stream_url,
370                can_seek=False,
371                allow_seek=False,
372            )
373        except RadioBrowserError as err:
374            raise ProviderUnavailableError(
375                f"Failed to get stream details for {item_id}: {err}"
376            ) from err
377
378    async def _parse_radio(self, radio_obj: Station) -> Radio:
379        """Parse Radio object from json obj returned from api."""
380        radio = Radio(
381            item_id=radio_obj.uuid,
382            provider=self.domain,
383            name=radio_obj.name,
384            provider_mappings={
385                ProviderMapping(
386                    item_id=radio_obj.uuid,
387                    provider_domain=self.domain,
388                    provider_instance=self.instance_id,
389                )
390            },
391        )
392        radio.metadata.popularity = radio_obj.click_count
393        if radio_obj.homepage:
394            radio.metadata.links = {MediaItemLink(type=LinkType.WEBSITE, url=radio_obj.homepage)}
395        if radio_obj.favicon:
396            radio.metadata.images = UniqueList(
397                [
398                    MediaItemImage(
399                        type=ImageType.THUMB,
400                        path=radio_obj.favicon,
401                        provider=self.instance_id,
402                        remotely_accessible=True,
403                    )
404                ]
405            )
406        return radio
407