/
/
/
1"""
2NTS Radio music provider for Music Assistant.
3
4Provides NTS Radio's two live channels and Infinite Mixtapes as
5browsable radio stations with live now-playing show metadata.
6"""
7
8from __future__ import annotations
9
10import html
11import re
12from collections.abc import Sequence
13from typing import TYPE_CHECKING, Any
14
15import aiohttp
16from music_assistant_models.enums import (
17 ContentType,
18 ImageType,
19 MediaType,
20 ProviderFeature,
21 StreamType,
22)
23from music_assistant_models.errors import (
24 MediaNotFoundError,
25 ProviderUnavailableError,
26)
27from music_assistant_models.media_items import (
28 AudioFormat,
29 BrowseFolder,
30 MediaItemImage,
31 MediaItemMetadata,
32 MediaItemType,
33 ProviderMapping,
34 Radio,
35)
36from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
37
38from music_assistant.controllers.cache import use_cache
39from music_assistant.models.music_provider import MusicProvider
40
41if TYPE_CHECKING:
42 from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
43 from music_assistant_models.provider import ProviderManifest
44
45 from music_assistant import MusicAssistant
46 from music_assistant.models import ProviderInstanceType
47
48SUPPORTED_FEATURES = {
49 ProviderFeature.BROWSE,
50}
51
52NTS_API_LIVE = "https://www.nts.live/api/v2/live"
53NTS_API_MIXTAPES = "https://www.nts.live/api/v2/mixtapes"
54
55NTS_LIVE_STREAMS = {
56 "1": "https://stream-relay-geo.ntslive.net/stream",
57 "2": "https://stream-relay-geo.ntslive.net/stream2",
58}
59
60CHANNEL_PREFIX = "nts_channel_"
61MIXTAPE_PREFIX = "nts_mixtape_"
62
63METADATA_REFRESH_INTERVAL = 60
64
65HTTP_TIMEOUT = aiohttp.ClientTimeout(total=10)
66
67# NTS source images are landscape; their CDN exposes /resize/ (preserves aspect)
68# and /crop/ (center-crop) endpoints. Rewriting picks the square variant so UIs
69# that expect square thumbnails don't get a letterboxed result.
70IMAGE_CROP_SIZE = 1000
71_NTS_IMAGE_OP_RE = re.compile(r"/(?:resize|crop)/\d+x\d+/")
72
73
74def _square_image_url(url: str | None) -> str | None:
75 """Rewrite an NTS image URL to a square center-crop."""
76 if not url:
77 return None
78 return _NTS_IMAGE_OP_RE.sub(f"/crop/{IMAGE_CROP_SIZE}x{IMAGE_CROP_SIZE}/", url, count=1)
79
80
81async def setup(
82 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
83) -> ProviderInstanceType:
84 """Initialize provider(instance) with given configuration."""
85 return NTSProvider(mass, manifest, config, SUPPORTED_FEATURES)
86
87
88class NTSProvider(MusicProvider):
89 """Provider implementation for NTS Radio."""
90
91 _mixtapes: dict[str, str]
92 _unknown_channels: set[str]
93
94 @property
95 def max_concurrent_streams(self) -> None:
96 """Allow unlimited concurrent upstream source streams."""
97 return None
98
99 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
100 """Return Config entries to setup this provider."""
101 return ()
102
103 @property
104 def is_streaming_provider(self) -> bool:
105 """Return True if the provider is a streaming provider."""
106 return True
107
108 async def handle_async_init(self) -> None:
109 """Handle async initialization of the provider."""
110 self._mixtapes = {}
111 self._unknown_channels = set()
112
113 # Live channels use static URLs; metadata enrichment is best-effort.
114 try:
115 await self._fetch_live_data()
116 except ProviderUnavailableError as err:
117 self.logger.debug("NTS live metadata unavailable at setup: %s", err)
118
119 # Mixtapes are best-effort too: a transient outage shouldn't take the
120 # whole provider offline (live channels still work). Will retry on demand.
121 try:
122 await self._refresh_mixtape_streams()
123 except ProviderUnavailableError as err:
124 self.logger.debug("NTS mixtapes unavailable at setup: %s", err)
125
126 async def browse(self, path: str) -> Sequence[MediaItemType | BrowseFolder]:
127 """Browse NTS radio stations."""
128 path_parts = [] if "://" not in path else path.split("://")[1].split("/")
129 subpath = path_parts[0] if path_parts else ""
130
131 if not subpath:
132 return [
133 BrowseFolder(
134 item_id="live",
135 provider=self.domain,
136 path=path + "live",
137 name="Live Channels",
138 translation_key="live_channels",
139 ),
140 BrowseFolder(
141 item_id="mixtapes",
142 provider=self.domain,
143 path=path + "mixtapes",
144 name="Infinite Mixtapes",
145 translation_key="infinite_mixtapes",
146 ),
147 ]
148
149 if subpath == "live":
150 return await self._get_live_channels()
151
152 if subpath == "mixtapes":
153 return await self._get_mixtapes()
154
155 return []
156
157 async def get_radio(self, prov_radio_id: str) -> Radio:
158 """Get full radio details by id."""
159 if prov_radio_id.startswith(CHANNEL_PREFIX):
160 channel_name = prov_radio_id.removeprefix(CHANNEL_PREFIX)
161 if channel_name in NTS_LIVE_STREAMS:
162 try:
163 live_data = await self._fetch_live_data()
164 except ProviderUnavailableError:
165 live_data = {}
166 api_channel = next(
167 (
168 ch
169 for ch in live_data.get("results", [])
170 if ch.get("channel_name") == channel_name
171 ),
172 None,
173 )
174 return self._build_channel_radio(channel_name, api_channel)
175 elif prov_radio_id.startswith(MIXTAPE_PREFIX):
176 alias = prov_radio_id.removeprefix(MIXTAPE_PREFIX)
177 try:
178 payload = await self._refresh_mixtape_streams()
179 except ProviderUnavailableError:
180 payload = {}
181 if alias in self._mixtapes:
182 mixtape = next(
183 (m for m in payload.get("results", []) if m.get("mixtape_alias") == alias),
184 None,
185 )
186 return self._build_mixtape_radio(alias, mixtape)
187 msg = f"NTS radio item {prov_radio_id} not found"
188 raise MediaNotFoundError(msg)
189
190 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
191 """Get stream details for an NTS radio station."""
192 stream_url = self._resolve_stream_url(item_id)
193 if not stream_url and item_id.startswith(MIXTAPE_PREFIX):
194 # mixtape map may be empty if the setup prefetch failed; retry on demand
195 try:
196 await self._refresh_mixtape_streams()
197 except ProviderUnavailableError as err:
198 self.logger.debug("NTS mixtape refresh failed: %s", err)
199 stream_url = self._resolve_stream_url(item_id)
200 if not stream_url:
201 msg = f"Could not resolve stream URL for {item_id}"
202 raise MediaNotFoundError(msg)
203
204 details = StreamDetails(
205 provider=self.instance_id,
206 item_id=item_id,
207 audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
208 media_type=MediaType.RADIO,
209 stream_type=StreamType.HTTP,
210 path=stream_url,
211 can_seek=False,
212 allow_seek=False,
213 )
214
215 if item_id.startswith(CHANNEL_PREFIX):
216 details.stream_metadata_update_callback = self._stream_metadata_callback
217 details.stream_metadata_update_interval = METADATA_REFRESH_INTERVAL
218 # populate initial metadata so the UI doesn't wait an interval
219 if (initial := await self._fetch_channel_stream_metadata(item_id)) is not None:
220 details.stream_metadata = initial
221
222 return details
223
224 # ------------------------------------------------------------------
225 # Internal helpers
226 # ------------------------------------------------------------------
227
228 async def _get_live_channels(self) -> list[Radio]:
229 """Build Radio objects for NTS live channels."""
230 try:
231 live_data = await self._fetch_live_data()
232 except ProviderUnavailableError as err:
233 self.logger.debug("NTS live metadata unavailable, returning bare channels: %s", err)
234 live_data = {}
235
236 api_channels: dict[str, dict[str, Any]] = {
237 ch.get("channel_name", ""): ch for ch in live_data.get("results", [])
238 }
239
240 for channel_name in api_channels:
241 if (
242 channel_name
243 and channel_name not in NTS_LIVE_STREAMS
244 and channel_name not in self._unknown_channels
245 ):
246 self.logger.warning(
247 "Unknown NTS channel %r â please report so it can be added",
248 channel_name,
249 )
250 self._unknown_channels.add(channel_name)
251
252 return [
253 self._build_channel_radio(name, api_channels.get(name)) for name in NTS_LIVE_STREAMS
254 ]
255
256 def _build_channel_radio(self, channel_name: str, api_channel: dict[str, Any] | None) -> Radio:
257 """Build a Radio for a static live channel, enriched with API metadata if available."""
258 description_text = ""
259 image_url: str | None = None
260 if api_channel:
261 title, location, description, image_url = self._extract_channel_info(api_channel)
262 desc_parts = [f"Now playing: {title}"]
263 if location:
264 desc_parts.append(f"Broadcasting from {location}")
265 if description:
266 desc_parts.append(description)
267 description_text = "\n".join(desc_parts)
268 return self._build_radio(
269 item_id=f"{CHANNEL_PREFIX}{channel_name}",
270 name=f"NTS {channel_name}",
271 description=description_text,
272 image_url=image_url,
273 )
274
275 @staticmethod
276 def _extract_channel_info(channel: dict[str, Any]) -> tuple[str, str, str, str | None]:
277 """Extract (title, location, description, image_url) from a live channel payload."""
278 channel_name = channel.get("channel_name", "")
279 now = channel.get("now", {})
280 details = now.get("embeds", {}).get("details", {})
281 media = details.get("media", {})
282 title = html.unescape(now.get("broadcast_title", f"NTS {channel_name}"))
283 location = details.get("location_long", "")
284 description = details.get("description", "")
285 image_url = _square_image_url(media.get("picture_large") or media.get("background_large"))
286 return title, location, description, image_url
287
288 async def _refresh_mixtape_streams(self) -> dict[str, Any]:
289 """Fetch the mixtapes payload and refresh the stream URL map. Returns the payload."""
290 payload = await self._fetch_mixtapes_data()
291 self._mixtapes = {
292 alias: endpoint
293 for mixtape in payload.get("results", [])
294 if (alias := mixtape.get("mixtape_alias"))
295 and (endpoint := mixtape.get("audio_stream_endpoint"))
296 }
297 return payload
298
299 async def _get_mixtapes(self) -> list[Radio]:
300 """Build Radio objects for all Infinite Mixtapes."""
301 mixtapes_data = await self._refresh_mixtape_streams()
302 radios: list[Radio] = []
303
304 for mixtape in mixtapes_data.get("results", []):
305 alias = mixtape.get("mixtape_alias", "")
306 if not alias or not mixtape.get("audio_stream_endpoint"):
307 continue
308 radios.append(self._build_mixtape_radio(alias, mixtape))
309
310 return radios
311
312 def _build_mixtape_radio(self, alias: str, mixtape: dict[str, Any] | None) -> Radio:
313 """Build a Radio for a mixtape, enriched with API metadata if available."""
314 if mixtape:
315 title = mixtape.get("title", alias)
316 subtitle = mixtape.get("subtitle", "")
317 description = mixtape.get("description", "")
318 return self._build_radio(
319 item_id=f"{MIXTAPE_PREFIX}{alias}",
320 name=f"NTS: {title}",
321 description=f"{subtitle}\n\n{description}" if subtitle else description,
322 image_url=_square_image_url(mixtape.get("media", {}).get("picture_large")),
323 )
324 return self._build_radio(
325 item_id=f"{MIXTAPE_PREFIX}{alias}",
326 name=f"NTS: {alias}",
327 description="",
328 image_url=None,
329 )
330
331 @use_cache(3600)
332 async def _fetch_mixtapes_data(self) -> dict[str, Any]:
333 """Fetch raw Infinite Mixtapes data from the NTS API (cached 1h)."""
334 try:
335 async with self.mass.http_session.get(NTS_API_MIXTAPES, timeout=HTTP_TIMEOUT) as resp:
336 resp.raise_for_status()
337 data: dict[str, Any] = await resp.json()
338 return data
339 except (aiohttp.ClientError, TimeoutError, ValueError) as err:
340 msg = f"NTS API unavailable: {err}"
341 raise ProviderUnavailableError(msg) from err
342
343 def _build_radio(
344 self,
345 item_id: str,
346 name: str,
347 description: str,
348 image_url: str | None,
349 ) -> Radio:
350 """Build a Radio object with standard provider mappings."""
351 radio = Radio(
352 provider=self.instance_id,
353 item_id=item_id,
354 name=name,
355 metadata=MediaItemMetadata(description=description),
356 provider_mappings={
357 ProviderMapping(
358 provider_domain=self.domain,
359 provider_instance=self.instance_id,
360 item_id=item_id,
361 available=True,
362 )
363 },
364 )
365 if image_url:
366 radio.metadata.add_image(
367 MediaItemImage(
368 type=ImageType.THUMB,
369 path=image_url,
370 provider=self.instance_id,
371 remotely_accessible=True,
372 )
373 )
374 return radio
375
376 def _resolve_stream_url(self, item_id: str) -> str | None:
377 """Resolve the stream URL for a given item ID."""
378 if item_id.startswith(CHANNEL_PREFIX):
379 return NTS_LIVE_STREAMS.get(item_id.removeprefix(CHANNEL_PREFIX))
380 if item_id.startswith(MIXTAPE_PREFIX):
381 return self._mixtapes.get(item_id.removeprefix(MIXTAPE_PREFIX))
382 return None
383
384 async def _stream_metadata_callback(self, stream_details: StreamDetails, _elapsed: int) -> None:
385 """Refresh stream metadata during playback (invoked by MA)."""
386 if (
387 metadata := await self._fetch_channel_stream_metadata(stream_details.item_id)
388 ) is not None:
389 stream_details.stream_metadata = metadata
390
391 async def _fetch_channel_stream_metadata(self, item_id: str) -> StreamMetadata | None:
392 """Fetch live data and build StreamMetadata for the given channel item_id."""
393 try:
394 live_data = await self._fetch_live_data()
395 except ProviderUnavailableError as err:
396 self.logger.debug("NTS live data fetch failed: %s", err)
397 return None
398 for channel in live_data.get("results", []):
399 if f"{CHANNEL_PREFIX}{channel.get('channel_name', '')}" == item_id:
400 return self._build_stream_metadata(channel)
401 return None
402
403 @classmethod
404 def _build_stream_metadata(cls, channel: dict[str, Any]) -> StreamMetadata:
405 """Build StreamMetadata from a live channel payload."""
406 title, location, description, image_url = cls._extract_channel_info(channel)
407 desc_parts = []
408 if location:
409 desc_parts.append(f"Broadcasting from {location}")
410 if description:
411 desc_parts.append(description)
412 return StreamMetadata(
413 title=title,
414 description="\n".join(desc_parts),
415 image_url=image_url,
416 )
417
418 @use_cache(METADATA_REFRESH_INTERVAL)
419 async def _fetch_live_data(self) -> dict[str, Any]:
420 """Fetch current live broadcast data from the NTS API."""
421 try:
422 async with self.mass.http_session.get(NTS_API_LIVE, timeout=HTTP_TIMEOUT) as resp:
423 resp.raise_for_status()
424 data: dict[str, Any] = await resp.json()
425 return data
426 except (aiohttp.ClientError, TimeoutError, ValueError) as err:
427 msg = f"NTS API unavailable: {err}"
428 raise ProviderUnavailableError(msg) from err
429