/
/
/
1"""
2Mamma Mi Radio music provider for Music Assistant.
3
4Exposes a self-hosted Mamma Mi Radio HA addon as a single Radio entry. Live
5now-playing metadata is read from the addon's versioned consumer contract
6``GET /api/integrations/v1/now-playing``; the provider requires addon 2.13
7or newer.
8
9See: https://github.com/florianhorner/mammamiradio
10"""
11
12from __future__ import annotations
13
14import re
15from typing import TYPE_CHECKING, Any
16from urllib.parse import urlsplit, urlunsplit
17
18import aiohttp
19from music_assistant_models.enums import (
20 ContentType,
21 MediaType,
22 ProviderFeature,
23 StreamType,
24)
25from music_assistant_models.errors import (
26 MediaNotFoundError,
27 ProviderUnavailableError,
28 SetupFailedError,
29)
30from music_assistant_models.media_items import (
31 AudioFormat,
32 BrowseFolder,
33 ItemMapping,
34 MediaItemMetadata,
35 MediaItemType,
36 ProviderMapping,
37 Radio,
38 SearchResults,
39 UniqueList,
40)
41from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
42
43from music_assistant.models.music_provider import MusicProvider
44
45if TYPE_CHECKING:
46 from collections.abc import Sequence
47
48 from music_assistant_models.config_entries import ProviderConfig
49 from music_assistant_models.provider import ProviderManifest
50
51 from music_assistant.mass import MusicAssistant
52 from music_assistant.models import ProviderInstanceType
53
54
55SUPPORTED_FEATURES = {
56 ProviderFeature.BROWSE,
57 ProviderFeature.SEARCH,
58}
59SUPPORTED_SCHEMA_VERSIONS = {"1"}
60
61CONF_MAMMAMIRADIO_URL = "mammamiradio_url"
62DEFAULT_URL = "http://localhost:8000"
63RADIO_ITEM_ID = "mammamiradio"
64RADIO_NAME = "Mamma Mi Radio"
65RADIO_DESCRIPTION = (
66 "Two Italian hosts. One very opinionated smart home. Self-hosted radio for "
67 "Home Assistant: music, banter, and ads, with your home's moments worked "
68 "into the show."
69)
70REACHABILITY_TIMEOUT = 5
71# How often Music Assistant invokes the live-metadata callback (seconds). 12s is
72# imperceptible on a now-playing card and keeps per-listener poll load on
73# mammamiradio's single-process addon modest.
74STREAM_METADATA_UPDATE_INTERVAL = 12
75# Short timeout for the metadata poll so a slow addon never eats most of the
76# metadata update interval.
77METADATA_TIMEOUT = 3
78
79# Addon endpoint paths.
80NOWPLAYING_PATH = "/api/integrations/v1/now-playing"
81STREAM_PATH = "/stream"
82
83# Published stream defaults (mammamiradio core AudioConfig) used when a format
84# field is missing from the contract.
85DEFAULT_CODEC = "mp3"
86DEFAULT_BITRATE_KBPS = 192
87DEFAULT_SAMPLE_RATE_HZ = 48000
88DEFAULT_CHANNELS = 2
89
90# v1 segment_class values that represent an actively-playing segment (i.e. a
91# segment for which an "Up next" description line is meaningful).
92_V1_ACTIVE_CLASSES = {"music", "voice", "interstitial"}
93
94
95def _clean_str(value: Any) -> str | None:
96 """Return ``value`` as a stripped non-empty string, or None."""
97 if isinstance(value, str):
98 return value.strip() or None
99 return None
100
101
102def _pos_int(value: Any, default: int) -> int:
103 """Return ``value`` if it is a positive (non-bool) int, else ``default``."""
104 if isinstance(value, int) and not isinstance(value, bool) and value > 0:
105 return value
106 return default
107
108
109def _supports_v1_schema(value: Any) -> bool:
110 """Return True if ``value`` identifies a supported now-playing schema version."""
111 return isinstance(value, str) and value.strip() in SUPPORTED_SCHEMA_VERSIONS
112
113
114def _normalize_base_url(value: Any) -> str:
115 """
116 Normalize a configured base URL to ``scheme://host[:port][/path]``.
117
118 Query strings, fragments, and userinfo are discarded; a reverse-proxy path
119 prefix is preserved.
120
121 :param value: The raw configured value.
122 :raises TypeError: if the value is not a string.
123 :raises ValueError: if the value is not a full http(s) URL with a hostname.
124 """
125 if not isinstance(value, str):
126 raise TypeError("base URL must be a string")
127 raw = value.strip()
128 if not raw:
129 raise ValueError("base URL is empty")
130 if any(ch.isspace() or not ch.isprintable() for ch in raw):
131 raise ValueError("base URL contains whitespace or control characters")
132 try:
133 parts = urlsplit(raw)
134 hostname = parts.hostname
135 _ = parts.port # a nonnumeric or out-of-range port raises ValueError
136 except ValueError as err:
137 msg = f"base URL is malformed: {err}"
138 raise ValueError(msg) from err
139 if parts.scheme not in ("http", "https"):
140 raise ValueError("base URL must start with http:// or https://")
141 if not hostname:
142 raise ValueError("base URL has no hostname")
143 netloc = parts.netloc.rsplit("@", 1)[-1]
144 # urlsplit accepts characters aiohttp later rejects (e.g. a backslash);
145 # rejecting them here keeps that mistake a localized setup error instead
146 # of a misleading probe failure.
147 if not re.fullmatch(r"[A-Za-z0-9._\-:\[\]]+", netloc):
148 raise ValueError("base URL host contains invalid characters")
149 return urlunsplit((parts.scheme, netloc, parts.path.rstrip("/"), "", ""))
150
151
152def _stream_path_from_contract(value: Any) -> str:
153 """Return a safe relative stream path from the v1 contract, else the default."""
154 raw = _clean_str(value)
155 if raw is None:
156 return STREAM_PATH
157 try:
158 parts = urlsplit(raw)
159 except ValueError:
160 # e.g. an invalid IPv6-looking value; a malformed contract field must
161 # never escape init as a non-MusicAssistantError (that would skip
162 # MA's automatic load retry).
163 return STREAM_PATH
164 path = parts.path.rstrip("/")
165 if parts.scheme or parts.netloc or not path.startswith("/") or not path:
166 return STREAM_PATH
167 return path
168
169
170def _host_display_names(hosts: Any) -> str | None:
171 """Join host display names from the contract's list of host objects."""
172 if not isinstance(hosts, list):
173 return None
174 names: list[str] = []
175 for host in hosts:
176 if isinstance(host, dict):
177 name = _clean_str(host.get("display_name")) or _clean_str(host.get("engine_host"))
178 else:
179 name = _clean_str(host)
180 if name:
181 names.append(name)
182 return ", ".join(names) or None
183
184
185def _audio_format_from_contract(fmt: Any) -> AudioFormat:
186 """Build an ``AudioFormat`` from ``stream.audio_format``, with published defaults."""
187 fmt = fmt if isinstance(fmt, dict) else {}
188 codec = _clean_str(fmt.get("codec")) or DEFAULT_CODEC
189 content_type = ContentType.try_parse(codec)
190 if content_type == ContentType.UNKNOWN:
191 content_type = ContentType.MP3
192 return AudioFormat(
193 content_type=content_type,
194 bit_rate=_pos_int(fmt.get("bitrate_kbps"), DEFAULT_BITRATE_KBPS),
195 sample_rate=_pos_int(fmt.get("sample_rate_hz"), DEFAULT_SAMPLE_RATE_HZ),
196 channels=_pos_int(fmt.get("channels"), DEFAULT_CHANNELS),
197 )
198
199
200def _v1_to_stream_metadata(payload: dict[str, Any], *, show_upcoming: bool) -> StreamMetadata:
201 """
202 Map a v1 now-playing payload onto a ``StreamMetadata``.
203
204 :param payload: The parsed now-playing response.
205 :param show_upcoming: Render the "Up next" frame instead of the "Now" frame.
206 """
207 station = payload.get("station")
208 station = station if isinstance(station, dict) else {}
209 station_name = _clean_str(station.get("name")) or RADIO_NAME
210
211 now = payload.get("now_playing")
212 now = now if isinstance(now, dict) else None
213 up_next = payload.get("up_next")
214 up_next = up_next if isinstance(up_next, list) else []
215
216 title: str | None = None
217 artist: str | None = None
218 image_url: str | None = None
219 # Album only applies to music segments.
220 album: str | None = None
221 seg_class: Any = None
222
223 if now is not None:
224 # Only string classes are meaningful; a non-str value must not reach the
225 # set-membership test below (unhashable types would raise).
226 seg_class = now.get("segment_class")
227 seg_class = seg_class if isinstance(seg_class, str) else None
228 np_title = _clean_str(now.get("title"))
229 if seg_class == "music":
230 title = np_title
231 artist = _clean_str(now.get("artist"))
232 image_url = _clean_str(now.get("artwork"))
233 album = _clean_str(now.get("album"))
234 elif seg_class == "voice":
235 title = np_title or "Host banter"
236 artist = (
237 _clean_str(now.get("host"))
238 or _host_display_names(station.get("hosts"))
239 or station_name
240 )
241 elif seg_class == "interstitial":
242 title = np_title or station_name
243 artist = station_name
244 else:
245 # "unavailable" or any future class: show a plain station frame.
246 title = station_name
247 else:
248 # session_state stopped / empty_queue: nothing playing.
249 title = station_name
250
251 # Ensure title is never empty.
252 title = _clean_str(title) or station_name
253
254 # Only http(s) artwork with a host may reach MA media surfaces.
255 if image_url is not None:
256 try:
257 art = urlsplit(image_url)
258 except ValueError:
259 image_url = None
260 else:
261 if art.scheme.lower() not in ("http", "https") or not art.netloc:
262 image_url = None
263
264 description: str | None = None
265 if now is not None and seg_class in _V1_ACTIVE_CLASSES and show_upcoming and up_next:
266 first = up_next[0]
267 # Skip idle "unavailable" up-next entries.
268 if isinstance(first, dict) and first.get("segment_class") != "unavailable":
269 up_label = _clean_str(first.get("title"))
270 if up_label:
271 description = f"Up next: {up_label}"
272
273 return StreamMetadata(
274 title=title,
275 artist=_clean_str(artist),
276 album=album,
277 image_url=image_url,
278 description=description,
279 )
280
281
282async def setup(
283 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
284) -> ProviderInstanceType:
285 """Initialize provider(instance) with given configuration."""
286 return MammamiradioProvider(mass, manifest, config, SUPPORTED_FEATURES)
287
288
289class MammamiradioProvider(MusicProvider):
290 """Provider implementation for mammamiradio."""
291
292 # All values are set in handle_async_init.
293 _base_url: str
294 _audio_format_dict: dict[str, Any] | None
295 _stream_path: str
296
297 @property
298 def max_concurrent_streams(self) -> None:
299 """Allow unlimited concurrent upstream source streams."""
300 return None
301
302 async def handle_async_init(self) -> None:
303 """Handle async initialization of the provider."""
304 raw = self.get_setup_value(CONF_MAMMAMIRADIO_URL)
305 try:
306 self._base_url = _normalize_base_url(DEFAULT_URL if raw is None else raw)
307 except (TypeError, ValueError) as err:
308 msg = "invalid base URL configured; enter a full http(s):// URL"
309 raise SetupFailedError(
310 msg,
311 translation_key="invalid_base_url",
312 translation_owner=self.translation_owner,
313 ) from err
314 payload = await self._probe_now_playing()
315 stream = payload.get("stream")
316 stream = stream if isinstance(stream, dict) else {}
317 audio_format = stream.get("audio_format")
318 self._audio_format_dict = audio_format if isinstance(audio_format, dict) else None
319 self._stream_path = _stream_path_from_contract(stream.get("relative_url"))
320 self.logger.info("now-playing contract reachable at %s", self._base_url)
321
322 async def loaded_in_mass(self) -> None:
323 """Call after the provider has been loaded."""
324 await super().loaded_in_mass()
325 await self.mass.music.add_item_to_library(self._build_radio())
326
327 async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
328 """Browse this provider's items."""
329 # mammamiradio exposes exactly one Radio entry; the path is irrelevant.
330 return [self._build_radio()]
331
332 async def search(
333 self,
334 search_query: str,
335 media_types: list[MediaType],
336 limit: int = 5,
337 ) -> SearchResults:
338 """Perform search on the single Mamma Mi Radio entry."""
339 results = SearchResults()
340 if MediaType.RADIO not in media_types:
341 return results
342 search_query_lower = search_query.lower().strip()
343 if not search_query_lower:
344 return results
345 # Match both the display name and the provider slug.
346 if search_query_lower in RADIO_NAME.lower() or search_query_lower in RADIO_ITEM_ID:
347 results.radio = [self._build_radio()]
348 return results
349
350 async def get_radio(self, prov_radio_id: str) -> Radio:
351 """Get full radio details by id."""
352 if prov_radio_id != RADIO_ITEM_ID:
353 msg = f"radio station {prov_radio_id} not found"
354 raise MediaNotFoundError(msg)
355 return self._build_radio()
356
357 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
358 """Return the streamdetails for the mammamiradio radio stream."""
359 if item_id != RADIO_ITEM_ID:
360 msg = f"radio station {item_id} not found"
361 raise MediaNotFoundError(msg)
362 # Liveness was checked at init; no probe at stream time.
363 return StreamDetails(
364 provider=self.instance_id,
365 item_id=item_id,
366 audio_format=self._audio_format(),
367 media_type=MediaType.RADIO,
368 stream_type=StreamType.HTTP,
369 path=f"{self._base_url}{self._stream_path}",
370 allow_seek=False,
371 can_seek=False,
372 stream_metadata_update_callback=self._update_stream_metadata,
373 stream_metadata_update_interval=STREAM_METADATA_UPDATE_INTERVAL,
374 )
375
376 async def _update_stream_metadata(
377 self, stream_details: StreamDetails, elapsed_time: int
378 ) -> None:
379 """
380 Refresh now-playing metadata for the active stream.
381
382 :param stream_details: StreamDetails object to update with metadata.
383 :param elapsed_time: Elapsed playback time in seconds (unused).
384 """
385 if stream_details.data is None:
386 stream_details.data = {}
387 # Namespace our per-stream state so it can never collide with keys MA core
388 # stashes in StreamDetails.data (e.g. hls_media_playlist_url for HLS).
389 data = stream_details.data.setdefault("mammamiradio", {})
390 payload = await self._fetch_now_playing(data)
391 if payload is None:
392 return
393
394 # Detect segment changes via a stable per-segment identity, not the
395 # contract's changed_at clock: the addon advances changed_at on any state
396 # change (e.g. a queue append mid-segment), which would snap the
397 # alternation back to "Now" mid-segment.
398 now = payload.get("now_playing")
399 now = now if isinstance(now, dict) else {}
400 # started_at is a stable per-segment start timestamp when the addon
401 # knows it (None otherwise); including it gives true per-segment identity.
402 seg_key = (
403 now.get("segment_type"),
404 now.get("title"),
405 now.get("artist"),
406 now.get("host"),
407 now.get("started_at"),
408 )
409 if seg_key != data.get("v1_segment"):
410 data["v1_segment"] = seg_key
411 data["show_upcoming"] = False
412
413 # Read the display mode before flipping it, so the first frame of every
414 # segment renders the "Now" view.
415 show_upcoming = data.get("show_upcoming", False)
416 stream_details.stream_metadata = _v1_to_stream_metadata(
417 payload, show_upcoming=show_upcoming
418 )
419 data["show_upcoming"] = not show_upcoming
420
421 async def _probe_now_playing(self) -> dict[str, Any]:
422 """
423 Probe the v1 now-playing endpoint and return its payload.
424
425 :raises ProviderUnavailableError: if the addon is unreachable, unhealthy,
426 or does not expose a supported v1 now-playing contract (addon 2.13+).
427 """
428 endpoint = f"{self._base_url}{NOWPLAYING_PATH}"
429 requires_msg = (
430 f"Mamma Mi Radio addon at {self._base_url} does not expose the now-playing "
431 "contract; this provider requires addon 2.13 or newer"
432 )
433 try:
434 timeout = aiohttp.ClientTimeout(total=REACHABILITY_TIMEOUT)
435 async with self.mass.http_session.get(endpoint, timeout=timeout) as response:
436 if response.status in (404, 405, 501):
437 raise ProviderUnavailableError(requires_msg)
438 if response.status >= 400:
439 msg = (
440 f"Mamma Mi Radio addon at {self._base_url} returned HTTP {response.status}"
441 )
442 raise ProviderUnavailableError(msg)
443 payload = await response.json()
444 if not isinstance(payload, dict):
445 raise ProviderUnavailableError(requires_msg)
446 schema = payload.get("schema_version")
447 if not _supports_v1_schema(schema):
448 if not isinstance(schema, str):
449 # No usable version field: treat as a pre-2.13 addon (or
450 # some other service answering on this port).
451 raise ProviderUnavailableError(requires_msg)
452 msg = (
453 f"Mamma Mi Radio addon at {self._base_url} publishes unsupported "
454 f"now-playing schema_version {str(schema)[:32]!r}; this provider "
455 "supports v1 (addon 2.13+)"
456 )
457 raise ProviderUnavailableError(msg)
458 return payload
459 # ContentTypeError subclasses ClientError but means the endpoint answered
460 # with a non-JSON body, so it must be caught first: an HTML splash page
461 # reports "requires addon 2.13+" instead of "unreachable".
462 except (aiohttp.ContentTypeError, ValueError) as err:
463 raise ProviderUnavailableError(requires_msg) from err
464 except (aiohttp.ClientError, TimeoutError) as err:
465 msg = f"Mamma Mi Radio addon unreachable at {self._base_url}: {err}"
466 raise ProviderUnavailableError(msg) from err
467
468 async def _fetch_now_playing(self, data: dict[str, Any]) -> dict[str, Any] | None:
469 """
470 Poll the now-playing endpoint, returning the payload or None on failure.
471
472 Sends a conditional request with the stored ETag; a 304 reuses the cached
473 payload. A 200 without an ETag header drops the stored validator so
474 polling becomes unconditional.
475
476 :param data: Per-stream state (ETag validator and cached payload).
477 """
478 url = f"{self._base_url}{NOWPLAYING_PATH}"
479 headers: dict[str, str] = {}
480 etag = data.get("v1_etag")
481 if isinstance(etag, str):
482 headers["If-None-Match"] = etag
483 try:
484 timeout = aiohttp.ClientTimeout(total=METADATA_TIMEOUT)
485 async with self.mass.http_session.get(
486 url, headers=headers, timeout=timeout
487 ) as response:
488 if response.status == 304:
489 cached = data.get("v1_last")
490 return cached if isinstance(cached, dict) else None
491 if response.status >= 400:
492 self.logger.debug("v1 now-playing returned HTTP %s", response.status)
493 return None
494 payload = await response.json()
495 if not isinstance(payload, dict):
496 return None
497 if not _supports_v1_schema(payload.get("schema_version")):
498 self.logger.debug(
499 "v1 now-playing returned unsupported schema_version %r",
500 payload.get("schema_version"),
501 )
502 return None
503 new_etag = response.headers.get("ETag")
504 if isinstance(new_etag, str):
505 data["v1_etag"] = new_etag
506 else:
507 # The server stopped emitting ETags: drop the stored validator
508 # so polling actually becomes unconditional.
509 data.pop("v1_etag", None)
510 data["v1_last"] = payload
511 return payload
512 except (aiohttp.ClientError, TimeoutError) as err:
513 # A poisoned stored ETag (e.g. control characters from a broken proxy)
514 # fails at request time on every tick; drop it so the next tick recovers.
515 data.pop("v1_etag", None)
516 self.logger.debug("v1 now-playing request failed: %s", err)
517 return None
518 except ValueError as err:
519 data.pop("v1_etag", None)
520 self.logger.debug("v1 now-playing returned bad JSON: %s", err)
521 return None
522
523 def _audio_format(self) -> AudioFormat:
524 """Return the shared stream AudioFormat (v1 contract if known, else defaults)."""
525 return _audio_format_from_contract(self._audio_format_dict)
526
527 def _build_radio(self) -> Radio:
528 """Construct the single Radio object for mammamiradio."""
529 return Radio(
530 provider=self.instance_id,
531 item_id=RADIO_ITEM_ID,
532 name=RADIO_NAME,
533 metadata=MediaItemMetadata(
534 description=RADIO_DESCRIPTION,
535 genres={"Italian", "Talk Radio"},
536 languages=UniqueList(["it"]),
537 ),
538 provider_mappings={
539 ProviderMapping(
540 item_id=RADIO_ITEM_ID,
541 provider_domain=self.domain,
542 provider_instance=self.instance_id,
543 available=True,
544 audio_format=self._audio_format(),
545 )
546 },
547 )
548