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