Compare commits

...
4 Commits
27 changed files with 2250 additions and 69 deletions

No files matched your search

+38
View File
@@ -0,0 +1,38 @@
# AGENTS.md
Notes for coding agents working in this repository.
## Re-sync the uv environment before doing anything
The checked-out `.venv` is **not** portable. It is built by `uv sync`, and its
`python` symlink points at a uv-managed interpreter path belonging to whichever
machine created it (for example `/home/node/.local/share/uv/python/...`). When
the repository is re-synced from the host, that interpreter is gone, the
symlink dangles, and every `.venv/bin/*` command fails with:
.venv/bin/python: No such file or directory
So always re-sync first — before tests, lint, the app, or any tooling:
```bash
uv sync
```
If it still fails, rebuild the environment once:
```bash
rm -rf .venv && uv sync
```
There is no project Python on `PATH`; use `uv run <command>` or the freshly
synced `.venv/bin/python` explicitly.
## Commands
```bash
uv sync # resync the environment (do this first)
uv run pytest # full test suite
uvx ruff check . # lint (ruff is not a project dependency)
uv run beefy-player --list # station ids for the current catalogue
uv run python tools/ansi_art.py ... # regenerate a provider's header art
```
+96 -16
View File
@@ -8,10 +8,12 @@ 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)** (its first and, at the moment, only provider),
and the UI never talks to a station catalogue directly: it asks a
[provider](#providers) for stations, song history and stream URLs. Adding
another internet-radio catalogue is a self-contained module, not a rewrite.
**[SomaFM](https://somafm.com)** (≈50 curated channels) and
**[tilderadio](https://tilderadio.org)** (the tildeverse's single community
stream), 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.
It runs the same way on **Linux** (PulseAudio/PipeWire) and **macOS**
(CoreAudio, iTerm2/Kitty/WezTerm): MPRIS on the desktop bus where that exists,
@@ -44,12 +46,23 @@ implemented here from public endpoints and open protocols.
"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
**Bandcamp ↗** link appears below it; otherwise the line stays hidden.
Catalogues can opt out with `bandcamp = False` (tilderadio does), and a
provider can fill the `Station.presenter` line instead — tilderadio shows
`auto dj · library rotation` between live shows and `dj <name>` during one.
- **Last played** — the full recent history, newest first.
- **Every station** — with SomaFM that is the whole `channels.json` roster
(≈50 channels), live listener counts and genre search.
(≈50 channels), live listener counts and genre search; tilderadio is a
single-station catalogue, and the list simply holds one.
- **Favourites** that float to the top, remembered between runs.
- **Pluggable catalogues** — one `providers` module per radio network, selected
with `--provider` and remembered in the config file.
- **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`.
- **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.
- **Desktop media integration** — on Linux an
`org.mpris.MediaPlayer2.beefyplayer.instance<pid>` player exposes
`PlaybackStatus`, `Metadata` (title/artist/album/artUrl), `Volume`, and
@@ -153,27 +166,69 @@ session — is provider-neutral.
class RadioProvider(ABC):
id, name, homepage, brand, tagline, station_source, history_source, user_agent
hero, hero_gradient, hero_lines # the header art and its caption
has_schedule # schedule pane instead of a list
bandcamp # look tracks up on Bandcamp?
async def fetch_stations(self, client) -> list[Station]: ...
def load_cached_stations(self) -> list[Station]: ...
def fallback_stations(self) -> list[Station]: ...
async def fetch_history(self, client, station_id) -> list[Song]: ...
def load_cached_history(self, station_id) -> list[Song]: ...
async def fetch_schedule(self, client) -> list[Broadcast]: ...
def load_cached_schedule(self) -> list[Broadcast]: ...
def fallback_urls(self, station) -> list[str]: ...
async def stream_candidates(self, client, station) -> list[str]: ...
```
`hero` rows are rich markup (half blocks); with `hero_gradient` the header
paints plain glyphs with the app's spectrum palette instead, which is how the
SomaFM wordmark stays theme-aware. `has_schedule` swaps the left pane for the
schedule view — the pane asks the provider for `Broadcast` objects (name, DJ,
start, end, live) and never learns where they came from.
SomaFM is the reference implementation, and it is quarantined in its own
package: the `channels.json` schema, the `songs/<station>.json` history, the PLS
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:
| 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) |
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.
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:
```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
```
`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 ▄.
Adding a catalogue:
1. create `beefy_player/providers/<slug>/` with a `RadioProvider` subclass,
2. register one instance in `beefy_player/providers/__init__.py`,
3. run it with `beefy-player --provider <slug>` (or set `"provider"` in
`~/.config/beefy-player/config.json`).
3. run it with `beefy-player --provider <slug>`, press `p` in the UI, or set
`"provider"` in `~/.config/beefy-player/config.json`.
`RadioProvider.stream_candidates()` already implements the common strategy —
resolve the station's playlists (with an on-disk cache), then try
@@ -211,6 +266,7 @@ Useful flags:
```bash
uv run beefy-player --provider somafm # pick a station catalogue
uv run beefy-player --provider tilderadio # one community stream, tildeverse
uv run beefy-player --play # start the last station immediately
uv run beefy-player --station defcon # pick a station
uv run beefy-player --volume 55
@@ -246,7 +302,8 @@ The UI wants roughly **84×24**; it also works smaller, just less pretty.
| `/` | filter by title, genre, description |
| `esc` | clear filter / close search |
| `v` | visualiser: mirror → bars → wave → waterfall → radial → braille |
| `r` | refresh stations + song history |
| `r` | refresh stations, schedule + song history |
| `p` | switch station catalogue (SomaFM / tilderadio)|
| `g` / `G` | top / bottom |
| `?` | help |
| `q` | quit |
@@ -290,7 +347,8 @@ beefy_player/
config.py XDG paths, settings, BEEFY_* env helpers
providers/
base.py RadioProvider interface + HTTP/playlist toolkit
somafm/ the SomaFM catalogue (API, bundled fallback list)
somafm/ the SomaFM catalogue (API, hero, bundled fallback list)
tilderadio/ the tilderadio catalogue (one station, hero, schedule)
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)
@@ -298,7 +356,10 @@ beefy_player/
mpris.py dbus-next MPRIS service
artwork.py Kitty TGP / iTerm2 images with an ANSI mosaic fallback
bandcamp.py "is this track on Bandcamp?" lookups
widgets/ the Textual widgets
widgets/ the Textual widgets (schedule pane, catalogue picker, …)
tools/
ansi_art.py logo PNG -> half-block ANSI hero (regenerates tilderadio's)
```
- `beefy_player/providers/base.py` — the provider interface and the shared
@@ -310,8 +371,20 @@ beefy_player/
- `songs/<station>.json` → last played tracks (cached on disk)
Playlist and channel caching means the UI opens instantly and still works
offline.
- `beefy_player/providers/tilderadio/api.py` — the same idea for a
single-station catalogue: `/api/now/` carries the station, the live DJ, the
listener count and both `now_playing` and `history[]`, while AzuraCast's
mount list contributes the bitrates (320k mp3, 192k mp3/ogg). Station and
history caches are per catalogue, so the two providers never overwrite each
other's files.
- `beefy_player/app.py` — holds one provider and one `Player`, ticks at 20 Hz,
and hands the widgets plain `Station`/`Song` objects.
and hands the widgets plain `Station`/`Song`/`Broadcast` objects. `p` swaps
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/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.
- `beefy_player/player.py` — mpv over a Unix-socket JSON IPC channel. Observes
`pause`/`volume`, handles mirror failover on `end-file`, auto-restarts a dead
engine, and uses `SIGSTOP`/`SIGCONT` when falling back to `ffplay`. On macOS
@@ -341,13 +414,18 @@ beefy_player/
### Files
```
~/.config/beefy-player/config.json favourites, volume, last station, mode
~/.cache/beefy-player/channels.json last good station list
~/.config/beefy-player/config.json favourites, volume, provider, last stations
~/.cache/beefy-player/channels.json last good SomaFM station list
~/.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/artwork/<sha1> cached album/channel artwork
```
`last_stations` remembers which station each catalogue was tuned to, so
switching back with `p` lands on the same channel.
These stay in the XDG locations on every platform (macOS included), so the
same dotfiles/config work everywhere; `XDG_CONFIG_HOME`, `XDG_CACHE_HOME` and
`XDG_RUNTIME_DIR` are honoured when set. The app started life as
@@ -408,6 +486,8 @@ so favourites and volume survive the rename.
metered link, use `--spectrum-source monitor` (or `--no-audio` for a UI-only
demo).
- 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.
## NixOS (optional aside)
@@ -435,6 +515,6 @@ uv run beefy-player
WTFPL — Do What The Fuck You Want To Public License, Version 2. See
[LICENSE](LICENSE). SomaFM and its channel names are trademarks of
SomaFM.com, LLC.
SomaFM.com, LLC. tilderadio belongs to the tildeverse community.
Terminal image rendering uses [Pillow](https://python-pillow.org/).
+172 -4
View File
@@ -25,13 +25,15 @@ from textual.widgets import Input
from . import artwork, bandcamp, config
from .mediasession import create_media_session, media_actions_queue
from .models import Station, parse_last_playing
from .models import Broadcast, Station, parse_last_playing
from .player import Player
from .providers import get_provider
from .providers.base import is_direct_stream, make_client
from .spectrum import SpectrumEngine
from .widgets.header import HeaderBar
from .widgets.help import HelpScreen
from .widgets.providers import ProviderScreen
from .widgets.schedule import SchedulePanel
from .widgets.search import SearchInput
from .widgets.spectrum import SpectrumPanel
from .widgets.stage import StagePanel
@@ -89,6 +91,7 @@ class BeefyPlayerApp(App[None]):
Binding("escape", "escape", "close", show=False),
Binding("v", "visualizer", "visualiser"),
Binding("r", "refresh", "refresh"),
Binding("p", "provider", "catalogue"),
Binding("question_mark", "help", "help"),
Binding("q", "quit", "quit"),
]
@@ -125,12 +128,17 @@ class BeefyPlayerApp(App[None]):
if volume is not None:
self.config.volume = max(0, min(100, int(volume)))
self.start_station_id = station_id or self.config.last_station
self.start_station_id = station_id or self.config.last_station_for(
self.provider.id
)
self.offline = offline
self.audio_enabled = audio
self.stations: list[Station] = []
self.view: list[Station] = []
#: The current catalogue's broadcast schedule, when it has one.
self.schedule: list[Broadcast] = []
self.schedule_loaded = False
self.index = 0
self.query = ""
self.current: Station | None = None
@@ -144,6 +152,8 @@ class BeefyPlayerApp(App[None]):
self._volume_before_mute = self.volume or 70
self._last_history_fetch = 0.0
self._history_refresh_seconds = 25.0
self._schedule_refresh_seconds = 300.0
self._last_schedule_fetch = 0.0
self._stream_url = ""
self.artwork_url = ""
self.artwork_bytes: bytes | None = None
@@ -174,6 +184,7 @@ class BeefyPlayerApp(App[None]):
with Vertical(id="left"):
yield SearchInput()
yield StationList()
yield SchedulePanel()
yield StagePanel()
yield SpectrumPanel(self.spectrum)
yield StatusBar()
@@ -188,6 +199,7 @@ class BeefyPlayerApp(App[None]):
self.media.start()
self._http = make_client(self.provider.user_agent)
self._apply_catalogue_chrome()
stations = self.provider.load_cached_stations()
if not stations:
@@ -208,6 +220,9 @@ class BeefyPlayerApp(App[None]):
self._refresh_stations()
if self.current is not None:
self._fetch_history(self.current.id)
if self.provider.has_schedule:
# Offline still shows the cached schedule — the worker decides.
self._refresh_schedule()
self.set_focus(None)
@@ -233,6 +248,7 @@ class BeefyPlayerApp(App[None]):
self.config.volume = self.volume
if self.current is not None:
self.config.last_station = self.current.id
self.config.last_stations[self.provider.id] = self.current.id
self.config.visualizer = self.visualizer
self.config.save()
@@ -315,6 +331,46 @@ class BeefyPlayerApp(App[None]):
if self.current is None and view:
self.current = view[self.index]
# ------------------------------------------------------------------ #
# the catalogue's chrome
# ------------------------------------------------------------------ #
def _apply_catalogue_chrome(self) -> None:
"""Point the chrome at the catalogue: hero height and the left pane."""
provider = self.provider
hero_rows = len(getattr(provider, "hero", ()) or ())
# Give the hero its rows, but never more than a quarter of the screen:
# on a short terminal the spectrum and the stage matter more.
screen_height = getattr(self, "size", None)
cap = max(4, (screen_height.height // 4) if screen_height else 10)
with contextlib.suppress(NoMatches):
header = self.query_one(HeaderBar)
header.styles.height = min(hero_rows + 1, cap) if hero_rows else 4
header.refresh()
self.query_one(StationList).display = not provider.has_schedule
self.query_one(SchedulePanel).display = provider.has_schedule
if provider.has_schedule:
self._reset_filter()
def _reset_filter(self) -> None:
"""Forget the station filter and put the search box away."""
self.query = ""
search = self.query_one(SearchInput)
search.value = ""
search.remove_class("visible")
search.disabled = True
self.query_one(StationList).set_filter("")
def on_resize(self, event) -> None:
self._apply_catalogue_chrome()
def _refresh_pane(self) -> None:
with contextlib.suppress(NoMatches):
if self.provider.has_schedule:
self.query_one(SchedulePanel).refresh()
else:
self.query_one(StationList).refresh()
# ------------------------------------------------------------------ #
# playback
# ------------------------------------------------------------------ #
@@ -455,6 +511,11 @@ class BeefyPlayerApp(App[None]):
song = station.now_playing
if song is None or not (song.artist.strip() or song.title.strip()):
return
if not getattr(self.provider, "bandcamp", True):
# Not a Bandcamp catalogue: never look the track up, and do not
# leave a stale link behind from the previous station.
song.bandcamp_url = ""
return
track_key = f"{station.id}|{song.key()}"
cached_url = self._bandcamp_urls.get(track_key)
@@ -615,8 +676,9 @@ class BeefyPlayerApp(App[None]):
self.query_one(HeaderBar).refresh()
if self.tick_count % 12 == 0:
self.query_one(StatusBar).refresh()
self.query_one(StationList).refresh()
self._refresh_pane()
self._maybe_refresh_history()
self._maybe_refresh_schedule()
def _drain_player(self) -> None:
while True:
@@ -693,6 +755,33 @@ class BeefyPlayerApp(App[None]):
continue
self._set_volume(value)
def _maybe_refresh_schedule(self) -> None:
if not self.provider.has_schedule or self.offline:
return
if (
time.monotonic() - self._last_schedule_fetch
>= self._schedule_refresh_seconds
):
self._refresh_schedule()
@work(exclusive=True, group="schedule", exit_on_error=False)
async def _refresh_schedule(self) -> None:
if not self.provider.has_schedule:
return
self._last_schedule_fetch = time.monotonic()
broadcasts: list[Broadcast] = []
if self.offline or self._http is None:
broadcasts = self.provider.load_cached_schedule()
else:
try:
broadcasts = await self.provider.fetch_schedule(self._http)
except Exception:
broadcasts = self.provider.load_cached_schedule()
self.schedule = broadcasts
self.schedule_loaded = True
with contextlib.suppress(NoMatches):
self.query_one(SchedulePanel).refresh()
# ------------------------------------------------------------------ #
# workers
# ------------------------------------------------------------------ #
@@ -714,7 +803,9 @@ class BeefyPlayerApp(App[None]):
self._set_stations(stations)
if before is not None:
self._select_station_id(before)
self._set_notice(f"{len(stations)} stations", 3)
self._set_notice(
"1 station" if len(stations) == 1 else f"{len(stations)} stations", 3
)
self.query_one(StationList).refresh()
# ------------------------------------------------------------------ #
@@ -810,6 +901,9 @@ class BeefyPlayerApp(App[None]):
self.query_one(StagePanel).refresh_stage()
def action_search(self) -> None:
if self.provider.has_schedule:
self._set_notice("this catalogue shows a schedule, not a station list", 3)
return
search = self.query_one(SearchInput)
search.disabled = False
search.add_class("visible")
@@ -857,10 +951,84 @@ class BeefyPlayerApp(App[None]):
return
self.artwork_url = ""
self._refresh_stations()
self._refresh_schedule()
if self.current is not None:
self._update_artwork()
self._fetch_history(self.current.id)
def action_provider(self) -> None:
self.push_screen(ProviderScreen(self.provider.id), self._provider_chosen)
def _provider_chosen(self, provider_id: str | None) -> None:
if provider_id:
self._switch_catalogue(provider_id)
def _cancel_workers(self) -> None:
for group in ("stream", "history", "stations", "schedule", "bandcamp", "artwork"):
with contextlib.suppress(Exception):
self.workers.cancel_group(self, group)
def _switch_catalogue(self, provider_id: str) -> None:
"""Move to another station catalogue without restarting the app."""
provider = get_provider(provider_id)
if provider is self.provider:
return
self._cancel_workers()
with contextlib.suppress(Exception):
self.player.stop()
self.spectrum.set_source(None)
self.playing_state = "stopped"
self.media.set_status("Stopped")
old_client = self._http
self.provider = provider
self._http = make_client(provider.user_agent)
if old_client is not None:
with contextlib.suppress(RuntimeError):
asyncio.get_running_loop().create_task(old_client.aclose())
self.config.provider = provider.id
self.config.save()
self.stations = []
self.view = []
self.index = 0
self.current = None
self.schedule = []
self.schedule_loaded = False
self._last_schedule_fetch = 0.0
self._last_history_fetch = 0.0
self._bandcamp_key = ""
self._bandcamp_urls.clear()
self._stream_url = ""
self._art_track_key = ""
self.artwork_url = ""
self.artwork_bytes = None
self.start_station_id = self.config.last_station_for(provider.id)
self._apply_catalogue_chrome()
with contextlib.suppress(NoMatches):
self._reset_filter()
stations = provider.load_cached_stations() or provider.fallback_stations()
self._set_stations(stations)
self._select_initial_station()
self._update_media()
self._update_artwork()
self._refresh_art()
with contextlib.suppress(NoMatches):
self.query_one(HeaderBar).refresh()
self.query_one(StagePanel).refresh_stage()
self.query_one(StatusBar).refresh()
self._set_notice(f"catalogue: {provider.name}", 3)
if not self.offline:
self._refresh_stations()
if self.current is not None:
self._fetch_history(self.current.id)
if provider.has_schedule:
self._refresh_schedule()
def action_help(self) -> None:
self.push_screen(HelpScreen())
+37 -1
View File
@@ -3,7 +3,9 @@
Everything lives in the usual XDG places:
~/.config/beefy-player/config.json favourites, volume, station
~/.cache/beefy-player/channels.json last good station list
~/.cache/beefy-player/channels.json last good station list (SomaFM)
~/.cache/beefy-player/channels-<id>.json ... for every other catalogue
~/.cache/beefy-player/schedule-<id>.json last good broadcast schedule
~/.cache/beefy-player/playlists.json resolved .pls -> stream URLs
~/.cache/beefy-player/songs/<id>.json last good song history
~/.cache/beefy-player/artwork/<sha1> cached album/channel artwork
@@ -17,6 +19,7 @@ from __future__ import annotations
import contextlib
import json
import os
import re
import shutil
from dataclasses import asdict, dataclass, field
from pathlib import Path
@@ -70,6 +73,24 @@ LEGACY_CONFIG_PATH = _xdg("XDG_CONFIG_HOME", ".config") / LEGACY_APP_NAME / "con
VISUALIZERS = ("both", "bars", "wave", "waterfall", "radial", "braille")
def channels_cache_path(provider_id: str) -> Path:
"""Where a catalogue caches its last good station list.
SomaFM keeps the historical ``channels.json``; every other provider gets
its own file, so two catalogues can never overwrite each other's cache.
"""
slug = re.sub(r"[^a-z0-9_-]", "_", provider_id.strip().lower())
if not slug or slug == "somafm":
return CHANNELS_CACHE
return CACHE_DIR / f"channels-{slug}.json"
def schedule_cache_path(provider_id: str) -> Path:
"""Where a catalogue caches its last good broadcast schedule."""
slug = re.sub(r"[^a-z0-9_-]", "_", provider_id.strip().lower()) or "provider"
return CACHE_DIR / f"schedule-{slug}.json"
def ensure_dirs() -> None:
for directory in (CONFIG_DIR, CACHE_DIR, SONGS_CACHE_DIR):
with contextlib.suppress(OSError):
@@ -115,6 +136,9 @@ def migrate_legacy_state() -> None:
class Config:
favorites: list[str] = field(default_factory=list)
last_station: str = "groovesalad"
#: Which station each catalogue was last tuned to, keyed by provider slug,
#: so switching back lands where you left off.
last_stations: dict[str, str] = field(default_factory=dict)
volume: int = 70
visualizer: str = "both"
autoplay: bool = False
@@ -141,6 +165,14 @@ class Config:
if isinstance(last_station, str) and last_station:
config.last_station = last_station
last_stations = raw.get("last_stations")
if isinstance(last_stations, dict):
config.last_stations = {
str(key): str(value)
for key, value in last_stations.items()
if isinstance(value, str) and value
}
volume = raw.get("volume")
if isinstance(volume, (int, float)):
config.volume = max(0, min(100, int(volume)))
@@ -168,6 +200,10 @@ class Config:
except OSError:
pass
def last_station_for(self, provider_id: str) -> str:
"""The station to re-tune for a catalogue (old configs have one only)."""
return self.last_stations.get(provider_id) or self.last_station
def is_favorite(self, station_id: str) -> bool:
return station_id in self.favorites
+38 -1
View File
@@ -40,6 +40,25 @@ class Song:
return f"{self.artist}\x1f{self.title}\x1f{self.album}"
@dataclass(slots=True)
class Broadcast:
"""One entry in a catalogue's schedule: a show, a DJ slot, a stream."""
name: str
dj: str = ""
start: datetime | None = None
end: datetime | None = None
#: True for the broadcast that is on air right now.
live: bool = False
@property
def display_name(self) -> str:
return self.name.strip() or self.dj.strip() or "untitled broadcast"
def key(self) -> str:
return f"{self.name}\x1f{self.start.isoformat() if self.start else ''}"
@dataclass(slots=True)
class Station:
id: str
@@ -47,6 +66,9 @@ class Station:
description: str = ""
genre: str = ""
dj: str = ""
#: Who is presenting right now when that is more than a DJ name —
#: ``"auto dj · library rotation"`` for a station between live shows.
presenter: str = ""
image: str = ""
listeners: int = 0
last_playing: str = ""
@@ -69,6 +91,15 @@ class Station:
def genre_label(self) -> str:
return self.genre.replace("|", " · ") or "eclectic"
@property
def presenter_label(self) -> str:
"""What the ON AIR metadata line says about who is on the desk."""
if self.presenter.strip():
return self.presenter.strip()
if self.dj.strip():
return f"dj {self.dj.strip()}"
return ""
def sorted_playlists(self) -> list[dict[str, str]]:
return sorted(
self.playlists,
@@ -87,7 +118,7 @@ class Station:
fmt = _FORMAT_LABEL.get(
first.get("format", ""), first.get("format") or "stream"
)
return f"{fmt} · {guess_bitrate(first.get('url', ''))}k"
return f"{fmt} · {playlist_bitrate(first)}k"
# --------------------------------------------------------------------------- #
@@ -95,6 +126,12 @@ class Station:
# --------------------------------------------------------------------------- #
def playlist_bitrate(playlist: dict[str, str]) -> int:
"""The bitrate a catalogue declared, else the best guess from the URL."""
declared = to_int(playlist.get("bitrate"))
return declared or guess_bitrate(playlist.get("url", ""))
def guess_bitrate(url: str) -> int:
match = re.search(
r"(256|192|130|128|96|64|48|32)(?=\.pls|\.m3u|-mp3|-aac)", url
+2 -1
View File
@@ -18,6 +18,7 @@ from __future__ import annotations
from .base import RadioProvider
from .somafm import SomaFMProvider
from .tilderadio import TilderadioProvider
__all__ = [
"DEFAULT_PROVIDER",
@@ -29,7 +30,7 @@ __all__ = [
#: Every catalogue the app can talk to, keyed by its slug.
PROVIDERS: dict[str, RadioProvider] = {
provider.id: provider for provider in (SomaFMProvider(),)
provider.id: provider for provider in (SomaFMProvider(), TilderadioProvider())
}
#: Provider used when nothing else is configured.
+33 -1
View File
@@ -24,7 +24,7 @@ from urllib.parse import quote
import httpx
from .. import config
from ..models import Song, Station
from ..models import Broadcast, Song, Station
__all__ = [
"PLAYLIST_TTL",
@@ -92,6 +92,28 @@ class RadioProvider(ABC):
#: ``User-Agent`` for this catalogue's requests (SomaFM likes to know).
user_agent: str = ""
# -- header hero ----------------------------------------------------- #
#: Rows of header art drawn at the top left (rich markup is allowed).
hero: tuple[str, ...] = ()
#: Paint :attr:`hero` with the app's spectrum gradient (a wordmark) instead
#: of the colours the art carries itself (a picture).
hero_gradient: bool = False
#: Short lines printed beside the hero, top to bottom.
hero_lines: tuple[str, ...] = ()
# -- left pane ------------------------------------------------------- #
#: Show a broadcast schedule in the left pane instead of a station list.
#: Set by single-stream catalogues, where a list of one station is a waste
#: of the pane and "what is on, what is next" is the useful view.
has_schedule: bool = False
#: Look the current track up on Bandcamp and offer a link when it is found.
#: Off for catalogues whose rotation is unlikely to be there (and for
#: stations that would rather not be scraped).
bandcamp: bool = True
# -- catalogue ------------------------------------------------------- #
@abstractmethod
@@ -118,6 +140,16 @@ class RadioProvider(ABC):
def load_cached_history(self, station_id: str) -> list[Song]:
"""The last good song history from disk, or ``[]``."""
# -- schedule -------------------------------------------------------- #
async def fetch_schedule(self, client: httpx.AsyncClient) -> list[Broadcast]:
"""Upcoming broadcasts, soonest first (the live one, if any, first)."""
return []
def load_cached_schedule(self) -> list[Broadcast]:
"""The last good schedule from disk, or ``[]``."""
return []
# -- streams --------------------------------------------------------- #
def fallback_urls(self, station: Station) -> list[str]:
@@ -15,6 +15,7 @@ from ... import config
from ...models import Song, Station
from ..base import RadioProvider
from . import api, fallback
from . import hero as hero_art
__all__ = ["SomaFMProvider"]
@@ -28,6 +29,10 @@ class SomaFMProvider(RadioProvider):
station_source = "somafm.com/channels.json"
history_source = "songs/<station>.json"
hero = hero_art.LOGO
hero_gradient = True
hero_lines = hero_art.TAGLINES
@property
def user_agent(self) -> str:
# SomaFM asks clients to identify themselves, so say who we are and
+20
View File
@@ -0,0 +1,20 @@
"""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
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, ...] = (
"▄▀▀▀ ▄▀▀▄ █▄ ▄█ ▄▀▀▄",
"▀▀▀▄ █ █ █ █ █ █▀▀█",
"▄▄▄▀ ▀▄▄▀ █ █ █ █",
)
TAGLINES: tuple[str, ...] = (
"SOMA FM RADIO",
"LISTENER SUPPORTED",
"SAN FRANCISCO",
)
@@ -0,0 +1,76 @@
"""tilderadio: the tildeverse's community station.
One stream, one catalogue entry, a live DJ when a show is on and the shared
library rotation the rest of the time. Everything tilderadio-specific — the
``/api/now/`` schema, the ``now_playing`` + ``history[]`` pairing, the AzuraCast
mount list and the bundled fallback — lives in this package; the app only sees
:class:`TilderadioProvider`.
"""
from __future__ import annotations
import httpx
from ... import config
from ...models import Broadcast, Song, Station
from ..base import RadioProvider
from . import api, fallback
from . import hero as hero_art
__all__ = ["TilderadioProvider"]
class TilderadioProvider(RadioProvider):
id = "tilderadio"
name = "tilderadio"
homepage = "tilderadio.org"
brand = "TILDE RADIO"
tagline = "community radio · tildeverse"
station_source = "tilderadio.org/api/now/"
history_source = "now_playing + history[]"
# The station's own radio icon, converted to half blocks by
# ``tools/ansi_art.py``; the caption is the tildeverse's motto.
hero = hero_art.HERO
hero_lines = (
"TILDERADIO",
"STREAMED BY / FOR USERS OF THE",
"TILDEVERSE",
)
# One stream, so the left pane is a schedule instead of a station list.
has_schedule = True
# A shared library rotation is not a Bandcamp catalogue; skip the lookups.
bandcamp = False
@property
def user_agent(self) -> str:
# A small volunteer station; introduce ourselves and say where we are.
return f"{config.USER_AGENT}; +https://tilderadio.org"
async def fetch_stations(self, client: httpx.AsyncClient) -> list[Station]:
return [await api.fetch_station(client)]
def load_cached_stations(self) -> list[Station]:
return api.load_cached_station()
def fallback_stations(self) -> list[Station]:
return fallback.fallback_stations()
async def fetch_history(
self, client: httpx.AsyncClient, station_id: str
) -> list[Song]:
return await api.fetch_songs(client, station_id)
def load_cached_history(self, station_id: str) -> list[Song]:
return api.load_cached_songs(station_id)
async def fetch_schedule(
self, client: httpx.AsyncClient
) -> list[Broadcast]:
return await api.fetch_schedule(client)
def load_cached_schedule(self) -> list[Broadcast]:
return api.load_cached_schedule()
def fallback_urls(self, station: Station) -> list[str]:
return fallback.known_stream_urls()
+401
View File
@@ -0,0 +1,401 @@
"""tilderadio's public endpoints, and nothing but.
* ``https://tilderadio.org/api/now/`` -> the station, listener count, the live
DJ, the current track and the last handful of tracks
* ``https://azuracast.tilderadio.org/api/nowplaying/tilderadio`` -> the mounts
(mp3/ogg, bitrates) behind the stream, so we can offer the best one first
tilderadio is a single-station catalogue: one community stream for the whole
tildeverse, run on an AzuraCast instance. That makes the "station list" a list
of one, and it means ``/now/`` is both the catalogue *and* the history feed —
the current track arrives as ``now_playing``, the previous ones as
``history[]``. Everything here is forgiving: the station is better served by a
slightly degraded panel than by a crash.
"""
from __future__ import annotations
import re
import time
from collections.abc import Iterable
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import httpx
from ... import config
from ...models import (
Broadcast,
Song,
Station,
parse_date,
parse_last_playing,
to_int,
)
from ..base import fetch_json, quote_station_id
PROVIDER_ID = "tilderadio"
API_ROOT = "https://tilderadio.org/api"
NOW_URL = f"{API_ROOT}/now/"
SCHEDULE_URL = f"{API_ROOT}/schedule/"
AZURACAST_URL = "https://azuracast.tilderadio.org/api/nowplaying/{station}"
DEFAULT_STATION_ID = "tilderadio"
STATION_IMAGE = "https://tilderadio.org/logos/tilderadio.png"
DESCRIPTION = "community radio for the tildeverse"
GENRE = "community|eclectic"
# --------------------------------------------------------------------------- #
# payload -> Station / Song
# --------------------------------------------------------------------------- #
def _artist_title(entry: dict[str, Any]) -> tuple[str, str]:
"""AzuraCast sends ``artist``/``title`` *or* one ``"Artist - Title"`` blob."""
artist = str(entry.get("artist") or "").strip()
title = str(entry.get("title") or "").strip()
if not artist:
text = str(entry.get("text") or "").strip()
split = parse_last_playing(text) if text else None
if split is not None:
artist, title = split.artist or artist, split.title or title
return artist, title
def song_from_track(entry: Any) -> Song | None:
if not isinstance(entry, dict):
return None
artist, title = _artist_title(entry)
if not (artist or title):
return None
return Song(
title=title,
artist=artist,
album=str(entry.get("album") or "").strip(),
art_url=str(entry.get("art") or entry.get("albumArt") or "").strip(),
played_at=parse_date(entry.get("played_at") or entry.get("date")),
)
def parse_tracks(entries: Any) -> list[Song]:
"""Newest first, and the same track only once even if the feed repeats it."""
if not isinstance(entries, list):
return []
songs: list[Song] = []
seen: set[tuple[str, Any]] = set()
for entry in entries:
song = song_from_track(entry)
if song is None:
continue
marker = (song.key(), song.played_at)
if marker in seen:
continue
seen.add(marker)
songs.append(song)
if all(song.played_at for song in songs):
songs.sort(key=lambda song: song.played_at, reverse=True) # type: ignore[arg-type,return-value]
return songs
def songs_from_now(payload: Any) -> list[Song]:
"""``now_playing`` followed by ``history[]``, as one newest-first list."""
if not isinstance(payload, dict):
return []
entries: list[Any] = [payload.get("now_playing")]
history = payload.get("history")
if isinstance(history, list):
entries.extend(history)
return parse_tracks(entries)
def _quality(bitrate: int) -> str:
if bitrate >= 256:
return "highest"
if bitrate >= 128:
return "high"
return "low" if bitrate else ""
def playlists_from_mounts(payload: Any) -> list[dict[str, str]]:
"""AzuraCast's mount list -> the playlist dicts :class:`Station` speaks."""
if not isinstance(payload, dict):
return []
info = payload.get("station")
mounts = info.get("mounts") if isinstance(info, dict) else None
if not isinstance(mounts, list):
return []
playlists: list[dict[str, str]] = []
seen: set[str] = set()
for mount in mounts:
if not isinstance(mount, dict):
continue
url = mount.get("url")
if not isinstance(url, str) or not url.startswith(("http://", "https://")):
continue
if url in seen:
continue
seen.add(url)
bitrate = to_int(mount.get("bitrate"))
playlists.append(
{
"url": url,
"format": str(mount.get("format") or "").strip().lower(),
"quality": _quality(bitrate),
"bitrate": str(bitrate) if bitrate else "",
}
)
return playlists
def _playlists_from_listen_url(info: dict[str, Any]) -> list[dict[str, str]]:
url = info.get("listen_url")
if not isinstance(url, str) or not url.startswith(("http://", "https://")):
return []
return [{"url": url, "format": "ogg", "quality": "high", "bitrate": ""}]
def _is_live(payload: dict[str, Any]) -> bool:
live = payload.get("live")
if isinstance(live, dict):
return bool(live.get("is_live"))
return bool(payload.get("is_live"))
def _live_dj(payload: dict[str, Any]) -> str:
"""Only name a DJ when the station says someone is actually on air."""
if not _is_live(payload):
return ""
live = payload.get("live")
if isinstance(live, dict):
return str(live.get("name") or live.get("dj") or "").strip()
return str(payload.get("dj") or "").strip()
def station_from_now(
payload: Any, playlists: Iterable[Any] = ()
) -> Station | None:
"""The whole catalogue, from one ``/api/now/`` response."""
if not isinstance(payload, dict):
return None
info = payload.get("station")
if not isinstance(info, dict):
return None
station_id = (
str(info.get("shortcode") or info.get("name") or DEFAULT_STATION_ID).strip()
or DEFAULT_STATION_ID
)
known = [dict(entry) for entry in playlists if isinstance(entry, dict)]
station = Station(
id=station_id,
title=str(info.get("name") or station_id).strip() or DEFAULT_STATION_ID,
description=str(info.get("description") or "").strip() or DESCRIPTION,
genre=GENRE,
dj=_live_dj(payload),
presenter="" if _is_live(payload) else "auto dj · library rotation",
image=STATION_IMAGE,
listeners=to_int(payload.get("listeners")),
playlists=known or _playlists_from_listen_url(info),
provider=PROVIDER_ID,
)
current = payload.get("now_playing")
if isinstance(current, dict):
song = song_from_track(current)
if song is not None:
station.now_playing = song
station.last_playing = str(current.get("text") or "").strip() or (
f"{song.artist} - {song.title}" if song.artist else song.title
)
return station
# --------------------------------------------------------------------------- #
# schedule
# --------------------------------------------------------------------------- #
def parse_schedule(payload: Any) -> list[Broadcast]:
"""``/api/schedule/`` -> broadcasts, soonest first."""
entries = payload.get("schedule") if isinstance(payload, dict) else None
if not isinstance(entries, list):
return []
broadcasts: list[Broadcast] = []
for entry in entries:
if not isinstance(entry, dict):
continue
name = str(entry.get("name") or entry.get("slug") or "").strip()
start = parse_date(entry.get("start_ts") or entry.get("start"))
end = parse_date(entry.get("end_ts") or entry.get("end"))
if not name or start is None:
continue
broadcasts.append(Broadcast(name=name, dj=name, start=start, end=end))
broadcasts.sort(key=lambda item: item.start or datetime.now(timezone.utc))
return broadcasts
def live_broadcast(payload: Any) -> Broadcast | None:
"""The show that is on air right now, if the feed says there is one."""
if not isinstance(payload, dict) or not payload.get("is_live"):
return None
name = _live_dj(payload) or "live broadcast"
return Broadcast(
name=name,
dj=name,
start=parse_date(payload.get("broadcast_start")),
end=parse_date(payload.get("broadcast_end")),
live=True,
)
# --------------------------------------------------------------------------- #
# caching
# --------------------------------------------------------------------------- #
def station_cache_path() -> Path:
return config.channels_cache_path(PROVIDER_ID)
def _songs_cache_path(station_id: str) -> Path:
safe = re.sub(r"[^A-Za-z0-9_.-]", "_", station_id or DEFAULT_STATION_ID)
return config.SONGS_CACHE_DIR / f"{PROVIDER_ID}-{safe}.json"
def _songs_to_cache(songs: Iterable[Song]) -> list[dict[str, Any]]:
return [
{
"title": song.title,
"artist": song.artist,
"album": song.album,
"art": song.art_url,
"played_at": song.played_at.isoformat() if song.played_at else "",
}
for song in songs
]
# --------------------------------------------------------------------------- #
# fetching
# --------------------------------------------------------------------------- #
async def _fetch_mounts(
client: httpx.AsyncClient, station_id: str
) -> list[dict[str, str]]:
url = AZURACAST_URL.format(station=quote_station_id(station_id))
try:
payload = await fetch_json(client, url, attempts=1)
except RuntimeError:
return []
return playlists_from_mounts(payload)
async def fetch_station(client: httpx.AsyncClient) -> Station:
try:
payload = await fetch_json(client, NOW_URL)
except RuntimeError as exc:
raise RuntimeError(f"could not fetch tilderadio now-playing: {exc}") from None
station = station_from_now(payload)
if station is None:
raise RuntimeError("tilderadio returned no station data")
# Bitrates and the mp3 mount only come from AzuraCast; losing them is not
# a reason to lose the station.
playlists = await _fetch_mounts(client, station.id)
if playlists:
station.playlists = playlists
config.write_json(
station_cache_path(),
{"fetched": time.time(), "now": payload, "mounts": station.playlists},
)
return station
def load_cached_station() -> list[Station]:
raw = config.read_json(station_cache_path())
if not isinstance(raw, dict):
return []
mounts = raw.get("mounts")
station = station_from_now(
raw.get("now"), mounts if isinstance(mounts, list) else []
)
return [station] if station is not None else []
async def fetch_songs(client: httpx.AsyncClient, station_id: str) -> list[Song]:
try:
payload = await fetch_json(client, NOW_URL, attempts=1)
except RuntimeError:
return load_cached_songs(station_id)
songs = songs_from_now(payload)
if not songs:
return load_cached_songs(station_id)
config.write_json(
_songs_cache_path(station_id),
{"fetched": time.time(), "songs": _songs_to_cache(songs)},
)
return songs
def load_cached_songs(station_id: str) -> list[Song]:
raw = config.read_json(_songs_cache_path(station_id))
if isinstance(raw, dict):
return parse_tracks(raw.get("songs"))
return []
def _schedule_cache_path() -> Path:
return config.schedule_cache_path(PROVIDER_ID)
def load_cached_schedule() -> list[Broadcast]:
raw = config.read_json(_schedule_cache_path())
entries = raw.get("schedule") if isinstance(raw, dict) else None
if not isinstance(entries, list):
return []
return parse_schedule({"schedule": entries})
async def fetch_schedule(client: httpx.AsyncClient) -> list[Broadcast]:
"""What is on air now (if anything) followed by what is coming up."""
live: Broadcast | None = None
try:
now_payload = await fetch_json(client, NOW_URL, attempts=1)
except RuntimeError:
now_payload = None
live = live_broadcast(now_payload)
try:
payload = await fetch_json(client, SCHEDULE_URL, attempts=1)
except RuntimeError:
upcoming = load_cached_schedule()
else:
upcoming = parse_schedule(payload)
config.write_json(
_schedule_cache_path(),
{
"fetched": time.time(),
"schedule": [
{
"name": item.name,
"start": item.start.isoformat() if item.start else "",
"end": item.end.isoformat() if item.end else "",
}
for item in upcoming
],
},
)
now = datetime.now(timezone.utc)
upcoming = [
item
for item in upcoming
if not item.live and (item.end or item.start or now) >= now
]
return ([live] if live is not None else []) + upcoming
@@ -0,0 +1,56 @@
"""Bundled station, so a cold offline start still makes sense.
tilderadio is a single-station catalogue — one community stream for the whole
tildeverse — so "the bundled station list" is that one station. The mounts
below are the ones the AzuraCast instance has served for years; the live list
from ``/api/now/`` always wins, and the URLs are additionally probed by the
player's normal failover.
"""
from __future__ import annotations
from typing import Any
# (url, format, quality, bitrate)
_MOUNTS: tuple[tuple[str, str, str, str], ...] = (
("https://azuracast.tilderadio.org/listen/tilderadio/radio.mp3", "mp3", "high", "192"),
("https://azuracast.tilderadio.org/listen/tilderadio/320k.mp3", "mp3", "highest", "320"),
("https://azuracast.tilderadio.org/listen/tilderadio/radio.ogg", "ogg", "high", "192"),
)
def fallback_payload() -> dict[str, Any]:
"""A ``/api/now/``-shaped payload, so the normal parser can eat it."""
return {
"api_version": 1,
"status": "ok",
"station": {
"id": 1,
"name": "tilderadio",
"shortcode": "tilderadio",
"listen_url": _MOUNTS[2][0],
},
"is_live": False,
"dj": None,
"listeners": 0,
"now_playing": None,
"history": [],
}
def fallback_playlists() -> list[dict[str, str]]:
return [
{"url": url, "format": fmt, "quality": quality, "bitrate": bitrate}
for url, fmt, quality, bitrate in _MOUNTS
]
def known_stream_urls() -> list[str]:
return [url for url, _format, _quality, _bitrate in _MOUNTS]
def fallback_stations() -> list[Any]:
from .api import station_from_now
station = station_from_now(fallback_payload(), fallback_playlists())
return [station] if station is not None else []
+24
View File
@@ -0,0 +1,24 @@
"""Generated header art — do not edit by hand.
Source: https://tilderadio.org/logos/tilderadio.png
Regenerate with::
uv run python tools/ansi_art.py --url https://tilderadio.org/logos/tilderadio.png --cols 22 --rows 7 --module beefy_player/providers/tilderadio/hero.py
Each row is rich markup; ``▀`` is a half block whose foreground is the top
pixel and whose background is the pixel below it. Colours are stock ANSI
palette names, so the art keeps the terminal's own palette.
"""
from __future__ import annotations
HERO: tuple[str, ...] = (
' [bright_black]▄[/][bright_black on bright_black]▀▀[/][bright_black]▀[/]',
' [bright_black]▄▄[/][bright_black on bright_black]▀▀▀[/][bright_black]▄▄▄▄▄▄▄▄▄▄▄[/]',
'[bright_black]▄[/][bright_black on bright_black]▀[/][bright_black on black]▀▀▀▀▀▀▀▀▀▀▀[/][bright_black on bright_black]▀▀▀▀▀▀[/][bright_black on black]▀[/][bright_black on bright_black]▀[/][bright_black]▄[/]',
'[bright_black on bright_black]▀▀[/][black on black]▀▀▀▀▀▀▀▀▀▀[/][bright_black on bright_black]▀▀[/][bright_black]▀[/] [bright_black on bright_black]▀▀▀▀[/]',
'[bright_black on bright_black]▀▀[/][black on black]▀[/][black on bright_black]▀▀[/][black on green]▀[/][black on bright_black]▀▀[/][black on black]▀▀[/][black on bright_black]▀[/][black on black]▀▀[/][bright_black on black]▀[/][bright_black on bright_black]▀[/][bright_black]▄▄[/][bright_black on bright_black]▀▀[/][bright_black on black]▀[/][bright_black on bright_black]▀▀[/]',
'[bright_black on bright_black]▀▀[/][black on black]▀[/][green on green]▀[/][green on black]▀▀▀[/][green on green]▀▀▀▀[/][bright_black on black]▀[/][black on black]▀▀▀[/][bright_black on bright_black]▀[/][bright_black on black]▀[/][bright_black on bright_black]▀▀[/][black on black]▀[/][bright_black on bright_black]▀▀[/]',
'[bright_black]▀▀[/][bright_black on bright_black]▀[/][black on bright_black]▀▀[/][bright_black on bright_black]▀[/][black on bright_black]▀▀▀▀▀[/][bright_black on bright_black]▀[/][black on bright_black]▀[/][bright_black on bright_black]▀[/][black on bright_black]▀[/][bright_black on bright_black]▀▀▀[/][black on bright_black]▀[/][bright_black on bright_black]▀[/][bright_black]▀▀[/]',
)
+21
View File
@@ -82,6 +82,17 @@ StationList {
background: ansi_default;
}
SchedulePanel {
color: ansi_default;
width: 1fr;
height: 1fr;
display: none;
border: round ansi_bright_black;
border-title-color: ansi_default;
padding: 0 1;
background: ansi_default;
}
StagePanel {
color: ansi_default;
layout: vertical;
@@ -151,6 +162,16 @@ ModalScreen {
background: ansi_default;
}
#providers {
color: ansi_default;
width: 66;
height: auto;
border: round ansi_bright_black;
border-title-color: ansi_default;
padding: 1 2;
background: ansi_default;
}
#help {
color: ansi_default;
width: 78;
+57 -32
View File
@@ -1,4 +1,4 @@
"""Header: logo, clock and a live ticker."""
"""Header: the catalogue's hero, the clock and a live ticker."""
from __future__ import annotations
@@ -21,13 +21,19 @@ from .common import (
if TYPE_CHECKING:
from ..app import BeefyPlayerApp
LOGO = (
"▄▀▀▀ ▄▀▀▄ █▄ ▄█ ▄▀▀▄",
"▀▀▀▄ █ █ █ █ █ █▀▀█",
"▄▄▄▀ ▀▄▄▀ █ █ █ █",
)
MIDDLE = ("SOMA FM RADIO", "LISTENER SUPPORTED", "SAN FRANCISCO")
def _hero_line(markup: str, index: int, width: int, gradient_fill: bool) -> Text:
"""One row of the hero: spectrum-painted glyphs, or the art's own colours."""
if not gradient_fill:
line = Text.from_markup(markup)
line.no_wrap = True
return line
line = Text(no_wrap=True)
for column, char in enumerate(markup):
position = (column / max(1, width - 1) + index * 0.055) % 1.0
line.append(char, style=f"bold {gradient(SPECTRUM_STOPS, position)}")
return line
class HeaderBar(Widget):
@@ -35,54 +41,63 @@ class HeaderBar(Widget):
def render(self) -> RenderableType:
app = cast("BeefyPlayerApp", self.app)
provider = getattr(app, "provider", None)
hero = tuple(getattr(provider, "hero", ()) or ())
hero_lines = tuple(getattr(provider, "hero_lines", ()) or ())
gradient_fill = bool(getattr(provider, "hero_gradient", False))
width = max(24, self.size.width)
height = max(1, self.size.height)
state = app.playing_state
glyph, colour = state_glyph(state)
logo_width = max(cell_len(row) for row in LOGO)
middle_width = max(cell_len(row) for row in MIDDLE)
# Measure what the hero *shows*: the rows are rich markup, so their
# raw length says nothing about the columns they occupy.
hero_width = max(
(cell_len(Text.from_markup(row).plain) for row in hero), default=0
)
middle_width = max((cell_len(row) for row in hero_lines), default=0)
right = (clock_time(), date_label(), f"vol {app.volume}%")
right_width = max(cell_len(item) for item in right)
show_middle = width >= logo_width + middle_width + 8
show_right = width >= logo_width + right_width + 8
show_middle = bool(hero_lines) and width >= hero_width + middle_width + 8
show_right = width >= hero_width + right_width + 8
if show_middle and show_right:
show_middle = width >= logo_width + 4 + middle_width + 2 + right_width
show_middle = (
width >= hero_width + 4 + middle_width + 2 + right_width
)
# Centre the short text block against however tall the hero is: a
# three-line caption beside a seven-row picture sits in the middle.
text_offset = max(0, (len(hero) - len(hero_lines)) // 2)
lines: list[Text] = []
for index, row in enumerate(LOGO):
line = Text(no_wrap=True)
for column, char in enumerate(row):
position = (
column / max(1, logo_width - 1) + index * 0.055
) % 1.0
line.append(
char, style=f"bold {gradient(SPECTRUM_STOPS, position)}"
)
if show_middle:
for index, row in enumerate(hero):
line = _hero_line(row, index, hero_width, gradient_fill)
if show_middle and text_offset <= index < text_offset + len(hero_lines):
middle = hero_lines[index - text_offset]
line.append(" " * 4)
middle = MIDDLE[index]
line.append(middle, style="dim")
# Keep the right block on one column even though the middle
# lines differ in length ("SOMA FM RADIO" is shorter than
# "LISTENER SUPPORTED").
# lines differ in length ("TILDERADIO" is shorter than
# "STREAMED BY / FOR USERS OF THE").
line.append(" " * max(0, middle_width - cell_len(middle)))
if show_right:
used = logo_width
if show_right and text_offset <= index < text_offset + len(right):
used = hero_width
if show_middle:
used += 4 + middle_width
# Right-align the block *and* every item inside it: the clock,
# the date and the volume each keep their own length, so pad
# the shortest ones instead of left-aligning the whole block.
pad = max(1, width - used - right_width)
item = right[index]
item = right[index - text_offset]
line.append(" " * pad)
line.append(" " * max(0, right_width - cell_len(item)))
style = "bold" if index == 0 else "dim"
style = "bold" if index == text_offset else "dim"
line.append(item, style=style)
lines.append(line)
ticker: Text | None = None
if height >= 4:
ticker = Text(no_wrap=True)
prefix = f"{glyph} {state.upper():<8} "
@@ -94,14 +109,24 @@ class HeaderBar(Widget):
elif station is not None:
song = station.now_playing
if song is not None:
text = f"{station.title} · {song.display_artist} — {song.display_title}"
text = (
f"{station.title} · "
f"{song.display_artist} — {song.display_title}"
)
else:
text = f"{station.title} · {station.description}"
text_style = ""
else:
text = "Pick a station — press enter to tune in"
text_style = "italic dim"
ticker.append(marquee(text, max(4, width - cell_len(prefix))), style=text_style)
lines.append(ticker)
ticker.append(
marquee(text, max(4, width - cell_len(prefix))), style=text_style
)
# A short terminal loses the bottom of the hero before it loses the
# ticker: the ticker carries the state, the art can be cropped.
if ticker is not None:
if len(lines) >= height:
lines = lines[: height - 1]
lines.append(ticker)
return Group(*lines[:height])
+43 -3
View File
@@ -13,6 +13,7 @@ from textual.app import ComposeResult
from textual.screen import ModalScreen
from textual.widgets import Static
from ..providers import PROVIDERS, available_providers
from .common import SPECTRUM_STOPS, gradient
if TYPE_CHECKING:
@@ -33,7 +34,8 @@ HELP_ROWS: tuple[tuple[str, str], ...] = (
"v",
"cycle visualiser: mirror → bars → wave → waterfall → radial → braille",
),
("r", "refresh station list and song history"),
("r", "refresh stations, schedule and song history"),
("p", "switch station catalogue (SomaFM / tilderadio)"),
("g / G", "jump to the top / bottom of the list"),
("? ", "this help"),
("q", "quit"),
@@ -68,12 +70,40 @@ ABOUT_MACOS = (
class HelpScreen(ModalScreen[None]):
def compose(self) -> ComposeResult:
provider = cast("BeefyPlayerApp", self.app).provider
app = cast("BeefyPlayerApp", self.app)
provider = app.provider
about = (
f"Stations and song history come from {provider.name} "
f"({provider.station_source} + {provider.history_source}).",
) + (ABOUT_MACOS if sys.platform == "darwin" else ABOUT_LINUX)
catalogues = Table.grid(padding=(0, 2), expand=False)
catalogues.add_column(style="bold cyan", no_wrap=True, justify="right")
catalogues.add_column(style="bold", no_wrap=True)
catalogues.add_column(style="dim")
for slug in available_providers():
catalogue = PROVIDERS[slug]
marker = "●" if slug == provider.id else " "
pane = "schedule pane" if catalogue.has_schedule else "station list"
catalogues.add_row(
marker, slug, f"{catalogue.name} — {pane}"
)
pane_note = Text()
if provider.has_schedule:
pane_note.append(
f"{provider.name} is a single stream, so the left pane is its "
"broadcast schedule; the station filter and favourites only "
"apply to multi-station catalogues.",
style="dim italic",
)
catalogue_title = Text()
catalogue_title.append("CATALOGUES", style="bold cyan")
catalogue_title.append(
" switch with p or --provider at start", style="dim"
)
table = Table.grid(padding=(0, 2), expand=False)
table.add_column(style="bold cyan", no_wrap=True, justify="right")
table.add_column(style="")
@@ -90,7 +120,17 @@ class HelpScreen(ModalScreen[None]):
for line in textwrap.wrap(paragraph, width=70):
footer.append(line + "\n", style="dim")
body = Group(header, Text(""), table, Text(""), footer)
body = Group(
header,
Text(""),
table,
Text(""),
pane_note,
catalogue_title,
catalogues,
Text(""),
footer,
)
yield Static(body, id="help")
def on_key(self, event) -> None:
+87
View File
@@ -0,0 +1,87 @@
"""Catalogue picker: switch between SomaFM, tilderadio and whatever comes next."""
from __future__ import annotations
from rich.console import Group
from rich.text import Text
from textual.app import ComposeResult
from textual.screen import ModalScreen
from textual.widgets import Static
from ..providers import PROVIDERS, available_providers
from .common import SPECTRUM_STOPS, gradient, truncate
class ProviderScreen(ModalScreen[str | None]):
"""A list of catalogues; enter selects, esc leaves everything as it was."""
def __init__(self, current_id: str = "") -> None:
super().__init__()
self.slugs: tuple[str, ...] = available_providers()
self.current_id = current_id
self.index = self.slugs.index(current_id) if current_id in self.slugs else 0
def compose(self) -> ComposeResult:
yield Static(self._body(), id="providers")
def _body(self) -> Group:
header = Text()
header.append("◤ ", style=f"bold {gradient(SPECTRUM_STOPS, 0.9)}")
header.append("CATALOGUES", style=f"bold {gradient(SPECTRUM_STOPS, 0.55)}")
header.append(" pick a station catalogue", style="dim")
rows: list[Text] = []
for index, slug in enumerate(self.slugs):
provider = PROVIDERS[slug]
selected = index == self.index
row = Text(no_wrap=True)
row.append("▶ " if selected else " ", style="bold magenta" if selected else "")
row.append(f"{index + 1} ", style="dim")
name_style = "bold underline" if selected else "bold"
row.append(f"{provider.name:<12}", style=name_style)
row.append(f" {provider.brand}", style="dim")
if slug == self.current_id:
row.append(" ●", style="bold green")
rows.append(row)
detail = Text(no_wrap=True)
detail.append(" ", style="")
detail.append(truncate(provider.tagline, 54), style="italic dim")
rows.append(detail)
current = PROVIDERS.get(self.current_id)
footer = Text()
footer.append("\n")
if current is not None:
footer.append(
f"now: {current.name} — {current.station_source}\n", style="dim"
)
footer.append(
"↑ ↓ choose · 1–9 jump · enter switch · esc cancel",
style="dim",
)
return Group(header, Text(""), *rows, footer)
# -- interaction ----------------------------------------------------- #
def _move(self, delta: int) -> None:
self.index = (self.index + delta) % len(self.slugs)
self.query_one("#providers", Static).update(self._body())
def on_key(self, event) -> None:
key = event.key
if key in ("up", "k", "shift+tab"):
self._move(-1)
elif key in ("down", "j", "tab"):
self._move(1)
elif key in ("enter", "space"):
self.dismiss(self.slugs[self.index])
elif key == "escape" or key == "q":
self.dismiss(None)
elif key.isdigit() and key != "0":
choice = int(key) - 1
if 0 <= choice < len(self.slugs):
self.dismiss(self.slugs[choice])
else:
return
event.stop()
+137
View File
@@ -0,0 +1,137 @@
"""The left pane for single-stream catalogues: what is on, what is next.
A catalogue with one station does not need a station list, but it usually has
a broadcast schedule. This pane answers the two questions a listener actually
has — "who is on air?" and "when is the next show?" — in the space the list
would have used.
"""
from __future__ import annotations
from datetime import date, datetime, timedelta
from typing import TYPE_CHECKING, cast
from rich.console import Group, RenderableType
from rich.text import Text
from textual.widget import Widget
from .common import local_time, local_zone, state_glyph, truncate
if TYPE_CHECKING:
from ..app import BeefyPlayerApp
from ..models import Broadcast
def _window(broadcast: Broadcast | None) -> str:
"""``since 22:00 · until 23:00`` — whichever ends the feed actually sent."""
if broadcast is None:
return ""
start = local_time(broadcast.start) if broadcast.start else ""
end = local_time(broadcast.end) if broadcast.end else ""
if start and end:
return f"since {start} · until {end}"
if start:
return f"since {start}"
if end:
return f"until {end}"
return ""
def _day_label(day: date, today: date) -> str:
if day == today:
return "TODAY"
if day == today + timedelta(days=1):
return "TOMORROW"
return day.strftime("%a %d %b").upper()
def _upcoming_rows(
upcoming: list[Broadcast], width: int, budget: int
) -> list[Text]:
"""Upcoming shows, grouped by local day, clipped to the pane."""
zone = local_zone()
today = datetime.now(zone).date()
# (is a show, its line) so the "… +N more" count talks about shows and
# not about the day headings in between.
rows: list[tuple[bool, Text]] = []
current_day: date | None = None
for item in upcoming:
if item.start is None:
continue
day = item.start.astimezone(zone).date()
if day != current_day:
current_day = day
rows.append((False, Text(_day_label(day, today), style="dim")))
row = Text(no_wrap=True)
row.append(" ")
row.append(local_time(item.start), style="bold")
row.append(" ")
row.append(truncate(item.display_name, max(4, width - 7)))
rows.append((True, row))
if len(rows) > budget:
kept = rows[: max(0, budget - 1)]
while kept and not kept[-1][0]:
kept.pop()
dropped = sum(1 for is_show, _ in rows[len(kept) :] if is_show)
lines = [text for _, text in kept]
if dropped:
lines.append(Text(f"… +{dropped} more", style="dim italic"))
return lines
return [text for _, text in rows]
class SchedulePanel(Widget):
can_focus = False
def __init__(self) -> None:
super().__init__()
self.border_title = "SCHEDULE"
def render(self) -> RenderableType:
app = cast("BeefyPlayerApp", self.app)
width = max(12, self.size.width)
height = max(1, self.size.height)
broadcasts = list(getattr(app, "schedule", ()) or ())
loaded = bool(getattr(app, "schedule_loaded", False))
live = next((item for item in broadcasts if item.live), None)
upcoming = [item for item in broadcasts if not item.live]
lines: list[Text] = []
head = Text(no_wrap=True)
glyph, colour = state_glyph("playing" if live else "stopped")
head.append(f"{glyph} ", style=f"bold {colour}")
head.append("ON AIR" if live else "AUTO DJ", style="bold magenta")
if live is not None:
head.append(" ", style="")
head.append(
truncate(live.display_name, max(4, width - 12)),
style="bold",
)
lines.append(head)
window = _window(live)
lines.append(
Text(truncate(window, width), style="dim")
if window
else Text("library rotation", style="dim italic")
)
lines.append(Text(""))
lines.append(Text("UPCOMING", style="bold cyan"))
if not loaded:
lines.append(Text("loading schedule…", style="dim italic"))
elif not upcoming:
lines.append(Text("nothing scheduled", style="dim italic"))
else:
lines.extend(
_upcoming_rows(upcoming, width, max(0, height - 1 - len(lines)))
)
lines.append(Text(""))
while len(lines) < height:
lines.append(Text(""))
return Group(*lines[:height])
+6 -3
View File
@@ -140,11 +140,14 @@ class StageInfo(Widget):
album = Text(no_wrap=True)
if song.album:
album.append(truncate(song.album, max(8, width - 22)), style="dim")
if station.dj:
presenter = station.presenter_label
if presenter:
if song.album:
album.append(" · ", style="dim")
album.append(f"dj {truncate(station.dj, 18)}", style="dim")
if song.art_url:
album.append(truncate(presenter, max(8, width - 22)), style="dim")
# The marker means "there is artwork", not "here is a lone box":
# without album or presenter there is nothing for it to mark.
if song.art_url and album.plain.strip():
album.append(" ▣", style="cyan")
else:
artist = Text("waiting for track data…", style="italic yellow")
+2 -2
View File
@@ -1,13 +1,13 @@
[project]
name = "beefy-player"
version = "0.1.0"
description = "A flashy terminal player for internet radio — live spectrum analyser, MPRIS / macOS Now Playing support, song history and pluggable station catalogues (SomaFM included)"
description = "A flashy terminal player for internet radio — live spectrum analyser, MPRIS / macOS Now Playing support, song history and pluggable station catalogues (SomaFM and tilderadio included)"
readme = "README.md"
license = "WTFPL"
license-files = ["LICENSE"]
requires-python = ">=3.10"
authors = [{ name = "beefy-player contributors" }]
keywords = ["radio", "internet-radio", "tui", "terminal", "mpris", "audio", "somafm"]
keywords = ["radio", "internet-radio", "tui", "terminal", "mpris", "audio", "somafm", "tilderadio"]
classifiers = [
"Environment :: Console",
"Intended Audience :: End Users/Desktop",
+51 -4
View File
@@ -6,6 +6,8 @@ import types
import pytest
from rich.cells import cell_len
from beefy_player.providers.somafm import SomaFMProvider
from beefy_player.providers.tilderadio import TilderadioProvider
from beefy_player.widgets.header import HeaderBar
TIME = re.compile(r"\d\d:\d\d:\d\d$")
@@ -13,11 +15,17 @@ DATE = re.compile(r"[A-Z][a-z]{2} \d\d [A-Z][a-z]{2}$")
VOLUME = re.compile(r"vol \d+%$")
def render_header(width: int, height: int = 3) -> list[str]:
def render_header(
width: int, height: int = 3, provider=None
) -> list[str]:
"""Render the header without a running app, as plain text per line."""
fake = type("Fake", (), {})()
fake.app = types.SimpleNamespace(
playing_state="playing", volume=70, current=None, notice=None
playing_state="playing",
volume=70,
current=None,
notice=None,
provider=provider or SomaFMProvider(),
)
fake.size = types.SimpleNamespace(width=width, height=height)
return [line.plain.rstrip() for line in HeaderBar.render(fake).renderables]
@@ -34,8 +42,8 @@ def test_right_block_is_flush_right(width: int) -> None:
def test_ragged_middle_does_not_shift_the_right_block() -> None:
""""SOMA FM RADIO" and "SAN FRANCISCO" are shorter than
"LISTENER SUPPORTED"; the right block must not follow their length."""
""""TILDERADIO" is shorter than "STREAMED BY / FOR USERS OF THE"; the
right block must not follow the length of the middle lines."""
clock, date, volume = render_header(143)
assert cell_len(clock) == cell_len(date) == cell_len(volume) == 143
@@ -44,3 +52,42 @@ def test_right_block_hidden_on_narrow_terminals() -> None:
"""Below the threshold the right block is dropped, not squashed."""
lines = render_header(30)
assert all(":" not in line for line in lines)
def test_somafm_hero_is_the_wordmark_and_its_taglines() -> None:
lines = render_header(143)
assert lines[0].startswith("▄▀▀▀ ▄▀▀▄ █▄ ▄█ ▄▀▀▄")
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 = [
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)
assert all("SOMA" not in line for line in lines)
def _fake(provider):
fake = type("Fake", (), {})()
fake.app = types.SimpleNamespace(
playing_state="playing",
volume=70,
current=None,
notice=None,
provider=provider,
)
fake.size = types.SimpleNamespace(width=143, height=len(provider.hero) or 3)
return fake
+115
View File
@@ -0,0 +1,115 @@
"""The ON AIR metadata line: who is on the desk, and no lone artwork glyph."""
from __future__ import annotations
import types
import pytest
from beefy_player.app import BeefyPlayerApp
from beefy_player.models import Song, Station
from beefy_player.providers.somafm import SomaFMProvider
from beefy_player.providers.tilderadio import TilderadioProvider, api
from beefy_player.widgets.stage import StageInfo
def details(station: Station, song: Song | None, width: int = 90) -> list[str]:
app = types.SimpleNamespace(playing_state="playing", notice="")
return [line.plain for line in StageInfo()._details(app, station, song, width)]
def test_somafm_still_prefixes_a_dj_name() -> None:
station = Station(id="gs", title="Groove Salad", dj="SomaFM", provider="somafm")
lines = details(station, Song(artist="A", title="T", album="Album"))
assert any("Album" in line and "dj SomaFM" in line for line in lines)
def test_lone_artwork_glyph_is_not_drawn() -> None:
"""No album, no presenter: the ▣ would be a box with nothing to mark."""
station = Station(id="x", title="X", provider="somafm")
lines = details(station, Song(artist="A", title="T", art_url="https://a/1.jpg"))
assert not any("▣" in line for line in lines)
def test_artwork_glyph_rides_along_with_real_information() -> None:
station = Station(id="x", title="X", dj="Someone", provider="somafm")
lines = details(station, Song(artist="A", title="T", art_url="https://a/1.jpg"))
assert any("dj Someone" in line and "▣" in line for line in lines)
def test_tilderadio_shows_the_rotation_when_nothing_is_live() -> None:
payload = {
"station": {"name": "tilderadio", "shortcode": "tilderadio"},
"is_live": False,
}
station = api.station_from_now(payload)
assert station is not None
assert station.presenter_label == "auto dj · library rotation"
lines = details(station, Song(artist="A", title="T", art_url="https://a/1.jpg"))
assert any("auto dj · library rotation" in line for line in lines)
def test_tilderadio_shows_the_live_dj_instead() -> None:
payload = {
"station": {"name": "tilderadio", "shortcode": "tilderadio"},
"is_live": True,
"dj": "dokuja",
}
station = api.station_from_now(payload)
assert station is not None
assert station.presenter_label == "dj dokuja"
assert "auto dj" not in station.presenter_label
def test_presenter_label_falls_back_to_the_dj_then_nothing() -> None:
assert Station(id="a", title="A").presenter_label == ""
assert Station(id="a", title="A", dj="kexp").presenter_label == "dj kexp"
assert (
Station(id="a", title="A", dj="kexp", presenter="guest host").presenter_label
== "guest host"
)
def test_tilderadio_opts_out_of_bandcamp_lookups() -> None:
assert TilderadioProvider().bandcamp is False
assert SomaFMProvider().bandcamp is True
@pytest.mark.asyncio
async def test_no_bandcamp_worker_is_started_for_tilderadio(monkeypatch) -> None:
calls: list[tuple] = []
async def fake_find(client, artist, album, title):
calls.append((artist, album, title))
return "https://bandcamp.test/track"
monkeypatch.setattr("beefy_player.bandcamp.find_track", fake_find)
app = BeefyPlayerApp(
provider="tilderadio", autoplay=False, offline=True, audio=False
)
async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause()
station = app.current
assert station is not None
station.now_playing = Song(artist="A", title="T")
app._update_bandcamp()
await pilot.pause()
assert calls == []
assert station.now_playing.bandcamp_url == ""
@pytest.mark.asyncio
async def test_a_stale_bandcamp_link_is_cleared_when_the_catalogue_does_not_want_one() -> None:
app = BeefyPlayerApp(
provider="tilderadio", autoplay=False, offline=True, audio=False
)
async with app.run_test(size=(100, 40)) as pilot:
await pilot.pause()
station = app.current
assert station is not None
station.now_playing = Song(
artist="A", title="T", bandcamp_url="https://bandcamp.test/old"
)
app._update_bandcamp()
assert station.now_playing.bandcamp_url == ""
+106 -1
View File
@@ -11,6 +11,8 @@ 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.somafm import SomaFMProvider
from beefy_player.providers.tilderadio import TilderadioProvider
from beefy_player.widgets.header import HeaderBar
def rendered_text(app) -> str:
@@ -19,9 +21,10 @@ def rendered_text(app) -> str:
def test_registry_lists_somafm_as_the_default() -> None:
assert available_providers() == ("somafm",)
assert available_providers() == ("somafm", "tilderadio")
assert isinstance(get_provider(), SomaFMProvider)
assert get_provider("somafm") is get_provider()
assert isinstance(get_provider("tilderadio"), TilderadioProvider)
# a stale slug in the config file must not stop the app from starting
assert get_provider("no-such-catalogue") is get_provider()
@@ -128,3 +131,105 @@ def test_environment_variables_prefer_the_beefy_prefix(monkeypatch) -> None:
monkeypatch.delenv("SOMAFM_IMAGES")
assert config.env_var("IMAGES") == ""
def test_tilderadio_replaces_the_station_list_with_a_schedule() -> None:
provider = TilderadioProvider()
assert provider.has_schedule is True
# the SomaFM wordmark's slot now holds tilderadio's own art and motto
assert provider.hero and not provider.hero_gradient
assert provider.hero_lines == (
"TILDERADIO",
"STREAMED BY / FOR USERS OF THE",
"TILDEVERSE",
)
assert SomaFMProvider().has_schedule is False
class _CatalogueApp:
"""A headless app on a fixed catalogue, for the picker tests."""
def __init__(self, provider: str = "somafm") -> None:
from beefy_player.app import BeefyPlayerApp
self.app = BeefyPlayerApp(
provider=provider, autoplay=False, offline=True, audio=False
)
@pytest.mark.asyncio
async def test_p_opens_the_picker_and_switches_catalogue() -> None:
from beefy_player.widgets.providers import ProviderScreen
from beefy_player.widgets.schedule import SchedulePanel
from beefy_player.widgets.station_list import StationList
app = _CatalogueApp().app
async with app.run_test(size=(110, 40)) as pilot:
await pilot.pause()
assert app.provider.id == "somafm"
assert app.query_one(StationList).display is True
assert app.query_one(SchedulePanel).display is False
assert app.query_one(HeaderBar).styles.height.value == 4
await pilot.press("p")
await pilot.pause()
assert isinstance(app.screen, ProviderScreen)
await pilot.press("2") # jump straight to tilderadio
await pilot.pause()
await pilot.pause()
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
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 "tilderadio" in app.title
@pytest.mark.asyncio
async def test_escape_closes_the_picker_without_switching() -> None:
app = _CatalogueApp().app
async with app.run_test(size=(110, 40)) as pilot:
await pilot.pause()
await pilot.press("p")
await pilot.pause()
await pilot.press("escape")
await pilot.pause()
assert app.provider.id == "somafm"
assert app.config.provider != "tilderadio"
@pytest.mark.asyncio
async def test_help_names_both_catalogues_and_the_switch_key() -> None:
app = _CatalogueApp().app
async with app.run_test(size=(120, 60)) as pilot:
await pilot.pause()
await pilot.press("question_mark")
await pilot.pause()
text = rendered_text(app)
assert "CATALOGUES" in text
assert "somafm" in text and "tilderadio" in text
assert "switch station catalogue" in text
def test_config_remembers_a_station_per_catalogue() -> None:
config.Config(last_stations={"somafm": "defcon", "tilderadio": "tilderadio"}).save()
loaded = config.Config.load()
assert loaded.last_station_for("somafm") == "defcon"
assert loaded.last_station_for("tilderadio") == "tilderadio"
# a catalogue nobody has tuned yet falls back to the legacy single field
assert loaded.last_station_for("no-such-catalogue") == loaded.last_station
@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)
+166
View File
@@ -0,0 +1,166 @@
"""tilderadio's schedule: parsing, the live slot, caching and the pane."""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import httpx
import pytest
from beefy_player.app import BeefyPlayerApp
from beefy_player.models import Broadcast
from beefy_player.providers.tilderadio import api
from beefy_player.widgets.schedule import SchedulePanel, _upcoming_rows
def iso(moment: datetime) -> str:
return moment.isoformat()
def test_parse_schedule_reads_the_documented_feed() -> None:
payload = {
"schedule": [
{
"name": "controlfreak",
"slug": "controlfreak",
"start": "2026-11-04T02:00:00+00:00",
"end": "2026-11-04T03:00:00+00:00",
"start_ts": 1793757600,
"end_ts": 1793761200,
},
{
"name": "pa2",
"slug": "pa2",
"start": "2026-11-04T14:00:00+00:00",
"end": "2026-11-04T15:00:00+00:00",
"start_ts": 1793800800,
"end_ts": 1793804400,
},
]
}
shows = api.parse_schedule(payload)
assert [show.name for show in shows] == ["controlfreak", "pa2"]
assert shows[0].start == datetime(2026, 11, 4, 2, tzinfo=timezone.utc)
assert shows[0].end == datetime(2026, 11, 4, 3, tzinfo=timezone.utc)
assert not shows[0].live
# junk is dropped rather than crashing the pane
assert api.parse_schedule({"schedule": [None, {}, {"name": "no start"}]}) == []
assert api.parse_schedule(None) == []
def test_live_broadcast_only_when_someone_is_on_air() -> None:
payload = {
"is_live": False,
"dj": None,
"broadcast_start": None,
}
assert api.live_broadcast(payload) is None
payload = {
"is_live": True,
"dj": "dokuja",
"broadcast_start": 1793599228,
}
live = api.live_broadcast(payload)
assert live is not None
assert (live.name, live.dj, live.live) == ("dokuja", "dokuja", True)
assert live.start is not None
def client_for(now_payload, schedule_payload, status: int = 200) -> httpx.AsyncClient:
def handler(request: httpx.Request) -> httpx.Response:
if str(request.url).endswith("/api/now/"):
return httpx.Response(200, json=now_payload)
if str(request.url).endswith("/api/schedule/"):
return httpx.Response(status, json=schedule_payload)
return httpx.Response(404, json={})
return httpx.AsyncClient(transport=httpx.MockTransport(handler))
@pytest.mark.asyncio
async def test_fetch_schedule_puts_the_live_show_first_and_drops_the_past() -> None:
now = datetime.now(timezone.utc)
now_payload = {
"is_live": True,
"dj": "dokuja",
"broadcast_start": iso(now - timedelta(minutes=25)),
}
schedule_payload = {
"schedule": [
{
"name": "yesterday",
"start": iso(now - timedelta(days=1)),
"end": iso(now - timedelta(days=1) + timedelta(hours=1)),
},
{
"name": "controlfreak",
"start": iso(now + timedelta(hours=2)),
"end": iso(now + timedelta(hours=3)),
},
]
}
async with client_for(now_payload, schedule_payload) as client:
broadcasts = await api.fetch_schedule(client)
assert broadcasts[0].live and broadcasts[0].name == "dokuja"
assert [item.name for item in broadcasts[1:]] == ["controlfreak"]
# ... and the schedule survives a dead endpoint
async with client_for({"is_live": False}, {}, status=500) as client:
offline = await api.fetch_schedule(client)
assert [item.name for item in offline] == ["controlfreak"]
def broadcast(name: str, minutes: int, *, live: bool = False) -> Broadcast:
start = datetime.now(timezone.utc) + timedelta(minutes=minutes)
return Broadcast(
name=name, dj=name, start=start, end=start + timedelta(hours=1), live=live
)
async def panel_lines(broadcasts: list[Broadcast], size=(100, 40)) -> list[str]:
app = BeefyPlayerApp(provider="tilderadio", offline=True, audio=False)
async with app.run_test(size=size) as pilot:
await pilot.pause()
app.schedule = broadcasts
app.schedule_loaded = True
panel = app.query_one(SchedulePanel)
return [line.plain for line in panel.render().renderables]
@pytest.mark.asyncio
async def test_panel_shows_the_transmitter_and_what_is_next() -> None:
lines = await panel_lines(
[broadcast("dokuja", -20, live=True), broadcast("controlfreak", 90)]
)
text = "\n".join(lines)
assert "ON AIR" in text and "dokuja" in text
assert "since " in text and "until " in text
assert "UPCOMING" in text
assert "controlfreak" in text
@pytest.mark.asyncio
async def test_panel_says_auto_dj_when_nothing_is_live() -> None:
lines = await panel_lines([broadcast("pa2", 120)])
text = "\n".join(lines)
assert "AUTO DJ" in text
assert "library rotation" in text
assert "ON AIR" not in text
@pytest.mark.asyncio
async def test_panel_handles_an_empty_schedule() -> None:
assert "nothing scheduled" in "\n".join(await panel_lines([]))
def test_upcoming_rows_count_the_shows_they_had_to_hide() -> None:
many = [broadcast(f"show {index}", 60 * (index + 1)) for index in range(30)]
lines = [line.plain for line in _upcoming_rows(many, 34, 10)]
assert len(lines) <= 10
assert any("more" in line for line in lines)
# every show is either listed or counted in the "… +N more" line
listed = sum(1 for line in lines if "show " in line)
hidden = int(lines[-1].split("+")[1].split()[0])
assert listed + hidden == 30
+2
View File
@@ -7,6 +7,7 @@ import pytest
from beefy_player.app import BeefyPlayerApp
from beefy_player.widgets.header import HeaderBar
from beefy_player.widgets.schedule import SchedulePanel
from beefy_player.widgets.search import SearchInput
from beefy_player.widgets.spectrum import SpectrumPanel
from beefy_player.widgets.stage import StagePanel
@@ -42,6 +43,7 @@ async def test_backgrounds_use_terminal_default_and_borders_are_ansi() -> None:
HeaderBar,
SearchInput,
StationList,
SchedulePanel,
StagePanel,
SpectrumPanel,
StatusBar,
+249
View File
@@ -0,0 +1,249 @@
"""tilderadio: parsing, caching and the offline fallback.
The payloads below are trimmed copies of real ``/api/now/`` and AzuraCast
``/api/nowplaying/<station>`` responses, so the tests keep working when the
station is off air.
"""
from __future__ import annotations
import httpx
import pytest
from beefy_player.providers.tilderadio import TilderadioProvider, api, fallback
NOW_PAYLOAD = {
"api_version": 1,
"status": "ok",
"updated_at": 1793741211,
"available": True,
"station": {
"id": 1,
"name": "tilderadio",
"shortcode": "tilderadio",
"listen_url": "https://azuracast.tilderadio.org/listen/tilderadio/radio.ogg",
},
"is_live": False,
"broadcast_start": None,
"dj": None,
"listeners": 5,
"now_playing": {
"artist": None,
"title": "souldies-dramatics-feel it",
"text": "souldies-dramatics-feel it",
"art": "https://azuracast.tilderadio.org/api/station/tilderadio/art/199f03",
"played_at": 1793741079,
"duration": 192,
},
"history": [
{
"artist": None,
"title": "souldies-everytime i see a pretty lady-gemini",
"text": "souldies-everytime i see a pretty lady-gemini",
"art": "https://azuracast.tilderadio.org/api/station/tilderadio/art/5e46a9",
"played_at": 1793740639,
},
{
"artist": "TILDE PROMO (station)",
"title": "owly (bots 3)",
"text": "TILDE PROMO (station) - owly (bots 3)",
"art": "https://azuracast.tilderadio.org/api/station/tilderadio/art/54ac9f",
"played_at": 1793740629,
},
],
}
MOUNTS_PAYLOAD = {
"station": {
"id": 1,
"name": "tilderadio",
"shortcode": "tilderadio",
"listen_url": "https://azuracast.tilderadio.org/listen/tilderadio/radio.ogg",
"mounts": [
{
"name": "/radio.mp3 (192kbps MP3)",
"url": "https://azuracast.tilderadio.org/listen/tilderadio/radio.mp3",
"bitrate": 192,
"format": "mp3",
"is_default": False,
},
{
"name": "/320k.mp3 (320kbps MP3)",
"url": "https://azuracast.tilderadio.org/listen/tilderadio/320k.mp3",
"bitrate": 320,
"format": "mp3",
"is_default": False,
},
{
"name": "192kbps OGG Vorbis",
"url": "https://azuracast.tilderadio.org/listen/tilderadio/radio.ogg",
"bitrate": 192,
"format": "ogg",
"is_default": True,
},
],
}
}
def client_for(*responses: tuple[str, int, object]) -> httpx.AsyncClient:
"""An httpx client that answers from ``(url_prefix, status, payload)``."""
table = list(responses)
def handler(request: httpx.Request) -> httpx.Response:
for prefix, status, payload in table:
if str(request.url).startswith(prefix):
return httpx.Response(status, json=payload)
return httpx.Response(404, json={})
return httpx.AsyncClient(transport=httpx.MockTransport(handler))
def test_provider_describes_its_catalogue() -> None:
provider = TilderadioProvider()
assert provider.id == "tilderadio"
assert provider.brand == "TILDE RADIO"
assert provider.station_source == "tilderadio.org/api/now/"
assert "tilderadio.org" in provider.user_agent
stations = provider.fallback_stations()
assert len(stations) == 1
station = stations[0]
assert station.uid == "tilderadio:tilderadio"
assert all("tilderadio.org" in url for url in provider.fallback_urls(station))
# the bundled list still advertises a real bitrate, not a URL guess
assert station.format_label == "mp3 · 320k"
def test_station_from_now_reads_the_single_station() -> None:
station = api.station_from_now(NOW_PAYLOAD, fallback.fallback_playlists())
assert station is not None
assert station.id == "tilderadio"
assert station.provider == "tilderadio"
assert station.listeners == 5
assert station.dj == "" # nobody is on air
assert station.genre_label == "community · eclectic"
assert station.now_playing is not None
assert station.now_playing.title == "souldies-dramatics-feel it"
assert station.now_playing.art_url.endswith("199f03")
assert station.last_playing == "souldies-dramatics-feel it"
def test_live_dj_is_only_named_when_someone_is_on_air() -> None:
live = dict(NOW_PAYLOAD, is_live=True, dj="ffog")
station = api.station_from_now(live)
assert station is not None and station.dj == "ffog"
# the AzuraCast flavour of the same field wins when present
nested = dict(NOW_PAYLOAD, live={"is_live": True, "name": "controlfreak"})
station = api.station_from_now(nested)
assert station is not None and station.dj == "controlfreak"
def test_artist_is_split_out_of_the_text_blob() -> None:
song = api.song_from_track(
{
"artist": "TILDE PROMO (station)",
"title": "owly (bots 3)",
"text": "TILDE PROMO (station) - owly (bots 3)",
"played_at": 1793740629,
}
)
assert song is not None
assert (song.artist, song.title) == ("TILDE PROMO (station)", "owly (bots 3)")
# no artist field: the "Artist - Title" text has to carry it
song = api.song_from_track(
{"title": "Kenny Ken", "text": "Kenny Ken - Amazon Jungle Collection 3"}
)
assert song is not None
assert (song.artist, song.title) == ("Kenny Ken", "Amazon Jungle Collection 3")
def test_songs_from_now_puts_the_current_track_first_and_dedupes() -> None:
payload = dict(NOW_PAYLOAD)
# a feed that repeats the current track in its own history
payload["history"] = [NOW_PAYLOAD["now_playing"], *NOW_PAYLOAD["history"]]
songs = api.songs_from_now(payload)
assert len(songs) == 3
assert songs[0].title == "souldies-dramatics-feel it"
assert songs[1].title == "souldies-everytime i see a pretty lady-gemini"
played = [song.played_at for song in songs]
assert played == sorted(played, reverse=True)
def test_mounts_become_playlists_best_first() -> None:
playlists = api.playlists_from_mounts(MOUNTS_PAYLOAD)
assert [entry["bitrate"] for entry in playlists] == ["192", "320", "192"]
station = api.station_from_now(
{**NOW_PAYLOAD, "now_playing": None}, playlists
)
assert station is not None
assert station.format_label == "mp3 · 320k"
assert station.sorted_playlists()[0]["url"].endswith("320k.mp3")
assert [entry["quality"] for entry in playlists] == ["high", "highest", "high"]
def test_station_without_mounts_falls_back_to_the_listen_url() -> None:
station = api.station_from_now(NOW_PAYLOAD)
assert station is not None
assert len(station.playlists) == 1
assert station.playlists[0]["url"].endswith("radio.ogg")
assert station.format_label == "ogg · 128k" # guessed, the feed didn't say
@pytest.mark.asyncio
async def test_fetch_station_caches_what_it_learned() -> None:
async with client_for(
("https://tilderadio.org/api/now/", 200, NOW_PAYLOAD),
("https://azuracast.tilderadio.org/api/nowplaying/", 200, MOUNTS_PAYLOAD),
) as client:
station = await api.fetch_station(client)
assert station.listeners == 5
assert station.format_label == "mp3 · 320k"
cached = api.load_cached_station()
assert len(cached) == 1
assert cached[0].id == "tilderadio"
assert cached[0].now_playing is not None
assert cached[0].format_label == "mp3 · 320k"
@pytest.mark.asyncio
async def test_fetch_station_survives_an_unreachable_azuracast() -> None:
async with client_for(("https://tilderadio.org/api/now/", 200, NOW_PAYLOAD)) as client:
station = await api.fetch_station(client)
# no mounts -> the documented listen_url is still a usable stream
assert station.playlists[0]["url"].endswith("radio.ogg")
@pytest.mark.asyncio
async def test_history_falls_back_to_the_cache_when_the_feed_is_down() -> None:
async with client_for(
("https://tilderadio.org/api/now/", 200, NOW_PAYLOAD)
) as client:
songs = await api.fetch_songs(client, "tilderadio")
assert len(songs) == 3
async with client_for(
("https://tilderadio.org/api/now/", 500, {})
) as client:
cached = await api.fetch_songs(client, "tilderadio")
assert [song.title for song in cached] == [song.title for song in songs]
def test_station_cache_is_per_provider() -> None:
from beefy_player import config
assert config.channels_cache_path("tilderadio").name == "channels-tilderadio.json"
# SomaFM keeps its historical file name, so upgrades don't lose the cache
assert config.channels_cache_path("somafm") == config.CHANNELS_CACHE
def test_fallback_station_is_usable_offline() -> None:
provider = TilderadioProvider()
stations = provider.fallback_stations()
assert [station.id for station in stations] == ["tilderadio"]
station = stations[0]
assert station.listeners == 0
assert station.now_playing is None
assert {entry["format"] for entry in station.playlists} == {"mp3", "ogg"}
+210
View File
@@ -0,0 +1,210 @@
#!/usr/bin/env python3
"""Turn a logo into half-block ANSI art for the header hero.
The header hero is a handful of text rows, so an image has to survive a brutal
downsample: two pixels per cell (upper/lower half block), one ANSI colour per
half. Colours are quantised to the stock 16-colour ANSI palette on purpose —
the rest of the UI never emits RGB, and a hero built from palette names still
reads as the station's art in any terminal theme.
Usage:
python tools/ansi_art.py logo.png --cols 22 --rows 7
python tools/ansi_art.py --url https://tilderadio.org/logos/tilderadio.png \\
--cols 22 --rows 7 --module beefy_player/providers/tilderadio/hero.py
``--preview out.png`` draws what the cells will look like, which is much easier
to judge than a wall of ▀ and ▄.
"""
from __future__ import annotations
import argparse
import sys
from pathlib import Path
from PIL import Image, ImageDraw
#: The stock xterm colours the names below resolve to in a terminal.
ANSI16: dict[str, tuple[int, int, int]] = {
"black": (0, 0, 0),
"red": (170, 0, 0),
"green": (0, 170, 0),
"yellow": (170, 85, 0),
"blue": (0, 0, 170),
"magenta": (170, 0, 170),
"cyan": (0, 170, 170),
"white": (170, 170, 170),
"bright_black": (85, 85, 85),
"bright_red": (255, 85, 85),
"bright_green": (85, 255, 85),
"bright_yellow": (255, 255, 85),
"bright_blue": (85, 85, 255),
"bright_magenta": (255, 85, 255),
"bright_cyan": (85, 255, 255),
"bright_white": (255, 255, 255),
}
#: Below this alpha a half-cell counts as empty (terminal background).
ALPHA_CUTOFF = 110
def nearest_colour(rgb: tuple[int, int, int]) -> str:
best, best_distance = "white", float("inf")
for name, reference in ANSI16.items():
distance = sum((a - b) ** 2 for a, b in zip(rgb, reference, strict=True))
if distance < best_distance:
best, best_distance = name, distance
return best
def load_image(path: Path | None, url: str | None) -> Image.Image:
if path is None and url:
import httpx
response = httpx.get(url, follow_redirects=True, timeout=30.0)
response.raise_for_status()
from io import BytesIO
image = Image.open(BytesIO(response.content))
elif path is not None:
image = Image.open(path)
else: # pragma: no cover - argparse enforces one of the two
raise SystemExit("pass an image path or --url")
image = image.convert("RGBA")
# Trim the transparent margin: the art should fill the box it is given.
opaque = image.getchannel("A").point(lambda value: 255 if value > 24 else 0)
box = opaque.getbbox()
return image.crop(box) if box else image
def art_rows(image: Image.Image, cols: int, rows: int) -> list[list[tuple[str, str, str]]]:
"""(glyph, foreground, background) per cell, two source pixels per cell."""
small = image.resize((cols, rows * 2), Image.LANCZOS)
pixels = small.load()
grid: list[list[tuple[str, str, str]]] = []
for row in range(rows):
cells: list[tuple[str, str, str]] = []
for column in range(cols):
top = pixels[column, row * 2]
bottom = pixels[column, row * 2 + 1]
top_ink = top[3] >= ALPHA_CUTOFF
bottom_ink = bottom[3] >= ALPHA_CUTOFF
if not top_ink and not bottom_ink:
cells.append((" ", "", ""))
elif top_ink and bottom_ink:
cells.append(("▀", nearest_colour(top[:3]), nearest_colour(bottom[:3])))
elif top_ink:
cells.append(("▀", nearest_colour(top[:3]), ""))
else:
cells.append(("▄", nearest_colour(bottom[:3]), ""))
grid.append(cells)
return grid
def markup(grid: list[list[tuple[str, str, str]]]) -> list[str]:
"""One rich-markup string per row, run-length encoded by style."""
lines: list[str] = []
for cells in grid:
line: list[str] = []
index = 0
while index < len(cells):
glyph, foreground, background = cells[index]
if not background and glyph == " ":
line.append(" ")
index += 1
continue
run = [glyph]
index += 1
while index < len(cells) and cells[index] == (glyph, foreground, background):
run.append(cells[index][0])
index += 1
style = " ".join(
part for part in (foreground, f"on {background}" if background else "") if part
)
body = "".join(run)
line.append(f"[{style}]{body}[/]" if style else body)
lines.append("".join(line).rstrip())
return lines
def preview(grid: list[list[tuple[str, str, str]]], path: Path, cell: tuple[int, int]) -> None:
width, height = cell
image = Image.new("RGB", (len(grid[0]) * width, len(grid) * height), (24, 24, 24))
draw = ImageDraw.Draw(image)
for row, cells in enumerate(grid):
for column, (glyph, foreground, background) in enumerate(cells):
x, y = column * width, row * height
if background:
draw.rectangle([x, y, x + width - 1, y + height - 1], fill=ANSI16[background])
if glyph == " ":
continue
colour = ANSI16.get(foreground, (140, 140, 140))
if glyph == "▀":
box = [x, y, x + width - 1, y + height // 2 - 1]
else:
box = [x, y + height // 2, x + width - 1, y + height - 1]
draw.rectangle(box, fill=colour)
image.resize((image.width * 3, image.height * 3), Image.NEAREST).save(path)
MODULE_TEMPLATE = '''"""Generated header art — do not edit by hand.
{provenance}
Regenerate with::
{command}
Each row is rich markup; ``▀`` is a half block whose foreground is the top
pixel and whose background is the pixel below it. Colours are stock ANSI
palette names, so the art keeps the terminal's own palette.
"""
from __future__ import annotations
HERO: tuple[str, ...] = (
{rows}
)
'''
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("image", nargs="?", type=Path, help="local image file")
parser.add_argument("--url", help="fetch the image from this URL instead")
parser.add_argument("--cols", type=int, default=22, help="terminal columns (default 22)")
parser.add_argument("--rows", type=int, default=7, help="terminal rows (default 7)")
parser.add_argument("--module", type=Path, help="write a Python module with HERO")
parser.add_argument("--preview", type=Path, help="write a PNG preview of the cells")
parser.add_argument("--provenance", default="", help="source note for the module header")
args = parser.parse_args(argv)
image = load_image(args.image, args.url)
grid = art_rows(image, args.cols, args.rows)
rows = markup(grid)
if args.preview:
preview(grid, args.preview, (10, 20))
if args.module:
source = args.provenance or args.url or str(args.image)
command = "uv run python tools/ansi_art.py " + " ".join(sys.argv[1:])
body = "\n".join(f" {row!r}," for row in rows)
args.module.write_text(
MODULE_TEMPLATE.format(
provenance=f"Source: {source}",
command=command,
rows=body,
),
encoding="utf-8",
)
print(f"wrote {args.module} ({len(rows)} rows x {args.cols} cols)")
else:
print("\n".join(rows))
return 0
if __name__ == "__main__":
raise SystemExit(main())