/
/
1"""Sveriges Radio music provider."""
2
3from __future__ import annotations
4
5from collections.abc import Sequence
6from typing import TYPE_CHECKING, Any, Final
7
8import aiohttp
9from music_assistant_models.enums import (
10 ContentType,
11 ImageType,
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 MediaItemType,
22 ProviderMapping,
23 Radio,
24 SearchResults,
25)
26from music_assistant_models.streamdetails import StreamDetails
27
28from music_assistant.controllers.cache import use_cache
29from music_assistant.models.music_provider import MusicProvider
30
31if TYPE_CHECKING:
32 from music_assistant_models.config_entries import (
33 ConfigEntry,
34 ProviderConfig,
35 )
36 from music_assistant_models.provider import ProviderManifest
37
38 from music_assistant.mass import MusicAssistant
39 from music_assistant.models import ProviderInstanceType
40
41API_BASE: Final = "https://api.sr.se/api/v2"
42
43# bound each request; the shared session's default timeout is too long for a metadata fetch
44HTTP_TIMEOUT: Final = aiohttp.ClientTimeout(total=10)
45
46SUPPORTED_FEATURES = {
47 ProviderFeature.BROWSE,
48 ProviderFeature.SEARCH,
49}
50
51
52async def setup(
53 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
54) -> ProviderInstanceType:
55 """Initialize instance."""
56 return SverigesRadio(mass, manifest, config, SUPPORTED_FEATURES)
57
58
59class SverigesRadio(MusicProvider):
60 """Sveriges Radio music provider."""
61
62 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
63 """Return Config entries to configure this provider (none required)."""
64 return ()
65
66 async def browse(self, path: str) -> Sequence[MediaItemType | BrowseFolder]:
67 """List all Sveriges Radio stations."""
68 if (subpath := path.split("://", 1)[1] if "://" in path else "") != "":
69 msg = f"Invalid subpath: {subpath}"
70 raise KeyError(msg)
71 return [self._parse_radio(channel) for channel in await self._get_channels()]
72
73 async def get_radio(self, prov_radio_id: str) -> Radio:
74 """Get full radio details by id."""
75 if channel := await self._get_channel(prov_radio_id):
76 return self._parse_radio(channel)
77 raise MediaNotFoundError(f"Radio station {prov_radio_id} not found")
78
79 async def search(
80 self,
81 search_query: str,
82 media_types: list[MediaType],
83 limit: int = 5,
84 ) -> SearchResults:
85 """Search Sveriges Radio channels by name."""
86 if media_types and MediaType.RADIO not in media_types:
87 return SearchResults()
88 query = search_query.lower()
89 radios = [
90 self._parse_radio(channel)
91 for channel in await self._get_channels()
92 if query in (channel.get("name") or "").lower()
93 ][:limit]
94 return SearchResults(radio=radios)
95
96 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
97 """Get stream details for a radio station."""
98 # the live audio URL is already part of the cached channel list, so no extra request
99 channel = await self._get_channel(item_id)
100 url = (channel.get("liveaudio") or {}).get("url") if channel else None
101 if not url:
102 raise MediaNotFoundError(f"Radio station {item_id} has no live audio URL")
103 return StreamDetails(
104 provider=self.domain,
105 item_id=item_id,
106 media_type=media_type,
107 stream_type=StreamType.HTTP,
108 path=url,
109 audio_format=AudioFormat(content_type=ContentType.MP3),
110 can_seek=False,
111 allow_seek=False,
112 )
113
114 @use_cache(3600 * 24) # Cache for 1 day
115 async def _get_channels(self) -> list[dict[str, Any]]:
116 """Fetch the full list of Sveriges Radio channels."""
117 params = {"format": "json", "size": "500"}
118 try:
119 async with self.mass.http_session.get(
120 f"{API_BASE}/channels", params=params, timeout=HTTP_TIMEOUT
121 ) as resp:
122 resp.raise_for_status()
123 data = await resp.json()
124 except (aiohttp.ClientError, TimeoutError, ValueError) as err:
125 raise ProviderUnavailableError("Sveriges Radio API unavailable") from err
126 channels: list[dict[str, Any]] = data.get("channels", [])
127 return channels
128
129 async def _get_channel(self, channel_id: str) -> dict[str, Any] | None:
130 """Return the cached channel payload for an id, or None if unknown."""
131 return next(
132 (channel for channel in await self._get_channels() if str(channel["id"]) == channel_id),
133 None,
134 )
135
136 def _parse_radio(self, channel: dict[str, Any]) -> Radio:
137 """Build a Radio object from an SR channel payload."""
138 channel_id = str(channel["id"])
139 radio = Radio(
140 name=channel.get("name") or channel.get("channeltype") or f"SR {channel_id}",
141 item_id=channel_id,
142 provider=self.domain,
143 provider_mappings={
144 ProviderMapping(
145 item_id=channel_id,
146 provider_domain=self.domain,
147 provider_instance=self.instance_id,
148 )
149 },
150 )
151 if image := channel.get("image"):
152 radio.metadata.add_image(
153 MediaItemImage(
154 type=ImageType.THUMB,
155 path=image,
156 provider=self.domain,
157 remotely_accessible=True,
158 )
159 )
160 return radio
161