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