/
/
/
1"""SomaFM Radio music provider support for MusicAssistant."""
2
3from __future__ import annotations
4
5import random
6from typing import TYPE_CHECKING, Any
7
8from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
9from music_assistant_models.enums import (
10 ConfigEntryType,
11 ContentType,
12 ImageType,
13 MediaType,
14 ProviderFeature,
15 StreamType,
16)
17from music_assistant_models.errors import MediaNotFoundError
18from music_assistant_models.media_items import (
19 AudioFormat,
20 BrowseFolder,
21 ItemMapping,
22 MediaItemImage,
23 MediaItemMetadata,
24 MediaItemType,
25 ProviderMapping,
26 Radio,
27 SearchResults,
28)
29from music_assistant_models.streamdetails import StreamDetails
30
31from music_assistant.controllers.cache import use_cache
32from music_assistant.helpers.playlists import PlaylistItem, fetch_playlist
33from music_assistant.models.music_provider import MusicProvider
34
35if TYPE_CHECKING:
36 from collections.abc import Sequence
37
38 from music_assistant_models.config_entries import ProviderConfig
39 from music_assistant_models.provider import ProviderManifest
40
41 from music_assistant import MusicAssistant
42 from music_assistant.models import ProviderInstanceType
43
44SUPPORTED_FEATURES = {
45 ProviderFeature.BROWSE,
46 ProviderFeature.SEARCH,
47}
48
49CONF_QUALITY = "quality"
50
51
52async def setup(
53 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
54) -> ProviderInstanceType:
55 """Initialize provider(instance) with given configuration."""
56 return SomaFMProvider(mass, manifest, config, SUPPORTED_FEATURES)
57
58
59class SomaFMProvider(MusicProvider):
60 """Provider implementation for SomaFM Radio."""
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."""
69 return (
70 ConfigEntry(
71 key=CONF_QUALITY,
72 advanced=True,
73 type=ConfigEntryType.STRING,
74 options=[
75 ConfigValueOption("highest"),
76 ConfigValueOption("high"),
77 ConfigValueOption("low"),
78 ],
79 default_value="highest",
80 ),
81 )
82
83 @property
84 def is_streaming_provider(self) -> bool:
85 """Return True if the provider is a streaming provider."""
86 return True
87
88 async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
89 """Browse this provider's radio stations."""
90 stations = await self._get_stations()
91 if stations:
92 return [self._parse_channel(channel_info) for channel_info in stations.values()]
93 return []
94
95 async def search(
96 self,
97 search_query: str,
98 media_types: list[MediaType],
99 limit: int = 5,
100 ) -> SearchResults:
101 """Perform search on SomaFM channels."""
102 results = SearchResults()
103 if MediaType.RADIO not in media_types:
104 return results
105 search_query_lower = search_query.lower().strip()
106 if not search_query_lower:
107 return results
108 stations = await self._get_stations()
109 if not stations:
110 return results
111 radios: list[Radio] = []
112 for channel_info in stations.values():
113 channel_name = str(channel_info.get("title", "")).lower()
114 if search_query_lower in channel_name:
115 radios.append(self._parse_channel(channel_info))
116 if len(radios) >= limit:
117 break
118 results.radio = radios
119 return results
120
121 async def get_radio(self, prov_radio_id: str) -> Radio:
122 """Get radio station details."""
123 stations = await self._get_stations() # May be cached
124 if stations:
125 radio = stations.get(prov_radio_id)
126 if radio:
127 return self._parse_channel(radio)
128 msg = f"Item {prov_radio_id} not found"
129 raise MediaNotFoundError(msg)
130
131 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
132 """Get stream details for a track/radio."""
133
134 async def _get_valid_playlist_item(playlist: list[PlaylistItem]) -> PlaylistItem:
135 """Randomly select stream URL from playlist and test it."""
136 random.shuffle(playlist)
137 for item in playlist:
138 async with self.mass.http_session.head(item.path, ssl=False) as response:
139 if response.status >= 100 and response.status < 300:
140 # Stream exists, return valid path
141 return item
142 self.logger.error("Could not find a working stream for playlist")
143 raise MediaNotFoundError("No valid SomaFM stream available")
144
145 def _get_playlist_url(station: dict[str, Any]) -> str:
146 """Pick playlist based on quality config value."""
147 req_quality = self.config.get_value(CONF_QUALITY)
148 playlists: list[dict[str, str]] = station.get("playlists", [])
149
150 # Remove MP3 playlist options for now; AAC is generally better
151 playlists = [
152 playlist for playlist in playlists if playlist["format"] in {"aac", "aacp"}
153 ]
154
155 # Sort by quality just in case they already aren't sorted highest/high/low
156 quality_map = {"highest": 0, "high": 1, "low": 2}
157 playlists.sort(key=lambda x: quality_map[x["quality"]])
158
159 # Detect empty playlist after sort and filter
160 if len(playlists) == 0:
161 raise MediaNotFoundError("No valid SomaFM playlist available")
162
163 # Find the first playlist item that has the requested quality
164 for playlist in playlists:
165 avail_quality = playlist.get("quality")
166 playlist_url = playlist.get("url")
167 if req_quality == avail_quality and playlist_url:
168 return playlist_url
169
170 self.logger.warning("Couldn't find SomaFM stream with requested quality and format")
171
172 # Get the first (highest quality) playlist if we couldn't find requested quality
173 playlist_url = playlists[0].get("url")
174 if playlist_url:
175 return playlist_url
176 raise MediaNotFoundError("No valid SomaFM playlist available")
177
178 async def _get_stream_path(item_id: str) -> str:
179 """Pick correct playlist, fetch the playlist, and extract stream URL."""
180 stations = await self._get_stations()
181 station = stations.get(item_id)
182 if station:
183 playlist_url = _get_playlist_url(station)
184 playlist = await fetch_playlist(self.mass, playlist_url)
185 playlist_item: PlaylistItem = await _get_valid_playlist_item(playlist)
186 return playlist_item.path
187 raise MediaNotFoundError
188
189 stream_path = await _get_stream_path(item_id)
190
191 return StreamDetails(
192 provider=self.instance_id,
193 item_id=item_id,
194 audio_format=AudioFormat(
195 content_type=ContentType.UNKNOWN,
196 ),
197 media_type=MediaType.RADIO,
198 path=stream_path,
199 stream_type=StreamType.HTTP,
200 allow_seek=False,
201 can_seek=False,
202 )
203
204 @use_cache(3600 * 24 * 1) # Cache for 1 day
205 async def _get_stations(self) -> dict[str, dict[str, Any]]:
206 url = "https://somafm.com/channels.json"
207 locale = self.mass.metadata.locale.replace("_", "-")
208 language = locale.split("-")[0]
209 headers = {"Accept-Language": f"{locale}, {language};q=0.9, *;q=0.5"}
210 async with (
211 self.mass.http_session.get(url, headers=headers, ssl=False) as response,
212 ):
213 result: Any = await response.json()
214 if not result or "error" in result:
215 self.logger.error(url)
216 elif isinstance(result, dict):
217 stations = result.get("channels")
218 if stations:
219 # Reformat into dict by channel id
220 return {info.get("id"): info for info in stations if info.get("id")}
221 raise MediaNotFoundError("Could not fetch SomaFM stations list")
222
223 def _parse_channel(self, channel_info: dict[str, Any]) -> Radio:
224 """Convert SomaFM channel info into a Radio object."""
225 # Construct radio station information
226 item_id = channel_info.get("id")
227 if not item_id:
228 raise MediaNotFoundError("Soma FM station generation failed")
229
230 radio = Radio(
231 provider=self.instance_id,
232 item_id=item_id,
233 name=f"SomaFM: {channel_info.get('title', 'Unknown Radio')}",
234 metadata=MediaItemMetadata(
235 description=channel_info.get("description", "No description"),
236 genres={channel_info.get("genre", "No genre")},
237 popularity=int(channel_info.get("listeners", "0")),
238 performers={
239 f"DJ: {channel_info.get('dj', 'No DJ info')}",
240 f"DJ Email: {channel_info.get('djmail', 'No DJ email')}",
241 },
242 ),
243 provider_mappings={
244 ProviderMapping(
245 provider_domain=self.domain,
246 provider_instance=self.instance_id,
247 item_id=item_id,
248 available=True,
249 )
250 },
251 )
252
253 # Add station image URL
254 station_icon_url = channel_info.get("largeimage")
255 if station_icon_url:
256 radio.metadata.add_image(
257 MediaItemImage(
258 provider=self.instance_id,
259 type=ImageType.THUMB,
260 path=station_icon_url,
261 remotely_accessible=True,
262 )
263 )
264 return radio
265