/
/
/
1"""Radio Paradise Music Provider for Music Assistant."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7from collections.abc import AsyncGenerator, Sequence
8from typing import Any
9
10import aiohttp
11from music_assistant_models.enums import MediaType, StreamType
12from music_assistant_models.errors import MediaNotFoundError, UnplayableMediaError
13from music_assistant_models.media_items import (
14 AudioFormat,
15 BrowseFolder,
16 ItemMapping,
17 MediaItemType,
18 Radio,
19)
20from music_assistant_models.streamdetails import StreamDetails
21
22from music_assistant.controllers.cache import use_cache
23from music_assistant.models.music_provider import MusicProvider
24
25from . import parsers
26from .constants import RADIO_PARADISE_CHANNELS
27from .helpers import build_stream_url, find_current_song, get_current_block_position, get_next_song
28
29
30class RadioParadiseProvider(MusicProvider):
31 """Radio Paradise Music Provider for Music Assistant."""
32
33 @property
34 def is_streaming_provider(self) -> bool:
35 """Return True if the provider is a streaming provider."""
36 return True
37
38 async def get_library_radios(self) -> AsyncGenerator[Radio, None]:
39 """Retrieve library/subscribed radio stations from the provider."""
40 for channel_id in RADIO_PARADISE_CHANNELS:
41 yield self._parse_radio(channel_id)
42
43 @use_cache(3600 * 3) # Cache for 3 hours
44 async def get_radio(self, prov_radio_id: str) -> Radio:
45 """Get full radio details by id."""
46 if prov_radio_id not in RADIO_PARADISE_CHANNELS:
47 raise MediaNotFoundError("Station not found")
48
49 return self._parse_radio(prov_radio_id)
50
51 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
52 """Get streamdetails for a radio station."""
53 if media_type != MediaType.RADIO:
54 raise UnplayableMediaError(f"Unsupported media type: {media_type}")
55 if item_id not in RADIO_PARADISE_CHANNELS:
56 raise MediaNotFoundError(f"Unknown radio channel: {item_id}")
57
58 stream_url = build_stream_url(item_id)
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 )
80
81 # Set initial metadata if available
82 metadata = await self._get_channel_metadata(item_id)
83 if metadata and metadata.get("current"):
84 current_song = metadata["current"]
85 stream_details.stream_metadata = parsers.build_stream_metadata(current_song, metadata)
86
87 # Store the monitoring task in streamdetails.data for cleanup in on_streamed
88 monitor_task = self.mass.create_task(self._monitor_stream_metadata(stream_details))
89 stream_details.data = {"monitor_task": monitor_task}
90
91 return stream_details
92
93 async def on_streamed(self, streamdetails: StreamDetails) -> None:
94 """Handle callback when given streamdetails completed streaming."""
95 self.logger.debug(
96 f"Radio Paradise channel {streamdetails.item_id} streamed for "
97 f"{streamdetails.seconds_streamed} seconds"
98 )
99
100 # Cancel and clean up the monitoring task
101 if "monitor_task" in streamdetails.data:
102 monitor_task = streamdetails.data["monitor_task"]
103 if not monitor_task.done():
104 monitor_task.cancel()
105 with contextlib.suppress(asyncio.CancelledError):
106 await monitor_task
107 del streamdetails.data["monitor_task"]
108
109 async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
110 """Browse this provider's items."""
111 return [self._parse_radio(channel_id) for channel_id in RADIO_PARADISE_CHANNELS]
112
113 def _parse_radio(self, channel_id: str) -> Radio:
114 """Create a Radio object from cached channel information."""
115 return parsers.parse_radio(channel_id, self.instance_id, self.domain)
116
117 async def _get_channel_metadata(self, channel_id: str) -> dict[str, Any] | None:
118 """Get current track and upcoming tracks from Radio Paradise's block API.
119
120 Args:
121 channel_id: Radio Paradise channel ID (0-5)
122
123 Returns:
124 Dict with current song, next song, and block data, or None if API fails
125 """
126 if channel_id not in RADIO_PARADISE_CHANNELS:
127 return None
128
129 try:
130 # Use block API for much richer data
131 api_url = (
132 f"https://api.radioparadise.com/api/get_block?bitrate=4&info=true&chan={channel_id}"
133 )
134 timeout = aiohttp.ClientTimeout(total=10)
135
136 async with self.mass.http_session.get(api_url, timeout=timeout) as response:
137 if response.status != 200:
138 self.logger.debug(f"Block API call failed with status {response.status}")
139 return None
140
141 data = await response.json()
142
143 # Find currently playing song based on elapsed time
144 current_time_ms = get_current_block_position(data)
145 current_song = find_current_song(data.get("song", {}), current_time_ms)
146
147 if not current_song:
148 self.logger.debug(f"No current song found for channel {channel_id}")
149 return None
150
151 # Get next song
152 next_song = get_next_song(data.get("song", {}), current_song)
153
154 return {"current": current_song, "next": next_song, "block_data": data}
155
156 except aiohttp.ClientError as exc:
157 self.logger.debug(f"Failed to get block metadata for channel {channel_id}: {exc}")
158 return None
159 except Exception as exc:
160 self.logger.debug(
161 f"Unexpected error getting block metadata for channel {channel_id}: {exc}"
162 )
163 return None
164
165 async def _monitor_stream_metadata(self, stream_details: StreamDetails) -> None:
166 """Monitor and update stream metadata in real-time during playback.
167
168 Fetches current track info from Radio Paradise's API every 10 seconds
169 and updates StreamDetails with track metadata and upcoming songs.
170
171 Args:
172 stream_details: StreamDetails object to update with metadata
173 """
174 last_track_event = ""
175 item_id = stream_details.item_id
176
177 try:
178 while True:
179 metadata = await self._get_channel_metadata(item_id)
180 if metadata and metadata.get("current"):
181 current_song = metadata["current"]
182 current_event = current_song.get("event", "")
183
184 if current_event != last_track_event:
185 # Create StreamMetadata object with full track info
186 stream_metadata = parsers.build_stream_metadata(current_song, metadata)
187
188 self.logger.debug(
189 f"Updating stream metadata for {item_id}: "
190 f"{stream_metadata.artist} - {stream_metadata.title}"
191 )
192 stream_details.stream_metadata = stream_metadata
193
194 last_track_event = current_event
195
196 await asyncio.sleep(15)
197 except asyncio.CancelledError:
198 self.logger.debug(f"Monitor task cancelled for {item_id}")
199 except aiohttp.ClientError as exc:
200 self.logger.debug(f"Network error while monitoring {item_id}: {exc}")
201 except Exception as exc:
202 self.logger.warning(f"Unexpected error monitoring {item_id}: {exc}")
203