feat(tilderadio): add the tildeverse community stream as a catalogue

beefy_player/config.py

@@ -3,7 +3,9 @@
 Everything lives in the usual XDG places:
 
     ~/.config/beefy-player/config.json          favourites, volume, station
-    ~/.cache/beefy-player/channels.json         last good station list
+    ~/.cache/beefy-player/channels.json         last good station list (SomaFM)
+    ~/.cache/beefy-player/channels-<id>.json    ... for every other catalogue
+    ~/.cache/beefy-player/schedule-<id>.json    last good broadcast schedule
     ~/.cache/beefy-player/playlists.json        resolved .pls -> stream URLs
     ~/.cache/beefy-player/songs/<id>.json       last good song history
     ~/.cache/beefy-player/artwork/<sha1>        cached album/channel artwork
@@ -17,6 +19,7 @@ from __future__ import annotations
 import contextlib
 import json
 import os
+import re
 import shutil
 from dataclasses import asdict, dataclass, field
 from pathlib import Path
@@ -70,6 +73,24 @@ LEGACY_CONFIG_PATH = _xdg("XDG_CONFIG_HOME", ".config") / LEGACY_APP_NAME / "con
 VISUALIZERS = ("both", "bars", "wave", "waterfall", "radial", "braille")
 
 
+def channels_cache_path(provider_id: str) -> Path:
+    """Where a catalogue caches its last good station list.
+
+    SomaFM keeps the historical ``channels.json``; every other provider gets
+    its own file, so two catalogues can never overwrite each other's cache.
+    """
+    slug = re.sub(r"[^a-z0-9_-]", "_", provider_id.strip().lower())
+    if not slug or slug == "somafm":
+        return CHANNELS_CACHE
+    return CACHE_DIR / f"channels-{slug}.json"
+
+
+def schedule_cache_path(provider_id: str) -> Path:
+    """Where a catalogue caches its last good broadcast schedule."""
+    slug = re.sub(r"[^a-z0-9_-]", "_", provider_id.strip().lower()) or "provider"
+    return CACHE_DIR / f"schedule-{slug}.json"
+
+
 def ensure_dirs() -> None:
     for directory in (CONFIG_DIR, CACHE_DIR, SONGS_CACHE_DIR):
         with contextlib.suppress(OSError):
@@ -115,6 +136,9 @@ def migrate_legacy_state() -> None:
 class Config:
     favorites: list[str] = field(default_factory=list)
     last_station: str = "groovesalad"
+    #: Which station each catalogue was last tuned to, keyed by provider slug,
+    #: so switching back lands where you left off.
+    last_stations: dict[str, str] = field(default_factory=dict)
     volume: int = 70
     visualizer: str = "both"
     autoplay: bool = False
@@ -141,6 +165,14 @@ class Config:
         if isinstance(last_station, str) and last_station:
             config.last_station = last_station
 
+        last_stations = raw.get("last_stations")
+        if isinstance(last_stations, dict):
+            config.last_stations = {
+                str(key): str(value)
+                for key, value in last_stations.items()
+                if isinstance(value, str) and value
+            }
+
         volume = raw.get("volume")
         if isinstance(volume, (int, float)):
             config.volume = max(0, min(100, int(volume)))
@@ -168,6 +200,10 @@ class Config:
         except OSError:
             pass
 
+    def last_station_for(self, provider_id: str) -> str:
+        """The station to re-tune for a catalogue (old configs have one only)."""
+        return self.last_stations.get(provider_id) or self.last_station
+
     def is_favorite(self, station_id: str) -> bool:
         return station_id in self.favorites
 

beefy_player/models.py

@@ -40,6 +40,25 @@ class Song:
         return f"{self.artist}\x1f{self.title}\x1f{self.album}"
 
 
+@dataclass(slots=True)
+class Broadcast:
+    """One entry in a catalogue's schedule: a show, a DJ slot, a stream."""
+
+    name: str
+    dj: str = ""
+    start: datetime | None = None
+    end: datetime | None = None
+    #: True for the broadcast that is on air right now.
+    live: bool = False
+
+    @property
+    def display_name(self) -> str:
+        return self.name.strip() or self.dj.strip() or "untitled broadcast"
+
+    def key(self) -> str:
+        return f"{self.name}\x1f{self.start.isoformat() if self.start else ''}"
+
+
 @dataclass(slots=True)
 class Station:
     id: str
@@ -47,6 +66,9 @@ class Station:
     description: str = ""
     genre: str = ""
     dj: str = ""
+    #: Who is presenting right now when that is more than a DJ name —
+    #: ``"auto dj · library rotation"`` for a station between live shows.
+    presenter: str = ""
     image: str = ""
     listeners: int = 0
     last_playing: str = ""
@@ -69,6 +91,15 @@ class Station:
     def genre_label(self) -> str:
         return self.genre.replace("|", " · ") or "eclectic"
 
