/
/
/
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 @property
63 def max_concurrent_streams(self) -> None:
64 """Allow unlimited concurrent upstream source streams."""
65 return None
66
67 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
68 """Return Config entries to configure this provider (none required)."""
69 return ()
70
71 async def browse(self, path: str) -> Sequence[MediaItemType | BrowseFolder]:
72 """List all Sveriges Radio stations."""
73 if (subpath := path.split("://", 1)[1] if "://" in path else "") != "":
74 msg = f"Invalid subpath: {subpath}"
75 raise KeyError(msg)
76 return [self._parse_radio(channel) for channel in await self._get_channels()]
77
78 async def get_radio(self, prov_radio_id: str) -> Radio:
79 """Get full radio details by id."""
80 if channel := await self._get_channel(prov_radio_id):
81 return self._parse_radio(channel)
82 raise MediaNotFoundError(f"Radio station {prov_radio_id} not found")
83
84 async def search(
85 self,
86 search_query: str,
87 media_types: list[MediaType],
88 limit: int = 5,
89 ) -> SearchResults:
90 """Search Sveriges Radio channels by name."""
91 if media_types and MediaType.RADIO not in media_types:
92 return SearchResults()
93 query = search_query.lower()
94 radios = [
95 self._parse_radio(channel)
96 for channel in await self._get_channels()
97 if query in (channel.get("name") or "").lower()
98 ][:limit]
99 return SearchResults(radio=radios)
100
101 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
102 """Get stream details for a radio station."""
103 # the live audio URL is already part of the cached channel list, so no extra request
104 channel = await self._get_channel(item_id)
105 url = (channel.get("liveaudio") or {}).get("url") if channel else None
106 if not url:
107 raise MediaNotFoundError(f"Radio station {item_id} has no live audio URL")
108 return StreamDetails(
109 provider=self.domain,
110 item_id=item_id,
111 media_type=media_type,
112 stream_type=StreamType.HTTP,
113 path=url,
114 audio_format=AudioFormat(content_type=ContentType.MP3),
115 can_seek=False,
116 allow_seek=False,
117 )
118
119 @use_cache(3600 * 24) # Cache for 1 day
120 async def _get_channels(self) -> list[dict[str, Any]]:
121 """Fetch the full list of Sveriges Radio channels."""
122 params = {"format": "json", "size": "500"}
123 try:
124 async with self.mass.http_session.get(
125 f"{API_BASE}/channels", params=params, timeout=HTTP_TIMEOUT
126 ) as resp:
127 resp.raise_for_status()
128 data = await resp.json()
129 except (aiohttp.ClientError, TimeoutError, ValueError) as err:
130 raise ProviderUnavailableError("Sveriges Radio API unavailable") from err
131 channels: list[dict[str, Any]] = data.get("channels", [])
132 return channels
133
134 async def _get_channel(self, channel_id: str) -> dict[str, Any] | None:
135 """Return the cached channel payload for an id, or None if unknown."""
136 return next(
137 (channel for channel in await self._get_channels() if str(channel["id"]) == channel_id),
138 None,
139 )
140
141 def _parse_radio(self, channel: dict[str, Any]) -> Radio:
142 """Build a Radio object from an SR channel payload."""
143 channel_id = str(channel["id"])
144 radio = Radio(
145 name=channel.get("name") or channel.get("channeltype") or f"SR {channel_id}",
146 item_id=channel_id,
147 provider=self.domain,
148 provider_mappings={
149 ProviderMapping(
150 item_id=channel_id,
151 provider_domain=self.domain,
152 provider_instance=self.instance_id,
153 )
154 },
155 )
156 if image := channel.get("image"):
157 radio.metadata.add_image(
158 MediaItemImage(
159 type=ImageType.THUMB,
160 path=image,
161 provider=self.domain,
162 remotely_accessible=True,
163 )
164 )
165 return radio
166