/
/
/
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 RADIO_PARADISE_CHANNELS
25
26
27class RadioParadiseProvider(MusicProvider):
28 """Radio Paradise Music Provider for Music Assistant."""
29
30 @property
31 def is_streaming_provider(self) -> bool:
32 """Return True if the provider is a streaming provider."""
33 return True
34
35 async def get_library_radios(self) -> AsyncGenerator[Radio, None]:
36 """Retrieve library/subscribed radio stations from the provider."""
37 for channel_id in RADIO_PARADISE_CHANNELS:
38 yield self._parse_radio(channel_id)
39
40 @use_cache(3600 * 3) # Cache for 3 hours
41 async def get_radio(self, prov_radio_id: str) -> Radio:
42 """Get full radio details by id."""
43 if prov_radio_id not in RADIO_PARADISE_CHANNELS:
44 raise MediaNotFoundError("Station not found")
45
46 return self._parse_radio(prov_radio_id)
47
48 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
49 """Get streamdetails for a radio station."""
50 if media_type != MediaType.RADIO:
51 raise UnplayableMediaError(f"Unsupported media type: {media_type}")
52 if item_id not in RADIO_PARADISE_CHANNELS:
53 raise MediaNotFoundError(f"Unknown radio channel: {item_id}")
54
55 # Get stream URL from channel configuration
56 channel_info = RADIO_PARADISE_CHANNELS[item_id]
57 stream_url = channel_info.get("stream_url")
58 if not stream_url:
59 raise UnplayableMediaError(f"No stream URL found for channel {item_id}")
60
61 # Get content type from channel configuration
62 channel_info = RADIO_PARADISE_CHANNELS[item_id]
63 content_type = channel_info["content_type"]
64
65 stream_details = StreamDetails(
66 item_id=item_id,
67 provider=self.instance_id,
68 audio_format=AudioFormat(
69 content_type=content_type,
70 channels=2,
71 ),
72 media_type=MediaType.RADIO,
73 stream_type=StreamType.HTTP,
74 path=stream_url,
75 allow_seek=False,
76 can_seek=False,
77 duration=0,
78 stream_metadata_update_callback=self._update_stream_metadata,
79 stream_metadata_update_interval=10, # Check every 10 seconds
80 )
81
82 # Set initial metadata if available
83 metadata = await self._get_channel_metadata(item_id)
84 if metadata and metadata.get("current"):
85 current_song = metadata["current"]
86 stream_details.stream_metadata = parsers.build_stream_metadata(current_song)
87
88 return stream_details
89
90 async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
91 """Browse this provider's items."""
92 return [self._parse_radio(channel_id) for channel_id in RADIO_PARADISE_CHANNELS]
93
94 def _parse_radio(self, channel_id: str) -> Radio:
95 """Create a Radio object from cached channel information."""
96 return parsers.parse_radio(channel_id, self.instance_id, self.domain)
97
98 async def _get_channel_metadata(self, channel_id: str) -> dict[str, Any] | None:
99 """Get current track metadata from Radio Paradise's now_playing API.
100
101 :param channel_id: Radio Paradise channel ID (0-5).
102 """
103 if channel_id not in RADIO_PARADISE_CHANNELS:
104 return None
105
106 try:
107 # Use now_playing API
108 channel_info = RADIO_PARADISE_CHANNELS[channel_id]
109 api_url = channel_info["api_url"]
110 timeout = aiohttp.ClientTimeout(total=10)
111
112 async with self.mass.http_session.get(api_url, timeout=timeout) as response:
113 if response.status != 200:
114 self.logger.debug(f"Now playing API call failed with status {response.status}")
115 return None
116
117 data = await response.json()
118
119 if not data:
120 self.logger.debug(f"No metadata returned for channel {channel_id}")
121 return None
122
123 return {"current": data, "next": None, "block_data": None}
124
125 except aiohttp.ClientError as exc:
126 self.logger.debug(f"Failed to get metadata for channel {channel_id}: {exc}")
127 return None
128 except Exception as exc:
129 self.logger.debug(f"Unexpected error getting metadata for channel {channel_id}: {exc}")
130 return None
131
132 async def _update_stream_metadata(
133 self, stream_details: StreamDetails, elapsed_time: int
134 ) -> None:
135 """Update stream metadata callback called by player queue controller.
136
137 Fetches current track info from Radio Paradise's API and updates
138 StreamDetails with track metadata.
139
140 :param stream_details: StreamDetails object to update with metadata.
141 :param elapsed_time: Elapsed playback time in seconds (unused for Radio Paradise).
142 """
143 item_id = stream_details.item_id
144
145 # Initialize data dict if needed
146 if stream_details.data is None:
147 stream_details.data = {}
148
149 try:
150 metadata = await self._get_channel_metadata(item_id)
151 if metadata and metadata.get("current"):
152 current_song = metadata["current"]
153 artist = current_song.get("artist", "")
154 title = current_song.get("title", "")
155 current_track_id = f"{artist}:{title}"
156
157 # Only update if track changed
158 if (
159 not stream_details.stream_metadata
160 or stream_details.data.get("last_track_id") != current_track_id
161 ):
162 # Create StreamMetadata object with full track info
163 stream_metadata = parsers.build_stream_metadata(current_song)
164
165 self.logger.debug(
166 f"Updating stream metadata for {item_id}: "
167 f"{stream_metadata.artist} - {stream_metadata.title}"
168 )
169 stream_details.stream_metadata = stream_metadata
170 stream_details.data["last_track_id"] = current_track_id
171
172 except aiohttp.ClientError as exc:
173 self.logger.debug(f"Network error while updating metadata for {item_id}: {exc}")
174 except Exception as exc:
175 self.logger.warning(f"Unexpected error updating metadata for {item_id}: {exc}")
176