/
/
/
1"""ABC Radio 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 ContentType, 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 .constants import (
24 ABC_RADIO_STATIONS,
25 API_TIMEOUT,
26 NOW_PLAYING_URL,
27 STREAM_METADATA_UPDATE_INTERVAL,
28)
29from .helpers import parse_now_playing, parse_radio
30
31if TYPE_CHECKING:
32 from music_assistant_models.streamdetails import StreamMetadata
33
34
35class ABCRadioProvider(MusicProvider):
36 """ABC Radio Music Provider for Music Assistant."""
37
38 @property
39 def max_concurrent_streams(self) -> None:
40 """Allow unlimited concurrent upstream source streams."""
41 return None
42
43 async def get_radio(self, prov_radio_id: str) -> Radio:
44 """Get full radio details by id."""
45 if prov_radio_id not in ABC_RADIO_STATIONS:
46 raise MediaNotFoundError(f"Unknown radio station: {prov_radio_id}")
47 return parse_radio(prov_radio_id, self.instance_id, self.domain)
48
49 async def search(
50 self,
51 search_query: str,
52 media_types: list[MediaType],
53 limit: int = 5,
54 ) -> SearchResults:
55 """Perform search on the ABC Radio stations."""
56 results = SearchResults()
57 if MediaType.RADIO not in media_types:
58 return results
59 search_query_lower = search_query.lower().strip()
60 if not search_query_lower:
61 return results
62 radios: list[Radio] = []
63 for station_id, station in ABC_RADIO_STATIONS.items():
64 if search_query_lower in station["name"].lower():
65 radios.append(parse_radio(station_id, self.instance_id, self.domain))
66 if len(radios) >= limit:
67 break
68 results.radio = radios
69 return results
70
71 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
72 """Get streamdetails for a radio station."""
73 if media_type != MediaType.RADIO:
74 raise UnplayableMediaError(f"Unsupported media type: {media_type}")
75 if item_id not in ABC_RADIO_STATIONS:
76 raise MediaNotFoundError(f"Unknown radio station: {item_id}")
77
78 stream_details = StreamDetails(
79 item_id=item_id,
80 provider=self.instance_id,
81 audio_format=AudioFormat(
82 content_type=ContentType.AAC,
83 channels=2,
84 ),
85 media_type=MediaType.RADIO,
86 stream_type=StreamType.HLS,
87 path=ABC_RADIO_STATIONS[item_id]["stream_url"],
88 allow_seek=False,
89 can_seek=False,
90 duration=0,
91 stream_metadata_update_callback=self._update_stream_metadata,
92 stream_metadata_update_interval=STREAM_METADATA_UPDATE_INTERVAL,
93 )
94 # Set initial metadata so the first frame the listener sees
95 # is the live track rather than an empty banner.
96 stream_details.stream_metadata = await self._get_now_playing(item_id)
97 return stream_details
98
99 async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
100 """Browse this provider's items."""
101 return [
102 parse_radio(station_id, self.instance_id, self.domain)
103 for station_id in ABC_RADIO_STATIONS
104 ]
105
106 async def _get_now_playing(self, station_id: str) -> StreamMetadata | None:
107 """
108 Get the currently playing track for a station, or None if unavailable.
109
110 :param station_id: A known ABC Radio station id.
111 """
112 url = NOW_PLAYING_URL.format(station_id=station_id)
113 try:
114 async with self.mass.http_session.get(url, timeout=API_TIMEOUT) as response:
115 if response.status != 200:
116 self.logger.debug(
117 "ABC Radio API returned status %s for station %s",
118 response.status,
119 station_id,
120 )
121 return None
122 data: dict[str, Any] = await response.json()
123 except (TimeoutError, aiohttp.ClientError, ValueError) as err:
124 self.logger.debug("ABC Radio API request failed for station %s: %s", station_id, err)
125 return None
126 return parse_now_playing(data)
127
128 async def _update_stream_metadata(
129 self, stream_details: StreamDetails, elapsed_time: int
130 ) -> None:
131 """
132 Update stream metadata callback called by player queue controller.
133
134 :param stream_details: StreamDetails object to update with metadata.
135 :param elapsed_time: Elapsed playback time in seconds (unused for ABC Radio).
136 """
137 metadata = await self._get_now_playing(stream_details.item_id)
138 if metadata:
139 self.logger.debug(
140 "Updating stream metadata for %s: %s - %s",
141 stream_details.item_id,
142 metadata.artist,
143 metadata.title,
144 )
145 stream_details.stream_metadata = metadata
146 elif stream_details.stream_metadata:
147 # No track playing (e.g. talk programming): clear stale track info.
148 stream_details.stream_metadata = None
149