From 058d336a2cf7d5a476eaa923a5c0f86217e6ed2c Mon Sep 17 00:00:00 2001 From: beefcarver Date: Tue, 3 Nov 2026 17:51:23 -0500 Subject: [PATCH] feat(tilderadio): add the tildeverse community stream as a catalogue --- beefy_player/config.py | 38 +- beefy_player/models.py | 39 +- beefy_player/providers/tilderadio/__init__.py | 76 ++++ beefy_player/providers/tilderadio/api.py | 401 ++++++++++++++++++ beefy_player/providers/tilderadio/fallback.py | 56 +++ beefy_player/providers/tilderadio/hero.py | 24 ++ pyproject.toml | 4 +- tests/test_tilderadio.py | 249 +++++++++++ 8 files changed, 883 insertions(+), 4 deletions(-) create mode 100644 beefy_player/providers/tilderadio/__init__.py create mode 100644 beefy_player/providers/tilderadio/api.py create mode 100644 beefy_player/providers/tilderadio/fallback.py create mode 100644 beefy_player/providers/tilderadio/hero.py create mode 100644 tests/test_tilderadio.py diff --git a/beefy_player/config.py b/beefy_player/config.py index e6f520f..f0b852b 100644 --- a/beefy_player/config.py +++ b/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-.json ... for every other catalogue + ~/.cache/beefy-player/schedule-.json last good broadcast schedule ~/.cache/beefy-player/playlists.json resolved .pls -> stream URLs ~/.cache/beefy-player/songs/.json last good song history ~/.cache/beefy-player/artwork/ 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 diff --git a/beefy_player/models.py b/beefy_player/models.py index a398808..a2ce0a7 100644 --- a/beefy_player/models.py +++ b/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 diff --git a/beefy_player/providers/tilderadio/__init__.py b/beefy_player/providers/tilderadio/__init__.py new file mode 100644 index 0000000..59f401b --- /dev/null +++ b/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() diff --git a/beefy_player/providers/tilderadio/api.py b/beefy_player/providers/tilderadio/api.py new file mode 100644 index 0000000..b94ced5 --- /dev/null +++ b/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 diff --git a/beefy_player/providers/tilderadio/fallback.py b/beefy_player/providers/tilderadio/fallback.py new file mode 100644 index 0000000..01133e0 --- /dev/null +++ b/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 [] diff --git a/beefy_player/providers/tilderadio/hero.py b/beefy_player/providers/tilderadio/hero.py new file mode 100644 index 0000000..3a731e7 --- /dev/null +++ b/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]▀▀[/]', +) diff --git a/pyproject.toml b/pyproject.toml index 739340e..2d1d90b 100644 --- a/pyproject.toml +++ b/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", diff --git a/tests/test_tilderadio.py b/tests/test_tilderadio.py new file mode 100644 index 0000000..40a3dea --- /dev/null +++ b/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/`` 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"}