music-assistant-server

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    async def handle_async_init(self) -> None:
298        """Handle async initialization of the provider."""
299        raw = self.get_setup_value(CONF_MAMMAMIRADIO_URL)
300        try:
301            self._base_url = _normalize_base_url(DEFAULT_URL if raw is None else raw)
302        except (TypeError, ValueError) as err:
303            msg = "invalid base URL configured; enter a full http(s):// URL"
304            raise SetupFailedError(
305                msg,
306                translation_key="invalid_base_url",
307                translation_owner=self.translation_owner,
308            ) from err
309        payload = await self._probe_now_playing()
310        stream = payload.get("stream")
311        stream = stream if isinstance(stream, dict) else {}
312        audio_format = stream.get("audio_format")
313        self._audio_format_dict = audio_format if isinstance(audio_format, dict) else None
314        self._stream_path = _stream_path_from_contract(stream.get("relative_url"))
315        self.logger.info("now-playing contract reachable at %s", self._base_url)
316
317    async def loaded_in_mass(self) -> None:
318        """Call after the provider has been loaded."""
319        await super().loaded_in_mass()
320        await self.mass.music.add_item_to_library(self._build_radio())
321
322    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
323        """Browse this provider's items."""
324        # mammamiradio exposes exactly one Radio entry; the path is irrelevant.
325        return [self._build_radio()]
326
327    async def search(
328        self,
329        search_query: str,
330        media_types: list[MediaType],
331        limit: int = 5,
332    ) -> SearchResults:
333        """Perform search on the single Mamma Mi Radio entry."""
334        results = SearchResults()
335        if MediaType.RADIO not in media_types:
336            return results
337        search_query_lower = search_query.lower().strip()
338        if not search_query_lower:
339            return results
340        # Match both the display name and the provider slug.
341        if search_query_lower in RADIO_NAME.lower() or search_query_lower in RADIO_ITEM_ID:
342            results.radio = [self._build_radio()]
343        return results
344
345    async def get_radio(self, prov_radio_id: str) -> Radio:
346        """Get full radio details by id."""
347        if prov_radio_id != RADIO_ITEM_ID:
348            msg = f"radio station {prov_radio_id} not found"
349            raise MediaNotFoundError(msg)
350        return self._build_radio()
351
352    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
353        """Return the streamdetails for the mammamiradio radio stream."""
354        if item_id != RADIO_ITEM_ID:
355            msg = f"radio station {item_id} not found"
356            raise MediaNotFoundError(msg)
357        # Liveness was checked at init; no probe at stream time.
358        return StreamDetails(
359            provider=self.instance_id,
360            item_id=item_id,
361            audio_format=self._audio_format(),
362            media_type=MediaType.RADIO,
363            stream_type=StreamType.HTTP,
364            path=f"{self._base_url}{self._stream_path}",
365            allow_seek=False,
366            can_seek=False,
367            stream_metadata_update_callback=self._update_stream_metadata,
368            stream_metadata_update_interval=STREAM_METADATA_UPDATE_INTERVAL,
369        )
370
371    async def _update_stream_metadata(
372        self, stream_details: StreamDetails, elapsed_time: int
373    ) -> None:
374        """
375        Refresh now-playing metadata for the active stream.
376
377        :param stream_details: StreamDetails object to update with metadata.
378        :param elapsed_time: Elapsed playback time in seconds (unused).
379        """
380        if stream_details.data is None:
381            stream_details.data = {}
382        # Namespace our per-stream state so it can never collide with keys MA core
383        # stashes in StreamDetails.data (e.g. hls_media_playlist_url for HLS).
384        data = stream_details.data.setdefault("mammamiradio", {})
385        payload = await self._fetch_now_playing(data)
386        if payload is None:
387            return
388
389        # Detect segment changes via a stable per-segment identity, not the
390        # contract's changed_at clock: the addon advances changed_at on any state
391        # change (e.g. a queue append mid-segment), which would snap the
392        # alternation back to "Now" mid-segment.
393        now = payload.get("now_playing")
394        now = now if isinstance(now, dict) else {}
395        # started_at is a stable per-segment start timestamp when the addon
396        # knows it (None otherwise); including it gives true per-segment identity.
397        seg_key = (
398            now.get("segment_type"),
399            now.get("title"),
400            now.get("artist"),
401            now.get("host"),
402            now.get("started_at"),
403        )
404        if seg_key != data.get("v1_segment"):
405            data["v1_segment"] = seg_key
406            data["show_upcoming"] = False
407
408        # Read the display mode before flipping it, so the first frame of every
409        # segment renders the "Now" view.
410        show_upcoming = data.get("show_upcoming", False)
411        stream_details.stream_metadata = _v1_to_stream_metadata(
412            payload, show_upcoming=show_upcoming
413        )
414        data["show_upcoming"] = not show_upcoming
415
416    async def _probe_now_playing(self) -> dict[str, Any]:
417        """
418        Probe the v1 now-playing endpoint and return its payload.
419
420        :raises ProviderUnavailableError: if the addon is unreachable, unhealthy,
421            or does not expose a supported v1 now-playing contract (addon 2.13+).
422        """
423        endpoint = f"{self._base_url}{NOWPLAYING_PATH}"
424        requires_msg = (
425            f"Mamma Mi Radio addon at {self._base_url} does not expose the now-playing "
426            "contract; this provider requires addon 2.13 or newer"
427        )
428        try:
429            timeout = aiohttp.ClientTimeout(total=REACHABILITY_TIMEOUT)
430            async with self.mass.http_session.get(endpoint, timeout=timeout) as response:
431                if response.status in (404, 405, 501):
432                    raise ProviderUnavailableError(requires_msg)
433                if response.status >= 400:
434                    msg = (
435                        f"Mamma Mi Radio addon at {self._base_url} returned HTTP {response.status}"
436                    )
437                    raise ProviderUnavailableError(msg)
438                payload = await response.json()
439                if not isinstance(payload, dict):
440                    raise ProviderUnavailableError(requires_msg)
441                schema = payload.get("schema_version")
442                if not _supports_v1_schema(schema):
443                    if not isinstance(schema, str):
444                        # No usable version field: treat as a pre-2.13 addon (or
445                        # some other service answering on this port).
446                        raise ProviderUnavailableError(requires_msg)
447                    msg = (
448                        f"Mamma Mi Radio addon at {self._base_url} publishes unsupported "
449                        f"now-playing schema_version {str(schema)[:32]!r}; this provider "
450                        "supports v1 (addon 2.13+)"
451                    )
452                    raise ProviderUnavailableError(msg)
453                return payload
454        # ContentTypeError subclasses ClientError but means the endpoint answered
455        # with a non-JSON body, so it must be caught first: an HTML splash page
456        # reports "requires addon 2.13+" instead of "unreachable".
457        except (aiohttp.ContentTypeError, ValueError) as err:
458            raise ProviderUnavailableError(requires_msg) from err
459        except (aiohttp.ClientError, TimeoutError) as err:
460            msg = f"Mamma Mi Radio addon unreachable at {self._base_url}: {err}"
461            raise ProviderUnavailableError(msg) from err
462
463    async def _fetch_now_playing(self, data: dict[str, Any]) -> dict[str, Any] | None:
464        """
465        Poll the now-playing endpoint, returning the payload or None on failure.
466
467        Sends a conditional request with the stored ETag; a 304 reuses the cached
468        payload. A 200 without an ETag header drops the stored validator so
469        polling becomes unconditional.
470
471        :param data: Per-stream state (ETag validator and cached payload).
472        """
473        url = f"{self._base_url}{NOWPLAYING_PATH}"
474        headers: dict[str, str] = {}
475        etag = data.get("v1_etag")
476        if isinstance(etag, str):
477            headers["If-None-Match"] = etag
478        try:
479            timeout = aiohttp.ClientTimeout(total=METADATA_TIMEOUT)
480            async with self.mass.http_session.get(
481                url, headers=headers, timeout=timeout
482            ) as response:
483                if response.status == 304:
484                    cached = data.get("v1_last")
485                    return cached if isinstance(cached, dict) else None
486                if response.status >= 400:
487                    self.logger.debug("v1 now-playing returned HTTP %s", response.status)
488                    return None
489                payload = await response.json()
490                if not isinstance(payload, dict):
491                    return None
492                if not _supports_v1_schema(payload.get("schema_version")):
493                    self.logger.debug(
494                        "v1 now-playing returned unsupported schema_version %r",
495                        payload.get("schema_version"),
496                    )
497                    return None
498                new_etag = response.headers.get("ETag")
499                if isinstance(new_etag, str):
500                    data["v1_etag"] = new_etag
501                else:
502                    # The server stopped emitting ETags: drop the stored validator
503                    # so polling actually becomes unconditional.
504                    data.pop("v1_etag", None)
505                data["v1_last"] = payload
506                return payload
507        except (aiohttp.ClientError, TimeoutError) as err:
508            # A poisoned stored ETag (e.g. control characters from a broken proxy)
509            # fails at request time on every tick; drop it so the next tick recovers.
510            data.pop("v1_etag", None)
511            self.logger.debug("v1 now-playing request failed: %s", err)
512            return None
513        except ValueError as err:
514            data.pop("v1_etag", None)
515            self.logger.debug("v1 now-playing returned bad JSON: %s", err)
516            return None
517
518    def _audio_format(self) -> AudioFormat:
519        """Return the shared stream AudioFormat (v1 contract if known, else defaults)."""
520        return _audio_format_from_contract(self._audio_format_dict)
521
522    def _build_radio(self) -> Radio:
523        """Construct the single Radio object for mammamiradio."""
524        return Radio(
525            provider=self.instance_id,
526            item_id=RADIO_ITEM_ID,
527            name=RADIO_NAME,
528            metadata=MediaItemMetadata(
529                description=RADIO_DESCRIPTION,
530                genres={"Italian", "Talk Radio"},
531                languages=UniqueList(["it"]),
532            ),
533            provider_mappings={
534                ProviderMapping(
535                    item_id=RADIO_ITEM_ID,
536                    provider_domain=self.domain,
537                    provider_instance=self.instance_id,
538                    available=True,
539                    audio_format=self._audio_format(),
540                )
541            },
542        )
543