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