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