music-assistant-server

1"""Radio Paradise Music Provider for Music Assistant."""
2
3from __future__ import annotations
4
5from collections.abc import Sequence
6from typing import TYPE_CHECKING, Any
7
8import aiohttp
9from music_assistant_models.enums import MediaType, StreamType
10from music_assistant_models.errors import MediaNotFoundError, UnplayableMediaError
11from music_assistant_models.media_items import (
12    AudioFormat,
13    BrowseFolder,
14    ItemMapping,
15    MediaItemType,
16    Radio,
17    SearchResults,
18)
19from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
20
21from music_assistant.controllers.streams.constants import (
22    STREAMDETAILS_INBAND_TITLE_HANDOFF_KEY,
23    STREAMDETAILS_INBAND_TITLE_KEY,
24)
25from music_assistant.models.music_provider import MusicProvider
26
27from . import parsers
28from .constants import (
29    API_TIMEOUT,
30    NOWPLAYING_API_URL,
31    PLAY_API_URL,
32    RADIO_PARADISE_CHANNELS,
33    STREAM_METADATA_UPDATE_INTERVAL,
34)
35from .helpers import (
36    find_current_song,
37    find_song_by_stream_title,
38    get_current_block_position,
39    get_next_song,
40)
41
42if TYPE_CHECKING:
43    from music_assistant_models.config_entries import ConfigEntry
44
45
46class RadioParadiseProvider(MusicProvider):
47    """Radio Paradise Music Provider for Music Assistant."""
48
49    @property
50    def max_concurrent_streams(self) -> None:
51        """Allow unlimited concurrent upstream source streams."""
52        return None
53
54    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
55        """Return Config entries to setup this provider."""
56        # we (currently) do not have any config entries to set up
57        return ()
58
59    @property
60    def is_streaming_provider(self) -> bool:
61        """Return True if the provider is a streaming provider."""
62        return True
63
64    async def get_radio(self, prov_radio_id: str) -> Radio:
65        """Get full radio details by id."""
66        if prov_radio_id not in RADIO_PARADISE_CHANNELS:
67            raise MediaNotFoundError("Station not found")
68        return self._parse_radio(prov_radio_id)
69
70    async def search(
71        self,
72        search_query: str,
73        media_types: list[MediaType],
74        limit: int = 5,
75    ) -> SearchResults:
76        """Perform search on Radio Paradise channels."""
77        results = SearchResults()
78        if MediaType.RADIO not in media_types:
79            return results
80        search_query_lower = search_query.lower().strip()
81        if not search_query_lower:
82            return results
83        radios: list[Radio] = []
84        for channel_id, channel_info in RADIO_PARADISE_CHANNELS.items():
85            if search_query_lower in channel_info["name"].lower():
86                radios.append(self._parse_radio(channel_id))
87                if len(radios) >= limit:
88                    break
89        results.radio = radios
90        return results
91
92    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
93        """Get streamdetails for a radio station."""
94        if media_type != MediaType.RADIO:
95            raise UnplayableMediaError(f"Unsupported media type: {media_type}")
96        if item_id not in RADIO_PARADISE_CHANNELS:
97            raise MediaNotFoundError(f"Unknown radio channel: {item_id}")
98
99        channel_info = RADIO_PARADISE_CHANNELS[item_id]
100        stream_url = channel_info["stream_url"]
101        content_type = channel_info["content_type"]
102
103        stream_details = StreamDetails(
104            item_id=item_id,
105            provider=self.instance_id,
106            audio_format=AudioFormat(
107                content_type=content_type,
108                channels=2,
109            ),
110            media_type=MediaType.RADIO,
111            stream_type=StreamType.HTTP,
112            path=stream_url,
113            allow_seek=False,
114            can_seek=False,
115            duration=0,
116            stream_metadata_update_callback=self._update_stream_metadata,
117            stream_metadata_update_interval=STREAM_METADATA_UPDATE_INTERVAL,
118            data={STREAMDETAILS_INBAND_TITLE_HANDOFF_KEY: True},
119        )
120
121        # Set initial metadata if available so the first frame the listener sees
122        # is the live track rather than an empty banner.
123        metadata = await self._get_channel_metadata(item_id)
124        if metadata and metadata.get("current"):
125            stream_details.stream_metadata = parsers.build_stream_metadata(
126                metadata["current"], metadata
127            )
128            if metadata.get("block_data"):
129                # Seed the block cache consumed by _update_stream_metadata.
130                stream_details.data = {
131                    STREAMDETAILS_INBAND_TITLE_HANDOFF_KEY: True,
132                    "block_data": metadata["block_data"],
133                }
134
135        return stream_details
136
137    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
138        """Browse this provider's items."""
139        return [self._parse_radio(channel_id) for channel_id in RADIO_PARADISE_CHANNELS]
140
141    def _parse_radio(self, channel_id: str) -> Radio:
142        """Create a Radio object from cached channel information."""
143        return parsers.parse_radio(channel_id, self.instance_id, self.domain)
144
145    async def _fetch_json(self, url: str, channel_id: str) -> dict[str, Any] | None:
146        """
147        Fetch JSON from a Radio Paradise endpoint, returning None on any failure.
148
149        :param url: Fully-qualified API URL to GET.
150        :param channel_id: Channel id, used for log context.
151        """
152        try:
153            async with self.mass.http_session.get(url, timeout=API_TIMEOUT) as response:
154                if response.status != 200:
155                    self.logger.debug(
156                        "Radio Paradise API returned status %s for channel %s",
157                        response.status,
158                        channel_id,
159                    )
160                    return None
161                data: dict[str, Any] = await response.json()
162                return data or None
163        except aiohttp.ClientError as exc:
164            self.logger.debug(
165                "Radio Paradise API request failed for channel %s: %s", channel_id, exc
166            )
167            return None
168        except (KeyError, ValueError, TypeError) as exc:
169            self.logger.debug(
170                "Error parsing Radio Paradise API response for channel %s: %s", channel_id, exc
171            )
172            return None
173
174    async def _get_channel_metadata(self, channel_id: str) -> dict[str, Any] | None:
175        """
176        Get current track and upcoming tracks from Radio Paradise's API.
177
178        Tries the enriched play API first, falls back to simple now_playing API if it fails.
179
180        :param channel_id: Radio Paradise channel ID (0-5).
181        """
182        if channel_id not in RADIO_PARADISE_CHANNELS:
183            return None
184
185        result = await self._get_play_api_metadata(channel_id)
186        if result:
187            return result
188
189        self.logger.debug("Falling back to now_playing API for channel %s", channel_id)
190        return await self._get_nowplaying_api_metadata(channel_id)
191
192    async def _get_play_api_metadata(self, channel_id: str) -> dict[str, Any] | None:
193        """
194        Get metadata from the enriched play API with upcoming track info.
195
196        :param channel_id: Radio Paradise channel ID (0-5).
197        """
198        data = await self._fetch_json(f"{PLAY_API_URL}{channel_id}", channel_id)
199        if not data or "song" not in data:
200            return None
201
202        songs = data.get("song", {})
203        current_time_ms = get_current_block_position(data)
204        current_song = find_current_song(songs, current_time_ms)
205        if not current_song:
206            self.logger.debug("No current song found for channel %s", channel_id)
207            return None
208
209        return {
210            "current": current_song,
211            "next": get_next_song(songs, current_song),
212            "block_data": data,
213        }
214
215    async def _get_nowplaying_api_metadata(self, channel_id: str) -> dict[str, Any] | None:
216        """
217        Get metadata from the simple now_playing API (fallback).
218
219        :param channel_id: Radio Paradise channel ID (0-5).
220        """
221        data = await self._fetch_json(f"{NOWPLAYING_API_URL}{channel_id}", channel_id)
222        if not data:
223            return None
224        # now_playing returns flat song data; no next song or block data is available.
225        return {"current": data, "next": None, "block_data": None}
226
227    async def _match_icy_title(
228        self, channel_id: str, icy_title: str, data: dict[str, Any]
229    ) -> dict[str, Any] | None:
230        """
231        Resolve an in-band ICY title against (cached) play API block data.
232
233        :param channel_id: Radio Paradise channel ID (0-5).
234        :param icy_title: Cleaned in-band stream title ("Artist - Title" form).
235        :param data: StreamDetails scratch dict holding the cached block.
236        :returns: Metadata dict in the shape of _get_channel_metadata, or None
237            when the title cannot be resolved to a block song.
238        """
239        block = data.get("block_data")
240        if block and (song := find_song_by_stream_title(block.get("song", {}), icy_title)):
241            return {
242                "current": song,
243                "next": get_next_song(block["song"], song),
244                "block_data": block,
245            }
246        fresh = await self._fetch_json(f"{PLAY_API_URL}{channel_id}", channel_id)
247        if not fresh or "song" not in fresh:
248            return None
249        song = find_song_by_stream_title(fresh["song"], icy_title)
250        if song is None:
251            # The API served a stale or future block; keep the cached one.
252            self.logger.debug(
253                "Play API block for channel %s does not contain current title %r; discarding",
254                channel_id,
255                icy_title,
256            )
257            return None
258        data["block_data"] = fresh
259        return {
260            "current": song,
261            "next": get_next_song(fresh["song"], song),
262            "block_data": fresh,
263        }
264
265    async def _update_stream_metadata(
266        self, stream_details: StreamDetails, elapsed_time: int
267    ) -> None:
268        """
269        Update stream metadata callback called by player queue controller.
270
271        The in-band ICY title identifies what is actually playing; the play API
272        provides enrichment (cover art, album/year, upcoming songs) and is only
273        trusted when its block contains that title. Falls back to the
274        schedule-derived guess until the first in-band title arrives. Alternates
275        between showing the artist and upcoming track info every interval.
276
277        :param stream_details: StreamDetails object to update with metadata.
278        :param elapsed_time: Elapsed playback time in seconds (unused for Radio Paradise).
279        """
280        item_id = stream_details.item_id
281        if stream_details.data is None:
282            stream_details.data = {}
283        data = stream_details.data
284
285        icy_title = (data.get(STREAMDETAILS_INBAND_TITLE_KEY) or "").strip()
286        if icy_title:
287            metadata = await self._match_icy_title(item_id, icy_title, data)
288            if metadata is None:
289                # Station break/PSA or block data unavailable: show the title verbatim.
290                if data.get("last_verbatim_title") != icy_title:
291                    data["last_verbatim_title"] = icy_title
292                    data["last_event"] = None
293                    stream_details.stream_metadata = StreamMetadata(title=icy_title)
294                return
295            data.pop("last_verbatim_title", None)
296        else:
297            metadata = await self._get_channel_metadata(item_id)
298        if not metadata or not metadata.get("current"):
299            return
300
301        current_song = metadata["current"]
302        current_event = current_song.get("event", "")
303
304        # On track change, restart the artist/upcoming alternation from "artist".
305        if stream_details.data.get("last_event") != current_event:
306            stream_details.data["last_event"] = current_event
307            stream_details.data["show_upcoming"] = False
308
309        show_upcoming = stream_details.data.get("show_upcoming", False)
310        stream_metadata = parsers.build_stream_metadata(
311            current_song, metadata, show_upcoming=show_upcoming
312        )
313
314        self.logger.debug(
315            "Updating stream metadata for %s: %s - %s",
316            item_id,
317            stream_metadata.artist,
318            stream_metadata.title,
319        )
320        stream_details.stream_metadata = stream_metadata
321        stream_details.data["show_upcoming"] = not show_upcoming
322