music-assistant-server

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    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
95        """Return Config entries to setup this provider."""
96        return ()
97
98    @property
99    def is_streaming_provider(self) -> bool:
100        """Return True if the provider is a streaming provider."""
101        return True
102
103    async def handle_async_init(self) -> None:
104        """Handle async initialization of the provider."""
105        self._mixtapes = {}
106        self._unknown_channels = set()
107
108        # Live channels use static URLs; metadata enrichment is best-effort.
109        try:
110            await self._fetch_live_data()
111        except ProviderUnavailableError as err:
112            self.logger.debug("NTS live metadata unavailable at setup: %s", err)
113
114        # Mixtapes are best-effort too: a transient outage shouldn't take the
115        # whole provider offline (live channels still work). Will retry on demand.
116        try:
117            await self._refresh_mixtape_streams()
118        except ProviderUnavailableError as err:
119            self.logger.debug("NTS mixtapes unavailable at setup: %s", err)
120
121    async def browse(self, path: str) -> Sequence[MediaItemType | BrowseFolder]:
122        """Browse NTS radio stations."""
123        path_parts = [] if "://" not in path else path.split("://")[1].split("/")
124        subpath = path_parts[0] if path_parts else ""
125
126        if not subpath:
127            return [
128                BrowseFolder(
129                    item_id="live",
130                    provider=self.domain,
131                    path=path + "live",
132                    name="Live Channels",
133                    translation_key="live_channels",
134                ),
135                BrowseFolder(
136                    item_id="mixtapes",
137                    provider=self.domain,
138                    path=path + "mixtapes",
139                    name="Infinite Mixtapes",
140                    translation_key="infinite_mixtapes",
141                ),
142            ]
143
144        if subpath == "live":
145            return await self._get_live_channels()
146
147        if subpath == "mixtapes":
148            return await self._get_mixtapes()
149
150        return []
151
152    async def get_radio(self, prov_radio_id: str) -> Radio:
153        """Get full radio details by id."""
154        if prov_radio_id.startswith(CHANNEL_PREFIX):
155            channel_name = prov_radio_id.removeprefix(CHANNEL_PREFIX)
156            if channel_name in NTS_LIVE_STREAMS:
157                try:
158                    live_data = await self._fetch_live_data()
159                except ProviderUnavailableError:
160                    live_data = {}
161                api_channel = next(
162                    (
163                        ch
164                        for ch in live_data.get("results", [])
165                        if ch.get("channel_name") == channel_name
166                    ),
167                    None,
168                )
169                return self._build_channel_radio(channel_name, api_channel)
170        elif prov_radio_id.startswith(MIXTAPE_PREFIX):
171            alias = prov_radio_id.removeprefix(MIXTAPE_PREFIX)
172            try:
173                payload = await self._refresh_mixtape_streams()
174            except ProviderUnavailableError:
175                payload = {}
176            if alias in self._mixtapes:
177                mixtape = next(
178                    (m for m in payload.get("results", []) if m.get("mixtape_alias") == alias),
179                    None,
180                )
181                return self._build_mixtape_radio(alias, mixtape)
182        msg = f"NTS radio item {prov_radio_id} not found"
183        raise MediaNotFoundError(msg)
184
185    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
186        """Get stream details for an NTS radio station."""
187        stream_url = self._resolve_stream_url(item_id)
188        if not stream_url and item_id.startswith(MIXTAPE_PREFIX):
189            # mixtape map may be empty if the setup prefetch failed; retry on demand
190            try:
191                await self._refresh_mixtape_streams()
192            except ProviderUnavailableError as err:
193                self.logger.debug("NTS mixtape refresh failed: %s", err)
194            stream_url = self._resolve_stream_url(item_id)
195        if not stream_url:
196            msg = f"Could not resolve stream URL for {item_id}"
197            raise MediaNotFoundError(msg)
198
199        details = StreamDetails(
200            provider=self.instance_id,
201            item_id=item_id,
202            audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
203            media_type=MediaType.RADIO,
204            stream_type=StreamType.HTTP,
205            path=stream_url,
206            can_seek=False,
207            allow_seek=False,
208        )
209
210        if item_id.startswith(CHANNEL_PREFIX):
211            details.stream_metadata_update_callback = self._stream_metadata_callback
212            details.stream_metadata_update_interval = METADATA_REFRESH_INTERVAL
213            # populate initial metadata so the UI doesn't wait an interval
214            if (initial := await self._fetch_channel_stream_metadata(item_id)) is not None:
215                details.stream_metadata = initial
216
217        return details
218
219    # ------------------------------------------------------------------
220    # Internal helpers
221    # ------------------------------------------------------------------
222
223    async def _get_live_channels(self) -> list[Radio]:
224        """Build Radio objects for NTS live channels."""
225        try:
226            live_data = await self._fetch_live_data()
227        except ProviderUnavailableError as err:
228            self.logger.debug("NTS live metadata unavailable, returning bare channels: %s", err)
229            live_data = {}
230
231        api_channels: dict[str, dict[str, Any]] = {
232            ch.get("channel_name", ""): ch for ch in live_data.get("results", [])
233        }
234
235        for channel_name in api_channels:
236            if (
237                channel_name
238                and channel_name not in NTS_LIVE_STREAMS
239                and channel_name not in self._unknown_channels
240            ):
241                self.logger.warning(
242                    "Unknown NTS channel %r — please report so it can be added",
243                    channel_name,
244                )
245                self._unknown_channels.add(channel_name)
246
247        return [
248            self._build_channel_radio(name, api_channels.get(name)) for name in NTS_LIVE_STREAMS
249        ]
250
251    def _build_channel_radio(self, channel_name: str, api_channel: dict[str, Any] | None) -> Radio:
252        """Build a Radio for a static live channel, enriched with API metadata if available."""
253        description_text = ""
254        image_url: str | None = None
255        if api_channel:
256            title, location, description, image_url = self._extract_channel_info(api_channel)
257            desc_parts = [f"Now playing: {title}"]
258            if location:
259                desc_parts.append(f"Broadcasting from {location}")
260            if description:
261                desc_parts.append(description)
262            description_text = "\n".join(desc_parts)
263        return self._build_radio(
264            item_id=f"{CHANNEL_PREFIX}{channel_name}",
265            name=f"NTS {channel_name}",
266            description=description_text,
267            image_url=image_url,
268        )
269
270    @staticmethod
271    def _extract_channel_info(channel: dict[str, Any]) -> tuple[str, str, str, str | None]:
272        """Extract (title, location, description, image_url) from a live channel payload."""
273        channel_name = channel.get("channel_name", "")
274        now = channel.get("now", {})
275        details = now.get("embeds", {}).get("details", {})
276        media = details.get("media", {})
277        title = html.unescape(now.get("broadcast_title", f"NTS {channel_name}"))
278        location = details.get("location_long", "")
279        description = details.get("description", "")
280        image_url = _square_image_url(media.get("picture_large") or media.get("background_large"))
281        return title, location, description, image_url
282
283    async def _refresh_mixtape_streams(self) -> dict[str, Any]:
284        """Fetch the mixtapes payload and refresh the stream URL map. Returns the payload."""
285        payload = await self._fetch_mixtapes_data()
286        self._mixtapes = {
287            alias: endpoint
288            for mixtape in payload.get("results", [])
289            if (alias := mixtape.get("mixtape_alias"))
290            and (endpoint := mixtape.get("audio_stream_endpoint"))
291        }
292        return payload
293
294    async def _get_mixtapes(self) -> list[Radio]:
295        """Build Radio objects for all Infinite Mixtapes."""
296        mixtapes_data = await self._refresh_mixtape_streams()
297        radios: list[Radio] = []
298
299        for mixtape in mixtapes_data.get("results", []):
300            alias = mixtape.get("mixtape_alias", "")
301            if not alias or not mixtape.get("audio_stream_endpoint"):
302                continue
303            radios.append(self._build_mixtape_radio(alias, mixtape))
304
305        return radios
306
307    def _build_mixtape_radio(self, alias: str, mixtape: dict[str, Any] | None) -> Radio:
308        """Build a Radio for a mixtape, enriched with API metadata if available."""
309        if mixtape:
310            title = mixtape.get("title", alias)
311            subtitle = mixtape.get("subtitle", "")
312            description = mixtape.get("description", "")
313            return self._build_radio(
314                item_id=f"{MIXTAPE_PREFIX}{alias}",
315                name=f"NTS: {title}",
316                description=f"{subtitle}\n\n{description}" if subtitle else description,
317                image_url=_square_image_url(mixtape.get("media", {}).get("picture_large")),
318            )
319        return self._build_radio(
320            item_id=f"{MIXTAPE_PREFIX}{alias}",
321            name=f"NTS: {alias}",
322            description="",
323            image_url=None,
324        )
325
326    @use_cache(3600)
327    async def _fetch_mixtapes_data(self) -> dict[str, Any]:
328        """Fetch raw Infinite Mixtapes data from the NTS API (cached 1h)."""
329        try:
330            async with self.mass.http_session.get(NTS_API_MIXTAPES, timeout=HTTP_TIMEOUT) as resp:
331                resp.raise_for_status()
332                data: dict[str, Any] = await resp.json()
333                return data
334        except (aiohttp.ClientError, TimeoutError, ValueError) as err:
335            msg = f"NTS API unavailable: {err}"
336            raise ProviderUnavailableError(msg) from err
337
338    def _build_radio(
339        self,
340        item_id: str,
341        name: str,
342        description: str,
343        image_url: str | None,
344    ) -> Radio:
345        """Build a Radio object with standard provider mappings."""
346        radio = Radio(
347            provider=self.instance_id,
348            item_id=item_id,
349            name=name,
350            metadata=MediaItemMetadata(description=description),
351            provider_mappings={
352                ProviderMapping(
353                    provider_domain=self.domain,
354                    provider_instance=self.instance_id,
355                    item_id=item_id,
356                    available=True,
357                )
358            },
359        )
360        if image_url:
361            radio.metadata.add_image(
362                MediaItemImage(
363                    type=ImageType.THUMB,
364                    path=image_url,
365                    provider=self.instance_id,
366                    remotely_accessible=True,
367                )
368            )
369        return radio
370
371    def _resolve_stream_url(self, item_id: str) -> str | None:
372        """Resolve the stream URL for a given item ID."""
373        if item_id.startswith(CHANNEL_PREFIX):
374            return NTS_LIVE_STREAMS.get(item_id.removeprefix(CHANNEL_PREFIX))
375        if item_id.startswith(MIXTAPE_PREFIX):
376            return self._mixtapes.get(item_id.removeprefix(MIXTAPE_PREFIX))
377        return None
378
379    async def _stream_metadata_callback(self, stream_details: StreamDetails, _elapsed: int) -> None:
380        """Refresh stream metadata during playback (invoked by MA)."""
381        if (
382            metadata := await self._fetch_channel_stream_metadata(stream_details.item_id)
383        ) is not None:
384            stream_details.stream_metadata = metadata
385
386    async def _fetch_channel_stream_metadata(self, item_id: str) -> StreamMetadata | None:
387        """Fetch live data and build StreamMetadata for the given channel item_id."""
388        try:
389            live_data = await self._fetch_live_data()
390        except ProviderUnavailableError as err:
391            self.logger.debug("NTS live data fetch failed: %s", err)
392            return None
393        for channel in live_data.get("results", []):
394            if f"{CHANNEL_PREFIX}{channel.get('channel_name', '')}" == item_id:
395                return self._build_stream_metadata(channel)
396        return None
397
398    @classmethod
399    def _build_stream_metadata(cls, channel: dict[str, Any]) -> StreamMetadata:
400        """Build StreamMetadata from a live channel payload."""
401        title, location, description, image_url = cls._extract_channel_info(channel)
402        desc_parts = []
403        if location:
404            desc_parts.append(f"Broadcasting from {location}")
405        if description:
406            desc_parts.append(description)
407        return StreamMetadata(
408            title=title,
409            description="\n".join(desc_parts),
410            image_url=image_url,
411        )
412
413    @use_cache(METADATA_REFRESH_INTERVAL)
414    async def _fetch_live_data(self) -> dict[str, Any]:
415        """Fetch current live broadcast data from the NTS API."""
416        try:
417            async with self.mass.http_session.get(NTS_API_LIVE, timeout=HTTP_TIMEOUT) as resp:
418                resp.raise_for_status()
419                data: dict[str, Any] = await resp.json()
420                return data
421        except (aiohttp.ClientError, TimeoutError, ValueError) as err:
422            msg = f"NTS API unavailable: {err}"
423            raise ProviderUnavailableError(msg) from err
424