+    @property
+    def presenter_label(self) -> str:
+        """What the ON AIR metadata line says about who is on the desk."""
+        if self.presenter.strip():
+            return self.presenter.strip()
+        if self.dj.strip():
+            return f"dj {self.dj.strip()}"
+        return ""
+
     def sorted_playlists(self) -> list[dict[str, str]]:
         return sorted(
             self.playlists,
@@ -87,7 +118,7 @@ class Station:
         fmt = _FORMAT_LABEL.get(
             first.get("format", ""), first.get("format") or "stream"
         )
-        return f"{fmt} · {guess_bitrate(first.get('url', ''))}k"
+        return f"{fmt} · {playlist_bitrate(first)}k"
 
 
 # --------------------------------------------------------------------------- #
@@ -95,6 +126,12 @@ class Station:
 # --------------------------------------------------------------------------- #
 
 
+def playlist_bitrate(playlist: dict[str, str]) -> int:
+    """The bitrate a catalogue declared, else the best guess from the URL."""
+    declared = to_int(playlist.get("bitrate"))
+    return declared or guess_bitrate(playlist.get("url", ""))
+
+
 def guess_bitrate(url: str) -> int:
     match = re.search(
         r"(256|192|130|128|96|64|48|32)(?=\.pls|\.m3u|-mp3|-aac)", url

beefy_player/providers/tilderadio/__init__.py

@@ -0,0 +1,76 @@
+"""tilderadio: the tildeverse's community station.
+
+One stream, one catalogue entry, a live DJ when a show is on and the shared
+library rotation the rest of the time.  Everything tilderadio-specific — the
+``/api/now/`` schema, the ``now_playing`` + ``history[]`` pairing, the AzuraCast
+mount list and the bundled fallback — lives in this package; the app only sees
+:class:`TilderadioProvider`.
+"""
+
+from __future__ import annotations
+
+import httpx
+
+from ... import config
+from ...models import Broadcast, Song, Station
+from ..base import RadioProvider
+from . import api, fallback
+from . import hero as hero_art
+
+__all__ = ["TilderadioProvider"]
+
+
+class TilderadioProvider(RadioProvider):
+    id = "tilderadio"
+    name = "tilderadio"
+    homepage = "tilderadio.org"
+    brand = "TILDE RADIO"
+    tagline = "community radio · tildeverse"
+    station_source = "tilderadio.org/api/now/"
+    history_source = "now_playing + history[]"
+
+    # The station's own radio icon, converted to half blocks by
+    # ``tools/ansi_art.py``; the caption is the tildeverse's motto.
+    hero = hero_art.HERO
+    hero_lines = (
+        "TILDERADIO",
+        "STREAMED BY / FOR USERS OF THE",
+        "TILDEVERSE",
+    )
+    # One stream, so the left pane is a schedule instead of a station list.
+    has_schedule = True
+    # A shared library rotation is not a Bandcamp catalogue; skip the lookups.
+    bandcamp = False
+
+    @property
+    def user_agent(self) -> str:
+        # A small volunteer station; introduce ourselves and say where we are.
+        return f"{config.USER_AGENT}; +https://tilderadio.org"
+
+    async def fetch_stations(self, client: httpx.AsyncClient) -> list[Station]:
+        return [await api.fetch_station(client)]
+
+    def load_cached_stations(self) -> list[Station]:
+        return api.load_cached_station()
+
+    def fallback_stations(self) -> list[Station]:
+        return fallback.fallback_stations()
+
+    async def fetch_history(
+        self, client: httpx.AsyncClient, station_id: str
+    ) -> list[Song]:
+        return await api.fetch_songs(client, station_id)
+
+    def load_cached_history(self, station_id: str) -> list[Song]:
+        return api.load_cached_songs(station_id)
+
+    async def fetch_schedule(
+        self, client: httpx.AsyncClient
+    ) -> list[Broadcast]:
+        return await api.fetch_schedule(client)
+
+    def load_cached_schedule(self) -> list[Broadcast]:
+        return api.load_cached_schedule()
+
+    def fallback_urls(self, station: Station) -> list[str]:
+        return fallback.known_stream_urls()

beefy_player/providers/tilderadio/api.py

@@ -0,0 +1,401 @@
+"""tilderadio's public endpoints, and nothing but.
+
+* ``https://tilderadio.org/api/now/``  -> the station, listener count, the live
+  DJ, the current track and the last handful of tracks
+* ``https://azuracast.tilderadio.org/api/nowplaying/tilderadio`` -> the mounts
+  (mp3/ogg, bitrates) behind the stream, so we can offer the best one first
+
+tilderadio is a single-station catalogue: one community stream for the whole
+tildeverse, run on an AzuraCast instance.  That makes the "station list" a list
+of one, and it means ``/now/`` is both the catalogue *and* the history feed —
+the current track arrives as ``now_playing``, the previous ones as
+``history[]``.  Everything here is forgiving: the station is better served by a
+slightly degraded panel than by a crash.
+"""
+
+from __future__ import annotations
+
+import re
+import time
+from collections.abc import Iterable
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+from ... import config
+from ...models import (
+    Broadcast,
+    Song,
+    Station,
+    parse_date,
+    parse_last_playing,
+    to_int,
+)
+from ..base import fetch_json, quote_station_id
+
+PROVIDER_ID = "tilderadio"
+API_ROOT = "https://tilderadio.org/api"
+NOW_URL = f"{API_ROOT}/now/"
+SCHEDULE_URL = f"{API_ROOT}/schedule/"
+AZURACAST_URL = "https://azuracast.tilderadio.org/api/nowplaying/{station}"
+
+DEFAULT_STATION_ID = "tilderadio"
+STATION_IMAGE = "https://tilderadio.org/logos/tilderadio.png"
+DESCRIPTION = "community radio for the tildeverse"
+GENRE = "community|eclectic"
+
+
+# --------------------------------------------------------------------------- #
+# payload -> Station / Song
+# --------------------------------------------------------------------------- #
+
+
+def _artist_title(entry: dict[str, Any]) -> tuple[str, str]:
+    """AzuraCast sends ``artist``/``title`` *or* one ``"Artist - Title"`` blob."""
+    artist = str(entry.get("artist") or "").strip()
+    title = str(entry.get("title") or "").strip()
+    if not artist:
+        text = str(entry.get("text") or "").strip()
+        split = parse_last_playing(text) if text else None
+        if split is not None:
+            artist, title = split.artist or artist, split.title or title
+    return artist, title
+
+
+def song_from_track(entry: Any) -> Song | None:
+    if not isinstance(entry, dict):
+        return None
+    artist, title = _artist_title(entry)
+    if not (artist or title):
+        return None
+    return Song(
+        title=title,
+        artist=artist,
+        album=str(entry.get("album") or "").strip(),
+        art_url=str(entry.get("art") or entry.get("albumArt") or "").strip(),
+        played_at=parse_date(entry.get("played_at") or entry.get("date")),
+    )
+
+
+def parse_tracks(entries: Any) -> list[Song]:
+    """Newest first, and the same track only once even if the feed repeats it."""
+    if not isinstance(entries, list):
+        return []
+    songs: list[Song] = []
+    seen: set[tuple[str, Any]] = set()
+    for entry in entries:
+        song = song_from_track(entry)
+        if song is None:
+            continue
+        marker = (song.key(), song.played_at)
+        if marker in seen:
+            continue
+        seen.add(marker)
+        songs.append(song)
+    if all(song.played_at for song in songs):
+        songs.sort(key=lambda song: song.played_at, reverse=True)  # type: ignore[arg-type,return-value]
+    return songs
+
+
+def songs_from_now(payload: Any) -> list[Song]:
+    """``now_playing`` followed by ``history[]``, as one newest-first list."""
+    if not isinstance(payload, dict):
+        return []
+    entries: list[Any] = [payload.get("now_playing")]
+    history = payload.get("history")
+    if isinstance(history, list):
+        entries.extend(history)
+    return parse_tracks(entries)
+
+
+def _quality(bitrate: int) -> str:
+    if bitrate >= 256:
+        return "highest"
+    if bitrate >= 128:
+        return "high"
+    return "low" if bitrate else ""
+
+
+def playlists_from_mounts(payload: Any) -> list[dict[str, str]]:
+    """AzuraCast's mount list -> the playlist dicts :class:`Station` speaks."""
+    if not isinstance(payload, dict):
+        return []
+    info = payload.get("station")
+    mounts = info.get("mounts") if isinstance(info, dict) else None
+    if not isinstance(mounts, list):
+        return []
+    playlists: list[dict[str, str]] = []
+    seen: set[str] = set()
+    for mount in mounts:
+        if not isinstance(mount, dict):
+            continue
+        url = mount.get("url")
+        if not isinstance(url, str) or not url.startswith(("http://", "https://")):
+            continue
+        if url in seen:
+            continue
+        seen.add(url)
+        bitrate = to_int(mount.get("bitrate"))
+        playlists.append(
+            {
+                "url": url,
+                "format": str(mount.get("format") or "").strip().lower(),
+                "quality": _quality(bitrate),
+                "bitrate": str(bitrate) if bitrate else "",
+            }
+        )
+    return playlists
+
+
+def _playlists_from_listen_url(info: dict[str, Any]) -> list[dict[str, str]]:
+    url = info.get("listen_url")
+    if not isinstance(url, str) or not url.startswith(("http://", "https://")):
+        return []
+    return [{"url": url, "format": "ogg", "quality": "high", "bitrate": ""}]
+
+
+def _is_live(payload: dict[str, Any]) -> bool:
+    live = payload.get("live")
+    if isinstance(live, dict):
+        return bool(live.get("is_live"))
+    return bool(payload.get("is_live"))
+
+
+def _live_dj(payload: dict[str, Any]) -> str:
+    """Only name a DJ when the station says someone is actually on air."""
+    if not _is_live(payload):
+        return ""
+    live = payload.get("live")
+    if isinstance(live, dict):
+        return str(live.get("name") or live.get("dj") or "").strip()
+    return str(payload.get("dj") or "").strip()
+
+
+def station_from_now(
+    payload: Any, playlists: Iterable[Any] = ()
+) -> Station | None:
+    """The whole catalogue, from one ``/api/now/`` response."""
+    if not isinstance(payload, dict):
+        return None
+    info = payload.get("station")
+    if not isinstance(info, dict):
+        return None
+    station_id = (
+        str(info.get("shortcode") or info.get("name") or DEFAULT_STATION_ID).strip()
+        or DEFAULT_STATION_ID
+    )
+
+    known = [dict(entry) for entry in playlists if isinstance(entry, dict)]
+    station = Station(
+        id=station_id,
+        title=str(info.get("name") or station_id).strip() or DEFAULT_STATION_ID,
+        description=str(info.get("description") or "").strip() or DESCRIPTION,
+        genre=GENRE,
+        dj=_live_dj(payload),
+        presenter="" if _is_live(payload) else "auto dj · library rotation",
+        image=STATION_IMAGE,
+        listeners=to_int(payload.get("listeners")),
+        playlists=known or _playlists_from_listen_url(info),
+        provider=PROVIDER_ID,
+    )
+
+    current = payload.get("now_playing")
+    if isinstance(current, dict):
+        song = song_from_track(current)
+        if song is not None:
+            station.now_playing = song
+            station.last_playing = str(current.get("text") or "").strip() or (
+                f"{song.artist} - {song.title}" if song.artist else song.title
+            )
+    return station
+
+
+# --------------------------------------------------------------------------- #
+# schedule
+# --------------------------------------------------------------------------- #
+
+
+def parse_schedule(payload: Any) -> list[Broadcast]:
+    """``/api/schedule/`` -> broadcasts, soonest first."""
+    entries = payload.get("schedule") if isinstance(payload, dict) else None
+    if not isinstance(entries, list):
+        return []
+    broadcasts: list[Broadcast] = []
+    for entry in entries:
+        if not isinstance(entry, dict):
+            continue
+        name = str(entry.get("name") or entry.get("slug") or "").strip()
+        start = parse_date(entry.get("start_ts") or entry.get("start"))
+        end = parse_date(entry.get("end_ts") or entry.get("end"))
+        if not name or start is None:
+            continue
+        broadcasts.append(Broadcast(name=name, dj=name, start=start, end=end))
+    broadcasts.sort(key=lambda item: item.start or datetime.now(timezone.utc))
+    return broadcasts
+
+
+def live_broadcast(payload: Any) -> Broadcast | None:
+    """The show that is on air right now, if the feed says there is one."""
+    if not isinstance(payload, dict) or not payload.get("is_live"):
+        return None
+    name = _live_dj(payload) or "live broadcast"
+    return Broadcast(
+        name=name,
+        dj=name,
+        start=parse_date(payload.get("broadcast_start")),
+        end=parse_date(payload.get("broadcast_end")),
+        live=True,
+    )
+
+
+# --------------------------------------------------------------------------- #
+# caching
+# --------------------------------------------------------------------------- #
+
+
+def station_cache_path() -> Path:
+    return config.channels_cache_path(PROVIDER_ID)
+
+
+def _songs_cache_path(station_id: str) -> Path:
+    safe = re.sub(r"[^A-Za-z0-9_.-]", "_", station_id or DEFAULT_STATION_ID)
+    return config.SONGS_CACHE_DIR / f"{PROVIDER_ID}-{safe}.json"
+
+
+def _songs_to_cache(songs: Iterable[Song]) -> list[dict[str, Any]]:
+    return [
+        {
+            "title": song.title,
+            "artist": song.artist,
+            "album": song.album,
+            "art": song.art_url,
+            "played_at": song.played_at.isoformat() if song.played_at else "",
+        }
+        for song in songs
+    ]
+
+
+# --------------------------------------------------------------------------- #
+# fetching
+# --------------------------------------------------------------------------- #
+
+
+async def _fetch_mounts(
+    client: httpx.AsyncClient, station_id: str
+) -> list[dict[str, str]]:
+    url = AZURACAST_URL.format(station=quote_station_id(station_id))
+    try:
+        payload = await fetch_json(client, url, attempts=1)
+    except RuntimeError:
+        return []
+    return playlists_from_mounts(payload)
+
+
+async def fetch_station(client: httpx.AsyncClient) -> Station:
+    try:
+        payload = await fetch_json(client, NOW_URL)
+    except RuntimeError as exc:
+        raise RuntimeError(f"could not fetch tilderadio now-playing: {exc}") from None
+
+    station = station_from_now(payload)
+    if station is None:
+        raise RuntimeError("tilderadio returned no station data")
+
+    # Bitrates and the mp3 mount only come from AzuraCast; losing them is not
+    # a reason to lose the station.
+    playlists = await _fetch_mounts(client, station.id)
+    if playlists:
+        station.playlists = playlists
+
+    config.write_json(
+        station_cache_path(),
+        {"fetched": time.time(), "now": payload, "mounts": station.playlists},
+    )
+    return station
+
+
+def load_cached_station() -> list[Station]:
+    raw = config.read_json(station_cache_path())
+    if not isinstance(raw, dict):
+        return []
+    mounts = raw.get("mounts")
+    station = station_from_now(
+        raw.get("now"), mounts if isinstance(mounts, list) else []
+    )
+    return [station] if station is not None else []
+
+
+async def fetch_songs(client: httpx.AsyncClient, station_id: str) -> list[Song]:
+    try:
+        payload = await fetch_json(client, NOW_URL, attempts=1)
+    except RuntimeError:
+        return load_cached_songs(station_id)
+
+    songs = songs_from_now(payload)
+    if not songs:
+        return load_cached_songs(station_id)
+    config.write_json(
+        _songs_cache_path(station_id),
+        {"fetched": time.time(), "songs": _songs_to_cache(songs)},
+    )
+    return songs
+
+
+def load_cached_songs(station_id: str) -> list[Song]:
+    raw = config.read_json(_songs_cache_path(station_id))
+    if isinstance(raw, dict):
+        return parse_tracks(raw.get("songs"))
+    return []
+
+
+def _schedule_cache_path() -> Path:
+    return config.schedule_cache_path(PROVIDER_ID)
+
+
+def load_cached_schedule() -> list[Broadcast]:
+    raw = config.read_json(_schedule_cache_path())
+    entries = raw.get("schedule") if isinstance(raw, dict) else None
+    if not isinstance(entries, list):
+        return []
+    return parse_schedule({"schedule": entries})
+
+
+async def fetch_schedule(client: httpx.AsyncClient) -> list[Broadcast]:
+    """What is on air now (if anything) followed by what is coming up."""
+    live: Broadcast | None = None
+    try:
+        now_payload = await fetch_json(client, NOW_URL, attempts=1)
+    except RuntimeError:
+        now_payload = None
+    live = live_broadcast(now_payload)
+
+    try:
+        payload = await fetch_json(client, SCHEDULE_URL, attempts=1)
+    except RuntimeError:
+        upcoming = load_cached_schedule()
+    else:
+        upcoming = parse_schedule(payload)
+        config.write_json(
+            _schedule_cache_path(),
+            {
+                "fetched": time.time(),
+                "schedule": [
+                    {
+                        "name": item.name,
+                        "start": item.start.isoformat() if item.start else "",
+                        "end": item.end.isoformat() if item.end else "",
+                    }
+                    for item in upcoming
+                ],
+            },
+        )
+
+    now = datetime.now(timezone.utc)
+    upcoming = [
+        item
+        for item in upcoming
+        if not item.live and (item.end or item.start or now) >= now
+    ]
+    return ([live] if live is not None else []) + upcoming

beefy_player/providers/tilderadio/fallback.py

@@ -0,0 +1,56 @@
+"""Bundled station, so a cold offline start still makes sense.
+
+tilderadio is a single-station catalogue — one community stream for the whole
+tildeverse — so "the bundled station list" is that one station.  The mounts
+below are the ones the AzuraCast instance has served for years; the live list
+from ``/api/now/`` always wins, and the URLs are additionally probed by the
+player's normal failover.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+# (url, format, quality, bitrate)
+_MOUNTS: tuple[tuple[str, str, str, str], ...] = (
+    ("https://azuracast.tilderadio.org/listen/tilderadio/radio.mp3", "mp3", "high", "192"),
+    ("https://azuracast.tilderadio.org/listen/tilderadio/320k.mp3", "mp3", "highest", "320"),
+    ("https://azuracast.tilderadio.org/listen/tilderadio/radio.ogg", "ogg", "high", "192"),
+)
+
+
+def fallback_payload() -> dict[str, Any]:
+    """A ``/api/now/``-shaped payload, so the normal parser can eat it."""
+    return {
+        "api_version": 1,
+        "status": "ok",
+        "station": {
+            "id": 1,
+            "name": "tilderadio",
+            "shortcode": "tilderadio",
+            "listen_url": _MOUNTS[2][0],
+        },
+        "is_live": False,
+        "dj": None,
+        "listeners": 0,
+        "now_playing": None,
+        "history": [],
+    }
+
+
+def fallback_playlists() -> list[dict[str, str]]:
+    return [
+        {"url": url, "format": fmt, "quality": quality, "bitrate": bitrate}
+        for url, fmt, quality, bitrate in _MOUNTS
+    ]
+
+
+def known_stream_urls() -> list[str]:
+    return [url for url, _format, _quality, _bitrate in _MOUNTS]
+
+
+def fallback_stations() -> list[Any]:
+    from .api import station_from_now
+
+    station = station_from_now(fallback_payload(), fallback_playlists())
+    return [station] if station is not None else []

beefy_player/providers/tilderadio/hero.py

@@ -0,0 +1,24 @@
+"""Generated header art — do not edit by hand.
+
+Source: https://tilderadio.org/logos/tilderadio.png
+
+Regenerate with::
+
+    uv run python tools/ansi_art.py --url https://tilderadio.org/logos/tilderadio.png --cols 22 --rows 7 --module beefy_player/providers/tilderadio/hero.py
+
+Each row is rich markup; ``▀`` is a half block whose foreground is the top
+pixel and whose background is the pixel below it.  Colours are stock ANSI
+palette names, so the art keeps the terminal's own palette.
+"""
+
+from __future__ import annotations
+
+HERO: tuple[str, ...] = (
+    '       [bright_black]▄[/][bright_black on bright_black]▀▀[/][bright_black]▀[/]',
+    '   [bright_black]▄▄[/][bright_black on bright_black]▀▀▀[/][bright_black]▄▄▄▄▄▄▄▄▄▄▄[/]',
+    '[bright_black]▄[/][bright_black on bright_black]▀[/][bright_black on black]▀▀▀▀▀▀▀▀▀▀▀[/][bright_black on bright_black]▀▀▀▀▀▀[/][bright_black on black]▀[/][bright_black on bright_black]▀[/][bright_black]▄[/]',
+    '[bright_black on bright_black]▀▀[/][black on black]▀▀▀▀▀▀▀▀▀▀[/][bright_black on bright_black]▀▀[/][bright_black]▀[/]   [bright_black on bright_black]▀▀▀▀[/]',
+    '[bright_black on bright_black]▀▀[/][black on black]▀[/][black on bright_black]▀▀[/][black on green]▀[/][black on bright_black]▀▀[/][black on black]▀▀[/][black on bright_black]▀[/][black on black]▀▀[/][bright_black on black]▀[/][bright_black on bright_black]▀[/][bright_black]▄▄[/][bright_black on bright_black]▀▀[/][bright_black on black]▀[/][bright_black on bright_black]▀▀[/]',
+    '[bright_black on bright_black]▀▀[/][black on black]▀[/][green on green]▀[/][green on black]▀▀▀[/][green on green]▀▀▀▀[/][bright_black on black]▀[/][black on black]▀▀▀[/][bright_black on bright_black]▀[/][bright_black on black]▀[/][bright_black on bright_black]▀▀[/][black on black]▀[/][bright_black on bright_black]▀▀[/]',
+    '[bright_black]▀▀[/][bright_black on bright_black]▀[/][black on bright_black]▀▀[/][bright_black on bright_black]▀[/][black on bright_black]▀▀▀▀▀[/][bright_black on bright_black]▀[/][black on bright_black]▀[/][bright_black on bright_black]▀[/][black on bright_black]▀[/][bright_black on bright_black]▀▀▀[/][black on bright_black]▀[/][bright_black on bright_black]▀[/][bright_black]▀▀[/]',
+)

pyproject.toml

@@ -1,13 +1,13 @@
 [project]
 name = "beefy-player"
 version = "0.1.0"
-description = "A flashy terminal player for internet radio — live spectrum analyser, MPRIS / macOS Now Playing support, song history and pluggable station catalogues (SomaFM included)"
+description = "A flashy terminal player for internet radio — live spectrum analyser, MPRIS / macOS Now Playing support, song history and pluggable station catalogues (SomaFM and tilderadio included)"
 readme = "README.md"
 license = "WTFPL"
 license-files = ["LICENSE"]
 requires-python = ">=3.10"
 authors = [{ name = "beefy-player contributors" }]
-keywords = ["radio", "internet-radio", "tui", "terminal", "mpris", "audio", "somafm"]
+keywords = ["radio", "internet-radio", "tui", "terminal", "mpris", "audio", "somafm", "tilderadio"]
 classifiers = [
   "Environment :: Console",
   "Intended Audience :: End Users/Desktop",

tests/test_tilderadio.py

@@ -0,0 +1,249 @@
+"""tilderadio: parsing, caching and the offline fallback.
+
+The payloads below are trimmed copies of real ``/api/now/`` and AzuraCast
+``/api/nowplaying/<station>`` responses, so the tests keep working when the
+station is off air.
+"""
+
+from __future__ import annotations
+
+import httpx
+import pytest
+
+from beefy_player.providers.tilderadio import TilderadioProvider, api, fallback
+
+NOW_PAYLOAD = {
+    "api_version": 1,
+    "status": "ok",
+    "updated_at": 1793741211,
+    "available": True,
+    "station": {
+        "id": 1,
+        "name": "tilderadio",
+        "shortcode": "tilderadio",
+        "listen_url": "https://azuracast.tilderadio.org/listen/tilderadio/radio.ogg",
+    },
+    "is_live": False,
+    "broadcast_start": None,
+    "dj": None,
+    "listeners": 5,
+    "now_playing": {
+        "artist": None,
+        "title": "souldies-dramatics-feel it",
+        "text": "souldies-dramatics-feel it",
+        "art": "https://azuracast.tilderadio.org/api/station/tilderadio/art/199f03",
+        "played_at": 1793741079,
+        "duration": 192,
+    },
+    "history": [
+        {
+            "artist": None,
+            "title": "souldies-everytime i see a pretty lady-gemini",
+            "text": "souldies-everytime i see a pretty lady-gemini",
+            "art": "https://azuracast.tilderadio.org/api/station/tilderadio/art/5e46a9",
+            "played_at": 1793740639,
+        },
+        {
+            "artist": "TILDE PROMO (station)",
+            "title": "owly (bots 3)",
+            "text": "TILDE PROMO (station) - owly (bots 3)",
+            "art": "https://azuracast.tilderadio.org/api/station/tilderadio/art/54ac9f",
+            "played_at": 1793740629,
+        },
+    ],
+}
+
+MOUNTS_PAYLOAD = {
+    "station": {
+        "id": 1,
+        "name": "tilderadio",
+        "shortcode": "tilderadio",
+        "listen_url": "https://azuracast.tilderadio.org/listen/tilderadio/radio.ogg",
+        "mounts": [
+            {
+                "name": "/radio.mp3 (192kbps MP3)",
+                "url": "https://azuracast.tilderadio.org/listen/tilderadio/radio.mp3",
+                "bitrate": 192,
+                "format": "mp3",
+                "is_default": False,
+            },
+            {
+                "name": "/320k.mp3 (320kbps MP3)",
+                "url": "https://azuracast.tilderadio.org/listen/tilderadio/320k.mp3",
+                "bitrate": 320,
+                "format": "mp3",
+                "is_default": False,
+            },
+            {
+                "name": "192kbps OGG Vorbis",
+                "url": "https://azuracast.tilderadio.org/listen/tilderadio/radio.ogg",
+                "bitrate": 192,
+                "format": "ogg",
+                "is_default": True,
+            },
+        ],
+    }
+}
+
+
+def client_for(*responses: tuple[str, int, object]) -> httpx.AsyncClient:
+    """An httpx client that answers from ``(url_prefix, status, payload)``."""
+    table = list(responses)
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        for prefix, status, payload in table:
+            if str(request.url).startswith(prefix):
+                return httpx.Response(status, json=payload)
+        return httpx.Response(404, json={})
+
+    return httpx.AsyncClient(transport=httpx.MockTransport(handler))
+
+
+def test_provider_describes_its_catalogue() -> None:
+    provider = TilderadioProvider()
+    assert provider.id == "tilderadio"
+    assert provider.brand == "TILDE RADIO"
+    assert provider.station_source == "tilderadio.org/api/now/"
+    assert "tilderadio.org" in provider.user_agent
+
+    stations = provider.fallback_stations()
+    assert len(stations) == 1
+    station = stations[0]
+    assert station.uid == "tilderadio:tilderadio"
+    assert all("tilderadio.org" in url for url in provider.fallback_urls(station))
+    # the bundled list still advertises a real bitrate, not a URL guess
+    assert station.format_label == "mp3 · 320k"
+
+
+def test_station_from_now_reads_the_single_station() -> None:
+    station = api.station_from_now(NOW_PAYLOAD, fallback.fallback_playlists())
+    assert station is not None
+    assert station.id == "tilderadio"
+    assert station.provider == "tilderadio"
+    assert station.listeners == 5
+    assert station.dj == ""  # nobody is on air
+    assert station.genre_label == "community · eclectic"
+    assert station.now_playing is not None
+    assert station.now_playing.title == "souldies-dramatics-feel it"
+    assert station.now_playing.art_url.endswith("199f03")
+    assert station.last_playing == "souldies-dramatics-feel it"
+
+
+def test_live_dj_is_only_named_when_someone_is_on_air() -> None:
+    live = dict(NOW_PAYLOAD, is_live=True, dj="ffog")
+    station = api.station_from_now(live)
+    assert station is not None and station.dj == "ffog"
+    # the AzuraCast flavour of the same field wins when present
+    nested = dict(NOW_PAYLOAD, live={"is_live": True, "name": "controlfreak"})
+    station = api.station_from_now(nested)
+    assert station is not None and station.dj == "controlfreak"
+
+
+def test_artist_is_split_out_of_the_text_blob() -> None:
+    song = api.song_from_track(
+        {
+            "artist": "TILDE PROMO (station)",
+            "title": "owly (bots 3)",
+            "text": "TILDE PROMO (station) - owly (bots 3)",
+            "played_at": 1793740629,
+        }
+    )
+    assert song is not None
+    assert (song.artist, song.title) == ("TILDE PROMO (station)", "owly (bots 3)")
+
+    # no artist field: the "Artist - Title" text has to carry it
+    song = api.song_from_track(
+        {"title": "Kenny Ken", "text": "Kenny Ken - Amazon Jungle Collection 3"}
+    )
+    assert song is not None
+    assert (song.artist, song.title) == ("Kenny Ken", "Amazon Jungle Collection 3")
+
+
+def test_songs_from_now_puts_the_current_track_first_and_dedupes() -> None:
+    payload = dict(NOW_PAYLOAD)
+    # a feed that repeats the current track in its own history
+    payload["history"] = [NOW_PAYLOAD["now_playing"], *NOW_PAYLOAD["history"]]
+    songs = api.songs_from_now(payload)
+    assert len(songs) == 3
+    assert songs[0].title == "souldies-dramatics-feel it"
+    assert songs[1].title == "souldies-everytime i see a pretty lady-gemini"
+    played = [song.played_at for song in songs]
+    assert played == sorted(played, reverse=True)
+
+
+def test_mounts_become_playlists_best_first() -> None:
+    playlists = api.playlists_from_mounts(MOUNTS_PAYLOAD)
+    assert [entry["bitrate"] for entry in playlists] == ["192", "320", "192"]
+    station = api.station_from_now(
+        {**NOW_PAYLOAD, "now_playing": None}, playlists
+    )
+    assert station is not None
+    assert station.format_label == "mp3 · 320k"
+    assert station.sorted_playlists()[0]["url"].endswith("320k.mp3")
+    assert [entry["quality"] for entry in playlists] == ["high", "highest", "high"]
+
+
+def test_station_without_mounts_falls_back_to_the_listen_url() -> None:
+    station = api.station_from_now(NOW_PAYLOAD)
+    assert station is not None
+    assert len(station.playlists) == 1
+    assert station.playlists[0]["url"].endswith("radio.ogg")
+    assert station.format_label == "ogg · 128k"  # guessed, the feed didn't say
+
+
+@pytest.mark.asyncio
+async def test_fetch_station_caches_what_it_learned() -> None:
+    async with client_for(
+        ("https://tilderadio.org/api/now/", 200, NOW_PAYLOAD),
+        ("https://azuracast.tilderadio.org/api/nowplaying/", 200, MOUNTS_PAYLOAD),
+    ) as client:
+        station = await api.fetch_station(client)
+    assert station.listeners == 5
+    assert station.format_label == "mp3 · 320k"
+
+    cached = api.load_cached_station()
+    assert len(cached) == 1
+    assert cached[0].id == "tilderadio"
+    assert cached[0].now_playing is not None
+    assert cached[0].format_label == "mp3 · 320k"
+
+
+@pytest.mark.asyncio
+async def test_fetch_station_survives_an_unreachable_azuracast() -> None:
+    async with client_for(("https://tilderadio.org/api/now/", 200, NOW_PAYLOAD)) as client:
+        station = await api.fetch_station(client)
+    # no mounts -> the documented listen_url is still a usable stream
+    assert station.playlists[0]["url"].endswith("radio.ogg")
+
+
+@pytest.mark.asyncio
+async def test_history_falls_back_to_the_cache_when_the_feed_is_down() -> None:
+    async with client_for(
+        ("https://tilderadio.org/api/now/", 200, NOW_PAYLOAD)
+    ) as client:
+        songs = await api.fetch_songs(client, "tilderadio")
+    assert len(songs) == 3
+
+    async with client_for(
+        ("https://tilderadio.org/api/now/", 500, {})
+    ) as client:
+        cached = await api.fetch_songs(client, "tilderadio")
+    assert [song.title for song in cached] == [song.title for song in songs]
+
+
+def test_station_cache_is_per_provider() -> None:
+    from beefy_player import config
+
+    assert config.channels_cache_path("tilderadio").name == "channels-tilderadio.json"
+    # SomaFM keeps its historical file name, so upgrades don't lose the cache
+    assert config.channels_cache_path("somafm") == config.CHANNELS_CACHE
+
+
+def test_fallback_station_is_usable_offline() -> None:
+    provider = TilderadioProvider()
+    stations = provider.fallback_stations()
+    assert [station.id for station in stations] == ["tilderadio"]
+    station = stations[0]
+    assert station.listeners == 0
+    assert station.now_playing is None
+    assert {entry["format"] for entry in station.playlists} == {"mp3", "ogg"}