feat(bassdrive): add the drum & bass relay and a shared wordmark font
This commit is contained in:
beefcarver committed
1 parent
84ec3b08f6
commit
50a9a735f7
14 files changed
+1166
-79
No files matched your search
@@ -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/<station>.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 `<local-time datetime>`
|
||||
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 <city>".
|
||||
- `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-<id>.json last good list, other catalogues
|
||||
~/.cache/beefy-player/schedule-<id>.json last good broadcast schedule
|
||||
~/.cache/beefy-player/playlists.json resolved stream URLs
|
||||
~/.cache/beefy-player/songs/<id>.json song history snapshots
|
||||
~/.cache/beefy-player/songs/<id>.json SomaFM song history snapshots
|
||||
~/.cache/beefy-player/songs/<p>-<id>.json history for every other catalogue
|
||||
~/.cache/beefy-player/artwork/<sha1> 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)
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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()
|
||||
@@ -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 ``<local-time datetime=...>``.
|
||||
|
||||
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'<span class="location">(.*?)</span>', 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 ``<li>`` per slot."""
|
||||
broadcasts: list[Broadcast] = []
|
||||
for chunk in re.split(r"<li\b", page):
|
||||
if 'class="title"' not in chunk or 'class="time"' not in chunk:
|
||||
continue
|
||||
stamps = re.findall(r'datetime="([^"]+)"', chunk)
|
||||
if len(stamps) < 2:
|
||||
continue
|
||||
start = parse_date(stamps[0])
|
||||
end = parse_date(stamps[1])
|
||||
if start is None:
|
||||
continue
|
||||
title = re.search(r'<p class="title">(.*?)</p>', 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
|
||||
@@ -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)
|
||||
@@ -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
|
||||
)
|
||||
@@ -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",
|
||||
|
||||
@@ -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")
|
||||
@@ -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)
|
||||
+2
-2
@@ -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",
|
||||
|
||||
@@ -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 = """
|
||||
<ul class="schedule">
|
||||
<li>
|
||||
<a href="radio/get-dipped">
|
||||
<p class="title">Get Dipped <span class="location">Los Angeles, CA</span></p>
|
||||
<span class="time">
|
||||
<local-time datetime="2026-11-03T19:00:00-05:00" hour="numeric"></local-time> -
|
||||
<local-time datetime="2026-11-03T21:00:00-05:00" hour="numeric"></local-time>
|
||||
</span>
|
||||
</a>
|
||||
</li>
|
||||
<li>
|
||||
<a href="radio/drumobsession">
|
||||
<p class="title">DrumObsession <span class="location">Poznan, Poland</span></p>
|
||||
<span class="time">
|
||||
<local-time datetime="2026-11-03T11:00:00-05:00" hour="numeric"></local-time> -
|
||||
<local-time datetime="2026-11-03T13:00:00-05:00" hour="numeric"></local-time>
|
||||
</span>
|
||||
</a>
|
||||
</li>
|
||||
</ul>
|
||||
"""
|
||||
|
||||
|
||||
@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("<ul><li>nothing here</li></ul>") == []
|
||||
|
||||
|
||||
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
|
||||
+49
-14
@@ -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(
|
||||
|
||||
+14
-11
@@ -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
|
||||
Reference in new issue
Block a user