music-assistant-server

1"""SomaFM Radio music provider support for MusicAssistant."""
2
3from __future__ import annotations
4
5import random
6from typing import TYPE_CHECKING, Any
7
8from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
9from music_assistant_models.enums import (
10    ConfigEntryType,
11    ContentType,
12    ImageType,
13    MediaType,
14    ProviderFeature,
15    StreamType,
16)
17from music_assistant_models.errors import MediaNotFoundError
18from music_assistant_models.media_items import (
19    AudioFormat,
20    BrowseFolder,
21    ItemMapping,
22    MediaItemImage,
23    MediaItemMetadata,
24    MediaItemType,
25    ProviderMapping,
26    Radio,
27    SearchResults,
28)
29from music_assistant_models.streamdetails import StreamDetails
30
31from music_assistant.controllers.cache import use_cache
32from music_assistant.helpers.playlists import PlaylistItem, fetch_playlist
33from music_assistant.models.music_provider import MusicProvider
34
35if TYPE_CHECKING:
36    from collections.abc import Sequence
37
38    from music_assistant_models.config_entries import ProviderConfig
39    from music_assistant_models.provider import ProviderManifest
40
41    from music_assistant import MusicAssistant
42    from music_assistant.models import ProviderInstanceType
43
44SUPPORTED_FEATURES = {
45    ProviderFeature.BROWSE,
46    ProviderFeature.SEARCH,
47}
48
49CONF_QUALITY = "quality"
50
51
52async def setup(
53    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
54) -> ProviderInstanceType:
55    """Initialize provider(instance) with given configuration."""
56    return SomaFMProvider(mass, manifest, config, SUPPORTED_FEATURES)
57
58
59class SomaFMProvider(MusicProvider):
60    """Provider implementation for SomaFM Radio."""
61
62    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
63        """Return Config entries to configure this provider."""
64        return (
65            ConfigEntry(
66                key=CONF_QUALITY,
67                advanced=True,
68                type=ConfigEntryType.STRING,
69                options=[
70                    ConfigValueOption("highest"),
71                    ConfigValueOption("high"),
72                    ConfigValueOption("low"),
73                ],
74                default_value="highest",
75            ),
76        )
77
78    @property
79    def is_streaming_provider(self) -> bool:
80        """Return True if the provider is a streaming provider."""
81        return True
82
83    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
84        """Browse this provider's radio stations."""
85        stations = await self._get_stations()
86        if stations:
87            return [self._parse_channel(channel_info) for channel_info in stations.values()]
88        return []
89
90    async def search(
91        self,
92        search_query: str,
93        media_types: list[MediaType],
94        limit: int = 5,
95    ) -> SearchResults:
96        """Perform search on SomaFM channels."""
97        results = SearchResults()
98        if MediaType.RADIO not in media_types:
99            return results
100        search_query_lower = search_query.lower().strip()
101        if not search_query_lower:
102            return results
103        stations = await self._get_stations()
104        if not stations:
105            return results
106        radios: list[Radio] = []
107        for channel_info in stations.values():
108            channel_name = str(channel_info.get("title", "")).lower()
109            if search_query_lower in channel_name:
110                radios.append(self._parse_channel(channel_info))
111                if len(radios) >= limit:
112                    break
113        results.radio = radios
114        return results
115
116    async def get_radio(self, prov_radio_id: str) -> Radio:
117        """Get radio station details."""
118        stations = await self._get_stations()  # May be cached
119        if stations:
120            radio = stations.get(prov_radio_id)
121        if radio:
122            return self._parse_channel(radio)
123        msg = f"Item {prov_radio_id} not found"
124        raise MediaNotFoundError(msg)
125
126    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
127        """Get stream details for a track/radio."""
128
129        async def _get_valid_playlist_item(playlist: list[PlaylistItem]) -> PlaylistItem:
130            """Randomly select stream URL from playlist and test it."""
131            random.shuffle(playlist)
132            for item in playlist:
133                async with self.mass.http_session.head(item.path, ssl=False) as response:
134                    if response.status >= 100 and response.status < 300:
135                        # Stream exists, return valid path
136                        return item
137            self.logger.error("Could not find a working stream for playlist")
138            raise MediaNotFoundError("No valid SomaFM stream available")
139
140        def _get_playlist_url(station: dict[str, Any]) -> str:
141            """Pick playlist based on quality config value."""
142            req_quality = self.config.get_value(CONF_QUALITY)
143            playlists: list[dict[str, str]] = station.get("playlists", [])
144
145            # Remove MP3 playlist options for now; AAC is generally better
146            playlists = [
147                playlist for playlist in playlists if playlist["format"] in {"aac", "aacp"}
148            ]
149
150            # Sort by quality just in case they already aren't sorted highest/high/low
151            quality_map = {"highest": 0, "high": 1, "low": 2}
152            playlists.sort(key=lambda x: quality_map[x["quality"]])
153
154            # Detect empty playlist after sort and filter
155            if len(playlists) == 0:
156                raise MediaNotFoundError("No valid SomaFM playlist available")
157
158            # Find the first playlist item that has the requested quality
159            for playlist in playlists:
160                avail_quality = playlist.get("quality")
161                playlist_url = playlist.get("url")
162                if req_quality == avail_quality and playlist_url:
163                    return playlist_url
164
165            self.logger.warning("Couldn't find SomaFM stream with requested quality and format")
166
167            # Get the first (highest quality) playlist if we couldn't find requested quality
168            playlist_url = playlists[0].get("url")
169            if playlist_url:
170                return playlist_url
171            raise MediaNotFoundError("No valid SomaFM playlist available")
172
173        async def _get_stream_path(item_id: str) -> str:
174            """Pick correct playlist, fetch the playlist, and extract stream URL."""
175            stations = await self._get_stations()
176            station = stations.get(item_id)
177            if station:
178                playlist_url = _get_playlist_url(station)
179                playlist = await fetch_playlist(self.mass, playlist_url)
180                playlist_item: PlaylistItem = await _get_valid_playlist_item(playlist)
181                return playlist_item.path
182            raise MediaNotFoundError
183
184        stream_path = await _get_stream_path(item_id)
185
186        return StreamDetails(
187            provider=self.instance_id,
188            item_id=item_id,
189            audio_format=AudioFormat(
190                content_type=ContentType.UNKNOWN,
191            ),
192            media_type=MediaType.RADIO,
193            path=stream_path,
194            stream_type=StreamType.HTTP,
195            allow_seek=False,
196            can_seek=False,
197        )
198
199    @use_cache(3600 * 24 * 1)  # Cache for 1 day
200    async def _get_stations(self) -> dict[str, dict[str, Any]]:
201        url = "https://somafm.com/channels.json"
202        locale = self.mass.metadata.locale.replace("_", "-")
203        language = locale.split("-")[0]
204        headers = {"Accept-Language": f"{locale}, {language};q=0.9, *;q=0.5"}
205        async with (
206            self.mass.http_session.get(url, headers=headers, ssl=False) as response,
207        ):
208            result: Any = await response.json()
209            if not result or "error" in result:
210                self.logger.error(url)
211            elif isinstance(result, dict):
212                stations = result.get("channels")
213                if stations:
214                    # Reformat into dict by channel id
215                    return {info.get("id"): info for info in stations if info.get("id")}
216            raise MediaNotFoundError("Could not fetch SomaFM stations list")
217
218    def _parse_channel(self, channel_info: dict[str, Any]) -> Radio:
219        """Convert SomaFM channel info into a Radio object."""
220        # Construct radio station information
221        item_id = channel_info.get("id")
222        if not item_id:
223            raise MediaNotFoundError("Soma FM station generation failed")
224
225        radio = Radio(
226            provider=self.instance_id,
227            item_id=item_id,
228            name=f"SomaFM: {channel_info.get('title', 'Unknown Radio')}",
229            metadata=MediaItemMetadata(
230                description=channel_info.get("description", "No description"),
231                genres={channel_info.get("genre", "No genre")},
232                popularity=int(channel_info.get("listeners", "0")),
233                performers={
234                    f"DJ: {channel_info.get('dj', 'No DJ info')}",
235                    f"DJ Email: {channel_info.get('djmail', 'No DJ email')}",
236                },
237            ),
238            provider_mappings={
239                ProviderMapping(
240                    provider_domain=self.domain,
241                    provider_instance=self.instance_id,
242                    item_id=item_id,
243                    available=True,
244                )
245            },
246        )
247
248        # Add station image URL
249        station_icon_url = channel_info.get("largeimage")
250        if station_icon_url:
251            radio.metadata.add_image(
252                MediaItemImage(
253                    provider=self.instance_id,
254                    type=ImageType.THUMB,
255                    path=station_icon_url,
256                    remotely_accessible=True,
257                )
258            )
259        return radio
260