Compare commits
4
Commits
84ec3b08f6
...
a3ad693132
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a3ad693132 | ||
|
|
b3bd8c1e9f | ||
|
|
e14a349c2d | ||
|
|
50a9a735f7 |
No files matched your search
@@ -8,9 +8,10 @@ own ANSI palette on the terminal's default background so it fits any theme
|
|||||||
(including transparent terminals).
|
(including transparent terminals).
|
||||||
|
|
||||||
The station catalogue is pluggable. Beefy Player ships with
|
The station catalogue is pluggable. Beefy Player ships with
|
||||||
**[SomaFM](https://somafm.com)** (≈50 curated channels) and
|
**[SomaFM](https://somafm.com)** (≈50 curated channels),
|
||||||
**[tilderadio](https://tilderadio.org)** (the tildeverse's single community
|
**[tilderadio](https://tilderadio.org)** (the tildeverse's single community
|
||||||
stream), and the UI never talks to a station catalogue directly: it asks a
|
stream) and **[Bassdrive](https://bassdrive.com)** (worldwide drum & bass), and
|
||||||
|
the UI never talks to a station catalogue directly: it asks a
|
||||||
[provider](#providers) for stations, song history and stream URLs. Press `p` to
|
[provider](#providers) for stations, song history and stream URLs. Press `p` to
|
||||||
switch catalogue without restarting. Adding another internet-radio catalogue is
|
switch catalogue without restarting. Adding another internet-radio catalogue is
|
||||||
a self-contained module, not a rewrite.
|
a self-contained module, not a rewrite.
|
||||||
@@ -41,7 +42,9 @@ implemented here from public endpoints and open protocols.
|
|||||||
iTerm2 inline-image protocol. Images occupy exactly the ON AIR art cell box
|
iTerm2 inline-image protocol. Images occupy exactly the ON AIR art cell box
|
||||||
and fall back to the procedural ANSI mosaic for other terminals, failed
|
and fall back to the procedural ANSI mosaic for other terminals, failed
|
||||||
loads, or blocked graphics passthrough. Images are cached in
|
loads, or blocked graphics passthrough. Images are cached in
|
||||||
`~/.cache/beefy-player/artwork`.
|
`~/.cache/beefy-player/artwork`, letterboxed rather than cropped, and an
|
||||||
|
image that carries an alpha channel is composited onto black — a transparent
|
||||||
|
PNG drawn over the mosaic reads as noise behind a station logo.
|
||||||
- **Now playing** — artist, title, album, DJ, listeners, format/bitrate and
|
- **Now playing** — artist, title, album, DJ, listeners, format/bitrate and
|
||||||
"played N minutes ago", with procedural ANSI-shaded album art that changes
|
"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
|
per track. When the current track can be found on Bandcamp, a clickable
|
||||||
@@ -57,9 +60,10 @@ implemented here from public endpoints and open protocols.
|
|||||||
- **Pluggable catalogues** — one `providers` module per radio network, with a
|
- **Pluggable catalogues** — one `providers` module per radio network, with a
|
||||||
`p` picker in the UI, `--provider` on the command line, and the choice
|
`p` picker in the UI, `--provider` on the command line, and the choice
|
||||||
remembered per catalogue (including the last station you tuned).
|
remembered per catalogue (including the last station you tuned).
|
||||||
- **A hero per catalogue** — the top-left art comes from the provider: SomaFM's
|
- **A hero per catalogue** — the top-left art is the catalogue's own name in a
|
||||||
block wordmark, or tilderadio's radio icon converted from its logo PNG to
|
shared half-block font: SomaFM in the spectrum gradient, tilderadio in the
|
||||||
half-block ANSI by `tools/ansi_art.py`.
|
terminal's default colour, Bassdrive in its two blues. Every hero is three
|
||||||
|
rows tall, so the header keeps one height.
|
||||||
- **Schedule pane** — a single-stream catalogue gets a broadcast schedule
|
- **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
|
instead of a list of one station: who is on air now, since/until when, and
|
||||||
the upcoming shows grouped by day.
|
the upcoming shows grouped by day.
|
||||||
@@ -193,35 +197,45 @@ layout and the guessed icecast mirrors all live in
|
|||||||
`beefy_player/providers/somafm/`. Nothing outside that package imports it; the
|
`beefy_player/providers/somafm/`. Nothing outside that package imports it; the
|
||||||
app only ever holds a `RadioProvider`.
|
app only ever holds a `RadioProvider`.
|
||||||
|
|
||||||
Two catalogues ship today:
|
Three catalogues ship today:
|
||||||
|
|
||||||
| slug | catalogue | station list | history | streams |
|
| slug | catalogue | station list | history | streams |
|
||||||
| --- | --- | --- | --- | --- |
|
| --- | --- | --- | --- | --- |
|
||||||
| `somafm` | [SomaFM](https://somafm.com) | `channels.json` (≈50) | `songs/<station>.json` | `.pls` → icecast mirrors |
|
| `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) |
|
| `tilderadio` | [tilderadio](https://tilderadio.org) | `/api/now/` (1) | `now_playing` + `history[]` | AzuraCast mounts (mp3/ogg, 320k) |
|
||||||
|
| `bassdrive` | [Bassdrive](https://bassdrive.com) | icecast status (1) | rolling `display-title` | three mounts + mirrors |
|
||||||
|
|
||||||
They are deliberately different shapes: one has fifty channels and PLS
|
They are deliberately different shapes: fifty channels with PLS playlists, a
|
||||||
playlists, the other has a single stream whose whole catalogue arrives in the
|
single AzuraCast stream that carries its own history, and an Icecast relay whose
|
||||||
same response as its history. That is the seam working.
|
"now playing" is the *show* rather than the track. That is the seam working.
|
||||||
|
|
||||||
The tilderadio package also exercises the two optional capabilities: its left
|
The single-stream catalogues also use the optional capabilities: their left
|
||||||
pane is the broadcast schedule (`/api/schedule/`, with the live DJ coming from
|
pane is the broadcast schedule rather than a list containing one station, and
|
||||||
`/api/now/`) rather than a list containing one station, and its header art is
|
they opt out of Bandcamp lookups. Bassdrive is the extreme case — Icecast
|
||||||
the station's own radio icon converted to half blocks. The icon is generated,
|
reports one title and keeps no history, so the provider maps the title to
|
||||||
not hand-drawn — regenerate it after the logo changes:
|
`Station.dj`/`presenter` and keeps a rolling record of the shows it has heard.
|
||||||
|
|
||||||
```bash
|
### Header wordmarks
|
||||||
uv run python tools/ansi_art.py \
|
|
||||||
--url https://tilderadio.org/logos/tilderadio.png \
|
All three heroes come from one font, `beefy_player/providers/wordmark.py`:
|
||||||
--cols 22 --rows 7 \
|
four-column capitals six pixels tall, emitted as three rows of half blocks
|
||||||
--module beefy_player/providers/tilderadio/hero.py
|
(`▀`/`▄`/`█`) — the shape of the original SomaFM logo. A provider's `hero.py`
|
||||||
|
is one line of data:
|
||||||
|
|
||||||
|
```python
|
||||||
|
# beefy_player/providers/bassdrive/hero.py
|
||||||
|
HERO = render("BASSDRIVE", ["bright_cyan"] * 4 + ["cyan"] * 5)
|
||||||
```
|
```
|
||||||
|
|
||||||
`tools/ansi_art.py` crops the transparent margin, resizes to 22×14 pixels (two
|
`render()` takes one optional style per character and groups equal neighbours
|
||||||
per text cell), quantises to the stock 16-colour ANSI palette and emits
|
into markup spans; with no styles the wordmark inherits the terminal's default
|
||||||
run-length encoded rich markup — so the art follows the terminal's palette
|
foreground (tilderadio). Add a glyph to `GLYPHS` when a catalogue needs a letter
|
||||||
instead of pinning RGB values. `--preview out.png` draws the result, which is
|
the font does not have yet — `▀█▀` is the I, one pixel thick on purpose.
|
||||||
far easier to judge than ▀ and ▄.
|
|
||||||
|
Image logos are a different problem: `tools/ansi_art.py` converts a PNG into
|
||||||
|
half-block art (`--key-white` for logos on white, `--preview out.png` to see the
|
||||||
|
result). No shipped catalogue needs it today, but it is the tool to reach for
|
||||||
|
when a station's identity is a picture rather than a word.
|
||||||
|
|
||||||
Adding a catalogue:
|
Adding a catalogue:
|
||||||
|
|
||||||
@@ -347,8 +361,10 @@ beefy_player/
|
|||||||
config.py XDG paths, settings, BEEFY_* env helpers
|
config.py XDG paths, settings, BEEFY_* env helpers
|
||||||
providers/
|
providers/
|
||||||
base.py RadioProvider interface + HTTP/playlist toolkit
|
base.py RadioProvider interface + HTTP/playlist toolkit
|
||||||
|
wordmark.py the half-block font every hero is drawn with
|
||||||
somafm/ the SomaFM catalogue (API, hero, bundled fallback list)
|
somafm/ the SomaFM catalogue (API, hero, bundled fallback list)
|
||||||
tilderadio/ the tilderadio catalogue (one station, hero, schedule)
|
tilderadio/ the tilderadio catalogue (one station, hero, schedule)
|
||||||
|
bassdrive/ Bassdrive (Icecast status, show schedule, rolling history)
|
||||||
player.py mpv over a Unix-socket JSON IPC channel
|
player.py mpv over a Unix-socket JSON IPC channel
|
||||||
spectrum.py analyser (parec / Core Audio tap / ffmpeg / sim)
|
spectrum.py analyser (parec / Core Audio tap / ffmpeg / sim)
|
||||||
macos_tap.py macOS output capture (ctypes + Core Audio)
|
macos_tap.py macOS output capture (ctypes + Core Audio)
|
||||||
@@ -382,6 +398,14 @@ tools/
|
|||||||
the provider in place: the old stream is stopped, workers for the previous
|
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
|
catalogue are cancelled, the hero height and left pane follow the new
|
||||||
provider, and the station list (or schedule) reloads from cache first.
|
provider, and the station list (or schedule) reloads from cache first.
|
||||||
|
- `beefy_player/providers/bassdrive/api.py` — Icecast's keyless
|
||||||
|
`status-json.xsl` (mounts, bitrates, listeners, `display-title`) plus the
|
||||||
|
homepage schedule grid, whose slots carry absolute `<local-time datetime>`
|
||||||
|
values. A memoised fetch and a lock keep a cold start to one homepage
|
||||||
|
download; the display-title's show is matched against the running slot for
|
||||||
|
"live from <city>".
|
||||||
|
- `beefy_player/providers/wordmark.py` — the half-block capitals behind every
|
||||||
|
hero, with per-character style spans.
|
||||||
- `beefy_player/widgets/schedule.py` — the left pane for single-stream
|
- `beefy_player/widgets/schedule.py` — the left pane for single-stream
|
||||||
catalogues: the live slot with its `since`/`until` window, then upcoming
|
catalogues: the live slot with its `since`/`until` window, then upcoming
|
||||||
shows grouped by local day, with `… +N more` when the pane runs out.
|
shows grouped by local day, with `… +N more` when the pane runs out.
|
||||||
@@ -419,7 +443,8 @@ tools/
|
|||||||
~/.cache/beefy-player/channels-<id>.json last good list, other catalogues
|
~/.cache/beefy-player/channels-<id>.json last good list, other catalogues
|
||||||
~/.cache/beefy-player/schedule-<id>.json last good broadcast schedule
|
~/.cache/beefy-player/schedule-<id>.json last good broadcast schedule
|
||||||
~/.cache/beefy-player/playlists.json resolved stream URLs
|
~/.cache/beefy-player/playlists.json resolved stream URLs
|
||||||
~/.cache/beefy-player/songs/<id>.json song history snapshots
|
~/.cache/beefy-player/songs/<id>.json SomaFM song history snapshots
|
||||||
|
~/.cache/beefy-player/songs/<p>-<id>.json history for every other catalogue
|
||||||
~/.cache/beefy-player/artwork/<sha1> cached album/channel artwork
|
~/.cache/beefy-player/artwork/<sha1> cached album/channel artwork
|
||||||
```
|
```
|
||||||
|
|
||||||
@@ -488,6 +513,8 @@ so favourites and volume survive the rename.
|
|||||||
- SomaFM is listener-supported: if you enjoy it, [donate](https://somafm.com/support/).
|
- 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),
|
- 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.
|
or show up on the [schedule](https://tilderadio.org/schedule/) with a set.
|
||||||
|
- Bassdrive is free and listener-supported too: [bassdrive.com](https://bassdrive.com)
|
||||||
|
has the shows, archives and the shop that keeps the relays running.
|
||||||
|
|
||||||
## NixOS (optional aside)
|
## NixOS (optional aside)
|
||||||
|
|
||||||
|
|||||||
+22
-6
@@ -230,22 +230,38 @@ async def fetch(client: httpx.AsyncClient, url: str) -> bytes | None:
|
|||||||
|
|
||||||
|
|
||||||
def fit_image(data: bytes, width: int, height: int):
|
def fit_image(data: bytes, width: int, height: int):
|
||||||
"""Decode and center-crop an image to exactly ``width`` x ``height`` cells."""
|
"""Decode and letterbox an image into exactly ``width`` x ``height`` cells.
|
||||||
|
|
||||||
|
Scale to fit, never crop. Square album art fills the box either way, but a
|
||||||
|
wide station wordmark (Bassdrive's logo is 3:1) used to be centre-cropped
|
||||||
|
down to a zoomed slice of its middle. The padding stays transparent, so
|
||||||
|
the mosaic behind the artwork shows through instead of a black bar.
|
||||||
|
"""
|
||||||
try:
|
try:
|
||||||
from PIL import Image, ImageOps
|
from PIL import Image, ImageOps
|
||||||
|
|
||||||
with Image.open(BytesIO(data)) as source:
|
with Image.open(BytesIO(data)) as source:
|
||||||
|
# A PNG with transparency is drawn *over* the mosaic, which reads
|
||||||
|
# as noise behind a logo; composite those onto black instead.
|
||||||
|
has_alpha = source.mode in ("RGBA", "LA", "PA", "RGBa", "La") or (
|
||||||
|
"transparency" in source.info
|
||||||
|
)
|
||||||
image = source.convert("RGBA")
|
image = source.convert("RGBA")
|
||||||
cell_width, cell_height = cell_size()
|
cell_width, cell_height = cell_size()
|
||||||
target = (
|
target = (
|
||||||
max(1, width * cell_width),
|
max(1, width * cell_width),
|
||||||
max(1, height * cell_height),
|
max(1, height * cell_height),
|
||||||
)
|
)
|
||||||
return ImageOps.fit(
|
fitted = ImageOps.contain(image, target, method=Image.Resampling.LANCZOS)
|
||||||
image,
|
background = (0, 0, 0, 255) if has_alpha else (0, 0, 0, 0)
|
||||||
target,
|
canvas = Image.new("RGBA", target, background)
|
||||||
method=Image.Resampling.LANCZOS,
|
canvas.alpha_composite(
|
||||||
centering=(0.5, 0.5),
|
fitted,
|
||||||
|
(
|
||||||
|
(target[0] - fitted.width) // 2,
|
||||||
|
(target[1] - fitted.height) // 2,
|
||||||
|
),
|
||||||
)
|
)
|
||||||
|
return canvas
|
||||||
except Exception:
|
except Exception:
|
||||||
return None
|
return None
|
||||||
@@ -46,6 +46,8 @@ class Broadcast:
|
|||||||
|
|
||||||
name: str
|
name: str
|
||||||
dj: str = ""
|
dj: str = ""
|
||||||
|
#: Where the broadcast comes from ("Poznan, Poland"), when the feed says.
|
||||||
|
location: str = ""
|
||||||
start: datetime | None = None
|
start: datetime | None = None
|
||||||
end: datetime | None = None
|
end: datetime | None = None
|
||||||
#: True for the broadcast that is on air right now.
|
#: True for the broadcast that is on air right now.
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ Adding a catalogue is meant to be a small, contained change:
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from .base import RadioProvider
|
from .base import RadioProvider
|
||||||
|
from .bassdrive import BassdriveProvider
|
||||||
from .somafm import SomaFMProvider
|
from .somafm import SomaFMProvider
|
||||||
from .tilderadio import TilderadioProvider
|
from .tilderadio import TilderadioProvider
|
||||||
|
|
||||||
@@ -30,7 +31,8 @@ __all__ = [
|
|||||||
|
|
||||||
#: Every catalogue the app can talk to, keyed by its slug.
|
#: Every catalogue the app can talk to, keyed by its slug.
|
||||||
PROVIDERS: dict[str, RadioProvider] = {
|
PROVIDERS: dict[str, RadioProvider] = {
|
||||||
provider.id: provider for provider in (SomaFMProvider(), TilderadioProvider())
|
provider.id: provider
|
||||||
|
for provider in (SomaFMProvider(), TilderadioProvider(), BassdriveProvider())
|
||||||
}
|
}
|
||||||
|
|
||||||
#: Provider used when nothing else is configured.
|
#: Provider used when nothing else is configured.
|
||||||
|
|||||||
@@ -0,0 +1,76 @@
|
|||||||
|
"""Bassdrive: worldwide drum & bass, on air since the dial-up days.
|
||||||
|
|
||||||
|
One stream with three mounts, a schedule full of live DJ shows and a
|
||||||
|
``display-title`` that names the show rather than the track. Everything
|
||||||
|
Bassdrive-specific — the Icecast status schema, the homepage schedule grid, the
|
||||||
|
rolling show history and the bundled fallback — lives in this package; the app
|
||||||
|
only sees :class:`BassdriveProvider`.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import httpx
|
||||||
|
|
||||||
|
from ... import config
|
||||||
|
from ...models import Broadcast, Song, Station
|
||||||
|
from ..base import RadioProvider
|
||||||
|
from . import api, fallback
|
||||||
|
from . import hero as hero_art
|
||||||
|
|
||||||
|
__all__ = ["BassdriveProvider"]
|
||||||
|
|
||||||
|
|
||||||
|
class BassdriveProvider(RadioProvider):
|
||||||
|
id = "bassdrive"
|
||||||
|
name = "Bassdrive"
|
||||||
|
homepage = "bassdrive.com"
|
||||||
|
brand = "BASSDRIVE"
|
||||||
|
tagline = "worldwide drum & bass · live shows"
|
||||||
|
station_source = "bassdrive.com + icecast status"
|
||||||
|
history_source = "icecast status (rolling)"
|
||||||
|
|
||||||
|
# The wordmark in the station's own two blues, in the shared half-block
|
||||||
|
# font; the caption is the station's own description.
|
||||||
|
hero = hero_art.HERO
|
||||||
|
hero_lines = (
|
||||||
|
"BASSDRIVE",
|
||||||
|
"WORLDWIDE DRUM & BASS",
|
||||||
|
"LIVE SHOWS 24/7",
|
||||||
|
)
|
||||||
|
|
||||||
|
# One stream and a schedule full of DJ slots: the left pane is the schedule.
|
||||||
|
has_schedule = True
|
||||||
|
# DJ-mix station: the rotation is not a Bandcamp catalogue.
|
||||||
|
bandcamp = False
|
||||||
|
|
||||||
|
@property
|
||||||
|
def user_agent(self) -> str:
|
||||||
|
return f"{config.USER_AGENT}; +https://bassdrive.com"
|
||||||
|
|
||||||
|
async def fetch_stations(self, client: httpx.AsyncClient) -> list[Station]:
|
||||||
|
return [await api.fetch_station(client)]
|
||||||
|
|
||||||
|
def load_cached_stations(self) -> list[Station]:
|
||||||
|
return api.load_cached_station()
|
||||||
|
|
||||||
|
def fallback_stations(self) -> list[Station]:
|
||||||
|
return fallback.fallback_stations()
|
||||||
|
|
||||||
|
async def fetch_history(
|
||||||
|
self, client: httpx.AsyncClient, station_id: str
|
||||||
|
) -> list[Song]:
|
||||||
|
return await api.fetch_songs(client, station_id)
|
||||||
|
|
||||||
|
def load_cached_history(self, station_id: str) -> list[Song]:
|
||||||
|
return api.load_cached_songs(station_id)
|
||||||
|
|
||||||
|
async def fetch_schedule(
|
||||||
|
self, client: httpx.AsyncClient
|
||||||
|
) -> list[Broadcast]:
|
||||||
|
return await api.fetch_schedule(client)
|
||||||
|
|
||||||
|
def load_cached_schedule(self) -> list[Broadcast]:
|
||||||
|
return api.load_cached_schedule()
|
||||||
|
|
||||||
|
def fallback_urls(self, station: Station) -> list[str]:
|
||||||
|
return fallback.known_stream_urls()
|
||||||
@@ -0,0 +1,517 @@
|
|||||||
|
"""Bassdrive's public endpoints, and nothing but.
|
||||||
|
|
||||||
|
* ``https://ice.bassdrive.net/status-json.xsl`` -> Icecast status: every mount
|
||||||
|
(mp3/aac+, bitrates), the listener counts and the current ``display-title``,
|
||||||
|
which for a DJ-mix stream is the *show*: ``"Ben XO :: LIVE :: Night Shift
|
||||||
|
Over"``.
|
||||||
|
* ``http://www.bassdrive.com/`` -> the homepage carries the whole week's
|
||||||
|
schedule, every slot stamped with an absolute ``<local-time datetime=...>``.
|
||||||
|
|
||||||
|
A DJ-mix station has no track history: Icecast reports one title and forgets
|
||||||
|
it. So "now playing" is the show (mapped onto ``Station.dj``/``presenter`` and
|
||||||
|
a :class:`Song`), and the history is a rolling record of the shows we have
|
||||||
|
heard since the app started — the unit a listener actually cares about here.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
import html as html_module
|
||||||
|
import re
|
||||||
|
import time
|
||||||
|
from collections.abc import Iterable
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
import httpx
|
||||||
|
|
||||||
|
from ... import config
|
||||||
|
from ...models import Broadcast, Song, Station, parse_date, to_int
|
||||||
|
from ..base import fetch_json
|
||||||
|
|
||||||
|
PROVIDER_ID = "bassdrive"
|
||||||
|
STATUS_URL = "https://ice.bassdrive.net/status-json.xsl"
|
||||||
|
SCHEDULE_URL = "https://www.bassdrive.com/"
|
||||||
|
|
||||||
|
DEFAULT_STATION_ID = "bassdrive"
|
||||||
|
STATION_IMAGE = "https://www.bassdrive.com/images/logo.png"
|
||||||
|
DESCRIPTION = "worldwide drum & bass radio — live DJ shows around the clock"
|
||||||
|
GENRE = "drum & bass|jungle"
|
||||||
|
|
||||||
|
WEEK = timedelta(days=7)
|
||||||
|
#: Upcoming slots kept in the pane (a week is plenty for a schedule).
|
||||||
|
SCHEDULE_LIMIT = 40
|
||||||
|
#: Shows remembered in LAST PLAYED.
|
||||||
|
HISTORY_LIMIT = 20
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
# status-json -> Station / Song
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
|
||||||
|
|
||||||
|
def _sources(payload: Any) -> list[dict[str, Any]]:
|
||||||
|
stats = payload.get("icestats") if isinstance(payload, dict) else None
|
||||||
|
source = stats.get("source") if isinstance(stats, dict) else None
|
||||||
|
if isinstance(source, dict):
|
||||||
|
return [source]
|
||||||
|
if isinstance(source, list):
|
||||||
|
return [entry for entry in source if isinstance(entry, dict)]
|
||||||
|
return []
|
||||||
|
|
||||||
|
|
||||||
|
def _quality(bitrate: int) -> str:
|
||||||
|
if bitrate >= 128:
|
||||||
|
return "highest"
|
||||||
|
if bitrate >= 48:
|
||||||
|
return "high"
|
||||||
|
return "low" if bitrate else ""
|
||||||
|
|
||||||
|
|
||||||
|
def playlists_from_status(payload: Any) -> list[dict[str, str]]:
|
||||||
|
playlists: list[dict[str, str]] = []
|
||||||
|
seen: set[str] = set()
|
||||||
|
for source in _sources(payload):
|
||||||
|
url = source.get("listenurl")
|
||||||
|
if not isinstance(url, str) or not url.startswith(("http://", "https://")):
|
||||||
|
continue
|
||||||
|
if url in seen:
|
||||||
|
continue
|
||||||
|
seen.add(url)
|
||||||
|
bitrate = to_int(source.get("bitrate"))
|
||||||
|
kind = str(source.get("server_type") or "").lower()
|
||||||
|
fmt = "mp3" if "mpeg" in kind else "aacp" if "aac" in kind else ""
|
||||||
|
playlists.append(
|
||||||
|
{
|
||||||
|
"url": url,
|
||||||
|
"format": fmt,
|
||||||
|
"quality": _quality(bitrate),
|
||||||
|
"bitrate": str(bitrate) if bitrate else "",
|
||||||
|
}
|
||||||
|
)
|
||||||
|
return playlists
|
||||||
|
|
||||||
|
|
||||||
|
def status_listeners(payload: Any) -> int:
|
||||||
|
"""Mounts split the audience, so the station's figure is their sum."""
|
||||||
|
return sum(to_int(source.get("listeners")) for source in _sources(payload))
|
||||||
|
|
||||||
|
|
||||||
|
def display_title(payload: Any) -> str:
|
||||||
|
for source in _sources(payload):
|
||||||
|
for key in ("display-title", "title"):
|
||||||
|
value = source.get(key)
|
||||||
|
if isinstance(value, str) and value.strip():
|
||||||
|
return value.strip()
|
||||||
|
return ""
|
||||||
|
|
||||||
|
|
||||||
|
def split_title(raw: str) -> tuple[str, str, bool]:
|
||||||
|
"""``"Ben XO :: LIVE :: Night Shift Over [promo]"`` -> DJ, show, live.
|
||||||
|
|
||||||
|
Bassdrive's title convention is ``DJ :: LIVE :: Show``, with promotional
|
||||||
|
brackets tacked on the end — but it is not enforced: plenty of shows go out
|
||||||
|
as ``"Sohl Patrol LIVE"``, so a trailing ``LIVE`` counts too. Anything
|
||||||
|
unexpected degrades to "the whole string is the show".
|
||||||
|
"""
|
||||||
|
cleaned = re.sub(r"\[[^\]]*\]", " ", raw)
|
||||||
|
parts = [part.strip() for part in cleaned.split("::") if part.strip()]
|
||||||
|
live = any(part.casefold() == "live" for part in parts)
|
||||||
|
parts = [part for part in parts if part.casefold() != "live"]
|
||||||
|
|
||||||
|
if parts:
|
||||||
|
trailing = re.search(r"(?i)\s+live$", parts[-1])
|
||||||
|
if trailing:
|
||||||
|
live = True
|
||||||
|
trimmed = parts[-1][: trailing.start()].strip()
|
||||||
|
parts = parts[:-1] + ([trimmed] if trimmed else [])
|
||||||
|
|
||||||
|
if not parts:
|
||||||
|
return "", raw.strip(), live
|
||||||
|
if len(parts) == 1:
|
||||||
|
return "", parts[0], live
|
||||||
|
return parts[0], parts[-1], live
|
||||||
|
|
||||||
|
|
||||||
|
def song_from_title(raw: str) -> Song | None:
|
||||||
|
dj, show, _live = split_title(raw)
|
||||||
|
if not (dj or show):
|
||||||
|
return None
|
||||||
|
# With no DJ in the title the station stands in as the "artist": a panel
|
||||||
|
# that says "Unknown artist" for every show would look broken.
|
||||||
|
return Song(artist=dj or "Bassdrive", title=show)
|
||||||
|
|
||||||
|
|
||||||
|
def _same_show(entry: Broadcast, show: str) -> bool:
|
||||||
|
return bool(show) and entry.name.strip().casefold() == show.strip().casefold()
|
||||||
|
|
||||||
|
|
||||||
|
def _slot_for(schedule: Iterable[Broadcast], show: str) -> Broadcast | None:
|
||||||
|
if not show:
|
||||||
|
return None
|
||||||
|
return next((entry for entry in schedule if _same_show(entry, show)), None)
|
||||||
|
|
||||||
|
|
||||||
|
def resolve_now(
|
||||||
|
payload: Any,
|
||||||
|
schedule: Iterable[Broadcast] = (),
|
||||||
|
now: datetime | None = None,
|
||||||
|
) -> tuple[str, str, bool, Broadcast | None]:
|
||||||
|
"""Who is on, according to the stream *and* the schedule.
|
||||||
|
|
||||||
|
The display-title often forgets the ``LIVE`` marker; a show whose slot
|
||||||
|
covers now counts as live even when the title is just its name.
|
||||||
|
"""
|
||||||
|
now = now or datetime.now(timezone.utc)
|
||||||
|
schedule = list(schedule)
|
||||||
|
dj, show, live = split_title(display_title(payload))
|
||||||
|
slot = _slot_for(schedule, show)
|
||||||
|
if (
|
||||||
|
slot is not None
|
||||||
|
and slot.start is not None
|
||||||
|
and slot.end is not None
|
||||||
|
and slot.start <= now < slot.end
|
||||||
|
):
|
||||||
|
live = True
|
||||||
|
return dj, show, live, slot
|
||||||
|
|
||||||
|
|
||||||
|
def station_from_status(
|
||||||
|
payload: Any, schedule: Iterable[Broadcast] = ()
|
||||||
|
) -> Station | None:
|
||||||
|
playlists = playlists_from_status(payload)
|
||||||
|
raw = display_title(payload)
|
||||||
|
if not playlists and not raw:
|
||||||
|
return None
|
||||||
|
|
||||||
|
dj, show, live, slot = resolve_now(payload, schedule)
|
||||||
|
location = slot.location if slot is not None else ""
|
||||||
|
if live:
|
||||||
|
presenter = f"live from {location}" if location else "live now"
|
||||||
|
else:
|
||||||
|
presenter = "rotation · drum & bass"
|
||||||
|
|
||||||
|
station = Station(
|
||||||
|
id=DEFAULT_STATION_ID,
|
||||||
|
title="Bassdrive",
|
||||||
|
description=DESCRIPTION,
|
||||||
|
genre=GENRE,
|
||||||
|
dj=dj,
|
||||||
|
presenter=presenter,
|
||||||
|
image=STATION_IMAGE,
|
||||||
|
listeners=status_listeners(payload),
|
||||||
|
playlists=playlists,
|
||||||
|
provider=PROVIDER_ID,
|
||||||
|
)
|
||||||
|
if show:
|
||||||
|
# The "artist" of a show-based stream is the station itself.
|
||||||
|
station.now_playing = Song(artist=dj or "Bassdrive", title=show)
|
||||||
|
station.last_playing = raw
|
||||||
|
return station
|
||||||
|
|
||||||
|
|
||||||
|
def live_broadcast(
|
||||||
|
payload: Any,
|
||||||
|
schedule: Iterable[Broadcast],
|
||||||
|
now: datetime | None = None,
|
||||||
|
) -> Broadcast | None:
|
||||||
|
dj, show, live, slot = resolve_now(payload, schedule, now)
|
||||||
|
if not live or not show:
|
||||||
|
return None
|
||||||
|
if slot is not None:
|
||||||
|
return Broadcast(
|
||||||
|
name=show,
|
||||||
|
dj=dj,
|
||||||
|
location=slot.location,
|
||||||
|
start=slot.start,
|
||||||
|
end=slot.end,
|
||||||
|
live=True,
|
||||||
|
)
|
||||||
|
return Broadcast(name=show, dj=dj, live=True)
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
# the homepage schedule
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
|
||||||
|
_SPAN_LOCATION = re.compile(r'<span class="location">(.*?)</span>', re.S)
|
||||||
|
|
||||||
|
|
||||||
|
def _text(value: str) -> str:
|
||||||
|
return re.sub(r"\s+", " ", html_module.unescape(re.sub(r"<[^>]+>", " ", value))).strip()
|
||||||
|
|
||||||
|
|
||||||
|
def parse_schedule(page: str) -> list[Broadcast]:
|
||||||
|
"""Pull the weekly grid out of the homepage, one ``<li>`` per slot."""
|
||||||
|
broadcasts: list[Broadcast] = []
|
||||||
|
for chunk in re.split(r"<li\b", page):
|
||||||
|
if 'class="title"' not in chunk or 'class="time"' not in chunk:
|
||||||
|
continue
|
||||||
|
stamps = re.findall(r'datetime="([^"]+)"', chunk)
|
||||||
|
if len(stamps) < 2:
|
||||||
|
continue
|
||||||
|
start = parse_date(stamps[0])
|
||||||
|
end = parse_date(stamps[1])
|
||||||
|
if start is None:
|
||||||
|
continue
|
||||||
|
title = re.search(r'<p class="title">(.*?)</p>', chunk, re.S)
|
||||||
|
if title is None:
|
||||||
|
continue
|
||||||
|
block = title.group(1)
|
||||||
|
location_match = _SPAN_LOCATION.search(block)
|
||||||
|
location = _text(location_match.group(1)) if location_match else ""
|
||||||
|
name = _text(_SPAN_LOCATION.sub(" ", block))
|
||||||
|
if not name:
|
||||||
|
continue
|
||||||
|
broadcasts.append(
|
||||||
|
Broadcast(name=name, location=location, start=start, end=end)
|
||||||
|
)
|
||||||
|
broadcasts.sort(key=lambda item: item.start or datetime.now(timezone.utc))
|
||||||
|
return broadcasts
|
||||||
|
|
||||||
|
|
||||||
|
def roll_forward(
|
||||||
|
entries: Iterable[Broadcast], now: datetime, limit: int = SCHEDULE_LIMIT
|
||||||
|
) -> list[Broadcast]:
|
||||||
|
"""The homepage only shows one week; repeat it so the pane is never empty."""
|
||||||
|
upcoming: list[Broadcast] = []
|
||||||
|
for entry in entries:
|
||||||
|
start, end = entry.start, entry.end
|
||||||
|
if start is None:
|
||||||
|
continue
|
||||||
|
for _ in range(3):
|
||||||
|
if end is None or end >= now:
|
||||||
|
break
|
||||||
|
start += WEEK
|
||||||
|
end = end + WEEK if end is not None else None
|
||||||
|
if end is not None and end < now:
|
||||||
|
continue
|
||||||
|
upcoming.append(
|
||||||
|
Broadcast(
|
||||||
|
name=entry.name,
|
||||||
|
dj=entry.dj,
|
||||||
|
location=entry.location,
|
||||||
|
start=start,
|
||||||
|
end=end,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
upcoming.sort(key=lambda item: item.start or now)
|
||||||
|
return upcoming[:limit]
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
# caching
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
|
||||||
|
#: One homepage fetch serves both the station and the schedule worker; the
|
||||||
|
#: lock keeps a cold start from downloading the same grid twice at once.
|
||||||
|
_MEMO_TTL = 120.0
|
||||||
|
_schedule_memo: tuple[float, list[Broadcast]] | None = None
|
||||||
|
_schedule_lock = asyncio.Lock()
|
||||||
|
|
||||||
|
|
||||||
|
def _memo_schedule() -> list[Broadcast]:
|
||||||
|
if _schedule_memo is None:
|
||||||
|
return []
|
||||||
|
fetched, entries = _schedule_memo
|
||||||
|
if time.time() - fetched > _MEMO_TTL:
|
||||||
|
return []
|
||||||
|
return list(entries)
|
||||||
|
|
||||||
|
|
||||||
|
def _remember_schedule(entries: list[Broadcast]) -> None:
|
||||||
|
global _schedule_memo
|
||||||
|
_schedule_memo = (time.time(), entries)
|
||||||
|
|
||||||
|
|
||||||
|
def reset_schedule_memo() -> None:
|
||||||
|
"""Forget the in-process schedule (the tests use this)."""
|
||||||
|
global _schedule_memo
|
||||||
|
_schedule_memo = None
|
||||||
|
|
||||||
|
|
||||||
|
def _songs_cache_path(station_id: str) -> Path:
|
||||||
|
safe = re.sub(r"[^A-Za-z0-9_.-]", "_", station_id or DEFAULT_STATION_ID)
|
||||||
|
return config.SONGS_CACHE_DIR / f"{PROVIDER_ID}-{safe}.json"
|
||||||
|
|
||||||
|
|
||||||
|
async def schedule_entries(client: httpx.AsyncClient) -> list[Broadcast]:
|
||||||
|
"""The weekly grid: memo, then the homepage, then the cached copy."""
|
||||||
|
remembered = _memo_schedule()
|
||||||
|
if remembered:
|
||||||
|
return remembered
|
||||||
|
async with _schedule_lock:
|
||||||
|
remembered = _memo_schedule()
|
||||||
|
if remembered:
|
||||||
|
return remembered
|
||||||
|
try:
|
||||||
|
response = await client.get(
|
||||||
|
SCHEDULE_URL, headers={"Accept": "text/html"}
|
||||||
|
)
|
||||||
|
response.raise_for_status()
|
||||||
|
entries = parse_schedule(response.text)
|
||||||
|
except httpx.HTTPError:
|
||||||
|
entries = []
|
||||||
|
if not entries:
|
||||||
|
return load_cached_schedule()
|
||||||
|
_remember_schedule(entries)
|
||||||
|
config.write_json(
|
||||||
|
config.schedule_cache_path(PROVIDER_ID),
|
||||||
|
{
|
||||||
|
"fetched": time.time(),
|
||||||
|
"schedule": [
|
||||||
|
{
|
||||||
|
"name": item.name,
|
||||||
|
"location": item.location,
|
||||||
|
"start": item.start.isoformat() if item.start else "",
|
||||||
|
"end": item.end.isoformat() if item.end else "",
|
||||||
|
}
|
||||||
|
for item in entries
|
||||||
|
],
|
||||||
|
},
|
||||||
|
)
|
||||||
|
return entries
|
||||||
|
|
||||||
|
|
||||||
|
def load_cached_station() -> list[Station]:
|
||||||
|
raw = config.read_json(config.channels_cache_path(PROVIDER_ID))
|
||||||
|
status = raw.get("status") if isinstance(raw, dict) else None
|
||||||
|
if status is None:
|
||||||
|
return []
|
||||||
|
station = station_from_status(status, load_cached_schedule())
|
||||||
|
return [station] if station is not None else []
|
||||||
|
|
||||||
|
|
||||||
|
def load_cached_schedule() -> list[Broadcast]:
|
||||||
|
raw = config.read_json(config.schedule_cache_path(PROVIDER_ID))
|
||||||
|
entries = raw.get("schedule") if isinstance(raw, dict) else None
|
||||||
|
if not isinstance(entries, list):
|
||||||
|
return []
|
||||||
|
return _schedule_from_cache(entries)
|
||||||
|
|
||||||
|
|
||||||
|
def _schedule_from_cache(entries: list[Any]) -> list[Broadcast]:
|
||||||
|
broadcasts: list[Broadcast] = []
|
||||||
|
for entry in entries:
|
||||||
|
if not isinstance(entry, dict):
|
||||||
|
continue
|
||||||
|
start = parse_date(entry.get("start"))
|
||||||
|
end = parse_date(entry.get("end"))
|
||||||
|
name = str(entry.get("name") or "").strip()
|
||||||
|
if not name or start is None:
|
||||||
|
continue
|
||||||
|
broadcasts.append(
|
||||||
|
Broadcast(
|
||||||
|
name=name,
|
||||||
|
location=str(entry.get("location") or "").strip(),
|
||||||
|
start=start,
|
||||||
|
end=end,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
broadcasts.sort(key=lambda item: item.start or datetime.now(timezone.utc))
|
||||||
|
return broadcasts
|
||||||
|
|
||||||
|
|
||||||
|
def load_cached_songs(station_id: str) -> list[Song]:
|
||||||
|
raw = config.read_json(_songs_cache_path(station_id))
|
||||||
|
entries = raw.get("songs") if isinstance(raw, dict) else None
|
||||||
|
if not isinstance(entries, list):
|
||||||
|
return []
|
||||||
|
songs: list[Song] = []
|
||||||
|
for entry in entries:
|
||||||
|
if not isinstance(entry, dict):
|
||||||
|
continue
|
||||||
|
song = Song(
|
||||||
|
title=str(entry.get("title") or "").strip(),
|
||||||
|
artist=str(entry.get("artist") or "").strip(),
|
||||||
|
played_at=parse_date(entry.get("played_at")),
|
||||||
|
)
|
||||||
|
if song.title or song.artist:
|
||||||
|
songs.append(song)
|
||||||
|
return songs
|
||||||
|
|
||||||
|
|
||||||
|
def _songs_to_cache(songs: Iterable[Song]) -> list[dict[str, Any]]:
|
||||||
|
return [
|
||||||
|
{
|
||||||
|
"title": song.title,
|
||||||
|
"artist": song.artist,
|
||||||
|
"played_at": song.played_at.isoformat() if song.played_at else "",
|
||||||
|
}
|
||||||
|
for song in songs
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
# fetching
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
|
||||||
|
|
||||||
|
async def fetch_station(client: httpx.AsyncClient) -> Station:
|
||||||
|
try:
|
||||||
|
payload = await fetch_json(client, STATUS_URL)
|
||||||
|
except RuntimeError as exc:
|
||||||
|
raise RuntimeError(f"could not fetch the Bassdrive status: {exc}") from None
|
||||||
|
|
||||||
|
# The schedule tells the station *where* a live show is coming from, so a
|
||||||
|
# cold start fetches it here; the schedule worker then reuses the memo.
|
||||||
|
schedule = load_cached_schedule() or await schedule_entries(client)
|
||||||
|
station = station_from_status(payload, schedule)
|
||||||
|
if station is None:
|
||||||
|
raise RuntimeError("Bassdrive returned no station data")
|
||||||
|
|
||||||
|
config.write_json(
|
||||||
|
config.channels_cache_path(PROVIDER_ID),
|
||||||
|
{"fetched": time.time(), "status": payload},
|
||||||
|
)
|
||||||
|
return station
|
||||||
|
|
||||||
|
|
||||||
|
async def fetch_songs(client: httpx.AsyncClient, station_id: str) -> list[Song]:
|
||||||
|
try:
|
||||||
|
payload = await fetch_json(client, STATUS_URL, attempts=1)
|
||||||
|
except RuntimeError:
|
||||||
|
return load_cached_songs(station_id)
|
||||||
|
|
||||||
|
current = song_from_title(display_title(payload))
|
||||||
|
if current is None:
|
||||||
|
return load_cached_songs(station_id)
|
||||||
|
|
||||||
|
history = load_cached_songs(station_id)
|
||||||
|
now = datetime.now(timezone.utc)
|
||||||
|
if history and history[0].key() == current.key():
|
||||||
|
# Same show still on air: keep the time we first heard it, so the pane
|
||||||
|
# says how long it has been running, not how long since the last poll.
|
||||||
|
current.played_at = history[0].played_at
|
||||||
|
history = [current, *history[1:]]
|
||||||
|
else:
|
||||||
|
current.played_at = now
|
||||||
|
history = [current] + [song for song in history if song.key() != current.key()]
|
||||||
|
history = history[:HISTORY_LIMIT]
|
||||||
|
|
||||||
|
config.write_json(
|
||||||
|
_songs_cache_path(station_id),
|
||||||
|
{"fetched": time.time(), "songs": _songs_to_cache(history)},
|
||||||
|
)
|
||||||
|
return history
|
||||||
|
|
||||||
|
|
||||||
|
async def fetch_schedule(client: httpx.AsyncClient) -> list[Broadcast]:
|
||||||
|
now = datetime.now(timezone.utc)
|
||||||
|
|
||||||
|
live: Broadcast | None = None
|
||||||
|
try:
|
||||||
|
status = await fetch_json(client, STATUS_URL, attempts=1)
|
||||||
|
except RuntimeError:
|
||||||
|
status = None
|
||||||
|
|
||||||
|
entries = await schedule_entries(client)
|
||||||
|
upcoming = roll_forward(entries, now)
|
||||||
|
if status is not None:
|
||||||
|
live = live_broadcast(status, entries, now)
|
||||||
|
if live is not None:
|
||||||
|
key = live.key()
|
||||||
|
upcoming = [item for item in upcoming if item.key() != key]
|
||||||
|
return [live, *upcoming]
|
||||||
|
return upcoming
|
||||||
@@ -0,0 +1,56 @@
|
|||||||
|
"""Bundled station, so a cold offline start still makes sense.
|
||||||
|
|
||||||
|
Bassdrive is one station with three mounts and a handful of mirror hosts; the
|
||||||
|
streams below are the ones its own ``.m3u`` files list. The live status-json
|
||||||
|
always wins when it is reachable.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
# (url, format, quality, bitrate)
|
||||||
|
_MOUNTS: tuple[tuple[str, str, str, str], ...] = (
|
||||||
|
("https://ice.bassdrive.net:443/stream", "mp3", "highest", "128"),
|
||||||
|
("https://ice.bassdrive.net:443/stream56", "aacp", "high", "56"),
|
||||||
|
("https://ice.bassdrive.net:443/stream32", "aacp", "low", "32"),
|
||||||
|
)
|
||||||
|
|
||||||
|
#: Alternative servers from bassdrive.m3u, tried when the primary is down.
|
||||||
|
_MIRRORS: tuple[str, ...] = (
|
||||||
|
"http://chi.bassdrive.co:80",
|
||||||
|
"http://bassdrive.radioca.st:8702",
|
||||||
|
"http://stream.bassdrive.uk:8200",
|
||||||
|
"http://au.bassdrive.co:8000",
|
||||||
|
)
|
||||||
|
|
||||||
|
_SERVER_TYPE = {"mp3": "audio/mpeg", "aacp": "audio/aacp"}
|
||||||
|
|
||||||
|
|
||||||
|
def fallback_status() -> dict[str, Any]:
|
||||||
|
"""A ``status-json.xsl``-shaped payload, so the normal parser can eat it."""
|
||||||
|
return {
|
||||||
|
"icestats": {
|
||||||
|
"source": [
|
||||||
|
{
|
||||||
|
"listenurl": url,
|
||||||
|
"bitrate": int(bitrate),
|
||||||
|
"server_type": _SERVER_TYPE.get(fmt, "audio/mpeg"),
|
||||||
|
"listeners": 0,
|
||||||
|
"display-title": "",
|
||||||
|
}
|
||||||
|
for url, fmt, _quality, bitrate in _MOUNTS
|
||||||
|
]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def fallback_stations() -> list[Any]:
|
||||||
|
from .api import station_from_status
|
||||||
|
|
||||||
|
station = station_from_status(fallback_status())
|
||||||
|
return [station] if station is not None else []
|
||||||
|
|
||||||
|
|
||||||
|
def known_stream_urls() -> list[str]:
|
||||||
|
return [url for url, _format, _quality, _bitrate in _MOUNTS] + list(_MIRRORS)
|
||||||
@@ -0,0 +1,14 @@
|
|||||||
|
"""Bassdrive's header hero: its wordmark in the station's two blues.
|
||||||
|
|
||||||
|
The logo paints "BASS" in the pale blue and "DRIVE" in the deeper one; those
|
||||||
|
map onto the stock palette's bright cyan and cyan, so the wordmark still reads
|
||||||
|
as Bassdrive without hard-coded RGB.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from ..wordmark import render
|
||||||
|
|
||||||
|
HERO: tuple[str, ...] = render(
|
||||||
|
"BASSDRIVE", ["bright_cyan"] * 4 + ["cyan"] * 5
|
||||||
|
)
|
||||||
@@ -1,17 +1,16 @@
|
|||||||
"""SomaFM's header hero: the block-letter wordmark and the three lines beside it.
|
"""SomaFM's header hero: the block-letter wordmark and the three lines beside it.
|
||||||
|
|
||||||
The glyphs are plain half-blocks; the header paints them with the app's
|
The glyphs come from the shared half-block font
|
||||||
|
(:mod:`beefy_player.providers.wordmark`); the header paints them with the app's
|
||||||
spectrum gradient (``hero_gradient``), so the logo takes on the terminal's own
|
spectrum gradient (``hero_gradient``), so the logo takes on the terminal's own
|
||||||
colours instead of carrying RGB values around.
|
colours instead of carrying RGB values around.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
LOGO: tuple[str, ...] = (
|
from ..wordmark import render
|
||||||
"▄▀▀▀ ▄▀▀▄ █▄ ▄█ ▄▀▀▄",
|
|
||||||
"▀▀▀▄ █ █ █ █ █ █▀▀█",
|
LOGO: tuple[str, ...] = render("SOMAFM")
|
||||||
"▄▄▄▀ ▀▄▄▀ █ █ █ █",
|
|
||||||
)
|
|
||||||
|
|
||||||
TAGLINES: tuple[str, ...] = (
|
TAGLINES: tuple[str, ...] = (
|
||||||
"SOMA FM RADIO",
|
"SOMA FM RADIO",
|
||||||
|
|||||||
@@ -1,24 +1,11 @@
|
|||||||
"""Generated header art — do not edit by hand.
|
"""tilderadio's header hero: its name in the shared half-block font.
|
||||||
|
|
||||||
Source: https://tilderadio.org/logos/tilderadio.png
|
No colours of its own — the wordmark inherits the terminal's default
|
||||||
|
foreground, so it looks like part of the chrome rather than a logo pasted in.
|
||||||
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
|
from __future__ import annotations
|
||||||
|
|
||||||
HERO: tuple[str, ...] = (
|
from ..wordmark import render
|
||||||
' [bright_black]▄[/][bright_black on bright_black]▀▀[/][bright_black]▀[/]',
|
|
||||||
' [bright_black]▄▄[/][bright_black on bright_black]▀▀▀[/][bright_black]▄▄▄▄▄▄▄▄▄▄▄[/]',
|
HERO: tuple[str, ...] = render("TILDERADIO")
|
||||||
'[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]▀▀[/]',
|
|
||||||
)
|
|
||||||
@@ -0,0 +1,91 @@
|
|||||||
|
"""A half-block wordmark font for the header hero.
|
||||||
|
|
||||||
|
Every shipped catalogue draws its brand the same way: four-column capitals,
|
||||||
|
six pixels tall, emitted as three rows of half blocks (``▀``/``▄``/``█``) — the
|
||||||
|
same shape as the original SomaFM logo. Catalogues differ only in the word
|
||||||
|
they spell and the ANSI colours they paint it with, so the font lives here and
|
||||||
|
each provider's ``hero.py`` is one line of data.
|
||||||
|
|
||||||
|
Glyphs are 6x4 bitmaps (``M`` is 6x5). Two pixel rows become one text row, so
|
||||||
|
a wordmark is always three rows tall and the header keeps a fixed height.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Sequence
|
||||||
|
|
||||||
|
#: 6-row bitmap per capital; all rows of one glyph are the same width.
|
||||||
|
GLYPHS: dict[str, tuple[str, ...]] = {
|
||||||
|
"A": (".##.", "#..#", "####", "#..#", "#..#", "#..#"),
|
||||||
|
"B": ("###.", "#..#", "###.", "#..#", "#..#", "###."),
|
||||||
|
"D": ("###.", "#..#", "#..#", "#..#", "#..#", "###."),
|
||||||
|
"E": ("####", "#...", "###.", "#...", "#...", "####"),
|
||||||
|
"F": ("####", "#...", "###.", "#...", "#...", "#..."),
|
||||||
|
"I": ("###", ".#.", ".#.", ".#.", ".#.", "###"),
|
||||||
|
"L": ("#...", "#...", "#...", "#...", "#...", "####"),
|
||||||
|
# 5 wide on purpose: the M is the odd one out in the original wordmark.
|
||||||
|
"M": ("#...#", "##.##", "#.#.#", "#.#.#", "#...#", "#...#"),
|
||||||
|
"O": (".##.", "#..#", "#..#", "#..#", "#..#", ".##."),
|
||||||
|
"R": ("###.", "#..#", "###.", "#.#.", "#..#", "#..#"),
|
||||||
|
"S": (".###", "#...", "###.", "...#", "...#", "###."),
|
||||||
|
"T": ("###", ".#.", ".#.", ".#.", ".#.", ".#."),
|
||||||
|
"V": ("#..#", "#..#", "#..#", "#..#", ".##.", ".##."),
|
||||||
|
}
|
||||||
|
|
||||||
|
ROWS = 3
|
||||||
|
_GAP = " "
|
||||||
|
|
||||||
|
|
||||||
|
def _cell(top: str, bottom: str) -> str:
|
||||||
|
if top == "#" and bottom == "#":
|
||||||
|
return "█"
|
||||||
|
if top == "#":
|
||||||
|
return "▀"
|
||||||
|
if bottom == "#":
|
||||||
|
return "▄"
|
||||||
|
return " "
|
||||||
|
|
||||||
|
|
||||||
|
def _glyph_rows(char: str) -> tuple[str, ...]:
|
||||||
|
bitmap = GLYPHS.get(char)
|
||||||
|
if bitmap is None:
|
||||||
|
raise KeyError(f"no wordmark glyph for {char!r}")
|
||||||
|
return tuple(
|
||||||
|
"".join(_cell(bitmap[row * 2][col], bitmap[row * 2 + 1][col]) for col in range(len(bitmap[0])))
|
||||||
|
for row in range(ROWS)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def render(text: str, styles: Sequence[str | None] | None = None) -> tuple[str, ...]:
|
||||||
|
"""Spell ``text`` as three rows of half blocks.
|
||||||
|
|
||||||
|
``styles`` gives one rich-markup style per character (``None`` leaves it in
|
||||||
|
the terminal's default colour). Contiguous runs of the same style share a
|
||||||
|
markup span, so ``render("BASSDRIVE", ...)`` paints the first four letters
|
||||||
|
and the last five separately.
|
||||||
|
"""
|
||||||
|
if styles is not None and len(styles) != len(text):
|
||||||
|
raise ValueError("styles must provide one entry per character")
|
||||||
|
|
||||||
|
glyphs = [_glyph_rows(char.upper()) for char in text]
|
||||||
|
rows: list[str] = []
|
||||||
|
for row in range(ROWS):
|
||||||
|
cells: list[tuple[str | None, str]] = []
|
||||||
|
for index, glyph in enumerate(glyphs):
|
||||||
|
style = styles[index] if styles is not None else None
|
||||||
|
if index:
|
||||||
|
cells.append((style, _GAP))
|
||||||
|
cells.extend((style, char) for char in glyph[row])
|
||||||
|
while cells and cells[-1][1] == " ":
|
||||||
|
cells.pop()
|
||||||
|
|
||||||
|
segments: list[tuple[str | None, str]] = []
|
||||||
|
for style, char in cells:
|
||||||
|
if segments and segments[-1][0] == style:
|
||||||
|
segments[-1] = (style, segments[-1][1] + char)
|
||||||
|
else:
|
||||||
|
segments.append((style, char))
|
||||||
|
rows.append(
|
||||||
|
"".join(f"[{style}]{body}[/]" if style else body for style, body in segments)
|
||||||
|
)
|
||||||
|
return tuple(rows)
|
||||||
@@ -28,7 +28,7 @@ class ProviderScreen(ModalScreen[str | None]):
|
|||||||
header = Text()
|
header = Text()
|
||||||
header.append("◤ ", style=f"bold {gradient(SPECTRUM_STOPS, 0.9)}")
|
header.append("◤ ", style=f"bold {gradient(SPECTRUM_STOPS, 0.9)}")
|
||||||
header.append("CATALOGUES", style=f"bold {gradient(SPECTRUM_STOPS, 0.55)}")
|
header.append("CATALOGUES", style=f"bold {gradient(SPECTRUM_STOPS, 0.55)}")
|
||||||
header.append(" pick a station catalogue", style="dim")
|
header.append(" · pick a station catalogue", style="dim")
|
||||||
|
|
||||||
rows: list[Text] = []
|
rows: list[Text] = []
|
||||||
for index, slug in enumerate(self.slugs):
|
for index, slug in enumerate(self.slugs):
|
||||||
|
|||||||
+2
-2
@@ -1,13 +1,13 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "beefy-player"
|
name = "beefy-player"
|
||||||
version = "0.1.0"
|
version = "0.1.0"
|
||||||
description = "A flashy terminal player for internet radio — live spectrum analyser, MPRIS / macOS Now Playing support, song history and pluggable station catalogues (SomaFM and tilderadio included)"
|
description = "A flashy terminal player for internet radio — live spectrum analyser, MPRIS / macOS Now Playing support, song history and pluggable station catalogues (SomaFM, tilderadio and Bassdrive included)"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
license = "WTFPL"
|
license = "WTFPL"
|
||||||
license-files = ["LICENSE"]
|
license-files = ["LICENSE"]
|
||||||
requires-python = ">=3.10"
|
requires-python = ">=3.10"
|
||||||
authors = [{ name = "beefy-player contributors" }]
|
authors = [{ name = "beefy-player contributors" }]
|
||||||
keywords = ["radio", "internet-radio", "tui", "terminal", "mpris", "audio", "somafm", "tilderadio"]
|
keywords = ["radio", "internet-radio", "tui", "terminal", "mpris", "audio", "somafm", "tilderadio", "bassdrive"]
|
||||||
classifiers = [
|
classifiers = [
|
||||||
"Environment :: Console",
|
"Environment :: Console",
|
||||||
"Intended Audience :: End Users/Desktop",
|
"Intended Audience :: End Users/Desktop",
|
||||||
|
|||||||
+63
-6
@@ -9,9 +9,20 @@ from PIL import Image
|
|||||||
from beefy_player import artwork
|
from beefy_player import artwork
|
||||||
|
|
||||||
|
|
||||||
def png_bytes(size=(400, 200)) -> bytes:
|
def png_bytes(size=(400, 200), alpha: int = 180) -> bytes:
|
||||||
buffer = BytesIO()
|
buffer = BytesIO()
|
||||||
Image.new("RGBA", size, (20, 180, 220, 180)).save(buffer, format="PNG")
|
Image.new("RGBA", size, (20, 180, 220, alpha)).save(buffer, format="PNG")
|
||||||
|
return buffer.getvalue()
|
||||||
|
|
||||||
|
|
||||||
|
def ink_bbox(image: Image.Image):
|
||||||
|
"""Where the artwork itself is, ignoring transparent or black padding."""
|
||||||
|
return image.convert("RGB").getbbox()
|
||||||
|
|
||||||
|
|
||||||
|
def opaque_png_bytes(size=(400, 200)) -> bytes:
|
||||||
|
buffer = BytesIO()
|
||||||
|
Image.new("RGB", size, (20, 180, 220)).save(buffer, format="PNG")
|
||||||
return buffer.getvalue()
|
return buffer.getvalue()
|
||||||
|
|
||||||
|
|
||||||
@@ -35,12 +46,58 @@ def test_protocol_detection() -> None:
|
|||||||
assert artwork.detect_protocol({"SOMAFM_IMAGES": "0"}) is None
|
assert artwork.detect_protocol({"SOMAFM_IMAGES": "0"}) is None
|
||||||
|
|
||||||
|
|
||||||
def test_fit_image_crops_to_the_cell_box() -> None:
|
def test_fit_image_letterboxes_inside_the_cell_box() -> None:
|
||||||
fitted = artwork.fit_image(png_bytes(), 13, 8)
|
fitted = artwork.fit_image(png_bytes((400, 200)), 13, 8)
|
||||||
assert fitted is not None
|
assert fitted is not None
|
||||||
cell_width, cell_height = artwork.cell_size()
|
cell_width, cell_height = artwork.cell_size()
|
||||||
expected = (13 * cell_width) / (8 * cell_height)
|
assert (fitted.width, fitted.height) == (13 * cell_width, 8 * cell_height)
|
||||||
assert fitted.width / fitted.height == pytest.approx(expected, rel=0.01)
|
|
||||||
|
# the ink keeps its own 2:1 aspect instead of being cropped to the box
|
||||||
|
ink = ink_bbox(fitted)
|
||||||
|
assert ink is not None
|
||||||
|
left, top, right, bottom = ink
|
||||||
|
assert (right - left) / (bottom - top) == pytest.approx(2.0, rel=0.02)
|
||||||
|
# a wide image is padded above and below, and spans the full width
|
||||||
|
assert left == 0 and right == fitted.width
|
||||||
|
assert top > 0 and bottom < fitted.height
|
||||||
|
|
||||||
|
|
||||||
|
def test_an_image_with_alpha_gets_a_black_backing() -> None:
|
||||||
|
"""Transparent artwork is drawn over the mosaic; black keeps it readable."""
|
||||||
|
fitted = artwork.fit_image(png_bytes((400, 200)), 13, 8)
|
||||||
|
assert fitted is not None
|
||||||
|
# every pixel is opaque now, and the padding is black rather than mosaic
|
||||||
|
assert fitted.getchannel("A").getextrema() == (255, 255)
|
||||||
|
assert fitted.getpixel((0, 0)) == (0, 0, 0, 255)
|
||||||
|
# the ink is still letterboxed, not cropped: its own 2:1 aspect survives
|
||||||
|
ink = ink_bbox(fitted)
|
||||||
|
assert ink is not None
|
||||||
|
left, top, right, bottom = ink
|
||||||
|
assert (right - left) / (bottom - top) == pytest.approx(2.0, rel=0.02)
|
||||||
|
|
||||||
|
|
||||||
|
def test_an_opaque_image_keeps_the_transparent_padding() -> None:
|
||||||
|
"""No alpha channel: nothing to hide, so the mosaic shows through."""
|
||||||
|
fitted = artwork.fit_image(opaque_png_bytes((400, 200)), 13, 8)
|
||||||
|
assert fitted is not None
|
||||||
|
assert fitted.getpixel((0, 0))[3] == 0
|
||||||
|
ink = ink_bbox(fitted)
|
||||||
|
assert ink is not None
|
||||||
|
left, top, right, bottom = ink
|
||||||
|
assert (right - left) / (bottom - top) == pytest.approx(2.0, rel=0.02)
|
||||||
|
|
||||||
|
|
||||||
|
def test_fit_image_centres_a_tall_image() -> None:
|
||||||
|
fitted = artwork.fit_image(png_bytes((100, 400)), 13, 8)
|
||||||
|
assert fitted is not None
|
||||||
|
ink = ink_bbox(fitted)
|
||||||
|
assert ink is not None
|
||||||
|
left, _, right, _ = ink
|
||||||
|
assert left > 0 and right < fitted.width
|
||||||
|
assert right - left == pytest.approx(fitted.height / 4, rel=0.05)
|
||||||
|
|
||||||
|
|
||||||
|
def test_fit_image_rejects_garbage() -> None:
|
||||||
assert artwork.fit_image(b"not an image", 13, 8) is None
|
assert artwork.fit_image(b"not an image", 13, 8) is None
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,278 @@
|
|||||||
|
"""Bassdrive: Icecast status, show titles, the homepage schedule, rolling history."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
|
|
||||||
|
import httpx
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from beefy_player.models import Broadcast
|
||||||
|
from beefy_player.providers.bassdrive import BassdriveProvider, api
|
||||||
|
|
||||||
|
STATUS = {
|
||||||
|
"icestats": {
|
||||||
|
"source": [
|
||||||
|
{
|
||||||
|
"listenurl": "https://ice.bassdrive.net:443/stream",
|
||||||
|
"bitrate": 128,
|
||||||
|
"server_type": "audio/mpeg",
|
||||||
|
"listeners": 31,
|
||||||
|
"display-title": "Ben XO :: LIVE :: Night Shift Over [discord.gg/bassdrive]",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"listenurl": "https://ice.bassdrive.net:443/stream56",
|
||||||
|
"bitrate": 56,
|
||||||
|
"server_type": "audio/aacp",
|
||||||
|
"listeners": 5,
|
||||||
|
"display-title": "Ben XO :: LIVE :: Night Shift Over [discord.gg/bassdrive]",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"listenurl": "https://ice.bassdrive.net:443/stream32",
|
||||||
|
"bitrate": 32,
|
||||||
|
"server_type": "audio/aacp",
|
||||||
|
"listeners": 2,
|
||||||
|
"display-title": "Ben XO :: LIVE :: Night Shift Over [discord.gg/bassdrive]",
|
||||||
|
},
|
||||||
|
]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
# Trimmed from the real homepage: the same markup the site ships.
|
||||||
|
SCHEDULE_PAGE = """
|
||||||
|
<ul class="schedule">
|
||||||
|
<li>
|
||||||
|
<a href="radio/get-dipped">
|
||||||
|
<p class="title">Get Dipped <span class="location">Los Angeles, CA</span></p>
|
||||||
|
<span class="time">
|
||||||
|
<local-time datetime="2026-11-03T19:00:00-05:00" hour="numeric"></local-time> -
|
||||||
|
<local-time datetime="2026-11-03T21:00:00-05:00" hour="numeric"></local-time>
|
||||||
|
</span>
|
||||||
|
</a>
|
||||||
|
</li>
|
||||||
|
<li>
|
||||||
|
<a href="radio/drumobsession">
|
||||||
|
<p class="title">DrumObsession <span class="location">Poznan, Poland</span></p>
|
||||||
|
<span class="time">
|
||||||
|
<local-time datetime="2026-11-03T11:00:00-05:00" hour="numeric"></local-time> -
|
||||||
|
<local-time datetime="2026-11-03T13:00:00-05:00" hour="numeric"></local-time>
|
||||||
|
</span>
|
||||||
|
</a>
|
||||||
|
</li>
|
||||||
|
</ul>
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def fresh_memo():
|
||||||
|
api.reset_schedule_memo()
|
||||||
|
yield
|
||||||
|
api.reset_schedule_memo()
|
||||||
|
|
||||||
|
|
||||||
|
def test_split_title_reads_both_conventions() -> None:
|
||||||
|
assert api.split_title("Ben XO :: LIVE :: Night Shift Over [promo]") == (
|
||||||
|
"Ben XO",
|
||||||
|
"Night Shift Over",
|
||||||
|
True,
|
||||||
|
)
|
||||||
|
# plenty of shows simply append LIVE
|
||||||
|
assert api.split_title("Sohl Patrol LIVE") == ("", "Sohl Patrol", True)
|
||||||
|
assert api.split_title("Bassdrive Auto DJ") == ("", "Bassdrive Auto DJ", False)
|
||||||
|
|
||||||
|
|
||||||
|
def test_mounts_become_playlists_best_first_and_listeners_add_up() -> None:
|
||||||
|
playlists = api.playlists_from_status(STATUS)
|
||||||
|
assert [entry["bitrate"] for entry in playlists] == ["128", "56", "32"]
|
||||||
|
assert [entry["quality"] for entry in playlists] == ["highest", "high", "low"]
|
||||||
|
assert playlists[0]["format"] == "mp3" and playlists[1]["format"] == "aacp"
|
||||||
|
assert api.status_listeners(STATUS) == 38
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_schedule_reads_the_homepage_grid() -> None:
|
||||||
|
entries = api.parse_schedule(SCHEDULE_PAGE)
|
||||||
|
assert [entry.name for entry in entries] == ["DrumObsession", "Get Dipped"]
|
||||||
|
assert entries[0].location == "Poznan, Poland"
|
||||||
|
assert entries[0].start == datetime(2026, 11, 3, 11, tzinfo=timezone(timedelta(hours=-5)))
|
||||||
|
assert entries[0].end == datetime(2026, 11, 3, 13, tzinfo=timezone(timedelta(hours=-5)))
|
||||||
|
# the schedule list itself is not a schedule slot
|
||||||
|
assert api.parse_schedule("<ul><li>nothing here</li></ul>") == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_roll_forward_repeats_the_week() -> None:
|
||||||
|
start = datetime(2026, 11, 3, 11, tzinfo=timezone(timedelta(hours=-5)))
|
||||||
|
entry = Broadcast(
|
||||||
|
name="DrumObsession",
|
||||||
|
start=start,
|
||||||
|
end=start + timedelta(hours=2),
|
||||||
|
)
|
||||||
|
now = start + timedelta(days=1)
|
||||||
|
rolled = api.roll_forward([entry], now)
|
||||||
|
assert len(rolled) == 1
|
||||||
|
assert rolled[0].start == start + timedelta(days=7)
|
||||||
|
assert rolled[0].start >= now
|
||||||
|
# an entry that is over and cannot be rolled into the future is dropped
|
||||||
|
far = start + timedelta(days=40)
|
||||||
|
assert api.roll_forward([entry], far) == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_station_maps_the_title_and_finds_the_live_location() -> None:
|
||||||
|
now = datetime.now(timezone.utc)
|
||||||
|
slot = Broadcast(
|
||||||
|
name="Night Shift Over",
|
||||||
|
location="London, UK",
|
||||||
|
start=now - timedelta(minutes=30),
|
||||||
|
end=now + timedelta(minutes=30),
|
||||||
|
)
|
||||||
|
station = api.station_from_status(STATUS, [slot])
|
||||||
|
assert station is not None
|
||||||
|
assert station.id == "bassdrive"
|
||||||
|
assert station.dj == "Ben XO"
|
||||||
|
assert station.presenter == "live from London, UK"
|
||||||
|
assert station.now_playing is not None
|
||||||
|
assert station.now_playing.title == "Night Shift Over"
|
||||||
|
assert station.now_playing.artist == "Ben XO"
|
||||||
|
assert station.listeners == 38
|
||||||
|
assert station.format_label == "mp3 · 128k"
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_title_without_a_live_marker_still_counts_when_the_slot_is_running() -> None:
|
||||||
|
now = datetime.now(timezone.utc)
|
||||||
|
status = {
|
||||||
|
"icestats": {
|
||||||
|
"source": [
|
||||||
|
{
|
||||||
|
"listenurl": "https://ice.bassdrive.net:443/stream",
|
||||||
|
"bitrate": 128,
|
||||||
|
"server_type": "audio/mpeg",
|
||||||
|
"listeners": 3,
|
||||||
|
"display-title": "Sohl Patrol",
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
slot = Broadcast(
|
||||||
|
name="Sohl Patrol",
|
||||||
|
location="White Plains, NY, USA",
|
||||||
|
start=now - timedelta(minutes=10),
|
||||||
|
end=now + timedelta(minutes=50),
|
||||||
|
)
|
||||||
|
live = api.live_broadcast(status, [slot], now)
|
||||||
|
assert live is not None and live.live is True
|
||||||
|
assert live.location == "White Plains, NY, USA"
|
||||||
|
|
||||||
|
|
||||||
|
def test_station_without_title_or_mounts_is_not_a_station() -> None:
|
||||||
|
assert api.station_from_status({"icestats": {}}) is None
|
||||||
|
|
||||||
|
|
||||||
|
def client_for(
|
||||||
|
status_payload,
|
||||||
|
page: str = "",
|
||||||
|
*,
|
||||||
|
status_code: int = 200,
|
||||||
|
page_code: int = 200,
|
||||||
|
) -> httpx.AsyncClient:
|
||||||
|
def handler(request: httpx.Request) -> httpx.Response:
|
||||||
|
if str(request.url).endswith("status-json.xsl"):
|
||||||
|
return httpx.Response(status_code, json=status_payload)
|
||||||
|
return httpx.Response(page_code, text=page)
|
||||||
|
|
||||||
|
return httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
||||||
|
|
||||||
|
|
||||||
|
async def songs_for(payload, station_id: str = "bassdrive"):
|
||||||
|
async with client_for(payload) as client:
|
||||||
|
return await api.fetch_songs(client, station_id)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_fetch_schedule_puts_the_live_show_first_and_caches_the_grid() -> None:
|
||||||
|
now = datetime.now(timezone.utc)
|
||||||
|
payload = {
|
||||||
|
"icestats": {
|
||||||
|
"source": [
|
||||||
|
{
|
||||||
|
"listenurl": "https://ice.bassdrive.net:443/stream",
|
||||||
|
"bitrate": 128,
|
||||||
|
"server_type": "audio/mpeg",
|
||||||
|
"listeners": 9,
|
||||||
|
"display-title": "Get Dipped LIVE",
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
# the grid is generated relative to now so the live slot is really running
|
||||||
|
local = now.astimezone(timezone(timedelta(hours=-5)))
|
||||||
|
start = local - timedelta(minutes=15)
|
||||||
|
end = local + timedelta(minutes=45)
|
||||||
|
page = SCHEDULE_PAGE.replace("2026-11-03T19:00:00-05:00", start.isoformat()).replace(
|
||||||
|
"2026-11-03T21:00:00-05:00", end.isoformat()
|
||||||
|
)
|
||||||
|
async with client_for(payload, page) as client:
|
||||||
|
broadcast_list = await api.fetch_schedule(client)
|
||||||
|
|
||||||
|
assert broadcast_list[0].live is True
|
||||||
|
assert broadcast_list[0].name == "Get Dipped"
|
||||||
|
assert broadcast_list[0].location == "Los Angeles, CA"
|
||||||
|
assert broadcast_list[0].start.hour == start.hour
|
||||||
|
assert all(not item.live for item in broadcast_list[1:])
|
||||||
|
# the grid was written for offline starts
|
||||||
|
assert api.load_cached_schedule()
|
||||||
|
# ... and the second fetch reuses the memo instead of the network
|
||||||
|
async with client_for(payload, page_code=500) as client:
|
||||||
|
assert await api.fetch_schedule(client)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_history_is_a_rolling_record_of_shows() -> None:
|
||||||
|
first = await songs_for(STATUS)
|
||||||
|
assert [song.title for song in first] == ["Night Shift Over"]
|
||||||
|
|
||||||
|
# same show: the first-seen time is kept
|
||||||
|
again = await songs_for(STATUS)
|
||||||
|
assert again[0].played_at == first[0].played_at
|
||||||
|
|
||||||
|
other = {
|
||||||
|
"icestats": {
|
||||||
|
"source": [
|
||||||
|
{
|
||||||
|
"listenurl": "https://ice.bassdrive.net:443/stream",
|
||||||
|
"bitrate": 128,
|
||||||
|
"server_type": "audio/mpeg",
|
||||||
|
"listeners": 1,
|
||||||
|
"display-title": "Sohl Patrol LIVE",
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
moved = await songs_for(other)
|
||||||
|
assert [song.title for song in moved] == ["Sohl Patrol", "Night Shift Over"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_provider_describes_its_catalogue() -> None:
|
||||||
|
provider = BassdriveProvider()
|
||||||
|
assert provider.has_schedule is True
|
||||||
|
assert provider.bandcamp is False
|
||||||
|
assert len(provider.hero) == 3
|
||||||
|
assert "bright_cyan" in "".join(provider.hero)
|
||||||
|
assert provider.hero_lines == (
|
||||||
|
"BASSDRIVE",
|
||||||
|
"WORLDWIDE DRUM & BASS",
|
||||||
|
"LIVE SHOWS 24/7",
|
||||||
|
)
|
||||||
|
|
||||||
|
stations = provider.fallback_stations()
|
||||||
|
assert [station.id for station in stations] == ["bassdrive"]
|
||||||
|
urls = provider.fallback_urls(stations[0])
|
||||||
|
assert urls[0].startswith("https://") and len(urls) == 7
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_cold_start_survives_an_unreachable_homepage() -> None:
|
||||||
|
"""`fetch_station` asks for the grid; a broken homepage must not stop it."""
|
||||||
|
async with client_for(STATUS, page_code=500) as client:
|
||||||
|
station = await api.fetch_station(client)
|
||||||
|
assert station.id == "bassdrive"
|
||||||
|
assert station.presenter == "live now" # live, but the location is unknown
|
||||||
|
assert station.now_playing is not None
|
||||||
+49
-14
@@ -6,8 +6,9 @@ import types
|
|||||||
import pytest
|
import pytest
|
||||||
from rich.cells import cell_len
|
from rich.cells import cell_len
|
||||||
|
|
||||||
|
from beefy_player.providers import PROVIDERS
|
||||||
from beefy_player.providers.somafm import SomaFMProvider
|
from beefy_player.providers.somafm import SomaFMProvider
|
||||||
from beefy_player.providers.tilderadio import TilderadioProvider
|
from beefy_player.providers.wordmark import render
|
||||||
from beefy_player.widgets.header import HeaderBar
|
from beefy_player.widgets.header import HeaderBar
|
||||||
|
|
||||||
TIME = re.compile(r"\d\d:\d\d:\d\d$")
|
TIME = re.compile(r"\d\d:\d\d:\d\d$")
|
||||||
@@ -54,32 +55,66 @@ def test_right_block_hidden_on_narrow_terminals() -> None:
|
|||||||
assert all(":" not in line for line in lines)
|
assert all(":" not in line for line in lines)
|
||||||
|
|
||||||
|
|
||||||
def test_somafm_hero_is_the_wordmark_and_its_taglines() -> None:
|
def test_somafm_hero_spells_the_whole_name_and_its_taglines() -> None:
|
||||||
lines = render_header(143)
|
lines = render_header(143)
|
||||||
assert lines[0].startswith("▄▀▀▀ ▄▀▀▄ █▄ ▄█ ▄▀▀▄")
|
assert lines[0].startswith(render("SOMAFM")[0])
|
||||||
assert "SOMA FM RADIO" in lines[0]
|
assert "SOMA FM RADIO" in lines[0]
|
||||||
assert "LISTENER SUPPORTED" in lines[1]
|
assert "LISTENER SUPPORTED" in lines[1]
|
||||||
assert "SAN FRANCISCO" in lines[2]
|
assert "SAN FRANCISCO" in lines[2]
|
||||||
|
|
||||||
|
|
||||||
def test_tilderadio_hero_is_its_own_art_and_motto() -> None:
|
def hero_styles(provider) -> list[str]:
|
||||||
provider = TilderadioProvider()
|
return [
|
||||||
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)
|
str(span.style)
|
||||||
for line in HeaderBar.render(_fake(provider)).renderables
|
for line in HeaderBar.render(_fake(provider)).renderables
|
||||||
for span in line.spans
|
for span in line.spans
|
||||||
]
|
]
|
||||||
assert any("green" in style for style in styles)
|
|
||||||
|
|
||||||
|
def test_the_i_and_t_have_thin_stems() -> None:
|
||||||
|
"""Two-pixel stems make an I look like an H and a T look like a block."""
|
||||||
|
assert render("I") == ("▀█▀", " █", "▄█▄")
|
||||||
|
assert render("T") == ("▀█▀", " █", " █")
|
||||||
|
|
||||||
|
|
||||||
|
def test_the_font_still_spells_soma_the_way_the_original_logo_did() -> None:
|
||||||
|
"""The wordmark font is the SomaFM logo, so the old art must be reproduced."""
|
||||||
|
assert render("SOMA") == (
|
||||||
|
"▄▀▀▀ ▄▀▀▄ █▄ ▄█ ▄▀▀▄",
|
||||||
|
"▀▀▀▄ █ █ █ █ █ █▀▀█",
|
||||||
|
"▄▄▄▀ ▀▄▄▀ █ █ █ █",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_every_catalogue_has_a_three_row_wordmark() -> None:
|
||||||
|
for slug, provider in PROVIDERS.items():
|
||||||
|
assert len(provider.hero) == 3, slug
|
||||||
|
assert provider.hero_lines, slug
|
||||||
|
|
||||||
|
|
||||||
|
def test_tilderadio_spells_its_own_name_in_the_default_colour() -> None:
|
||||||
|
provider = PROVIDERS["tilderadio"]
|
||||||
|
lines = render_header(143, height=len(provider.hero), provider=provider)
|
||||||
|
assert len(lines) == 3
|
||||||
|
assert "TILDERADIO" in lines[0] # the wordmark
|
||||||
|
assert "STREAMED BY / FOR USERS OF THE" in lines[1]
|
||||||
|
assert "TILDEVERSE" in lines[2]
|
||||||
|
# no colour of its own: the art inherits the terminal foreground
|
||||||
|
assert all("[" not in row for row in provider.hero)
|
||||||
|
assert not {"bright_cyan", "cyan"} & set(hero_styles(provider))
|
||||||
assert all("SOMA" not in line for line in lines)
|
assert all("SOMA" not in line for line in lines)
|
||||||
|
|
||||||
|
|
||||||
|
def test_bassdrive_spells_its_name_in_its_own_two_blues() -> None:
|
||||||
|
provider = PROVIDERS["bassdrive"]
|
||||||
|
lines = render_header(143, height=len(provider.hero), provider=provider)
|
||||||
|
assert "BASSDRIVE" in lines[0]
|
||||||
|
assert "WORLDWIDE DRUM & BASS" in lines[1]
|
||||||
|
assert "LIVE SHOWS 24/7" in lines[2]
|
||||||
|
styles = hero_styles(provider)
|
||||||
|
assert "bright_cyan" in styles and "cyan" in styles
|
||||||
|
|
||||||
|
|
||||||
def _fake(provider):
|
def _fake(provider):
|
||||||
fake = type("Fake", (), {})()
|
fake = type("Fake", (), {})()
|
||||||
fake.app = types.SimpleNamespace(
|
fake.app = types.SimpleNamespace(
|
||||||
|
|||||||
+11
-8
@@ -10,6 +10,7 @@ from beefy_player import config
|
|||||||
from beefy_player.models import Song, Station
|
from beefy_player.models import Song, Station
|
||||||
from beefy_player.providers import PROVIDERS, available_providers, get_provider
|
from beefy_player.providers import PROVIDERS, available_providers, get_provider
|
||||||
from beefy_player.providers.base import RadioProvider
|
from beefy_player.providers.base import RadioProvider
|
||||||
|
from beefy_player.providers.bassdrive import BassdriveProvider
|
||||||
from beefy_player.providers.somafm import SomaFMProvider
|
from beefy_player.providers.somafm import SomaFMProvider
|
||||||
from beefy_player.providers.tilderadio import TilderadioProvider
|
from beefy_player.providers.tilderadio import TilderadioProvider
|
||||||
from beefy_player.widgets.header import HeaderBar
|
from beefy_player.widgets.header import HeaderBar
|
||||||
@@ -21,10 +22,11 @@ def rendered_text(app) -> str:
|
|||||||
|
|
||||||
|
|
||||||
def test_registry_lists_somafm_as_the_default() -> None:
|
def test_registry_lists_somafm_as_the_default() -> None:
|
||||||
assert available_providers() == ("somafm", "tilderadio")
|
assert available_providers() == ("somafm", "tilderadio", "bassdrive")
|
||||||
assert isinstance(get_provider(), SomaFMProvider)
|
assert isinstance(get_provider(), SomaFMProvider)
|
||||||
assert get_provider("somafm") is get_provider()
|
assert get_provider("somafm") is get_provider()
|
||||||
assert isinstance(get_provider("tilderadio"), TilderadioProvider)
|
assert isinstance(get_provider("tilderadio"), TilderadioProvider)
|
||||||
|
assert isinstance(get_provider("bassdrive"), BassdriveProvider)
|
||||||
# a stale slug in the config file must not stop the app from starting
|
# a stale slug in the config file must not stop the app from starting
|
||||||
assert get_provider("no-such-catalogue") is get_provider()
|
assert get_provider("no-such-catalogue") is get_provider()
|
||||||
|
|
||||||
@@ -182,10 +184,10 @@ async def test_p_opens_the_picker_and_switches_catalogue() -> None:
|
|||||||
assert app.provider.id == "tilderadio"
|
assert app.provider.id == "tilderadio"
|
||||||
assert app.config.provider == "tilderadio"
|
assert app.config.provider == "tilderadio"
|
||||||
assert [station.id for station in app.stations] == ["tilderadio"]
|
assert [station.id for station in app.stations] == ["tilderadio"]
|
||||||
# the left pane becomes the schedule, and the hero grows
|
# the left pane becomes the schedule; every hero is the same height
|
||||||
assert app.query_one(StationList).display is False
|
assert app.query_one(StationList).display is False
|
||||||
assert app.query_one(SchedulePanel).display is True
|
assert app.query_one(SchedulePanel).display is True
|
||||||
assert app.query_one(HeaderBar).styles.height.value == 8
|
assert app.query_one(HeaderBar).styles.height.value == 4
|
||||||
assert "tilderadio" in app.title
|
assert "tilderadio" in app.title
|
||||||
|
|
||||||
|
|
||||||
@@ -225,11 +227,12 @@ def test_config_remembers_a_station_per_catalogue() -> None:
|
|||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_short_terminals_crop_the_hero_but_keep_the_ticker() -> None:
|
async def test_every_catalogue_header_is_the_same_height() -> None:
|
||||||
app = _CatalogueApp("tilderadio").app
|
for slug in available_providers():
|
||||||
async with app.run_test(size=(90, 24)) as pilot:
|
app = _CatalogueApp(slug).app
|
||||||
|
async with app.run_test(size=(110, 40)) as pilot:
|
||||||
await pilot.pause()
|
await pilot.pause()
|
||||||
header = app.query_one(HeaderBar)
|
header = app.query_one(HeaderBar)
|
||||||
assert header.styles.height.value <= 6
|
assert header.styles.height.value == 4, slug
|
||||||
lines = [line.plain for line in header.render().renderables]
|
lines = [line.plain for line in header.render().renderables]
|
||||||
assert any("STOPPED" in line or "PLAYING" in line for line in lines)
|
assert any("STOPPED" in line or "PLAYING" in line for line in lines), slug
|
||||||
+58
-4
@@ -48,6 +48,9 @@ ANSI16: dict[str, tuple[int, int, int]] = {
|
|||||||
#: Below this alpha a half-cell counts as empty (terminal background).
|
#: Below this alpha a half-cell counts as empty (terminal background).
|
||||||
ALPHA_CUTOFF = 110
|
ALPHA_CUTOFF = 110
|
||||||
|
|
||||||
|
#: Ink weaker than this is treated as fringe when cropping the image.
|
||||||
|
CROP_CUTOFF = 96
|
||||||
|
|
||||||
|
|
||||||
def nearest_colour(rgb: tuple[int, int, int]) -> str:
|
def nearest_colour(rgb: tuple[int, int, int]) -> str:
|
||||||
best, best_distance = "white", float("inf")
|
best, best_distance = "white", float("inf")
|
||||||
@@ -58,7 +61,51 @@ def nearest_colour(rgb: tuple[int, int, int]) -> str:
|
|||||||
return best
|
return best
|
||||||
|
|
||||||
|
|
||||||
def load_image(path: Path | None, url: str | None) -> Image.Image:
|
def key_white(image: Image.Image, tolerance: int) -> Image.Image:
|
||||||
|
"""Make near-white pixels transparent.
|
||||||
|
|
||||||
|
A logo exported on a white background (or one whose anti-aliasing bleeds
|
||||||
|
white) otherwise survives the downsample as a glaring box.
|
||||||
|
"""
|
||||||
|
image = image.copy()
|
||||||
|
pixels = image.load()
|
||||||
|
width, height = image.size
|
||||||
|
threshold = 255 - max(0, tolerance)
|
||||||
|
for y in range(height):
|
||||||
|
for x in range(width):
|
||||||
|
red, green, blue, alpha = pixels[x, y]
|
||||||
|
if alpha and min(red, green, blue) >= threshold:
|
||||||
|
pixels[x, y] = (red, green, blue, 0)
|
||||||
|
return image
|
||||||
|
|
||||||
|
|
||||||
|
def resize_premultiplied(image: Image.Image, size: tuple[int, int]) -> Image.Image:
|
||||||
|
"""Resize in premultiplied alpha, so transparent pixels cannot bleed in.
|
||||||
|
|
||||||
|
Plain ``Image.resize`` averages the RGB of fully transparent pixels (often
|
||||||
|
white) with the surrounding ink, which is how a clean logo turns into a
|
||||||
|
grey haze at 20x14.
|
||||||
|
"""
|
||||||
|
import numpy as np
|
||||||
|
|
||||||
|
data = np.asarray(image.convert("RGBA"), dtype=np.float32)
|
||||||
|
alpha = data[..., 3:4] / 255.0
|
||||||
|
premultiplied = np.concatenate([data[..., :3] * alpha, data[..., 3:4]], axis=-1)
|
||||||
|
small = Image.fromarray(premultiplied.astype(np.uint8), "RGBA").resize(
|
||||||
|
size, Image.LANCZOS
|
||||||
|
)
|
||||||
|
out = np.asarray(small, dtype=np.float32)
|
||||||
|
out_alpha = out[..., 3:4] / 255.0
|
||||||
|
rgb = out[..., :3] / np.maximum(out_alpha, 1e-6)
|
||||||
|
return Image.fromarray(
|
||||||
|
np.concatenate([np.clip(rgb, 0, 255), out[..., 3:4]], axis=-1).astype(np.uint8),
|
||||||
|
"RGBA",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def load_image(
|
||||||
|
path: Path | None, url: str | None, key_white_background: bool = False
|
||||||
|
) -> Image.Image:
|
||||||
if path is None and url:
|
if path is None and url:
|
||||||
import httpx
|
import httpx
|
||||||
|
|
||||||
@@ -73,15 +120,17 @@ def load_image(path: Path | None, url: str | None) -> Image.Image:
|
|||||||
raise SystemExit("pass an image path or --url")
|
raise SystemExit("pass an image path or --url")
|
||||||
|
|
||||||
image = image.convert("RGBA")
|
image = image.convert("RGBA")
|
||||||
|
if key_white_background:
|
||||||
|
image = key_white(image, 24)
|
||||||
# Trim the transparent margin: the art should fill the box it is given.
|
# 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)
|
opaque = image.getchannel("A").point(lambda value: 255 if value > CROP_CUTOFF else 0)
|
||||||
box = opaque.getbbox()
|
box = opaque.getbbox()
|
||||||
return image.crop(box) if box else image
|
return image.crop(box) if box else image
|
||||||
|
|
||||||
|
|
||||||
def art_rows(image: Image.Image, cols: int, rows: int) -> list[list[tuple[str, str, str]]]:
|
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."""
|
"""(glyph, foreground, background) per cell, two source pixels per cell."""
|
||||||
small = image.resize((cols, rows * 2), Image.LANCZOS)
|
small = resize_premultiplied(image, (cols, rows * 2))
|
||||||
pixels = small.load()
|
pixels = small.load()
|
||||||
grid: list[list[tuple[str, str, str]]] = []
|
grid: list[list[tuple[str, str, str]]] = []
|
||||||
for row in range(rows):
|
for row in range(rows):
|
||||||
@@ -179,9 +228,14 @@ def main(argv: list[str] | None = None) -> int:
|
|||||||
parser.add_argument("--module", type=Path, help="write a Python module with HERO")
|
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("--preview", type=Path, help="write a PNG preview of the cells")
|
||||||
parser.add_argument("--provenance", default="", help="source note for the module header")
|
parser.add_argument("--provenance", default="", help="source note for the module header")
|
||||||
|
parser.add_argument(
|
||||||
|
"--key-white",
|
||||||
|
action="store_true",
|
||||||
|
help="treat near-white pixels as transparent (logos on white)",
|
||||||
|
)
|
||||||
args = parser.parse_args(argv)
|
args = parser.parse_args(argv)
|
||||||
|
|
||||||
image = load_image(args.image, args.url)
|
image = load_image(args.image, args.url, args.key_white)
|
||||||
grid = art_rows(image, args.cols, args.rows)
|
grid = art_rows(image, args.cols, args.rows)
|
||||||
rows = markup(grid)
|
rows = markup(grid)
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user