From 50a9a735f735e0aa67ba1281d44e1e9cfc8bcb91 Mon Sep 17 00:00:00 2001 From: beefcarver Date: Tue, 3 Nov 2026 18:44:40 -0500 Subject: [PATCH] feat(bassdrive): add the drum & bass relay and a shared wordmark font --- README.md | 79 ++- beefy_player/models.py | 2 + beefy_player/providers/__init__.py | 4 +- beefy_player/providers/bassdrive/__init__.py | 76 +++ beefy_player/providers/bassdrive/api.py | 517 +++++++++++++++++++ beefy_player/providers/bassdrive/fallback.py | 56 ++ beefy_player/providers/bassdrive/hero.py | 14 + beefy_player/providers/somafm/hero.py | 11 +- beefy_player/providers/tilderadio/hero.py | 25 +- beefy_player/providers/wordmark.py | 91 ++++ pyproject.toml | 4 +- tests/test_bassdrive.py | 278 ++++++++++ tests/test_header.py | 63 ++- tests/test_providers.py | 25 +- 14 files changed, 1166 insertions(+), 79 deletions(-) create mode 100644 beefy_player/providers/bassdrive/__init__.py create mode 100644 beefy_player/providers/bassdrive/api.py create mode 100644 beefy_player/providers/bassdrive/fallback.py create mode 100644 beefy_player/providers/bassdrive/hero.py create mode 100644 beefy_player/providers/wordmark.py create mode 100644 tests/test_bassdrive.py diff --git a/README.md b/README.md index 9850a5b..7e6173e 100644 --- a/README.md +++ b/README.md @@ -8,9 +8,10 @@ own ANSI palette on the terminal's default background so it fits any theme (including transparent terminals). The station catalogue is pluggable. Beefy Player ships with -**[SomaFM](https://somafm.com)** (≈50 curated channels) and +**[SomaFM](https://somafm.com)** (≈50 curated channels), **[tilderadio](https://tilderadio.org)** (the tildeverse's single community -stream), and the UI never talks to a station catalogue directly: it asks a +stream) and **[Bassdrive](https://bassdrive.com)** (worldwide drum & bass), and +the UI never talks to a station catalogue directly: it asks a [provider](#providers) for stations, song history and stream URLs. Press `p` to switch catalogue without restarting. Adding another internet-radio catalogue is a self-contained module, not a rewrite. @@ -41,7 +42,9 @@ implemented here from public endpoints and open protocols. iTerm2 inline-image protocol. Images occupy exactly the ON AIR art cell box and fall back to the procedural ANSI mosaic for other terminals, failed loads, or blocked graphics passthrough. Images are cached in - `~/.cache/beefy-player/artwork`. + `~/.cache/beefy-player/artwork`, letterboxed rather than cropped, and an + image that carries an alpha channel is composited onto black — a transparent + PNG drawn over the mosaic reads as noise behind a station logo. - **Now playing** — artist, title, album, DJ, listeners, format/bitrate and "played N minutes ago", with procedural ANSI-shaded album art that changes per track. When the current track can be found on Bandcamp, a clickable @@ -57,9 +60,10 @@ implemented here from public endpoints and open protocols. - **Pluggable catalogues** — one `providers` module per radio network, with a `p` picker in the UI, `--provider` on the command line, and the choice remembered per catalogue (including the last station you tuned). -- **A hero per catalogue** — the top-left art comes from the provider: SomaFM's - block wordmark, or tilderadio's radio icon converted from its logo PNG to - half-block ANSI by `tools/ansi_art.py`. +- **A hero per catalogue** — the top-left art is the catalogue's own name in a + shared half-block font: SomaFM in the spectrum gradient, tilderadio in the + terminal's default colour, Bassdrive in its two blues. Every hero is three + rows tall, so the header keeps one height. - **Schedule pane** — a single-stream catalogue gets a broadcast schedule instead of a list of one station: who is on air now, since/until when, and the upcoming shows grouped by day. @@ -193,35 +197,45 @@ layout and the guessed icecast mirrors all live in `beefy_player/providers/somafm/`. Nothing outside that package imports it; the app only ever holds a `RadioProvider`. -Two catalogues ship today: +Three catalogues ship today: | slug | catalogue | station list | history | streams | | --- | --- | --- | --- | --- | | `somafm` | [SomaFM](https://somafm.com) | `channels.json` (≈50) | `songs/.json` | `.pls` → icecast mirrors | | `tilderadio` | [tilderadio](https://tilderadio.org) | `/api/now/` (1) | `now_playing` + `history[]` | AzuraCast mounts (mp3/ogg, 320k) | +| `bassdrive` | [Bassdrive](https://bassdrive.com) | icecast status (1) | rolling `display-title` | three mounts + mirrors | -They are deliberately different shapes: one has fifty channels and PLS -playlists, the other has a single stream whose whole catalogue arrives in the -same response as its history. That is the seam working. +They are deliberately different shapes: fifty channels with PLS playlists, a +single AzuraCast stream that carries its own history, and an Icecast relay whose +"now playing" is the *show* rather than the track. That is the seam working. -The tilderadio package also exercises the two optional capabilities: its left -pane is the broadcast schedule (`/api/schedule/`, with the live DJ coming from -`/api/now/`) rather than a list containing one station, and its header art is -the station's own radio icon converted to half blocks. The icon is generated, -not hand-drawn — regenerate it after the logo changes: +The single-stream catalogues also use the optional capabilities: their left +pane is the broadcast schedule rather than a list containing one station, and +they opt out of Bandcamp lookups. Bassdrive is the extreme case — Icecast +reports one title and keeps no history, so the provider maps the title to +`Station.dj`/`presenter` and keeps a rolling record of the shows it has heard. -```bash -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 +### Header wordmarks + +All three heroes come from one font, `beefy_player/providers/wordmark.py`: +four-column capitals six pixels tall, emitted as three rows of half blocks +(`▀`/`▄`/`█`) — the shape of the original SomaFM logo. A provider's `hero.py` +is one line of data: + +```python +# beefy_player/providers/bassdrive/hero.py +HERO = render("BASSDRIVE", ["bright_cyan"] * 4 + ["cyan"] * 5) ``` -`tools/ansi_art.py` crops the transparent margin, resizes to 22×14 pixels (two -per text cell), quantises to the stock 16-colour ANSI palette and emits -run-length encoded rich markup — so the art follows the terminal's palette -instead of pinning RGB values. `--preview out.png` draws the result, which is -far easier to judge than ▀ and ▄. +`render()` takes one optional style per character and groups equal neighbours +into markup spans; with no styles the wordmark inherits the terminal's default +foreground (tilderadio). Add a glyph to `GLYPHS` when a catalogue needs a letter +the font does not have yet — `▀█▀` is the I, one pixel thick on purpose. + +Image logos are a different problem: `tools/ansi_art.py` converts a PNG into +half-block art (`--key-white` for logos on white, `--preview out.png` to see the +result). No shipped catalogue needs it today, but it is the tool to reach for +when a station's identity is a picture rather than a word. Adding a catalogue: @@ -347,8 +361,10 @@ beefy_player/ config.py XDG paths, settings, BEEFY_* env helpers providers/ base.py RadioProvider interface + HTTP/playlist toolkit + wordmark.py the half-block font every hero is drawn with somafm/ the SomaFM catalogue (API, hero, bundled fallback list) tilderadio/ the tilderadio catalogue (one station, hero, schedule) + bassdrive/ Bassdrive (Icecast status, show schedule, rolling history) player.py mpv over a Unix-socket JSON IPC channel spectrum.py analyser (parec / Core Audio tap / ffmpeg / sim) macos_tap.py macOS output capture (ctypes + Core Audio) @@ -382,6 +398,14 @@ tools/ the provider in place: the old stream is stopped, workers for the previous catalogue are cancelled, the hero height and left pane follow the new provider, and the station list (or schedule) reloads from cache first. +- `beefy_player/providers/bassdrive/api.py` — Icecast's keyless + `status-json.xsl` (mounts, bitrates, listeners, `display-title`) plus the + homepage schedule grid, whose slots carry absolute `` + values. A memoised fetch and a lock keep a cold start to one homepage + download; the display-title's show is matched against the running slot for + "live from ". +- `beefy_player/providers/wordmark.py` — the half-block capitals behind every + hero, with per-character style spans. - `beefy_player/widgets/schedule.py` — the left pane for single-stream catalogues: the live slot with its `since`/`until` window, then upcoming shows grouped by local day, with `… +N more` when the pane runs out. @@ -419,7 +443,8 @@ tools/ ~/.cache/beefy-player/channels-.json last good list, other catalogues ~/.cache/beefy-player/schedule-.json last good broadcast schedule ~/.cache/beefy-player/playlists.json resolved stream URLs -~/.cache/beefy-player/songs/.json song history snapshots +~/.cache/beefy-player/songs/.json SomaFM song history snapshots +~/.cache/beefy-player/songs/

-.json history for every other catalogue ~/.cache/beefy-player/artwork/ cached album/channel artwork ``` @@ -488,6 +513,8 @@ so favourites and volume survive the rename. - SomaFM is listener-supported: if you enjoy it, [donate](https://somafm.com/support/). - tilderadio is a volunteer-run community station: [drop by](https://tilderadio.org), or show up on the [schedule](https://tilderadio.org/schedule/) with a set. +- Bassdrive is free and listener-supported too: [bassdrive.com](https://bassdrive.com) + has the shows, archives and the shop that keeps the relays running. ## NixOS (optional aside) diff --git a/beefy_player/models.py b/beefy_player/models.py index a2ce0a7..67c8649 100644 --- a/beefy_player/models.py +++ b/beefy_player/models.py @@ -46,6 +46,8 @@ class Broadcast: name: str dj: str = "" + #: Where the broadcast comes from ("Poznan, Poland"), when the feed says. + location: str = "" start: datetime | None = None end: datetime | None = None #: True for the broadcast that is on air right now. diff --git a/beefy_player/providers/__init__.py b/beefy_player/providers/__init__.py index cda0b84..ead4713 100644 --- a/beefy_player/providers/__init__.py +++ b/beefy_player/providers/__init__.py @@ -17,6 +17,7 @@ Adding a catalogue is meant to be a small, contained change: from __future__ import annotations from .base import RadioProvider +from .bassdrive import BassdriveProvider from .somafm import SomaFMProvider from .tilderadio import TilderadioProvider @@ -30,7 +31,8 @@ __all__ = [ #: Every catalogue the app can talk to, keyed by its slug. PROVIDERS: dict[str, RadioProvider] = { - provider.id: provider for provider in (SomaFMProvider(), TilderadioProvider()) + provider.id: provider + for provider in (SomaFMProvider(), TilderadioProvider(), BassdriveProvider()) } #: Provider used when nothing else is configured. diff --git a/beefy_player/providers/bassdrive/__init__.py b/beefy_player/providers/bassdrive/__init__.py new file mode 100644 index 0000000..006a52d --- /dev/null +++ b/beefy_player/providers/bassdrive/__init__.py @@ -0,0 +1,76 @@ +"""Bassdrive: worldwide drum & bass, on air since the dial-up days. + +One stream with three mounts, a schedule full of live DJ shows and a +``display-title`` that names the show rather than the track. Everything +Bassdrive-specific — the Icecast status schema, the homepage schedule grid, the +rolling show history and the bundled fallback — lives in this package; the app +only sees :class:`BassdriveProvider`. +""" + +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__ = ["BassdriveProvider"] + + +class BassdriveProvider(RadioProvider): + id = "bassdrive" + name = "Bassdrive" + homepage = "bassdrive.com" + brand = "BASSDRIVE" + tagline = "worldwide drum & bass · live shows" + station_source = "bassdrive.com + icecast status" + history_source = "icecast status (rolling)" + + # The wordmark in the station's own two blues, in the shared half-block + # font; the caption is the station's own description. + hero = hero_art.HERO + hero_lines = ( + "BASSDRIVE", + "WORLDWIDE DRUM & BASS", + "LIVE SHOWS 24/7", + ) + + # One stream and a schedule full of DJ slots: the left pane is the schedule. + has_schedule = True + # DJ-mix station: the rotation is not a Bandcamp catalogue. + bandcamp = False + + @property + def user_agent(self) -> str: + return f"{config.USER_AGENT}; +https://bassdrive.com" + + 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/bassdrive/api.py b/beefy_player/providers/bassdrive/api.py new file mode 100644 index 0000000..4148f57 --- /dev/null +++ b/beefy_player/providers/bassdrive/api.py @@ -0,0 +1,517 @@ +"""Bassdrive's public endpoints, and nothing but. + +* ``https://ice.bassdrive.net/status-json.xsl`` -> Icecast status: every mount + (mp3/aac+, bitrates), the listener counts and the current ``display-title``, + which for a DJ-mix stream is the *show*: ``"Ben XO :: LIVE :: Night Shift + Over"``. +* ``http://www.bassdrive.com/`` -> the homepage carries the whole week's + schedule, every slot stamped with an absolute ````. + +A DJ-mix station has no track history: Icecast reports one title and forgets +it. So "now playing" is the show (mapped onto ``Station.dj``/``presenter`` and +a :class:`Song`), and the history is a rolling record of the shows we have +heard since the app started — the unit a listener actually cares about here. +""" + +from __future__ import annotations + +import asyncio +import html as html_module +import re +import time +from collections.abc import Iterable +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +import httpx + +from ... import config +from ...models import Broadcast, Song, Station, parse_date, to_int +from ..base import fetch_json + +PROVIDER_ID = "bassdrive" +STATUS_URL = "https://ice.bassdrive.net/status-json.xsl" +SCHEDULE_URL = "https://www.bassdrive.com/" + +DEFAULT_STATION_ID = "bassdrive" +STATION_IMAGE = "https://www.bassdrive.com/images/logo.png" +DESCRIPTION = "worldwide drum & bass radio — live DJ shows around the clock" +GENRE = "drum & bass|jungle" + +WEEK = timedelta(days=7) +#: Upcoming slots kept in the pane (a week is plenty for a schedule). +SCHEDULE_LIMIT = 40 +#: Shows remembered in LAST PLAYED. +HISTORY_LIMIT = 20 + + +# --------------------------------------------------------------------------- # +# status-json -> Station / Song +# --------------------------------------------------------------------------- # + + +def _sources(payload: Any) -> list[dict[str, Any]]: + stats = payload.get("icestats") if isinstance(payload, dict) else None + source = stats.get("source") if isinstance(stats, dict) else None + if isinstance(source, dict): + return [source] + if isinstance(source, list): + return [entry for entry in source if isinstance(entry, dict)] + return [] + + +def _quality(bitrate: int) -> str: + if bitrate >= 128: + return "highest" + if bitrate >= 48: + return "high" + return "low" if bitrate else "" + + +def playlists_from_status(payload: Any) -> list[dict[str, str]]: + playlists: list[dict[str, str]] = [] + seen: set[str] = set() + for source in _sources(payload): + url = source.get("listenurl") + if not isinstance(url, str) or not url.startswith(("http://", "https://")): + continue + if url in seen: + continue + seen.add(url) + bitrate = to_int(source.get("bitrate")) + kind = str(source.get("server_type") or "").lower() + fmt = "mp3" if "mpeg" in kind else "aacp" if "aac" in kind else "" + playlists.append( + { + "url": url, + "format": fmt, + "quality": _quality(bitrate), + "bitrate": str(bitrate) if bitrate else "", + } + ) + return playlists + + +def status_listeners(payload: Any) -> int: + """Mounts split the audience, so the station's figure is their sum.""" + return sum(to_int(source.get("listeners")) for source in _sources(payload)) + + +def display_title(payload: Any) -> str: + for source in _sources(payload): + for key in ("display-title", "title"): + value = source.get(key) + if isinstance(value, str) and value.strip(): + return value.strip() + return "" + + +def split_title(raw: str) -> tuple[str, str, bool]: + """``"Ben XO :: LIVE :: Night Shift Over [promo]"`` -> DJ, show, live. + + Bassdrive's title convention is ``DJ :: LIVE :: Show``, with promotional + brackets tacked on the end — but it is not enforced: plenty of shows go out + as ``"Sohl Patrol LIVE"``, so a trailing ``LIVE`` counts too. Anything + unexpected degrades to "the whole string is the show". + """ + cleaned = re.sub(r"\[[^\]]*\]", " ", raw) + parts = [part.strip() for part in cleaned.split("::") if part.strip()] + live = any(part.casefold() == "live" for part in parts) + parts = [part for part in parts if part.casefold() != "live"] + + if parts: + trailing = re.search(r"(?i)\s+live$", parts[-1]) + if trailing: + live = True + trimmed = parts[-1][: trailing.start()].strip() + parts = parts[:-1] + ([trimmed] if trimmed else []) + + if not parts: + return "", raw.strip(), live + if len(parts) == 1: + return "", parts[0], live + return parts[0], parts[-1], live + + +def song_from_title(raw: str) -> Song | None: + dj, show, _live = split_title(raw) + if not (dj or show): + return None + # With no DJ in the title the station stands in as the "artist": a panel + # that says "Unknown artist" for every show would look broken. + return Song(artist=dj or "Bassdrive", title=show) + + +def _same_show(entry: Broadcast, show: str) -> bool: + return bool(show) and entry.name.strip().casefold() == show.strip().casefold() + + +def _slot_for(schedule: Iterable[Broadcast], show: str) -> Broadcast | None: + if not show: + return None + return next((entry for entry in schedule if _same_show(entry, show)), None) + + +def resolve_now( + payload: Any, + schedule: Iterable[Broadcast] = (), + now: datetime | None = None, +) -> tuple[str, str, bool, Broadcast | None]: + """Who is on, according to the stream *and* the schedule. + + The display-title often forgets the ``LIVE`` marker; a show whose slot + covers now counts as live even when the title is just its name. + """ + now = now or datetime.now(timezone.utc) + schedule = list(schedule) + dj, show, live = split_title(display_title(payload)) + slot = _slot_for(schedule, show) + if ( + slot is not None + and slot.start is not None + and slot.end is not None + and slot.start <= now < slot.end + ): + live = True + return dj, show, live, slot + + +def station_from_status( + payload: Any, schedule: Iterable[Broadcast] = () +) -> Station | None: + playlists = playlists_from_status(payload) + raw = display_title(payload) + if not playlists and not raw: + return None + + dj, show, live, slot = resolve_now(payload, schedule) + location = slot.location if slot is not None else "" + if live: + presenter = f"live from {location}" if location else "live now" + else: + presenter = "rotation · drum & bass" + + station = Station( + id=DEFAULT_STATION_ID, + title="Bassdrive", + description=DESCRIPTION, + genre=GENRE, + dj=dj, + presenter=presenter, + image=STATION_IMAGE, + listeners=status_listeners(payload), + playlists=playlists, + provider=PROVIDER_ID, + ) + if show: + # The "artist" of a show-based stream is the station itself. + station.now_playing = Song(artist=dj or "Bassdrive", title=show) + station.last_playing = raw + return station + + +def live_broadcast( + payload: Any, + schedule: Iterable[Broadcast], + now: datetime | None = None, +) -> Broadcast | None: + dj, show, live, slot = resolve_now(payload, schedule, now) + if not live or not show: + return None + if slot is not None: + return Broadcast( + name=show, + dj=dj, + location=slot.location, + start=slot.start, + end=slot.end, + live=True, + ) + return Broadcast(name=show, dj=dj, live=True) + + +# --------------------------------------------------------------------------- # +# the homepage schedule +# --------------------------------------------------------------------------- # + +_SPAN_LOCATION = re.compile(r'(.*?)', re.S) + + +def _text(value: str) -> str: + return re.sub(r"\s+", " ", html_module.unescape(re.sub(r"<[^>]+>", " ", value))).strip() + + +def parse_schedule(page: str) -> list[Broadcast]: + """Pull the weekly grid out of the homepage, one ``

  • `` per slot.""" + broadcasts: list[Broadcast] = [] + for chunk in re.split(r"(.*?)

    ', chunk, re.S) + if title is None: + continue + block = title.group(1) + location_match = _SPAN_LOCATION.search(block) + location = _text(location_match.group(1)) if location_match else "" + name = _text(_SPAN_LOCATION.sub(" ", block)) + if not name: + continue + broadcasts.append( + Broadcast(name=name, location=location, start=start, end=end) + ) + broadcasts.sort(key=lambda item: item.start or datetime.now(timezone.utc)) + return broadcasts + + +def roll_forward( + entries: Iterable[Broadcast], now: datetime, limit: int = SCHEDULE_LIMIT +) -> list[Broadcast]: + """The homepage only shows one week; repeat it so the pane is never empty.""" + upcoming: list[Broadcast] = [] + for entry in entries: + start, end = entry.start, entry.end + if start is None: + continue + for _ in range(3): + if end is None or end >= now: + break + start += WEEK + end = end + WEEK if end is not None else None + if end is not None and end < now: + continue + upcoming.append( + Broadcast( + name=entry.name, + dj=entry.dj, + location=entry.location, + start=start, + end=end, + ) + ) + upcoming.sort(key=lambda item: item.start or now) + return upcoming[:limit] + + +# --------------------------------------------------------------------------- # +# caching +# --------------------------------------------------------------------------- # + +#: One homepage fetch serves both the station and the schedule worker; the +#: lock keeps a cold start from downloading the same grid twice at once. +_MEMO_TTL = 120.0 +_schedule_memo: tuple[float, list[Broadcast]] | None = None +_schedule_lock = asyncio.Lock() + + +def _memo_schedule() -> list[Broadcast]: + if _schedule_memo is None: + return [] + fetched, entries = _schedule_memo + if time.time() - fetched > _MEMO_TTL: + return [] + return list(entries) + + +def _remember_schedule(entries: list[Broadcast]) -> None: + global _schedule_memo + _schedule_memo = (time.time(), entries) + + +def reset_schedule_memo() -> None: + """Forget the in-process schedule (the tests use this).""" + global _schedule_memo + _schedule_memo = None + + +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" + + +async def schedule_entries(client: httpx.AsyncClient) -> list[Broadcast]: + """The weekly grid: memo, then the homepage, then the cached copy.""" + remembered = _memo_schedule() + if remembered: + return remembered + async with _schedule_lock: + remembered = _memo_schedule() + if remembered: + return remembered + try: + response = await client.get( + SCHEDULE_URL, headers={"Accept": "text/html"} + ) + response.raise_for_status() + entries = parse_schedule(response.text) + except httpx.HTTPError: + entries = [] + if not entries: + return load_cached_schedule() + _remember_schedule(entries) + config.write_json( + config.schedule_cache_path(PROVIDER_ID), + { + "fetched": time.time(), + "schedule": [ + { + "name": item.name, + "location": item.location, + "start": item.start.isoformat() if item.start else "", + "end": item.end.isoformat() if item.end else "", + } + for item in entries + ], + }, + ) + return entries + + +def load_cached_station() -> list[Station]: + raw = config.read_json(config.channels_cache_path(PROVIDER_ID)) + status = raw.get("status") if isinstance(raw, dict) else None + if status is None: + return [] + station = station_from_status(status, load_cached_schedule()) + return [station] if station is not None else [] + + +def load_cached_schedule() -> list[Broadcast]: + raw = config.read_json(config.schedule_cache_path(PROVIDER_ID)) + entries = raw.get("schedule") if isinstance(raw, dict) else None + if not isinstance(entries, list): + return [] + return _schedule_from_cache(entries) + + +def _schedule_from_cache(entries: list[Any]) -> list[Broadcast]: + broadcasts: list[Broadcast] = [] + for entry in entries: + if not isinstance(entry, dict): + continue + start = parse_date(entry.get("start")) + end = parse_date(entry.get("end")) + name = str(entry.get("name") or "").strip() + if not name or start is None: + continue + broadcasts.append( + Broadcast( + name=name, + location=str(entry.get("location") or "").strip(), + start=start, + end=end, + ) + ) + broadcasts.sort(key=lambda item: item.start or datetime.now(timezone.utc)) + return broadcasts + + +def load_cached_songs(station_id: str) -> list[Song]: + raw = config.read_json(_songs_cache_path(station_id)) + entries = raw.get("songs") if isinstance(raw, dict) else None + if not isinstance(entries, list): + return [] + songs: list[Song] = [] + for entry in entries: + if not isinstance(entry, dict): + continue + song = Song( + title=str(entry.get("title") or "").strip(), + artist=str(entry.get("artist") or "").strip(), + played_at=parse_date(entry.get("played_at")), + ) + if song.title or song.artist: + songs.append(song) + return songs + + +def _songs_to_cache(songs: Iterable[Song]) -> list[dict[str, Any]]: + return [ + { + "title": song.title, + "artist": song.artist, + "played_at": song.played_at.isoformat() if song.played_at else "", + } + for song in songs + ] + + +# --------------------------------------------------------------------------- # +# fetching +# --------------------------------------------------------------------------- # + + +async def fetch_station(client: httpx.AsyncClient) -> Station: + try: + payload = await fetch_json(client, STATUS_URL) + except RuntimeError as exc: + raise RuntimeError(f"could not fetch the Bassdrive status: {exc}") from None + + # The schedule tells the station *where* a live show is coming from, so a + # cold start fetches it here; the schedule worker then reuses the memo. + schedule = load_cached_schedule() or await schedule_entries(client) + station = station_from_status(payload, schedule) + if station is None: + raise RuntimeError("Bassdrive returned no station data") + + config.write_json( + config.channels_cache_path(PROVIDER_ID), + {"fetched": time.time(), "status": payload}, + ) + return station + + +async def fetch_songs(client: httpx.AsyncClient, station_id: str) -> list[Song]: + try: + payload = await fetch_json(client, STATUS_URL, attempts=1) + except RuntimeError: + return load_cached_songs(station_id) + + current = song_from_title(display_title(payload)) + if current is None: + return load_cached_songs(station_id) + + history = load_cached_songs(station_id) + now = datetime.now(timezone.utc) + if history and history[0].key() == current.key(): + # Same show still on air: keep the time we first heard it, so the pane + # says how long it has been running, not how long since the last poll. + current.played_at = history[0].played_at + history = [current, *history[1:]] + else: + current.played_at = now + history = [current] + [song for song in history if song.key() != current.key()] + history = history[:HISTORY_LIMIT] + + config.write_json( + _songs_cache_path(station_id), + {"fetched": time.time(), "songs": _songs_to_cache(history)}, + ) + return history + + +async def fetch_schedule(client: httpx.AsyncClient) -> list[Broadcast]: + now = datetime.now(timezone.utc) + + live: Broadcast | None = None + try: + status = await fetch_json(client, STATUS_URL, attempts=1) + except RuntimeError: + status = None + + entries = await schedule_entries(client) + upcoming = roll_forward(entries, now) + if status is not None: + live = live_broadcast(status, entries, now) + if live is not None: + key = live.key() + upcoming = [item for item in upcoming if item.key() != key] + return [live, *upcoming] + return upcoming diff --git a/beefy_player/providers/bassdrive/fallback.py b/beefy_player/providers/bassdrive/fallback.py new file mode 100644 index 0000000..aa3190b --- /dev/null +++ b/beefy_player/providers/bassdrive/fallback.py @@ -0,0 +1,56 @@ +"""Bundled station, so a cold offline start still makes sense. + +Bassdrive is one station with three mounts and a handful of mirror hosts; the +streams below are the ones its own ``.m3u`` files list. The live status-json +always wins when it is reachable. +""" + +from __future__ import annotations + +from typing import Any + +# (url, format, quality, bitrate) +_MOUNTS: tuple[tuple[str, str, str, str], ...] = ( + ("https://ice.bassdrive.net:443/stream", "mp3", "highest", "128"), + ("https://ice.bassdrive.net:443/stream56", "aacp", "high", "56"), + ("https://ice.bassdrive.net:443/stream32", "aacp", "low", "32"), +) + +#: Alternative servers from bassdrive.m3u, tried when the primary is down. +_MIRRORS: tuple[str, ...] = ( + "http://chi.bassdrive.co:80", + "http://bassdrive.radioca.st:8702", + "http://stream.bassdrive.uk:8200", + "http://au.bassdrive.co:8000", +) + +_SERVER_TYPE = {"mp3": "audio/mpeg", "aacp": "audio/aacp"} + + +def fallback_status() -> dict[str, Any]: + """A ``status-json.xsl``-shaped payload, so the normal parser can eat it.""" + return { + "icestats": { + "source": [ + { + "listenurl": url, + "bitrate": int(bitrate), + "server_type": _SERVER_TYPE.get(fmt, "audio/mpeg"), + "listeners": 0, + "display-title": "", + } + for url, fmt, _quality, bitrate in _MOUNTS + ] + } + } + + +def fallback_stations() -> list[Any]: + from .api import station_from_status + + station = station_from_status(fallback_status()) + return [station] if station is not None else [] + + +def known_stream_urls() -> list[str]: + return [url for url, _format, _quality, _bitrate in _MOUNTS] + list(_MIRRORS) diff --git a/beefy_player/providers/bassdrive/hero.py b/beefy_player/providers/bassdrive/hero.py new file mode 100644 index 0000000..715277c --- /dev/null +++ b/beefy_player/providers/bassdrive/hero.py @@ -0,0 +1,14 @@ +"""Bassdrive's header hero: its wordmark in the station's two blues. + +The logo paints "BASS" in the pale blue and "DRIVE" in the deeper one; those +map onto the stock palette's bright cyan and cyan, so the wordmark still reads +as Bassdrive without hard-coded RGB. +""" + +from __future__ import annotations + +from ..wordmark import render + +HERO: tuple[str, ...] = render( + "BASSDRIVE", ["bright_cyan"] * 4 + ["cyan"] * 5 +) diff --git a/beefy_player/providers/somafm/hero.py b/beefy_player/providers/somafm/hero.py index d2ee4b1..bd226c5 100644 --- a/beefy_player/providers/somafm/hero.py +++ b/beefy_player/providers/somafm/hero.py @@ -1,17 +1,16 @@ """SomaFM's header hero: the block-letter wordmark and the three lines beside it. -The glyphs are plain half-blocks; the header paints them with the app's +The glyphs come from the shared half-block font +(:mod:`beefy_player.providers.wordmark`); the header paints them with the app's spectrum gradient (``hero_gradient``), so the logo takes on the terminal's own colours instead of carrying RGB values around. """ from __future__ import annotations -LOGO: tuple[str, ...] = ( - "▄▀▀▀ ▄▀▀▄ █▄ ▄█ ▄▀▀▄", - "▀▀▀▄ █ █ █ █ █ █▀▀█", - "▄▄▄▀ ▀▄▄▀ █ █ █ █", -) +from ..wordmark import render + +LOGO: tuple[str, ...] = render("SOMAFM") TAGLINES: tuple[str, ...] = ( "SOMA FM RADIO", diff --git a/beefy_player/providers/tilderadio/hero.py b/beefy_player/providers/tilderadio/hero.py index 3a731e7..d852a49 100644 --- a/beefy_player/providers/tilderadio/hero.py +++ b/beefy_player/providers/tilderadio/hero.py @@ -1,24 +1,11 @@ -"""Generated header art — do not edit by hand. +"""tilderadio's header hero: its name in the shared half-block font. -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. +No colours of its own — the wordmark inherits the terminal's default +foreground, so it looks like part of the chrome rather than a logo pasted in. """ 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]▀▀[/]', -) +from ..wordmark import render + +HERO: tuple[str, ...] = render("TILDERADIO") diff --git a/beefy_player/providers/wordmark.py b/beefy_player/providers/wordmark.py new file mode 100644 index 0000000..8416e7f --- /dev/null +++ b/beefy_player/providers/wordmark.py @@ -0,0 +1,91 @@ +"""A half-block wordmark font for the header hero. + +Every shipped catalogue draws its brand the same way: four-column capitals, +six pixels tall, emitted as three rows of half blocks (``▀``/``▄``/``█``) — the +same shape as the original SomaFM logo. Catalogues differ only in the word +they spell and the ANSI colours they paint it with, so the font lives here and +each provider's ``hero.py`` is one line of data. + +Glyphs are 6x4 bitmaps (``M`` is 6x5). Two pixel rows become one text row, so +a wordmark is always three rows tall and the header keeps a fixed height. +""" + +from __future__ import annotations + +from collections.abc import Sequence + +#: 6-row bitmap per capital; all rows of one glyph are the same width. +GLYPHS: dict[str, tuple[str, ...]] = { + "A": (".##.", "#..#", "####", "#..#", "#..#", "#..#"), + "B": ("###.", "#..#", "###.", "#..#", "#..#", "###."), + "D": ("###.", "#..#", "#..#", "#..#", "#..#", "###."), + "E": ("####", "#...", "###.", "#...", "#...", "####"), + "F": ("####", "#...", "###.", "#...", "#...", "#..."), + "I": ("###", ".#.", ".#.", ".#.", ".#.", "###"), + "L": ("#...", "#...", "#...", "#...", "#...", "####"), + # 5 wide on purpose: the M is the odd one out in the original wordmark. + "M": ("#...#", "##.##", "#.#.#", "#.#.#", "#...#", "#...#"), + "O": (".##.", "#..#", "#..#", "#..#", "#..#", ".##."), + "R": ("###.", "#..#", "###.", "#.#.", "#..#", "#..#"), + "S": (".###", "#...", "###.", "...#", "...#", "###."), + "T": ("###", ".#.", ".#.", ".#.", ".#.", ".#."), + "V": ("#..#", "#..#", "#..#", "#..#", ".##.", ".##."), +} + +ROWS = 3 +_GAP = " " + + +def _cell(top: str, bottom: str) -> str: + if top == "#" and bottom == "#": + return "█" + if top == "#": + return "▀" + if bottom == "#": + return "▄" + return " " + + +def _glyph_rows(char: str) -> tuple[str, ...]: + bitmap = GLYPHS.get(char) + if bitmap is None: + raise KeyError(f"no wordmark glyph for {char!r}") + return tuple( + "".join(_cell(bitmap[row * 2][col], bitmap[row * 2 + 1][col]) for col in range(len(bitmap[0]))) + for row in range(ROWS) + ) + + +def render(text: str, styles: Sequence[str | None] | None = None) -> tuple[str, ...]: + """Spell ``text`` as three rows of half blocks. + + ``styles`` gives one rich-markup style per character (``None`` leaves it in + the terminal's default colour). Contiguous runs of the same style share a + markup span, so ``render("BASSDRIVE", ...)`` paints the first four letters + and the last five separately. + """ + if styles is not None and len(styles) != len(text): + raise ValueError("styles must provide one entry per character") + + glyphs = [_glyph_rows(char.upper()) for char in text] + rows: list[str] = [] + for row in range(ROWS): + cells: list[tuple[str | None, str]] = [] + for index, glyph in enumerate(glyphs): + style = styles[index] if styles is not None else None + if index: + cells.append((style, _GAP)) + cells.extend((style, char) for char in glyph[row]) + while cells and cells[-1][1] == " ": + cells.pop() + + segments: list[tuple[str | None, str]] = [] + for style, char in cells: + if segments and segments[-1][0] == style: + segments[-1] = (style, segments[-1][1] + char) + else: + segments.append((style, char)) + rows.append( + "".join(f"[{style}]{body}[/]" if style else body for style, body in segments) + ) + return tuple(rows) diff --git a/pyproject.toml b/pyproject.toml index 2d1d90b..8cc8beb 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 and tilderadio included)" +description = "A flashy terminal player for internet radio — live spectrum analyser, MPRIS / macOS Now Playing support, song history and pluggable station catalogues (SomaFM, tilderadio and Bassdrive 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", "tilderadio"] +keywords = ["radio", "internet-radio", "tui", "terminal", "mpris", "audio", "somafm", "tilderadio", "bassdrive"] classifiers = [ "Environment :: Console", "Intended Audience :: End Users/Desktop", diff --git a/tests/test_bassdrive.py b/tests/test_bassdrive.py new file mode 100644 index 0000000..6440d4e --- /dev/null +++ b/tests/test_bassdrive.py @@ -0,0 +1,278 @@ +"""Bassdrive: Icecast status, show titles, the homepage schedule, rolling history.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import httpx +import pytest + +from beefy_player.models import Broadcast +from beefy_player.providers.bassdrive import BassdriveProvider, api + +STATUS = { + "icestats": { + "source": [ + { + "listenurl": "https://ice.bassdrive.net:443/stream", + "bitrate": 128, + "server_type": "audio/mpeg", + "listeners": 31, + "display-title": "Ben XO :: LIVE :: Night Shift Over [discord.gg/bassdrive]", + }, + { + "listenurl": "https://ice.bassdrive.net:443/stream56", + "bitrate": 56, + "server_type": "audio/aacp", + "listeners": 5, + "display-title": "Ben XO :: LIVE :: Night Shift Over [discord.gg/bassdrive]", + }, + { + "listenurl": "https://ice.bassdrive.net:443/stream32", + "bitrate": 32, + "server_type": "audio/aacp", + "listeners": 2, + "display-title": "Ben XO :: LIVE :: Night Shift Over [discord.gg/bassdrive]", + }, + ] + } +} + +# Trimmed from the real homepage: the same markup the site ships. +SCHEDULE_PAGE = """ + +""" + + +@pytest.fixture(autouse=True) +def fresh_memo(): + api.reset_schedule_memo() + yield + api.reset_schedule_memo() + + +def test_split_title_reads_both_conventions() -> None: + assert api.split_title("Ben XO :: LIVE :: Night Shift Over [promo]") == ( + "Ben XO", + "Night Shift Over", + True, + ) + # plenty of shows simply append LIVE + assert api.split_title("Sohl Patrol LIVE") == ("", "Sohl Patrol", True) + assert api.split_title("Bassdrive Auto DJ") == ("", "Bassdrive Auto DJ", False) + + +def test_mounts_become_playlists_best_first_and_listeners_add_up() -> None: + playlists = api.playlists_from_status(STATUS) + assert [entry["bitrate"] for entry in playlists] == ["128", "56", "32"] + assert [entry["quality"] for entry in playlists] == ["highest", "high", "low"] + assert playlists[0]["format"] == "mp3" and playlists[1]["format"] == "aacp" + assert api.status_listeners(STATUS) == 38 + + +def test_parse_schedule_reads_the_homepage_grid() -> None: + entries = api.parse_schedule(SCHEDULE_PAGE) + assert [entry.name for entry in entries] == ["DrumObsession", "Get Dipped"] + assert entries[0].location == "Poznan, Poland" + assert entries[0].start == datetime(2026, 11, 3, 11, tzinfo=timezone(timedelta(hours=-5))) + assert entries[0].end == datetime(2026, 11, 3, 13, tzinfo=timezone(timedelta(hours=-5))) + # the schedule list itself is not a schedule slot + assert api.parse_schedule("
    • nothing here
    ") == [] + + +def test_roll_forward_repeats_the_week() -> None: + start = datetime(2026, 11, 3, 11, tzinfo=timezone(timedelta(hours=-5))) + entry = Broadcast( + name="DrumObsession", + start=start, + end=start + timedelta(hours=2), + ) + now = start + timedelta(days=1) + rolled = api.roll_forward([entry], now) + assert len(rolled) == 1 + assert rolled[0].start == start + timedelta(days=7) + assert rolled[0].start >= now + # an entry that is over and cannot be rolled into the future is dropped + far = start + timedelta(days=40) + assert api.roll_forward([entry], far) == [] + + +def test_station_maps_the_title_and_finds_the_live_location() -> None: + now = datetime.now(timezone.utc) + slot = Broadcast( + name="Night Shift Over", + location="London, UK", + start=now - timedelta(minutes=30), + end=now + timedelta(minutes=30), + ) + station = api.station_from_status(STATUS, [slot]) + assert station is not None + assert station.id == "bassdrive" + assert station.dj == "Ben XO" + assert station.presenter == "live from London, UK" + assert station.now_playing is not None + assert station.now_playing.title == "Night Shift Over" + assert station.now_playing.artist == "Ben XO" + assert station.listeners == 38 + assert station.format_label == "mp3 · 128k" + + +def test_a_title_without_a_live_marker_still_counts_when_the_slot_is_running() -> None: + now = datetime.now(timezone.utc) + status = { + "icestats": { + "source": [ + { + "listenurl": "https://ice.bassdrive.net:443/stream", + "bitrate": 128, + "server_type": "audio/mpeg", + "listeners": 3, + "display-title": "Sohl Patrol", + } + ] + } + } + slot = Broadcast( + name="Sohl Patrol", + location="White Plains, NY, USA", + start=now - timedelta(minutes=10), + end=now + timedelta(minutes=50), + ) + live = api.live_broadcast(status, [slot], now) + assert live is not None and live.live is True + assert live.location == "White Plains, NY, USA" + + +def test_station_without_title_or_mounts_is_not_a_station() -> None: + assert api.station_from_status({"icestats": {}}) is None + + +def client_for( + status_payload, + page: str = "", + *, + status_code: int = 200, + page_code: int = 200, +) -> httpx.AsyncClient: + def handler(request: httpx.Request) -> httpx.Response: + if str(request.url).endswith("status-json.xsl"): + return httpx.Response(status_code, json=status_payload) + return httpx.Response(page_code, text=page) + + return httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + +async def songs_for(payload, station_id: str = "bassdrive"): + async with client_for(payload) as client: + return await api.fetch_songs(client, station_id) + + +@pytest.mark.asyncio +async def test_fetch_schedule_puts_the_live_show_first_and_caches_the_grid() -> None: + now = datetime.now(timezone.utc) + payload = { + "icestats": { + "source": [ + { + "listenurl": "https://ice.bassdrive.net:443/stream", + "bitrate": 128, + "server_type": "audio/mpeg", + "listeners": 9, + "display-title": "Get Dipped LIVE", + } + ] + } + } + # the grid is generated relative to now so the live slot is really running + local = now.astimezone(timezone(timedelta(hours=-5))) + start = local - timedelta(minutes=15) + end = local + timedelta(minutes=45) + page = SCHEDULE_PAGE.replace("2026-11-03T19:00:00-05:00", start.isoformat()).replace( + "2026-11-03T21:00:00-05:00", end.isoformat() + ) + async with client_for(payload, page) as client: + broadcast_list = await api.fetch_schedule(client) + + assert broadcast_list[0].live is True + assert broadcast_list[0].name == "Get Dipped" + assert broadcast_list[0].location == "Los Angeles, CA" + assert broadcast_list[0].start.hour == start.hour + assert all(not item.live for item in broadcast_list[1:]) + # the grid was written for offline starts + assert api.load_cached_schedule() + # ... and the second fetch reuses the memo instead of the network + async with client_for(payload, page_code=500) as client: + assert await api.fetch_schedule(client) + + +@pytest.mark.asyncio +async def test_history_is_a_rolling_record_of_shows() -> None: + first = await songs_for(STATUS) + assert [song.title for song in first] == ["Night Shift Over"] + + # same show: the first-seen time is kept + again = await songs_for(STATUS) + assert again[0].played_at == first[0].played_at + + other = { + "icestats": { + "source": [ + { + "listenurl": "https://ice.bassdrive.net:443/stream", + "bitrate": 128, + "server_type": "audio/mpeg", + "listeners": 1, + "display-title": "Sohl Patrol LIVE", + } + ] + } + } + moved = await songs_for(other) + assert [song.title for song in moved] == ["Sohl Patrol", "Night Shift Over"] + + +def test_provider_describes_its_catalogue() -> None: + provider = BassdriveProvider() + assert provider.has_schedule is True + assert provider.bandcamp is False + assert len(provider.hero) == 3 + assert "bright_cyan" in "".join(provider.hero) + assert provider.hero_lines == ( + "BASSDRIVE", + "WORLDWIDE DRUM & BASS", + "LIVE SHOWS 24/7", + ) + + stations = provider.fallback_stations() + assert [station.id for station in stations] == ["bassdrive"] + urls = provider.fallback_urls(stations[0]) + assert urls[0].startswith("https://") and len(urls) == 7 + + +@pytest.mark.asyncio +async def test_cold_start_survives_an_unreachable_homepage() -> None: + """`fetch_station` asks for the grid; a broken homepage must not stop it.""" + async with client_for(STATUS, page_code=500) as client: + station = await api.fetch_station(client) + assert station.id == "bassdrive" + assert station.presenter == "live now" # live, but the location is unknown + assert station.now_playing is not None diff --git a/tests/test_header.py b/tests/test_header.py index 5edc3ef..be6e1c1 100644 --- a/tests/test_header.py +++ b/tests/test_header.py @@ -6,8 +6,9 @@ import types import pytest from rich.cells import cell_len +from beefy_player.providers import PROVIDERS from beefy_player.providers.somafm import SomaFMProvider -from beefy_player.providers.tilderadio import TilderadioProvider +from beefy_player.providers.wordmark import render from beefy_player.widgets.header import HeaderBar TIME = re.compile(r"\d\d:\d\d:\d\d$") @@ -54,32 +55,66 @@ def test_right_block_hidden_on_narrow_terminals() -> None: assert all(":" not in line for line in lines) -def test_somafm_hero_is_the_wordmark_and_its_taglines() -> None: +def test_somafm_hero_spells_the_whole_name_and_its_taglines() -> None: lines = render_header(143) - assert lines[0].startswith("▄▀▀▀ ▄▀▀▄ █▄ ▄█ ▄▀▀▄") + assert lines[0].startswith(render("SOMAFM")[0]) assert "SOMA FM RADIO" in lines[0] assert "LISTENER SUPPORTED" in lines[1] assert "SAN FRANCISCO" in lines[2] -def test_tilderadio_hero_is_its_own_art_and_motto() -> None: - provider = TilderadioProvider() - lines = render_header(143, height=len(provider.hero), provider=provider) - assert len(lines) == len(provider.hero) - # the caption block is centred against the seven-row radio icon - assert "TILDERADIO" in lines[2] - assert "STREAMED BY / FOR USERS OF THE" in lines[3] - assert "TILDEVERSE" in lines[4] - # the radio icon's signature green wave survives the conversion - styles = [ +def hero_styles(provider) -> list[str]: + return [ str(span.style) for line in HeaderBar.render(_fake(provider)).renderables for span in line.spans ] - assert any("green" in style for style in styles) + + +def test_the_i_and_t_have_thin_stems() -> None: + """Two-pixel stems make an I look like an H and a T look like a block.""" + assert render("I") == ("▀█▀", " █", "▄█▄") + assert render("T") == ("▀█▀", " █", " █") + + +def test_the_font_still_spells_soma_the_way_the_original_logo_did() -> None: + """The wordmark font is the SomaFM logo, so the old art must be reproduced.""" + assert render("SOMA") == ( + "▄▀▀▀ ▄▀▀▄ █▄ ▄█ ▄▀▀▄", + "▀▀▀▄ █ █ █ █ █ █▀▀█", + "▄▄▄▀ ▀▄▄▀ █ █ █ █", + ) + + +def test_every_catalogue_has_a_three_row_wordmark() -> None: + for slug, provider in PROVIDERS.items(): + assert len(provider.hero) == 3, slug + assert provider.hero_lines, slug + + +def test_tilderadio_spells_its_own_name_in_the_default_colour() -> None: + provider = PROVIDERS["tilderadio"] + lines = render_header(143, height=len(provider.hero), provider=provider) + assert len(lines) == 3 + assert "TILDERADIO" in lines[0] # the wordmark + assert "STREAMED BY / FOR USERS OF THE" in lines[1] + assert "TILDEVERSE" in lines[2] + # no colour of its own: the art inherits the terminal foreground + assert all("[" not in row for row in provider.hero) + assert not {"bright_cyan", "cyan"} & set(hero_styles(provider)) assert all("SOMA" not in line for line in lines) +def test_bassdrive_spells_its_name_in_its_own_two_blues() -> None: + provider = PROVIDERS["bassdrive"] + lines = render_header(143, height=len(provider.hero), provider=provider) + assert "BASSDRIVE" in lines[0] + assert "WORLDWIDE DRUM & BASS" in lines[1] + assert "LIVE SHOWS 24/7" in lines[2] + styles = hero_styles(provider) + assert "bright_cyan" in styles and "cyan" in styles + + def _fake(provider): fake = type("Fake", (), {})() fake.app = types.SimpleNamespace( diff --git a/tests/test_providers.py b/tests/test_providers.py index 569ba61..f11b32e 100644 --- a/tests/test_providers.py +++ b/tests/test_providers.py @@ -10,6 +10,7 @@ from beefy_player import config from beefy_player.models import Song, Station from beefy_player.providers import PROVIDERS, available_providers, get_provider from beefy_player.providers.base import RadioProvider +from beefy_player.providers.bassdrive import BassdriveProvider from beefy_player.providers.somafm import SomaFMProvider from beefy_player.providers.tilderadio import TilderadioProvider from beefy_player.widgets.header import HeaderBar @@ -21,10 +22,11 @@ def rendered_text(app) -> str: def test_registry_lists_somafm_as_the_default() -> None: - assert available_providers() == ("somafm", "tilderadio") + assert available_providers() == ("somafm", "tilderadio", "bassdrive") assert isinstance(get_provider(), SomaFMProvider) assert get_provider("somafm") is get_provider() assert isinstance(get_provider("tilderadio"), TilderadioProvider) + assert isinstance(get_provider("bassdrive"), BassdriveProvider) # a stale slug in the config file must not stop the app from starting assert get_provider("no-such-catalogue") is get_provider() @@ -182,10 +184,10 @@ async def test_p_opens_the_picker_and_switches_catalogue() -> None: assert app.provider.id == "tilderadio" assert app.config.provider == "tilderadio" assert [station.id for station in app.stations] == ["tilderadio"] - # the left pane becomes the schedule, and the hero grows + # the left pane becomes the schedule; every hero is the same height assert app.query_one(StationList).display is False assert app.query_one(SchedulePanel).display is True - assert app.query_one(HeaderBar).styles.height.value == 8 + assert app.query_one(HeaderBar).styles.height.value == 4 assert "tilderadio" in app.title @@ -225,11 +227,12 @@ def test_config_remembers_a_station_per_catalogue() -> None: @pytest.mark.asyncio -async def test_short_terminals_crop_the_hero_but_keep_the_ticker() -> None: - app = _CatalogueApp("tilderadio").app - async with app.run_test(size=(90, 24)) as pilot: - await pilot.pause() - header = app.query_one(HeaderBar) - assert header.styles.height.value <= 6 - lines = [line.plain for line in header.render().renderables] - assert any("STOPPED" in line or "PLAYING" in line for line in lines) +async def test_every_catalogue_header_is_the_same_height() -> None: + for slug in available_providers(): + app = _CatalogueApp(slug).app + async with app.run_test(size=(110, 40)) as pilot: + await pilot.pause() + header = app.query_one(HeaderBar) + assert header.styles.height.value == 4, slug + lines = [line.plain for line in header.render().renderables] + assert any("STOPPED" in line or "PLAYING" in line for line in lines), slug