/
/
/
1"""Radio Paradise Music Provider for Music Assistant."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator, Sequence
6from typing import 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)
18from music_assistant_models.streamdetails import StreamDetails
19
20from music_assistant.controllers.cache import use_cache
21from music_assistant.models.music_provider import MusicProvider
22
23from . import parsers
24from .constants import NOWPLAYING_API_URL, PLAY_API_URL, RADIO_PARADISE_CHANNELS
25from .helpers import find_current_song, get_current_block_position, get_next_song
26
27
28class RadioParadiseProvider(MusicProvider):
29 """Radio Paradise Music Provider for Music Assistant."""
30
31 @property
32 def is_streaming_provider(self) -> bool:
33 """Return True if the provider is a streaming provider."""
34 return True
35
36 async def get_library_radios(self) -> AsyncGenerator[Radio, None]:
37 """Retrieve library/subscribed radio stations from the provider."""
38 for channel_id in RADIO_PARADISE_CHANNELS:
39 yield self._parse_radio(channel_id)
40
41 @use_cache(3600 * 3) # Cache for 3 hours
42 async def get_radio(self, prov_radio_id: str) -> Radio:
43 """Get full radio details by id."""
44 if prov_radio_id not in RADIO_PARADISE_CHANNELS:
45 raise MediaNotFoundError("Station not found")
46
47 return self._parse_radio(prov_radio_id)
48
49 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
50 """Get streamdetails for a radio station."""
51 if media_type != MediaType.RADIO:
52 raise UnplayableMediaError(f"Unsupported media type: {media_type}")
53 if item_id not in RADIO_PARADISE_CHANNELS:
54 raise MediaNotFoundError(f"Unknown radio channel: {item_id}")
55
56 # Get stream URL from channel configuration
57 channel_info = RADIO_PARADISE_CHANNELS[item_id]
58 stream_url = channel_info.get("stream_url")
59 if not stream_url:
60 raise UnplayableMediaError(f"No stream URL found for channel {item_id}")
61
62 # Get content type from channel configuration
63 channel_info = RADIO_PARADISE_CHANNELS[item_id]
64 content_type = channel_info["content_type"]
65
66 stream_details = StreamDetails(
67 item_id=item_id,
68 provider=self.instance_id,
69 audio_format=AudioFormat(
70 content_type=content_type,
71 channels=2,
72 ),
73 media_type=MediaType.RADIO,
74 stream_type=StreamType.HTTP,
75 path=stream_url,
76 allow_seek=False,
77 can_seek=False,
78 duration=0,
79 stream_metadata_update_callback=self._update_stream_metadata,
80 stream_metadata_update_interval=10, # Check every 10 seconds
81 )
82
83 # Set initial metadata if available
84 metadata = await self._get_channel_metadata(item_id)
85 if metadata and metadata.get("current"):
86 current_song = metadata["current"]
87 stream_details.stream_metadata = parsers.build_stream_metadata(current_song, metadata)
88
89 return stream_details
90
91 async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
92 """Browse this provider's items."""
93 return [self._parse_radio(channel_id) for channel_id in RADIO_PARADISE_CHANNELS]
94
95 def _parse_radio(self, channel_id: str) -> Radio:
96 """Create a Radio object from cached channel information."""
97 return parsers.parse_radio(channel_id, self.instance_id, self.domain)
98
99 async def _get_channel_metadata(self, channel_id: str) -> dict[str, Any] | None:
100 """Get current track and upcoming tracks from Radio Paradise's API.
101
102 Tries the enriched play API first, falls back to simple now_playing API if it fails.
103
104 :param channel_id: Radio Paradise channel ID (0-5).
105 """
106 if channel_id not in RADIO_PARADISE_CHANNELS:
107 return None
108
109 # Try enriched play API first
110 result = await self._get_play_api_metadata(channel_id)
111 if result:
112 return result
113
114 # Fallback to simple now_playing API
115 self.logger.debug(f"Falling back to now_playing API for channel {channel_id}")
116 return await self._get_nowplaying_api_metadata(channel_id)
117
118 async def _get_play_api_metadata(self, channel_id: str) -> dict[str, Any] | None:
119 """Get metadata from the enriched play API with upcoming track info.
120
121 :param channel_id: Radio Paradise channel ID (0-5).
122 """
123 try:
124 api_url = f"{PLAY_API_URL}{channel_id}"
125 timeout = aiohttp.ClientTimeout(total=10)
126
127 async with self.mass.http_session.get(api_url, timeout=timeout) as response:
128 if response.status != 200:
129 self.logger.debug(f"Play API call failed with status {response.status}")
130 return None
131
132 data = await response.json()
133
134 if not data or "song" not in data:
135 self.logger.debug(f"No song data in play API response for channel {channel_id}")
136 return None
137
138 # Find currently playing song based on elapsed time
139 current_time_ms = get_current_block_position(data)
140 current_song = find_current_song(data.get("song", {}), current_time_ms)
141
142 if not current_song:
143 self.logger.debug(f"No current song found for channel {channel_id}")
144 return None
145
146 # Get next song
147 next_song = get_next_song(data.get("song", {}), current_song)
148
149 return {"current": current_song, "next": next_song, "block_data": data}
150
151 except aiohttp.ClientError as exc:
152 self.logger.debug(f"Play API request failed for channel {channel_id}: {exc}")
153 return None
154 except (KeyError, ValueError, TypeError) as exc:
155 self.logger.debug(f"Error parsing play API response for channel {channel_id}: {exc}")
156 return None
157
158 async def _get_nowplaying_api_metadata(self, channel_id: str) -> dict[str, Any] | None:
159 """Get metadata from the simple now_playing API (fallback).
160
161 :param channel_id: Radio Paradise channel ID (0-5).
162 """
163 try:
164 api_url = f"{NOWPLAYING_API_URL}{channel_id}"
165 timeout = aiohttp.ClientTimeout(total=10)
166
167 async with self.mass.http_session.get(api_url, timeout=timeout) as response:
168 if response.status != 200:
169 self.logger.debug(f"Now playing API failed with status {response.status}")
170 return None
171
172 data = await response.json()
173
174 if not data:
175 self.logger.debug(f"No data from now_playing API for channel {channel_id}")
176 return None
177
178 # now_playing API returns flat song data, no next song or block data
179 return {"current": data, "next": None, "block_data": None}
180
181 except aiohttp.ClientError as exc:
182 self.logger.debug(f"Now playing API request failed for channel {channel_id}: {exc}")
183 return None
184 except (KeyError, ValueError, TypeError) as exc:
185 self.logger.debug(f"Error parsing now_playing response for channel {channel_id}: {exc}")
186 return None
187
188 async def _update_stream_metadata(
189 self, stream_details: StreamDetails, elapsed_time: int
190 ) -> None:
191 """Update stream metadata callback called by player queue controller.
192
193 Fetches current track info from Radio Paradise's API and updates
194 StreamDetails with track metadata. Alternates between showing the artist
195 and upcoming track info every 10 seconds.
196
197 :param stream_details: StreamDetails object to update with metadata.
198 :param elapsed_time: Elapsed playback time in seconds (unused for Radio Paradise).
199 """
200 item_id = stream_details.item_id
201
202 # Initialize data dict if needed
203 if stream_details.data is None:
204 stream_details.data = {}
205
206 try:
207 metadata = await self._get_channel_metadata(item_id)
208 if metadata and metadata.get("current"):
209 current_song = metadata["current"]
210 current_event = current_song.get("event", "")
211
212 # Track changed - reset to show artist first
213 if stream_details.data.get("last_event") != current_event:
214 stream_details.data["last_event"] = current_event
215 stream_details.data["show_upcoming"] = False
216
217 # Toggle between artist and upcoming info
218 show_upcoming = stream_details.data.get("show_upcoming", False)
219
220 # Create StreamMetadata object with full track info
221 stream_metadata = parsers.build_stream_metadata(
222 current_song, metadata, show_upcoming=show_upcoming
223 )
224
225 self.logger.debug(
226 f"Updating stream metadata for {item_id}: "
227 f"{stream_metadata.artist} - {stream_metadata.title}"
228 )
229 stream_details.stream_metadata = stream_metadata
230
231 # Toggle for next update
232 stream_details.data["show_upcoming"] = not show_upcoming
233
234 except aiohttp.ClientError as exc:
235 self.logger.debug(f"Network error updating metadata for {item_id}: {exc}")
236