337 lines
12 KiB
Python
337 lines
12 KiB
Python
"""Community stations: one relay each, sometimes with a status document.
|
|
|
|
Most independent stations are a single stream on Icecast, radio.co or Airtime,
|
|
plus — if you are lucky — a status document that names what is on. This module
|
|
turns that into a provider: the station, its relays, and a rolling history of
|
|
whatever the relay has announced while you were listening.
|
|
|
|
A subclass only carries data. Stations with a schedule or a multi-station
|
|
catalogue (Dublab, Rinse, Radio Paradise) override a hook or two.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import time
|
|
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 RadioProvider, fetch_json
|
|
from ..wordmark import render
|
|
|
|
#: ``public.radio.co`` status document: keyless, no station id in the body.
|
|
RADIO_CO_STATUS = "https://public.radio.co/stations/{station}/status"
|
|
|
|
#: How many shows/tracks LAST PLAYED remembers.
|
|
HISTORY_LIMIT = 20
|
|
|
|
|
|
class CommunityProvider(RadioProvider):
|
|
"""A catalogue that broadcasts one stream (or a fixed handful)."""
|
|
|
|
# -- catalogue data (set by each station module) ---------------------- #
|
|
|
|
station_title = ""
|
|
station_description = ""
|
|
station_genre = "eclectic"
|
|
station_image = ""
|
|
|
|
#: ``(url, format, quality, bitrate)`` — the relays, best first.
|
|
streams: tuple[tuple[str, str, str, str], ...] = ()
|
|
#: Keyless Icecast ``status-json.xsl``, when the relay publishes one.
|
|
status_url = ""
|
|
#: ``public.radio.co`` station id, when the relay is radio.co.
|
|
radio_co_id = ""
|
|
#: Wordmark the header spells; empty leaves the provider without a hero.
|
|
wordmark = ""
|
|
|
|
# Independent stations are not Bandcamp catalogues.
|
|
bandcamp = False
|
|
|
|
def __init__(self) -> None:
|
|
if self.wordmark:
|
|
self.hero = render(self.wordmark)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# station
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def _playlists(self) -> list[dict[str, str]]:
|
|
return [
|
|
{"url": url, "format": fmt, "quality": quality, "bitrate": bitrate}
|
|
for url, fmt, quality, bitrate in self.streams
|
|
]
|
|
|
|
def station(self) -> Station:
|
|
return Station(
|
|
id=self.id,
|
|
title=self.station_title or self.name,
|
|
description=self.station_description or self.tagline,
|
|
genre=self.station_genre,
|
|
image=self.station_image,
|
|
playlists=self._playlists(),
|
|
provider=self.id,
|
|
)
|
|
|
|
def stations(self) -> list[Station]:
|
|
return [self.station()]
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# what the relay says is on
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@property
|
|
def has_status(self) -> bool:
|
|
return bool(self.status_url or self.radio_co_id)
|
|
|
|
async def _status(self, client: httpx.AsyncClient, attempts: int = 2) -> Any:
|
|
url = (
|
|
RADIO_CO_STATUS.format(station=self.radio_co_id)
|
|
if self.radio_co_id
|
|
else self.status_url
|
|
)
|
|
return await fetch_json(client, url, attempts=attempts)
|
|
|
|
def _listeners(self, payload: Any) -> int:
|
|
if not isinstance(payload, dict):
|
|
return 0
|
|
if self.radio_co_id:
|
|
return to_int(payload.get("listeners"))
|
|
stats = payload.get("icestats")
|
|
source = stats.get("source") if isinstance(stats, dict) else None
|
|
sources = (
|
|
source
|
|
if isinstance(source, list)
|
|
else [source]
|
|
if isinstance(source, dict)
|
|
else []
|
|
)
|
|
return sum(
|
|
to_int(item.get("listeners")) for item in sources if isinstance(item, dict)
|
|
)
|
|
|
|
def _icecast_title(self, payload: Any) -> str:
|
|
stats = payload.get("icestats") if isinstance(payload, dict) else None
|
|
source = stats.get("source") if isinstance(stats, dict) else None
|
|
sources = (
|
|
source
|
|
if isinstance(source, list)
|
|
else [source]
|
|
if isinstance(source, dict)
|
|
else []
|
|
)
|
|
for item in sources:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
for key in ("display-title", "title", "yp_currently_playing"):
|
|
value = item.get(key)
|
|
if isinstance(value, str) and value.strip():
|
|
return value.strip()
|
|
return ""
|
|
|
|
def _now(self, payload: Any) -> Song | None:
|
|
"""The relay's current track/show, as a :class:`Song`."""
|
|
if not isinstance(payload, dict):
|
|
return None
|
|
if self.radio_co_id:
|
|
track = payload.get("current_track")
|
|
if not isinstance(track, dict):
|
|
return None
|
|
title = str(track.get("title") or "").strip()
|
|
if not title:
|
|
return None
|
|
song = parse_last_playing(title) or Song(title=title)
|
|
song.art_url = str(track.get("artwork_url") or "").strip()
|
|
song.played_at = parse_date(track.get("start_time"))
|
|
return song
|
|
|
|
title = self._icecast_title(payload)
|
|
if not title:
|
|
return None
|
|
return parse_last_playing(title)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# the schedule, when the catalogue has one (Dublab)
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def _scheduled_now(self) -> Broadcast | None:
|
|
now = datetime.now(timezone.utc)
|
|
for entry in self.load_cached_schedule():
|
|
if entry.start is None or entry.end is None:
|
|
continue
|
|
if entry.start <= now < entry.end:
|
|
return entry
|
|
return None
|
|
|
|
def _mark_live(self, entries: list[Broadcast]) -> list[Broadcast]:
|
|
"""Flag the slot that is running now, so the pane can say ON AIR."""
|
|
now = datetime.now(timezone.utc)
|
|
marked: list[Broadcast] = []
|
|
for entry in entries:
|
|
if (
|
|
entry.start is not None
|
|
and entry.end is not None
|
|
and entry.start <= now < entry.end
|
|
):
|
|
marked.append(
|
|
Broadcast(
|
|
name=entry.name,
|
|
dj=entry.dj,
|
|
location=entry.location,
|
|
start=entry.start,
|
|
end=entry.end,
|
|
live=True,
|
|
)
|
|
)
|
|
else:
|
|
marked.append(entry)
|
|
return marked
|
|
|
|
def _scheduled_song(self) -> Song | None:
|
|
entry = self._scheduled_now()
|
|
if entry is None:
|
|
return None
|
|
return Song(artist=self.name, title=entry.name)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# caching
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def _station_cache(self) -> Path:
|
|
return config.channels_cache_path(self.id)
|
|
|
|
def _songs_cache(self, station_id: str) -> Path:
|
|
safe = re.sub(r"[^A-Za-z0-9_.-]", "_", station_id or self.id)
|
|
return config.SONGS_CACHE_DIR / f"{self.id}-{safe}.json"
|
|
|
|
def _songs_to_cache(self, songs: list[Song]) -> list[dict[str, str]]:
|
|
return [
|
|
{
|
|
"title": song.title,
|
|
"artist": song.artist,
|
|
"art": song.art_url,
|
|
"played_at": song.played_at.isoformat() if song.played_at else "",
|
|
}
|
|
for song in songs
|
|
]
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# RadioProvider
|
|
# ------------------------------------------------------------------ #
|
|
|
|
async def fetch_stations(self, client: httpx.AsyncClient) -> list[Station]:
|
|
stations = self.stations()
|
|
payload: Any = None
|
|
if self.has_status:
|
|
try:
|
|
payload = await self._status(client)
|
|
except RuntimeError:
|
|
payload = None
|
|
if payload is None:
|
|
payload = self._cached_status()
|
|
else:
|
|
config.write_json(
|
|
self._station_cache(), {"fetched": time.time(), "status": payload}
|
|
)
|
|
|
|
song = self._now(payload) if payload is not None else None
|
|
if song is None:
|
|
song = self._scheduled_song()
|
|
listeners = self._listeners(payload) if payload is not None else 0
|
|
entry = self._scheduled_now()
|
|
|
|
for station in stations:
|
|
if song is not None:
|
|
station.now_playing = song
|
|
station.last_playing = (
|
|
f"{song.artist} - {song.title}" if song.artist else song.title
|
|
)
|
|
if listeners:
|
|
station.listeners = listeners
|
|
if entry is not None:
|
|
station.presenter = (
|
|
f"live from {entry.location}" if entry.location else "live now"
|
|
)
|
|
return stations
|
|
|
|
def _cached_status(self) -> Any:
|
|
raw = config.read_json(self._station_cache())
|
|
return raw.get("status") if isinstance(raw, dict) else None
|
|
|
|
def load_cached_stations(self) -> list[Station]:
|
|
stations = self.stations()
|
|
payload = self._cached_status()
|
|
song = self._now(payload) if payload is not None else None
|
|
if song is None:
|
|
song = self._scheduled_song()
|
|
listeners = self._listeners(payload) if payload is not None else 0
|
|
for station in stations:
|
|
if song is not None:
|
|
station.now_playing = song
|
|
if listeners:
|
|
station.listeners = listeners
|
|
return stations
|
|
|
|
def fallback_stations(self) -> list[Station]:
|
|
return self.stations()
|
|
|
|
async def fetch_history(
|
|
self, client: httpx.AsyncClient, station_id: str
|
|
) -> list[Song]:
|
|
current: Song | None = None
|
|
if self.has_status:
|
|
try:
|
|
payload = await self._status(client, attempts=1)
|
|
except RuntimeError:
|
|
payload = None
|
|
if payload is not None:
|
|
current = self._now(payload)
|
|
if current is None:
|
|
current = self._scheduled_song()
|
|
if current is None:
|
|
return self.load_cached_history(station_id)
|
|
|
|
history = self.load_cached_history(station_id)
|
|
now = datetime.now(timezone.utc)
|
|
if history and history[0].key() == current.key():
|
|
# Same show or track: keep the time we first heard it.
|
|
current.played_at = history[0].played_at
|
|
history = [current, *history[1:]]
|
|
else:
|
|
current.played_at = current.played_at or now
|
|
history = [current, *[s for s in history if s.key() != current.key()]]
|
|
history = history[:HISTORY_LIMIT]
|
|
|
|
config.write_json(
|
|
self._songs_cache(station_id),
|
|
{"fetched": time.time(), "songs": self._songs_to_cache(history)},
|
|
)
|
|
return history
|
|
|
|
def load_cached_history(self, station_id: str) -> list[Song]:
|
|
raw = config.read_json(self._songs_cache(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(),
|
|
art_url=str(entry.get("art") or "").strip(),
|
|
played_at=parse_date(entry.get("played_at")),
|
|
)
|
|
if song.title or song.artist:
|
|
songs.append(song)
|
|
return songs
|
|
|
|
def fallback_urls(self, station: Station) -> list[str]:
|
|
return [url for url, _format, _quality, _bitrate in self.streams]
|