diff --git a/beefy_player/__main__.py b/beefy_player/__main__.py index 51369dd..f333c77 100644 --- a/beefy_player/__main__.py +++ b/beefy_player/__main__.py @@ -6,16 +6,23 @@ import argparse import asyncio from collections.abc import Sequence -from . import __version__, api -from .fallback import fallback_channels +from . import __version__ +from .providers import available_providers, get_provider +from .providers.base import make_client from .spectrum import SOURCE_MODES def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( - prog="somafm-tui", - description="A flashy terminal radio for somafm.com — live spectrum, " - "song history and desktop media integration.", + prog="beefy-player", + description="Beefy Player — a flashy terminal player for internet " + "radio: live spectrum, song history and desktop media integration.", + ) + parser.add_argument( + "--provider", + choices=available_providers(), + metavar="ID", + help="station catalogue to use (default: the one in the config file)", ) parser.add_argument("--station", metavar="ID", help="station id to select") parser.add_argument( @@ -72,13 +79,15 @@ def _build_parser() -> argparse.ArgumentParser: return parser -def _list_stations(offline: bool) -> int: - stations = api.load_cached_channels() +def _list_stations(offline: bool, provider_id: str | None = None) -> int: + provider = get_provider(provider_id) + stations = provider.load_cached_stations() if not stations and not offline: + async def load(): - client = api.make_client() + client = make_client(provider.user_agent) try: - return await api.fetch_channels(client) + return await provider.fetch_stations(client) finally: await client.aclose() @@ -87,7 +96,8 @@ def _list_stations(offline: bool) -> int: except Exception: stations = [] if not stations: - stations = api.parse_channels({"channels": fallback_channels()}) + stations = provider.fallback_stations() + print(f"# {provider.name} ({provider.id}) — {provider.station_source}") for station in sorted(stations, key=lambda item: item.title.casefold()): print(f"{station.id:<20} {station.title}") return 0 @@ -103,13 +113,14 @@ def main(argv: Sequence[str] | None = None) -> int: return 0 if args.list: - return _list_stations(args.offline) + return _list_stations(args.offline, args.provider) # Import the app lazily so --list/--version do not probe the terminal for # image protocols. - from .app import SomaFMApp + from .app import BeefyPlayerApp - app = SomaFMApp( + app = BeefyPlayerApp( + provider=args.provider, autoplay=args.autoplay, station_id=args.station, volume=args.volume, diff --git a/beefy_player/app.py b/beefy_player/app.py index 6c83014..92ed34b 100644 --- a/beefy_player/app.py +++ b/beefy_player/app.py @@ -1,4 +1,10 @@ -"""The Textual application: layout, key bindings and orchestration.""" +"""The Textual application: layout, key bindings and orchestration. + +The app is deliberately catalogue-agnostic: it holds one +:class:`~beefy_player.providers.base.RadioProvider` and asks it for stations, +song history and stream URLs. SomaFM is one such provider; see +:mod:`beefy_player.providers`. +""" from __future__ import annotations @@ -17,11 +23,12 @@ from textual.css.query import NoMatches from textual.theme import Theme from textual.widgets import Input -from . import api, artwork, bandcamp, config -from .api import Station -from .fallback import fallback_channels +from . import artwork, bandcamp, config from .mediasession import create_media_session, media_actions_queue +from .models import Station, parse_last_playing from .player import Player +from .providers import get_provider +from .providers.base import is_direct_stream, make_client from .spectrum import SpectrumEngine from .widgets.header import HeaderBar from .widgets.help import HelpScreen @@ -56,11 +63,11 @@ TERMINAL_THEME = Theme( ) -class SomaFMApp(App[None]): +class BeefyPlayerApp(App[None]): CSS_PATH = "theme.tcss" ENABLE_COMMAND_PALETTE = False - TITLE = "SomaFM" - SUB_TITLE = "terminal radio" + TITLE = "Beefy Player" + SUB_TITLE = "terminal internet radio" BINDINGS: ClassVar[list[Binding]] = [ Binding("up,k", "cursor_up", "up", show=False), @@ -89,6 +96,7 @@ class SomaFMApp(App[None]): def __init__( self, *, + provider: str | None = None, autoplay: bool | None = None, station_id: str | None = None, volume: int | None = None, @@ -104,6 +112,8 @@ class SomaFMApp(App[None]): self.register_theme(TERMINAL_THEME) self.theme = TERMINAL_THEME.name self.config = config.Config.load() + #: The station catalogue in use (see :mod:`beefy_player.providers`). + self.provider = get_provider(provider or self.config.provider) # One-off command line overrides are deliberately *not* written back # into the config file (only an explicit --volume is remembered). self.autoplay = self.config.autoplay if autoplay is None else autoplay @@ -177,16 +187,16 @@ class SomaFMApp(App[None]): self.spectrum.start() self.media.start() - self._http = api.make_client() + self._http = make_client(self.provider.user_agent) - stations = api.load_cached_channels() + stations = self.provider.load_cached_stations() if not stations: - stations = api.parse_channels({"channels": fallback_channels()}) + stations = self.provider.fallback_stations() self._set_stations(stations) self._select_initial_station() if self.current is not None: - cached = api.load_cached_songs(self.current.id) + cached = self.provider.load_cached_history(self.current.id) if cached: self.current.history = cached self.current.now_playing = cached[0] @@ -195,7 +205,7 @@ class SomaFMApp(App[None]): self._tick_timer = self.set_interval(0.05, self._tick) if not self.offline: - self._refresh_channels() + self._refresh_stations() if self.current is not None: self._fetch_history(self.current.id) @@ -238,7 +248,7 @@ class SomaFMApp(App[None]): station.favorite = previous.favorite station.favorite = self.config.is_favorite(station.id) if station.now_playing is None and station.last_playing: - station.now_playing = api.parse_last_playing(station.last_playing) + station.now_playing = parse_last_playing(station.last_playing) self.stations = stations wanted = self.current.id if self.current is not None else self.start_station_id @@ -337,10 +347,10 @@ class SomaFMApp(App[None]): return if self.offline or self._http is None: - urls = station.synthetic_mirrors() + urls = self.provider.fallback_urls(station) else: try: - urls = await api.stream_candidates(self._http, station) + urls = await self.provider.stream_candidates(self._http, station) except Exception as exc: self.playing_state = "error" self._set_notice(f"could not resolve stream: {exc}", 8) @@ -357,7 +367,7 @@ class SomaFMApp(App[None]): self.query_one(StagePanel).refresh_stage() return - direct = next((url for url in urls if api.is_direct_stream(url)), None) + direct = next((url for url in urls if is_direct_stream(url)), None) self._stream_url = direct or urls[0] if direct is not None: self.spectrum.set_source(direct) @@ -371,9 +381,9 @@ class SomaFMApp(App[None]): if station is None or self._http is None: return try: - songs = await api.fetch_history(self._http, station_id) + songs = await self.provider.fetch_history(self._http, station_id) except Exception: - songs = api.load_cached_songs(station_id) + songs = self.provider.load_cached_history(station_id) if not songs: return @@ -408,9 +418,9 @@ class SomaFMApp(App[None]): return song = station.now_playing self.title = ( - f"SomaFM — {song.display_artist} · {song.display_title}" + f"{self.provider.name} — {song.display_artist} · {song.display_title}" if song is not None - else "SomaFM" + else self.provider.name ) if song is not None: self.media.set_track( @@ -426,7 +436,7 @@ class SomaFMApp(App[None]): self.media.set_track( station=station.title, title=station.last_playing or station.title, - artist=station.dj or "SomaFM", + artist=station.dj or self.provider.name, album=station.description, art_url=station.image, stream_url=self._stream_url, @@ -687,17 +697,15 @@ class SomaFMApp(App[None]): # workers # ------------------------------------------------------------------ # - @work(exclusive=True, group="channels", exit_on_error=False) - async def _refresh_channels(self) -> None: + @work(exclusive=True, group="stations", exit_on_error=False) + async def _refresh_stations(self) -> None: if self.offline or self._http is None: return try: - stations = await api.fetch_channels(self._http) + stations = await self.provider.fetch_stations(self._http) except Exception: if not self.stations: - self._set_stations( - api.parse_channels({"channels": fallback_channels()}) - ) + self._set_stations(self.provider.fallback_stations()) self._set_notice("offline — using the built-in station list", 8) else: self._set_notice("station refresh failed — using the cached list", 6) @@ -848,7 +856,7 @@ class SomaFMApp(App[None]): self._set_notice("offline mode — nothing to refresh", 3) return self.artwork_url = "" - self._refresh_channels() + self._refresh_stations() if self.current is not None: self._update_artwork() self._fetch_history(self.current.id) diff --git a/beefy_player/models.py b/beefy_player/models.py new file mode 100644 index 0000000..a398808 --- /dev/null +++ b/beefy_player/models.py @@ -0,0 +1,171 @@ +"""Provider-neutral data types. + +Every radio catalogue — SomaFM today, anything else later — is reduced to +:class:`Station` and :class:`Song` before the UI sees it, so the widgets and +the app never have to know which catalogue a station came from. The parsers +at the bottom are the small, boring helpers (numbers, dates, "Artist - Title" +strings) that every provider ends up needing. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any + +_QUALITY_RANK = {"highest": 0, "high": 1, "low": 2} +_FORMAT_RANK = {"mp3": 0, "aac": 1, "aacp": 2} +_FORMAT_LABEL = {"aacp": "aac+", "aac": "aac", "mp3": "mp3"} + + +@dataclass(slots=True) +class Song: + title: str = "" + artist: str = "" + album: str = "" + art_url: str = "" + bandcamp_url: str = "" + played_at: datetime | None = None + + @property + def display_title(self) -> str: + return self.title.strip() or "Unknown track" + + @property + def display_artist(self) -> str: + return self.artist.strip() or "Unknown artist" + + def key(self) -> str: + return f"{self.artist}\x1f{self.title}\x1f{self.album}" + + +@dataclass(slots=True) +class Station: + id: str + title: str + description: str = "" + genre: str = "" + dj: str = "" + image: str = "" + listeners: int = 0 + last_playing: str = "" + playlists: list[dict[str, str]] = field(default_factory=list) + now_playing: Song | None = None + history: list[Song] = field(default_factory=list) + favorite: bool = False + #: Id of the provider that produced this station (see + #: :mod:`beefy_player.providers`). Station ids are only unique within a + #: catalogue, so anything persisted across providers should use + #: :attr:`uid` instead. + provider: str = "" + + @property + def uid(self) -> str: + """Provider-qualified id, stable once more than one catalogue exists.""" + return f"{self.provider}:{self.id}" if self.provider else self.id + + @property + def genre_label(self) -> str: + return self.genre.replace("|", " · ") or "eclectic" + + def sorted_playlists(self) -> list[dict[str, str]]: + return sorted( + self.playlists, + key=lambda pl: ( + _QUALITY_RANK.get(pl.get("quality", ""), 3), + _FORMAT_RANK.get(pl.get("format", ""), 3), + ), + ) + + @property + def format_label(self) -> str: + playlists = self.sorted_playlists() + if not playlists: + return "stream" + first = playlists[0] + fmt = _FORMAT_LABEL.get( + first.get("format", ""), first.get("format") or "stream" + ) + return f"{fmt} · {guess_bitrate(first.get('url', ''))}k" + + +# --------------------------------------------------------------------------- # +# small, provider-agnostic parsers +# --------------------------------------------------------------------------- # + + +def guess_bitrate(url: str) -> int: + match = re.search( + r"(256|192|130|128|96|64|48|32)(?=\.pls|\.m3u|-mp3|-aac)", url + ) + if not match: + return 128 + value = int(match.group(1)) + return 128 if value == 130 else value + + +def to_int(value: Any, default: int = 0) -> int: + """Forgiving ``int`` — catalogues like to send ``"1,873"`` as text.""" + if isinstance(value, bool): + return default + if isinstance(value, int): + return value + if isinstance(value, float): + return int(value) + if isinstance(value, str): + digits = re.sub(r"[^0-9]", "", value) + if digits: + try: + return int(digits) + except ValueError: + return default + return default + + +def parse_date(value: Any) -> datetime | None: + """Unix timestamp, ISO 8601 or one of the usual ``%Y-%m-%d`` shapes.""" + if value is None or value == "": + return None + if isinstance(value, (int, float)): + try: + return datetime.fromtimestamp(float(value), tz=timezone.utc) + except (OverflowError, OSError, ValueError): + return None + if isinstance(value, str): + text = value.strip() + if text.isdigit(): + try: + return datetime.fromtimestamp(int(text), tz=timezone.utc) + except (OverflowError, OSError, ValueError): + return None + try: + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + except ValueError: + parsed = None + for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y/%m/%d %H:%M:%S"): + try: + parsed = datetime.strptime(text, fmt) + break + except ValueError: + continue + if parsed is not None: + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + return None + + +_SEPARATORS = (" - ", " – ", " — ", " -- ") + + +def parse_last_playing(text: str | None) -> Song | None: + """Split the ``"Artist - Title"`` blurb most catalogues send.""" + if not isinstance(text, str): + return None + cleaned = text.strip() + if not cleaned: + return None + for separator in _SEPARATORS: + if separator in cleaned: + artist, title = cleaned.split(separator, 1) + return Song(title=title.strip(), artist=artist.strip()) + return Song(title=cleaned) diff --git a/beefy_player/providers/__init__.py b/beefy_player/providers/__init__.py new file mode 100644 index 0000000..e25de24 --- /dev/null +++ b/beefy_player/providers/__init__.py @@ -0,0 +1,53 @@ +"""The catalogue registry. + +Beefy Player speaks to internet radio through :class:`RadioProvider` +implementations (see :mod:`beefy_player.providers.base`). Each provider is +registered here once; the app looks its provider up by slug, and ``--provider`` +on the command line is validated against :func:`available_providers`. + +Adding a catalogue is meant to be a small, contained change: + +1. write the package (``beefy_player/providers//``) with a + :class:`~beefy_player.providers.base.RadioProvider` subclass, +2. add one instance to :data:`PROVIDERS` below, +3. that is it — the UI, the player, the spectrum and the media session do not + care where the stations came from. +""" + +from __future__ import annotations + +from .base import RadioProvider +from .somafm import SomaFMProvider + +__all__ = [ + "DEFAULT_PROVIDER", + "PROVIDERS", + "RadioProvider", + "available_providers", + "get_provider", +] + +#: Every catalogue the app can talk to, keyed by its slug. +PROVIDERS: dict[str, RadioProvider] = { + provider.id: provider for provider in (SomaFMProvider(),) +} + +#: Provider used when nothing else is configured. +DEFAULT_PROVIDER = SomaFMProvider.id + + +def available_providers() -> tuple[str, ...]: + """Registered provider slugs, for ``--provider`` and the config file.""" + return tuple(PROVIDERS) + + +def get_provider(provider_id: str | None = None) -> RadioProvider: + """Look a provider up by slug, falling back to the default one. + + An unknown slug is a stale config file or a typo, not a reason to refuse to + start; the help screen and ``--list`` always name what is in use. + """ + key = (provider_id or "").strip().lower() + if key in PROVIDERS: + return PROVIDERS[key] + return PROVIDERS[DEFAULT_PROVIDER] diff --git a/beefy_player/providers/base.py b/beefy_player/providers/base.py new file mode 100644 index 0000000..a8e991e --- /dev/null +++ b/beefy_player/providers/base.py @@ -0,0 +1,249 @@ +"""The provider seam: what a radio catalogue has to implement. + +Beefy Player is an internet-radio player. Everything the UI needs from a +catalogue — the station list, the song history, the stream URLs — goes through +:class:`RadioProvider`, and the app only ever holds one of those. SomaFM +(:mod:`beefy_player.providers.somafm`) is the first implementation; adding +another station catalogue means writing another subclass and registering it in +:mod:`beefy_player.providers`, with no changes to the app or the widgets. + +This module also carries the toolkit a provider needs in practice: an HTTP +client factory and playlist handling (``.pls``/``.m3u`` resolution with an +on-disk cache), which are the same for every catalogue. +""" + +from __future__ import annotations + +import asyncio +import re +import time +from abc import ABC, abstractmethod +from typing import Any +from urllib.parse import quote + +import httpx + +from .. import config +from ..models import Song, Station + +__all__ = [ + "PLAYLIST_TTL", + "RadioProvider", + "fetch_json", + "is_direct_stream", + "is_playlist", + "make_client", + "parse_pls", + "quote_station_id", + "reset_playlist_cache", + "resolve_playlist", +] + +REQUEST_TIMEOUT = httpx.Timeout(15.0, connect=8.0) +PLAYLIST_TTL = 8 * 3600 + +_PLAYLIST_CACHE: dict[str, dict[str, Any]] | None = None + + +def make_client(user_agent: str = "") -> httpx.AsyncClient: + return httpx.AsyncClient( + headers={ + "User-Agent": user_agent or config.USER_AGENT, + "Accept": "application/json, */*", + }, + timeout=REQUEST_TIMEOUT, + follow_redirects=True, + ) + + +def reset_playlist_cache() -> None: + """Forget the in-memory playlist cache (the tests use this).""" + global _PLAYLIST_CACHE + _PLAYLIST_CACHE = None + + +# --------------------------------------------------------------------------- # +# the interface +# --------------------------------------------------------------------------- # + + +class RadioProvider(ABC): + """A station catalogue plus its streams and song history. + + Providers are stateless singletons registered in + :data:`beefy_player.providers.PROVIDERS`; the app keeps one for its whole + lifetime and never instantiates another. + """ + + #: Stable slug, used in caches, config and ``--provider``. + id: str = "" + #: Display name, e.g. ``"SomaFM"``. + name: str = "" + #: Bare domain, for docs and the help screen. + homepage: str = "" + #: Brand shown above the station list before anything is tuned in. + brand: str = "" + #: One-liner under the brand. + tagline: str = "" + #: Where stations come from, e.g. ``"somafm.com/channels.json"``. + station_source: str = "" + #: Where the song history comes from, e.g. ``"songs/.json"``. + history_source: str = "" + #: ``User-Agent`` for this catalogue's requests (SomaFM likes to know). + user_agent: str = "" + + # -- catalogue ------------------------------------------------------- # + + @abstractmethod + async def fetch_stations(self, client: httpx.AsyncClient) -> list[Station]: + """Fetch the live station list and refresh the on-disk cache.""" + + @abstractmethod + def load_cached_stations(self) -> list[Station]: + """The last good station list from disk, or ``[]``.""" + + @abstractmethod + def fallback_stations(self) -> list[Station]: + """A bundled list, so a fresh offline start still works.""" + + # -- metadata -------------------------------------------------------- # + + @abstractmethod + async def fetch_history( + self, client: httpx.AsyncClient, station_id: str + ) -> list[Song]: + """Recently played songs, newest first (falls back to the cache).""" + + @abstractmethod + def load_cached_history(self, station_id: str) -> list[Song]: + """The last good song history from disk, or ``[]``.""" + + # -- streams --------------------------------------------------------- # + + def fallback_urls(self, station: Station) -> list[str]: + """Guessed stream URLs, used when the catalogue is unreachable.""" + return [] + + async def stream_candidates( + self, client: httpx.AsyncClient, station: Station + ) -> list[str]: + """Direct stream URLs to try, best first. + + Resolves the station's own playlists (a couple of ``.pls`` files at + most) and appends :meth:`fallback_urls` in case none of them answer. + """ + urls: list[str] = [] + playlists = station.sorted_playlists() + + for entry in playlists[:3]: + url = entry.get("url", "") + resolved = await resolve_playlist(client, url) + for candidate in resolved: + if candidate not in urls: + urls.append(candidate) + if not resolved and url and url not in urls: + urls.append(url) + + for entry in playlists[3:]: + url = entry.get("url", "") + if url and url not in urls: + urls.append(url) + + for url in self.fallback_urls(station): + if url not in urls: + urls.append(url) + + return urls[:10] + + +# --------------------------------------------------------------------------- # +# playlists +# --------------------------------------------------------------------------- # + + +def is_playlist(url: str) -> bool: + path = url.split("?", 1)[0].split("#", 1)[0].lower() + return path.endswith((".pls", ".m3u", ".m3u8")) + + +def is_direct_stream(url: str) -> bool: + return url.startswith(("http://", "https://")) and not is_playlist(url) + + +def parse_pls(text: str) -> list[str]: + urls: list[str] = [] + for line in text.splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + match = re.match(r"(?i)^file\d+\s*=\s*(\S+)$", line) + candidate = match.group(1) if match else line + if candidate.startswith(("http://", "https://")) and candidate not in urls: + urls.append(candidate) + return urls + + +def _playlist_cache() -> dict[str, dict[str, Any]]: + global _PLAYLIST_CACHE + if _PLAYLIST_CACHE is None: + raw = config.read_json(config.PLAYLIST_CACHE) + _PLAYLIST_CACHE = raw if isinstance(raw, dict) else {} + return _PLAYLIST_CACHE + + +async def resolve_playlist(client: httpx.AsyncClient, url: str) -> list[str]: + """Turn a ``.pls`` URL into direct icecast URLs (cached for a few hours).""" + if not url: + return [] + if not is_playlist(url): + return [url] + + cache = _playlist_cache() + now = time.time() + entry = cache.get(url) + if isinstance(entry, dict): + fetched = entry.get("ts") + urls = entry.get("urls") + if ( + isinstance(fetched, (int, float)) + and now - fetched < PLAYLIST_TTL + and isinstance(urls, list) + ): + cached_urls = [item for item in urls if isinstance(item, str)] + if cached_urls: + return cached_urls + + try: + response = await client.get(url, headers={"Accept": "*/*"}) + response.raise_for_status() + text = response.text + except httpx.HTTPError: + return [] + + urls = parse_pls(text) + if urls: + cache[url] = {"ts": now, "urls": urls} + config.write_json(config.PLAYLIST_CACHE, cache) + return urls + + +def quote_station_id(station_id: str) -> str: + """Percent-encode a station id for use in a provider URL.""" + return quote(station_id, safe="") + + +async def fetch_json( + client: httpx.AsyncClient, url: str, *, attempts: int = 2, pause: float = 0.6 +) -> Any: + """``GET url`` and decode JSON, retrying once: flaky mirrors are normal.""" + last_error: Exception | None = None + for attempt in range(attempts): + try: + response = await client.get(url) + response.raise_for_status() + return response.json() + except (httpx.HTTPError, ValueError) as exc: # includes JSON errors + last_error = exc + if attempt + 1 < attempts: + await asyncio.sleep(pause) + raise RuntimeError(f"could not fetch {url}: {last_error}") diff --git a/beefy_player/providers/somafm/__init__.py b/beefy_player/providers/somafm/__init__.py new file mode 100644 index 0000000..cdf1d48 --- /dev/null +++ b/beefy_player/providers/somafm/__init__.py @@ -0,0 +1,55 @@ +"""SomaFM: the station catalogue Beefy Player was born with. + +Everything SomaFM-specific lives in this package — the ``channels.json`` +schema, the ``songs/.json`` history, the ``.pls`` layout and the +guessed icecast mirrors. The rest of the app only sees +:class:`SomaFMProvider`, which implements +:class:`beefy_player.providers.base.RadioProvider`. +""" + +from __future__ import annotations + +import httpx + +from ... import config +from ...models import Song, Station +from ..base import RadioProvider +from . import api, fallback + +__all__ = ["SomaFMProvider"] + + +class SomaFMProvider(RadioProvider): + id = "somafm" + name = "SomaFM" + homepage = "somafm.com" + brand = "SOMA FM" + tagline = "listener-supported radio · san francisco" + station_source = "somafm.com/channels.json" + history_source = "songs/.json" + + @property + def user_agent(self) -> str: + # SomaFM asks clients to identify themselves, so say who we are and + # where we come from. + return f"{config.USER_AGENT}; +https://somafm.com" + + async def fetch_stations(self, client: httpx.AsyncClient) -> list[Station]: + return await api.fetch_channels(client) + + def load_cached_stations(self) -> list[Station]: + return api.load_cached_channels() + + def fallback_stations(self) -> list[Station]: + return api.parse_channels({"channels": fallback.fallback_channels()}) + + async def fetch_history( + self, client: httpx.AsyncClient, station_id: str + ) -> list[Song]: + return await api.fetch_history(client, station_id) + + def load_cached_history(self, station_id: str) -> list[Song]: + return api.load_cached_songs(station_id) + + def fallback_urls(self, station: Station) -> list[str]: + return api.synthetic_mirrors(station.id) diff --git a/beefy_player/providers/somafm/api.py b/beefy_player/providers/somafm/api.py index 46b4dd5..79cc248 100644 --- a/beefy_player/providers/somafm/api.py +++ b/beefy_player/providers/somafm/api.py @@ -1,197 +1,30 @@ -"""Async client for the public SomaFM endpoints. - -The app talks to three things: +"""The public SomaFM endpoints, and nothing but. * ``https://somafm.com/channels.json`` -> every station + listener counts * ``https://somafm.com/.pls`` -> tiny PLS playlists (resolved to ice URLs) * ``https://somafm.com/songs/.json`` -> the "last played" song history All parsing is deliberately forgiving: SomaFM has changed field names before -and we would rather show a slightly degraded station than crash. +and we would rather show a slightly degraded station than crash. Nothing +outside :mod:`beefy_player.providers.somafm` should import this module — the +app talks to :class:`beefy_player.providers.somafm.SomaFMProvider` instead. """ from __future__ import annotations -import asyncio import re import time from collections.abc import Iterable -from dataclasses import dataclass, field -from datetime import datetime, timezone from typing import Any -from urllib.parse import quote import httpx -from . import config +from ... import config +from ...models import Song, Station, parse_date, parse_last_playing, to_int +from ..base import fetch_json, quote_station_id CHANNELS_URL = "https://somafm.com/channels.json" SONGS_URL = "https://somafm.com/songs/{station}.json" -USER_AGENT = "somafm-tui/0.1 (terminal radio; +https://somafm.com)" -REQUEST_TIMEOUT = httpx.Timeout(15.0, connect=8.0) -PLAYLIST_TTL = 8 * 3600 - -_QUALITY_RANK = {"highest": 0, "high": 1, "low": 2} -_FORMAT_RANK = {"mp3": 0, "aac": 1, "aacp": 2} - -_PLAYLIST_CACHE: dict[str, dict[str, Any]] | None = None - - -def make_client() -> httpx.AsyncClient: - return httpx.AsyncClient( - headers={"User-Agent": USER_AGENT, "Accept": "application/json, */*"}, - timeout=REQUEST_TIMEOUT, - follow_redirects=True, - ) - - -# --------------------------------------------------------------------------- # -# models -# --------------------------------------------------------------------------- # - - -@dataclass(slots=True) -class Song: - title: str = "" - artist: str = "" - album: str = "" - art_url: str = "" - bandcamp_url: str = "" - played_at: datetime | None = None - - @property - def display_title(self) -> str: - return self.title.strip() or "Unknown track" - - @property - def display_artist(self) -> str: - return self.artist.strip() or "Unknown artist" - - def key(self) -> str: - return f"{self.artist}\x1f{self.title}\x1f{self.album}" - - -@dataclass(slots=True) -class Station: - id: str - title: str - description: str = "" - genre: str = "" - dj: str = "" - image: str = "" - listeners: int = 0 - last_playing: str = "" - playlists: list[dict[str, str]] = field(default_factory=list) - now_playing: Song | None = None - history: list[Song] = field(default_factory=list) - favorite: bool = False - - @property - def genre_label(self) -> str: - return self.genre.replace("|", " · ") or "eclectic" - - def sorted_playlists(self) -> list[dict[str, str]]: - return sorted( - self.playlists, - key=lambda pl: ( - _QUALITY_RANK.get(pl.get("quality", ""), 3), - _FORMAT_RANK.get(pl.get("format", ""), 3), - ), - ) - - @property - def format_label(self) -> str: - playlists = self.sorted_playlists() - if not playlists: - return "stream" - first = playlists[0] - fmt = {"aacp": "aac+", "aac": "aac", "mp3": "mp3"}.get( - first.get("format", ""), first.get("format") or "stream" - ) - return f"{fmt} · {_guess_bitrate(first.get('url', ''))}k" - - def synthetic_mirrors(self) -> list[str]: - """Direct icecast URLs, handy when the PLS files are unreachable.""" - urls: list[str] = [] - for suffix in ("-128-mp3", "-64-aac"): - for host in ("ice1", "ice2", "ice3", "ice4"): - urls.append(f"https://{host}.somafm.com/{self.id}{suffix}") - return urls - - -# --------------------------------------------------------------------------- # -# small helpers -# --------------------------------------------------------------------------- # - - -def _guess_bitrate(url: str) -> int: - match = re.search( - r"(256|192|130|128|96|64|48|32)(?=\.pls|\.m3u|-mp3|-aac)", url - ) - if not match: - return 128 - value = int(match.group(1)) - return 128 if value == 130 else value - - -def _to_int(value: Any, default: int = 0) -> int: - if isinstance(value, bool): - return default - if isinstance(value, int): - return value - if isinstance(value, float): - return int(value) - if isinstance(value, str): - digits = re.sub(r"[^0-9]", "", value) - if digits: - try: - return int(digits) - except ValueError: - return default - return default - - -def parse_date(value: Any) -> datetime | None: - if value is None or value == "": - return None - if isinstance(value, (int, float)): - try: - return datetime.fromtimestamp(float(value), tz=timezone.utc) - except (OverflowError, OSError, ValueError): - return None - if isinstance(value, str): - text = value.strip() - if text.isdigit(): - try: - return datetime.fromtimestamp(int(text), tz=timezone.utc) - except (OverflowError, OSError, ValueError): - return None - try: - parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) - except ValueError: - parsed = None - for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y/%m/%d %H:%M:%S"): - try: - parsed = datetime.strptime(text, fmt) - break - except ValueError: - continue - if parsed is not None: - return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) - return None - - -def parse_last_playing(text: str | None) -> Song | None: - if not isinstance(text, str): - return None - cleaned = text.strip() - if not cleaned: - return None - for separator in (" - ", " – ", " — ", " -- "): - if separator in cleaned: - artist, title = cleaned.split(separator, 1) - return Song(title=title.strip(), artist=artist.strip()) - return Song(title=cleaned) # --------------------------------------------------------------------------- # @@ -231,8 +64,9 @@ def station_from_payload(item: dict[str, Any]) -> Station | None: genre=str(item.get("genre") or "").strip(), dj=str(item.get("dj") or "").strip(), image=str(item.get("image") or item.get("largeimage") or "").strip(), - listeners=_to_int(item.get("listeners")), + listeners=to_int(item.get("listeners")), playlists=playlists, + provider="somafm", ) last_playing = item.get("lastPlaying") if isinstance(last_playing, str): @@ -264,20 +98,10 @@ def parse_channels(payload: Any) -> list[Station]: async def fetch_channels(client: httpx.AsyncClient) -> list[Station]: - payload: Any = None - last_error: Exception | None = None - for attempt in range(2): - try: - response = await client.get(CHANNELS_URL) - response.raise_for_status() - payload = response.json() - break - except (httpx.HTTPError, ValueError) as exc: # includes JSON errors - last_error = exc - if attempt == 0: - await asyncio.sleep(0.6) - if payload is None: - raise RuntimeError(f"could not fetch SomaFM channels: {last_error}") + try: + payload = await fetch_json(client, CHANNELS_URL) + except RuntimeError as exc: + raise RuntimeError(f"could not fetch SomaFM channels: {exc}") from None stations = parse_channels(payload) if not stations: raise RuntimeError("SomaFM returned an empty channel list") @@ -350,7 +174,7 @@ def _songs_to_cache(songs: Iterable[Song]) -> list[dict[str, Any]]: async def fetch_history(client: httpx.AsyncClient, station_id: str) -> list[Song]: - url = SONGS_URL.format(station=quote(station_id, safe="")) + url = SONGS_URL.format(station=quote_station_id(station_id)) try: response = await client.get(url) response.raise_for_status() @@ -369,95 +193,14 @@ async def fetch_history(client: httpx.AsyncClient, station_id: str) -> list[Song # --------------------------------------------------------------------------- # -# playlist resolution +# streams # --------------------------------------------------------------------------- # -def is_playlist(url: str) -> bool: - path = url.split("?", 1)[0].split("#", 1)[0].lower() - return path.endswith((".pls", ".m3u", ".m3u8")) - - -def is_direct_stream(url: str) -> bool: - return url.startswith(("http://", "https://")) and not is_playlist(url) - - -def parse_pls(text: str) -> list[str]: +def synthetic_mirrors(station_id: str) -> list[str]: + """Direct icecast URLs, handy when the PLS files are unreachable.""" urls: list[str] = [] - for line in text.splitlines(): - line = line.strip() - if not line or line.startswith("#"): - continue - match = re.match(r"(?i)^file\d+\s*=\s*(\S+)$", line) - candidate = match.group(1) if match else line - if candidate.startswith(("http://", "https://")) and candidate not in urls: - urls.append(candidate) + for suffix in ("-128-mp3", "-64-aac"): + for host in ("ice1", "ice2", "ice3", "ice4"): + urls.append(f"https://{host}.somafm.com/{station_id}{suffix}") return urls - - -def _playlist_cache() -> dict[str, dict[str, Any]]: - global _PLAYLIST_CACHE - if _PLAYLIST_CACHE is None: - raw = config.read_json(config.PLAYLIST_CACHE) - _PLAYLIST_CACHE = raw if isinstance(raw, dict) else {} - return _PLAYLIST_CACHE - - -async def resolve_playlist(client: httpx.AsyncClient, url: str) -> list[str]: - """Turn a ``.pls`` URL into direct icecast URLs (cached for a few hours).""" - if not url: - return [] - if not is_playlist(url): - return [url] - - cache = _playlist_cache() - now = time.time() - entry = cache.get(url) - if isinstance(entry, dict): - fetched = entry.get("ts") - urls = entry.get("urls") - if ( - isinstance(fetched, (int, float)) - and now - fetched < PLAYLIST_TTL - and isinstance(urls, list) - ): - cached_urls = [item for item in urls if isinstance(item, str)] - if cached_urls: - return cached_urls - - try: - response = await client.get(url, headers={"Accept": "*/*"}) - response.raise_for_status() - text = response.text - except httpx.HTTPError: - return [] - - urls = parse_pls(text) - if urls: - cache[url] = {"ts": now, "urls": urls} - config.write_json(config.PLAYLIST_CACHE, cache) - return urls - - -async def stream_candidates(client: httpx.AsyncClient, station: Station) -> list[str]: - """Direct URLs first, then unresolved playlists, then guessed mirrors.""" - playlists = station.sorted_playlists() - urls: list[str] = [] - - for entry in playlists[:3]: - resolved = await resolve_playlist(client, entry["url"]) - for url in resolved: - if url not in urls: - urls.append(url) - if not resolved and entry["url"] not in urls: - urls.append(entry["url"]) - - for entry in playlists[3:]: - if entry["url"] not in urls: - urls.append(entry["url"]) - - for url in station.synthetic_mirrors(): - if url not in urls: - urls.append(url) - - return urls[:10] diff --git a/beefy_player/widgets/help.py b/beefy_player/widgets/help.py index 05b06ac..febbff8 100644 --- a/beefy_player/widgets/help.py +++ b/beefy_player/widgets/help.py @@ -4,6 +4,7 @@ from __future__ import annotations import sys import textwrap +from typing import TYPE_CHECKING, cast from rich.console import Group from rich.table import Table @@ -14,6 +15,9 @@ from textual.widgets import Static from .common import SPECTRUM_STOPS, gradient +if TYPE_CHECKING: + from ..app import BeefyPlayerApp + HELP_ROWS: tuple[tuple[str, str], ...] = ( ("↑ / ↓ j / k", "move through the station list"), ("enter", "tune in to the selected station"), @@ -35,13 +39,6 @@ HELP_ROWS: tuple[tuple[str, str], ...] = ( ("q", "quit"), ) -ABOUT_COMMON = ( - ( - "Stations and song history come from somafm.com " - "(channels.json + songs/.json)." - ), -) - ABOUT_LINUX = ( ( "The spectrum captures your actual PulseAudio/PipeWire output with " @@ -50,7 +47,7 @@ ABOUT_LINUX = ( ), ( "The desktop picks us up over MPRIS as a " - "somafmtui media player — try `playerctl -l` / `playerctl metadata`." + "beefyplayer media player — try `playerctl -l` / `playerctl metadata`." ), ) @@ -68,11 +65,15 @@ ABOUT_MACOS = ( ), ) -ABOUT = ABOUT_COMMON + (ABOUT_MACOS if sys.platform == "darwin" else ABOUT_LINUX) - class HelpScreen(ModalScreen[None]): def compose(self) -> ComposeResult: + provider = cast("BeefyPlayerApp", self.app).provider + about = ( + f"Stations and song history come from {provider.name} " + f"({provider.station_source} + {provider.history_source}).", + ) + (ABOUT_MACOS if sys.platform == "darwin" else ABOUT_LINUX) + table = Table.grid(padding=(0, 2), expand=False) table.add_column(style="bold cyan", no_wrap=True, justify="right") table.add_column(style="") @@ -81,11 +82,11 @@ class HelpScreen(ModalScreen[None]): header = Text() header.append("◤ ", style=f"bold {gradient(SPECTRUM_STOPS, 0.9)}") - header.append("SOMAFM TUI", style=f"bold {gradient(SPECTRUM_STOPS, 0.55)}") + header.append("BEEFY PLAYER", style=f"bold {gradient(SPECTRUM_STOPS, 0.55)}") header.append(" keyboard & credits", style="dim") footer = Text() - for paragraph in ABOUT: + for paragraph in about: for line in textwrap.wrap(paragraph, width=70): footer.append(line + "\n", style="dim") diff --git a/beefy_player/widgets/stage.py b/beefy_player/widgets/stage.py index 5f552ca..2dd609f 100644 --- a/beefy_player/widgets/stage.py +++ b/beefy_player/widgets/stage.py @@ -11,7 +11,7 @@ from textual.containers import Horizontal from textual.css.query import NoMatches from textual.widget import Widget -from ..api import Song, Station +from ..models import Song, Station from .art import ArtPane from .common import ( compact_count, @@ -23,7 +23,7 @@ from .common import ( ) if TYPE_CHECKING: - from ..app import SomaFMApp + from ..app import BeefyPlayerApp class StagePanel(Widget): @@ -75,27 +75,30 @@ class StageInfo(Widget): can_focus = False def render(self) -> RenderableType: - app = cast("SomaFMApp", self.app) + app = cast("BeefyPlayerApp", self.app) width = max(8, self.size.width) height = max(1, self.size.height) station = app.current if station is None: - return Group(*self._splash(width, height)) + return Group(*self._splash(app, width, height)) details = self._details(app, station, station.now_playing, width) return Group(*details[:height]) - def _splash(self, width: int, height: int) -> list[Text]: + def _splash( + self, app: BeefyPlayerApp, width: int, height: int + ) -> list[Text]: + provider = app.provider lines = [ - Text("◤ SOMA FM ◢", style="bold magenta"), - Text("listener-supported radio · san francisco", style="italic dim"), + Text(f"◤ {provider.brand} ◢", style="bold magenta"), + Text(provider.tagline, style="italic dim"), Text(""), Text("↑ ↓ choose a station", style=""), Text("⏎ tune in", style=""), Text("? all keyboard shortcuts", style=""), Text(""), - Text("Stations load from somafm.com/channels.json", style="dim"), + Text(f"Stations load from {provider.station_source}", style="dim"), ] return [ Text(truncate(line.plain, width), style=line.style) @@ -104,7 +107,7 @@ class StageInfo(Widget): def _details( self, - app: SomaFMApp, + app: BeefyPlayerApp, station: Station, song: Song | None, width: int, @@ -188,7 +191,7 @@ class HistoryPane(Widget): can_focus = False def render(self) -> RenderableType: - app = cast("SomaFMApp", self.app) + app = cast("BeefyPlayerApp", self.app) width = max(8, self.size.width) height = max(1, self.size.height) station = app.current diff --git a/tests/conftest.py b/tests/conftest.py index b72666e..171f1e1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,11 +2,12 @@ from __future__ import annotations import os -os.environ["SOMAFM_IMAGES"] = "0" +os.environ["BEEFY_IMAGES"] = "0" import pytest -from somafm_tui import api, config +from beefy_player import config +from beefy_player.providers import base @pytest.fixture(autouse=True) @@ -25,6 +26,8 @@ def isolated_state(tmp_path, monkeypatch): ("CONFIG_PATH", config_dir / "config.json"), ("CHANNELS_CACHE", cache_dir / "channels.json"), ("PLAYLIST_CACHE", cache_dir / "playlists.json"), + ("ARTWORK_CACHE_DIR", cache_dir / "artwork"), + ("LEGACY_CONFIG_PATH", tmp_path / "legacy" / "config.json"), ): monkeypatch.setattr(config, attr, value) @@ -33,6 +36,6 @@ def isolated_state(tmp_path, monkeypatch): path.mkdir(parents=True, exist_ok=True) monkeypatch.setattr(config, "ensure_dirs", ensure_dirs) - monkeypatch.setattr(api, "_PLAYLIST_CACHE", None) + base.reset_playlist_cache() ensure_dirs() yield diff --git a/tests/test_api.py b/tests/test_api.py index d8c5d79..53ac61c 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -3,7 +3,12 @@ from __future__ import annotations import httpx import pytest -from somafm_tui import api, config +from beefy_player import config +from beefy_player.models import parse_last_playing +from beefy_player.providers import base, get_provider +from beefy_player.providers.somafm import api + +SOMAFM = get_provider("somafm") CHANNELS_PAYLOAD = { "channels": [ @@ -50,14 +55,14 @@ def test_parse_channels() -> None: def test_parse_last_playing() -> None: - song = api.parse_last_playing("Jens Buchert - Rauschebart") + song = parse_last_playing("Jens Buchert - Rauschebart") assert song is not None and song.artist == "Jens Buchert" and song.title == "Rauschebart" - en_dash = api.parse_last_playing("Tosca – Amber November") + en_dash = parse_last_playing("Tosca – Amber November") assert en_dash is not None and en_dash.artist == "Tosca" - plain = api.parse_last_playing("Just A Title") + plain = parse_last_playing("Just A Title") assert plain is not None and plain.title == "Just A Title" and plain.artist == "" - assert api.parse_last_playing("") is None - assert api.parse_last_playing(None) is None + assert parse_last_playing("") is None + assert parse_last_playing(None) is None def test_parse_pls() -> None: @@ -67,16 +72,16 @@ File1=https://ice2.somafm.com/groovesalad-256-mp3 File2=https://ice6.somafm.com/groovesalad-256-mp3 Version=2 """ - assert api.parse_pls(text) == [ + assert base.parse_pls(text) == [ "https://ice2.somafm.com/groovesalad-256-mp3", "https://ice6.somafm.com/groovesalad-256-mp3", ] - assert api.parse_pls("https://ice1.somafm.com/x-128-mp3\n# comment\n") == [ + assert base.parse_pls("https://ice1.somafm.com/x-128-mp3\n# comment\n") == [ "https://ice1.somafm.com/x-128-mp3" ] - assert api.is_playlist("https://x/y.pls?foo=1") is True - assert api.is_direct_stream("https://ice1.somafm.com/x-128-mp3") is True - assert api.is_direct_stream("https://x/y.m3u") is False + assert base.is_playlist("https://x/y.pls?foo=1") is True + assert base.is_direct_stream("https://ice1.somafm.com/x-128-mp3") is True + assert base.is_direct_stream("https://x/y.m3u") is False def test_parse_songs_and_dates() -> None: @@ -143,7 +148,7 @@ async def test_stream_candidates_with_mock_transport() -> None: station = api.parse_channels(CHANNELS_PAYLOAD)[0] async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: - urls = await api.stream_candidates(client, station) + urls = await SOMAFM.stream_candidates(client, station) assert urls[:2] == [ "https://ice2.somafm.com/x-256-mp3", @@ -152,7 +157,7 @@ async def test_stream_candidates_with_mock_transport() -> None: assert len(requests) == 3 # three playlist URLs, cached by URL # second run hits the playlist cache async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: - await api.stream_candidates(client, station) + await SOMAFM.stream_candidates(client, station) assert len(requests) == 3 diff --git a/tests/test_app.py b/tests/test_app.py index 227a3d3..2d6bfda 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -6,22 +6,31 @@ from datetime import datetime, timedelta, timezone import pytest -from somafm_tui import config -from somafm_tui.api import Song -from somafm_tui.app import SomaFMApp -from somafm_tui.widgets.help import HelpScreen -from somafm_tui.widgets.stage import StagePanel -from somafm_tui.widgets.station_list import StationList +from beefy_player import config +from beefy_player.app import BeefyPlayerApp +from beefy_player.models import Song +from beefy_player.widgets.help import HelpScreen +from beefy_player.widgets.stage import StagePanel +from beefy_player.widgets.station_list import StationList -def rendered_text(app: SomaFMApp) -> str: +def _history_method(func): + """Adapt a ``(client, station_id)`` fake to the provider method signature.""" + + async def method(self, client, station_id): + return await func(client, station_id) + + return method + + +def rendered_text(app: BeefyPlayerApp) -> str: svg = html.unescape(app.export_screenshot()).replace("\xa0", " ") return re.sub(r"<[^>]+>", " ", svg) @pytest.mark.asyncio async def test_boot_navigation_filter_favourite_visualizer_help() -> None: - app = SomaFMApp(autoplay=False, offline=True, audio=False) + app = BeefyPlayerApp(autoplay=False, offline=True, audio=False) async with app.run_test(size=(110, 36)) as pilot: await pilot.pause() @@ -92,7 +101,7 @@ async def test_boot_navigation_filter_favourite_visualizer_help() -> None: @pytest.mark.asyncio async def test_stage_renders_history_and_listener_count() -> None: - app = SomaFMApp(autoplay=False, offline=True, audio=False) + app = BeefyPlayerApp(autoplay=False, offline=True, audio=False) async with app.run_test(size=(110, 40)) as pilot: await pilot.pause() station = app.current @@ -123,9 +132,22 @@ async def test_stage_renders_history_and_listener_count() -> None: assert needle in text +@pytest.mark.asyncio +async def test_splash_names_the_provider() -> None: + app = BeefyPlayerApp(autoplay=False, offline=True, audio=False) + async with app.run_test(size=(100, 32)) as pilot: + await pilot.pause() + app.current = None + app.query_one(StagePanel).refresh_stage() + await pilot.pause() + text = rendered_text(app) + assert "SOMA FM" in text + assert "somafm.com/channels.json" in text + + @pytest.mark.asyncio async def test_narrow_layout() -> None: - app = SomaFMApp(autoplay=False, offline=True, audio=False) + app = BeefyPlayerApp(autoplay=False, offline=True, audio=False) async with app.run_test(size=(64, 20)) as pilot: await pilot.pause() app.query_one(StagePanel).refresh_stage() @@ -138,7 +160,7 @@ async def test_narrow_layout() -> None: @pytest.mark.asyncio async def test_stage_shows_bandcamp_link_only_when_present() -> None: - app = SomaFMApp(autoplay=False, offline=True, audio=False) + app = BeefyPlayerApp(autoplay=False, offline=True, audio=False) async with app.run_test(size=(110, 40)) as pilot: await pilot.pause() station = app.current @@ -162,7 +184,7 @@ async def test_stage_shows_bandcamp_link_only_when_present() -> None: @pytest.mark.asyncio async def test_bandcamp_link_survives_history_refresh(monkeypatch) -> None: - app = SomaFMApp(autoplay=False, offline=True, audio=False) + app = BeefyPlayerApp(autoplay=False, offline=True, audio=False) async with app.run_test(size=(110, 40)) as pilot: await pilot.pause() station = app.current @@ -187,7 +209,9 @@ async def test_bandcamp_link_survives_history_refresh(monkeypatch) -> None: ) ] - monkeypatch.setattr("somafm_tui.app.api.fetch_history", fake_fetch_history) + monkeypatch.setattr( + type(app.provider), "fetch_history", _history_method(fake_fetch_history) + ) worker = app._fetch_history(station.id) await worker.wait() await pilot.pause() @@ -200,7 +224,7 @@ async def test_bandcamp_link_survives_history_refresh(monkeypatch) -> None: @pytest.mark.asyncio async def test_cached_bandcamp_link_is_reapplied_to_new_song() -> None: - app = SomaFMApp(autoplay=False, offline=True, audio=False) + app = BeefyPlayerApp(autoplay=False, offline=True, audio=False) async with app.run_test(size=(110, 40)) as pilot: await pilot.pause() station = app.current @@ -219,7 +243,7 @@ async def test_cached_bandcamp_link_is_reapplied_to_new_song() -> None: @pytest.mark.asyncio async def test_media_session_commands_drive_the_app() -> None: - app = SomaFMApp(autoplay=False, offline=True, audio=False) + app = BeefyPlayerApp(autoplay=False, offline=True, audio=False) async with app.run_test(size=(100, 30)) as pilot: await pilot.pause() # The status bar names whichever backend this platform uses. diff --git a/tests/test_main.py b/tests/test_main.py index 170c0ec..6398aa0 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -1,6 +1,8 @@ from __future__ import annotations -from somafm_tui.__main__ import _build_parser, _list_stations +import pytest + +from beefy_player.__main__ import _build_parser, _list_stations def test_parser_defaults_and_flags() -> None: @@ -19,6 +21,14 @@ def test_parser_defaults_and_flags() -> None: assert args.autoplay is True assert args.no_mpris is True assert args.spectrum_source == "monitor" + assert args.provider is None # the config file decides + + +def test_provider_flag_is_validated() -> None: + parser = _build_parser() + assert parser.parse_args(["--provider", "somafm"]).provider == "somafm" + with pytest.raises(SystemExit): + parser.parse_args(["--provider", "no-such-catalogue"]) def test_list_stations_offline(capsys) -> None: