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
20
21from music_assistant.models.music_provider import MusicProvider
22
23from . import parsers
24from .constants import (
25    API_TIMEOUT,
26    NOWPLAYING_API_URL,
27    PLAY_API_URL,
28    RADIO_PARADISE_CHANNELS,
29    STREAM_METADATA_UPDATE_INTERVAL,
30)
31from .helpers import find_current_song, get_current_block_position, get_next_song
32
33if TYPE_CHECKING:
34    from music_assistant_models.config_entries import ConfigEntry
35
36
37class RadioParadiseProvider(MusicProvider):
38    """Radio Paradise Music Provider for Music Assistant."""
39
40    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
41        """Return Config entries to setup this provider."""
42        # we (currently) do not have any config entries to set up
43        return ()
44
45    @property
46    def is_streaming_provider(self) -> bool:
47        """Return True if the provider is a streaming provider."""
48        return True
49
50    async def get_radio(self, prov_radio_id: str) -> Radio:
51        """Get full radio details by id."""
52        if prov_radio_id not in RADIO_PARADISE_CHANNELS:
53            raise MediaNotFoundError("Station not found")
54        return self._parse_radio(prov_radio_id)
55
56    async def search(
57        self,
58        search_query: str,
59        media_types: list[MediaType],
60        limit: int = 5,
61    ) -> SearchResults:
62        """Perform search on Radio Paradise channels."""
63        results = SearchResults()
64        if MediaType.RADIO not in media_types:
65            return results
66        search_query_lower = search_query.lower().strip()
67        if not search_query_lower:
68            return results
69        radios: list[Radio] = []
70        for channel_id, channel_info in RADIO_PARADISE_CHANNELS.items():
71            if search_query_lower in channel_info["name"].lower():
72                radios.append(self._parse_radio(channel_id))
73                if len(radios) >= limit:
74                    break
75        results.radio = radios
76        return results
77
78    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
79        """Get streamdetails for a radio station."""
80        if media_type != MediaType.RADIO:
81            raise UnplayableMediaError(f"Unsupported media type: {media_type}")
82        if item_id not in RADIO_PARADISE_CHANNELS:
83            raise MediaNotFoundError(f"Unknown radio channel: {item_id}")
84
85        channel_info = RADIO_PARADISE_CHANNELS[item_id]
86        stream_url = channel_info["stream_url"]
87        content_type = channel_info["content_type"]
88
89        stream_details = StreamDetails(
90            item_id=item_id,
91            provider=self.instance_id,
92            audio_format=AudioFormat(
93                content_type=content_type,
94                channels=2,
95            ),
96            media_type=MediaType.RADIO,
97            stream_type=StreamType.HTTP,
98            path=stream_url,
99            allow_seek=False,
100            can_seek=False,
101            duration=0,
102            stream_metadata_update_callback=self._update_stream_metadata,
103            stream_metadata_update_interval=STREAM_METADATA_UPDATE_INTERVAL,
104        )
105
106        # Set initial metadata if available so the first frame the listener sees
107        # is the live track rather than an empty banner.
108        metadata = await self._get_channel_metadata(item_id)
109        if metadata and metadata.get("current"):
110            stream_details.stream_metadata = parsers.build_stream_metadata(
111                metadata["current"], metadata
112            )
113
114        return stream_details
115
116    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
117        """Browse this provider's items."""
118        return [self._parse_radio(channel_id) for channel_id in RADIO_PARADISE_CHANNELS]
119
120    def _parse_radio(self, channel_id: str) -> Radio:
121        """Create a Radio object from cached channel information."""
122        return parsers.parse_radio(channel_id, self.instance_id, self.domain)
123
124    async def _fetch_json(self, url: str, channel_id: str) -> dict[str, Any] | None:
125        """
126        Fetch JSON from a Radio Paradise endpoint, returning None on any failure.
127
128        :param url: Fully-qualified API URL to GET.
129        :param channel_id: Channel id, used for log context.
130        """
131        try:
132            async with self.mass.http_session.get(url, timeout=API_TIMEOUT) as response:
133                if response.status != 200:
134                    self.logger.debug(
135                        "Radio Paradise API returned status %s for channel %s",
136                        response.status,
137                        channel_id,
138                    )
139                    return None
140                data: dict[str, Any] = await response.json()
141                return data or None
142        except aiohttp.ClientError as exc:
143            self.logger.debug(
144                "Radio Paradise API request failed for channel %s: %s", channel_id, exc
145            )
146            return None
147        except (KeyError, ValueError, TypeError) as exc:
148            self.logger.debug(
149                "Error parsing Radio Paradise API response for channel %s: %s", channel_id, exc
150            )
151            return None
152
153    async def _get_channel_metadata(self, channel_id: str) -> dict[str, Any] | None:
154        """
155        Get current track and upcoming tracks from Radio Paradise's API.
156
157        Tries the enriched play API first, falls back to simple now_playing API if it fails.
158
159        :param channel_id: Radio Paradise channel ID (0-5).
160        """
161        if channel_id not in RADIO_PARADISE_CHANNELS:
162            return None
163
164        result = await self._get_play_api_metadata(channel_id)
165        if result:
166            return result
167
168        self.logger.debug("Falling back to now_playing API for channel %s", channel_id)
169        return await self._get_nowplaying_api_metadata(channel_id)
170
171    async def _get_play_api_metadata(self, channel_id: str) -> dict[str, Any] | None:
172        """
173        Get metadata from the enriched play API with upcoming track info.
174
175        :param channel_id: Radio Paradise channel ID (0-5).
176        """
177        data = await self._fetch_json(f"{PLAY_API_URL}{channel_id}", channel_id)
178        if not data or "song" not in data:
179            return None
180
181        songs = data.get("song", {})
182        current_time_ms = get_current_block_position(data)
183        current_song = find_current_song(songs, current_time_ms)
184        if not current_song:
185            self.logger.debug("No current song found for channel %s", channel_id)
186            return None
187
188        return {
189            "current": current_song,
190            "next": get_next_song(songs, current_song),
191            "block_data": data,
192        }
193
194    async def _get_nowplaying_api_metadata(self, channel_id: str) -> dict[str, Any] | None:
195        """
196        Get metadata from the simple now_playing API (fallback).
197
198        :param channel_id: Radio Paradise channel ID (0-5).
199        """
200        data = await self._fetch_json(f"{NOWPLAYING_API_URL}{channel_id}", channel_id)
201        if not data:
202            return None
203        # now_playing returns flat song data; no next song or block data is available.
204        return {"current": data, "next": None, "block_data": None}
205
206    async def _update_stream_metadata(
207        self, stream_details: StreamDetails, elapsed_time: int
208    ) -> None:
209        """
210        Update stream metadata callback called by player queue controller.
211
212        Fetches current track info from Radio Paradise's API and updates
213        StreamDetails with track metadata. Alternates between showing the artist
214        and upcoming track info every interval.
215
216        :param stream_details: StreamDetails object to update with metadata.
217        :param elapsed_time: Elapsed playback time in seconds (unused for Radio Paradise).
218        """
219        item_id = stream_details.item_id
220        if stream_details.data is None:
221            stream_details.data = {}
222
223        metadata = await self._get_channel_metadata(item_id)
224        if not metadata or not metadata.get("current"):
225            return
226
227        current_song = metadata["current"]
228        current_event = current_song.get("event", "")
229
230        # On track change, restart the artist/upcoming alternation from "artist".
231        if stream_details.data.get("last_event") != current_event:
232            stream_details.data["last_event"] = current_event
233            stream_details.data["show_upcoming"] = False
234
235        show_upcoming = stream_details.data.get("show_upcoming", False)
236        stream_metadata = parsers.build_stream_metadata(
237            current_song, metadata, show_upcoming=show_upcoming
238        )
239
240        self.logger.debug(
241            "Updating stream metadata for %s: %s - %s",
242            item_id,
243            stream_metadata.artist,
244            stream_metadata.title,
245        )
246        stream_details.stream_metadata = stream_metadata
247        stream_details.data["show_upcoming"] = not show_upcoming
248