From 237de2efe96f3b429c0280d8ef1468c29d27a417 Mon Sep 17 00:00:00 2001 From: beefcarver Date: Fri, 30 Oct 2026 04:17:53 +0000 Subject: [PATCH] feat(tui): implement SomaFM terminal radio player --- somafm_tui/__init__.py | 3 + somafm_tui/__main__.py | 122 +++++ somafm_tui/api.py | 462 ++++++++++++++++ somafm_tui/app.py | 818 +++++++++++++++++++++++++++++ somafm_tui/artwork.py | 237 +++++++++ somafm_tui/config.py | 123 +++++ somafm_tui/fallback.py | 96 ++++ somafm_tui/mpris.py | 420 +++++++++++++++ somafm_tui/player.py | 551 +++++++++++++++++++ somafm_tui/spectrum.py | 534 +++++++++++++++++++ somafm_tui/theme.tcss | 162 ++++++ somafm_tui/widgets/__init__.py | 1 + somafm_tui/widgets/art.py | 375 +++++++++++++ somafm_tui/widgets/common.py | 212 ++++++++ somafm_tui/widgets/header.py | 97 ++++ somafm_tui/widgets/help.py | 74 +++ somafm_tui/widgets/search.py | 14 + somafm_tui/widgets/spectrum.py | 202 +++++++ somafm_tui/widgets/stage.py | 238 +++++++++ somafm_tui/widgets/station_list.py | 106 ++++ somafm_tui/widgets/status.py | 90 ++++ 21 files changed, 4937 insertions(+) create mode 100644 somafm_tui/__init__.py create mode 100644 somafm_tui/__main__.py create mode 100644 somafm_tui/api.py create mode 100644 somafm_tui/app.py create mode 100644 somafm_tui/artwork.py create mode 100644 somafm_tui/config.py create mode 100644 somafm_tui/fallback.py create mode 100644 somafm_tui/mpris.py create mode 100644 somafm_tui/player.py create mode 100644 somafm_tui/spectrum.py create mode 100644 somafm_tui/theme.tcss create mode 100644 somafm_tui/widgets/__init__.py create mode 100644 somafm_tui/widgets/art.py create mode 100644 somafm_tui/widgets/common.py create mode 100644 somafm_tui/widgets/header.py create mode 100644 somafm_tui/widgets/help.py create mode 100644 somafm_tui/widgets/search.py create mode 100644 somafm_tui/widgets/spectrum.py create mode 100644 somafm_tui/widgets/stage.py create mode 100644 somafm_tui/widgets/station_list.py create mode 100644 somafm_tui/widgets/status.py diff --git a/somafm_tui/__init__.py b/somafm_tui/__init__.py new file mode 100644 index 0000000..f5f3989 --- /dev/null +++ b/somafm_tui/__init__.py @@ -0,0 +1,3 @@ +"""A flashy terminal radio player for somafm.com.""" + +__version__ = "0.1.0" diff --git a/somafm_tui/__main__.py b/somafm_tui/__main__.py new file mode 100644 index 0000000..ddf1433 --- /dev/null +++ b/somafm_tui/__main__.py @@ -0,0 +1,122 @@ +"""Console entry point.""" + +from __future__ import annotations + +import argparse +import asyncio +from collections.abc import Sequence + +from . import __version__, api +from .fallback import fallback_channels +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.", + ) + parser.add_argument("--station", metavar="ID", help="station id to select") + parser.add_argument( + "--play", + dest="autoplay", + action="store_true", + default=None, + help="start playing immediately", + ) + parser.add_argument( + "--no-autoplay", + dest="autoplay", + action="store_false", + help="never start playing on launch", + ) + parser.add_argument("--volume", type=int, metavar="0-100", help="initial volume") + parser.add_argument( + "--no-mpris", + action="store_true", + help="do not announce playback over MPRIS", + ) + parser.add_argument( + "--offline", + action="store_true", + help="do not touch the network (cached/bundled stations only)", + ) + parser.add_argument( + "--no-audio", + action="store_true", + help="UI only; skip mpv/ffmpeg/MPRIS (useful for demos)", + ) + parser.add_argument( + "--spectrum-source", + choices=SOURCE_MODES, + default="auto", + help="spectrum audio source: auto (monitor→stream→sim), monitor, " + "stream (muted second ffmpeg decode), or sim", + ) + parser.add_argument( + "--diagnose-images", + action="store_true", + help="print terminal/image-protocol diagnostics and exit", + ) + parser.add_argument( + "--list", + action="store_true", + help="print known station ids and exit", + ) + parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}") + return parser + + +def _list_stations(offline: bool) -> int: + stations = api.load_cached_channels() + if not stations and not offline: + async def load(): + client = api.make_client() + try: + return await api.fetch_channels(client) + finally: + await client.aclose() + + try: + stations = asyncio.run(load()) + except Exception: + stations = [] + if not stations: + stations = api.parse_channels({"channels": fallback_channels()}) + for station in sorted(stations, key=lambda item: item.title.casefold()): + print(f"{station.id:<20} {station.title}") + return 0 + + +def main(argv: Sequence[str] | None = None) -> int: + args = _build_parser().parse_args(argv) + if args.diagnose_images: + from . import artwork + + for key, value in artwork.diagnostics().items(): + print(f"{key}: {value}") + return 0 + + if args.list: + return _list_stations(args.offline) + + # Import the app lazily so --list/--version do not probe the terminal for + # image protocols. + from .app import SomaFMApp + + app = SomaFMApp( + autoplay=args.autoplay, + station_id=args.station, + volume=args.volume, + mpris=not args.no_mpris, + offline=args.offline, + audio=not args.no_audio, + spectrum_source=args.spectrum_source, + ) + app.run() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/somafm_tui/api.py b/somafm_tui/api.py new file mode 100644 index 0000000..61122c8 --- /dev/null +++ b/somafm_tui/api.py @@ -0,0 +1,462 @@ +"""Async client for the public SomaFM endpoints. + +The app talks to three things: + +* ``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. +""" + +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 + +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 = "" + 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) + + +# --------------------------------------------------------------------------- # +# channels +# --------------------------------------------------------------------------- # + + +def station_from_payload(item: dict[str, Any]) -> Station | None: + if not isinstance(item, dict): + return None + channel_id = item.get("id") + if not isinstance(channel_id, str) or not channel_id.strip(): + return None + channel_id = channel_id.strip() + + playlists: list[dict[str, str]] = [] + raw_playlists = item.get("playlists") + if isinstance(raw_playlists, list): + for entry in raw_playlists: + if not isinstance(entry, dict): + continue + url = entry.get("url") + if not isinstance(url, str) or not url.startswith(("http://", "https://")): + continue + playlists.append( + { + "url": url, + "format": str(entry.get("format") or "").strip().lower(), + "quality": str(entry.get("quality") or "").strip().lower(), + } + ) + + station = Station( + id=channel_id, + title=str(item.get("title") or channel_id).strip(), + description=str(item.get("description") or "").strip(), + 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")), + playlists=playlists, + ) + last_playing = item.get("lastPlaying") + if isinstance(last_playing, str): + station.last_playing = last_playing.strip() + station.now_playing = parse_last_playing(station.last_playing) + if not station.image: + station.image = f"https://somafm.com/logos/256/{channel_id}256.png" + return station + + +def parse_channels(payload: Any) -> list[Station]: + if isinstance(payload, dict): + items = payload.get("channels") + elif isinstance(payload, list): + items = payload + else: + items = None + if not isinstance(items, list): + return [] + stations: list[Station] = [] + seen: set[str] = set() + for item in items: + station = station_from_payload(item) + if station is None or station.id in seen: + continue + seen.add(station.id) + stations.append(station) + return stations + + +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}") + stations = parse_channels(payload) + if not stations: + raise RuntimeError("SomaFM returned an empty channel list") + config.write_json(config.CHANNELS_CACHE, payload) + return stations + + +def load_cached_channels() -> list[Station]: + return parse_channels(config.read_json(config.CHANNELS_CACHE)) + + +# --------------------------------------------------------------------------- # +# song history +# --------------------------------------------------------------------------- # + + +def parse_songs(payload: Any) -> list[Song]: + if isinstance(payload, dict): + items = payload.get("songs") + elif isinstance(payload, list): + items = payload + else: + items = None + if not isinstance(items, list): + return [] + songs: list[Song] = [] + for entry in items: + if not isinstance(entry, dict): + continue + song = Song( + title=str(entry.get("title") or "").strip(), + artist=str(entry.get("artist") or "").strip(), + album=str(entry.get("album") or "").strip(), + art_url=str( + entry.get("albumArt") or entry.get("albumart") or entry.get("art") or "" + ).strip(), + played_at=parse_date(entry.get("date")), + ) + if song.title or song.artist: + songs.append(song) + # Newest first; the upstream feed is already sorted but let's be sure. + if all(song.played_at for song in songs): + songs.sort(key=lambda song: song.played_at, reverse=True) # type: ignore[arg-type,return-value] + return songs + + +def _songs_cache_path(station_id: str): + safe = re.sub(r"[^A-Za-z0-9_.-]", "_", station_id) + return config.SONGS_CACHE_DIR / f"{safe}.json" + + +def load_cached_songs(station_id: str) -> list[Song]: + raw = config.read_json(_songs_cache_path(station_id)) + if isinstance(raw, dict): + return parse_songs(raw.get("songs")) + return [] + + +def _songs_to_cache(songs: Iterable[Song]) -> list[dict[str, Any]]: + return [ + { + "title": song.title, + "artist": song.artist, + "album": song.album, + "albumArt": song.art_url, + "date": song.played_at.isoformat() if song.played_at else "", + } + for song in songs + ] + + +async def fetch_history(client: httpx.AsyncClient, station_id: str) -> list[Song]: + url = SONGS_URL.format(station=quote(station_id, safe="")) + try: + response = await client.get(url) + response.raise_for_status() + payload = response.json() + except (httpx.HTTPError, ValueError): + return load_cached_songs(station_id) + + songs = parse_songs(payload) + if songs: + config.write_json( + _songs_cache_path(station_id), + {"fetched": time.time(), "songs": _songs_to_cache(songs)}, + ) + return songs + return load_cached_songs(station_id) + + +# --------------------------------------------------------------------------- # +# playlist resolution +# --------------------------------------------------------------------------- # + + +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 + + +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/somafm_tui/app.py b/somafm_tui/app.py new file mode 100644 index 0000000..a16f142 --- /dev/null +++ b/somafm_tui/app.py @@ -0,0 +1,818 @@ +"""The Textual application: layout, key bindings and orchestration.""" + +from __future__ import annotations + +import asyncio +import contextlib +import queue +import time +from typing import ClassVar + +import httpx +from textual import work +from textual.app import App, ComposeResult +from textual.binding import Binding +from textual.containers import Horizontal, Vertical +from textual.css.query import NoMatches +from textual.theme import Theme +from textual.widgets import Input + +from . import api, artwork, config +from .api import Station +from .fallback import fallback_channels +from .mpris import MprisService +from .player import Player +from .spectrum import SpectrumEngine +from .widgets.header import HeaderBar +from .widgets.help import HelpScreen +from .widgets.search import SearchInput +from .widgets.spectrum import SpectrumPanel +from .widgets.stage import StagePanel +from .widgets.station_list import StationList +from .widgets.status import StatusBar + +VOLUME_STEP = 5 + +# A theme made only of terminal defaults / ANSI palette entries. Textual's +# built-in themes paint an RGB background and map ANSI names to their own RGB +# palette, which breaks transparency and ignores the user's colours. +TERMINAL_THEME = Theme( + name="terminal", + primary="ansi_cyan", + secondary="ansi_blue", + accent="ansi_magenta", + warning="ansi_yellow", + error="ansi_red", + success="ansi_green", + # Leave foreground/background unset: Textual's ANSI theme generator then + # resolves them to "ansi_default" (SGR 49) instead of RGB black. + dark=True, + ansi=True, + text_alpha=1.0, + variables={ + "ansi-background": "ansi_default", + "ansi-foreground": "ansi_default", + }, +) + + +class SomaFMApp(App[None]): + CSS_PATH = "theme.tcss" + ENABLE_COMMAND_PALETTE = False + TITLE = "SomaFM" + SUB_TITLE = "terminal radio" + + BINDINGS: ClassVar[list[Binding]] = [ + Binding("up,k", "cursor_up", "up", show=False), + Binding("down,j", "cursor_down", "down", show=False), + Binding("g", "cursor_top", "top", show=False), + Binding("shift+g", "cursor_bottom", "bottom", show=False), + Binding("enter", "tune", "tune in"), + Binding("space", "toggle_play", "play/pause"), + Binding("s", "stop", "stop", show=False), + Binding("right,bracketright", "next_station", "next", show=False), + Binding("left,bracketleft", "prev_station", "previous", show=False), + Binding("plus", "volume_up", "vol +", show=False), + Binding("equals_sign", "volume_up", "vol +", show=False), + Binding("minus", "volume_down", "vol -", show=False), + Binding("underscore", "volume_down", "vol -", show=False), + Binding("m", "mute", "mute", show=False), + Binding("f", "favorite", "favourite"), + Binding("slash", "search", "filter"), + Binding("escape", "escape", "close", show=False), + Binding("v", "visualizer", "visualiser"), + Binding("r", "refresh", "refresh"), + Binding("question_mark", "help", "help"), + Binding("q", "quit", "quit"), + ] + + def __init__( + self, + *, + autoplay: bool | None = None, + station_id: str | None = None, + volume: int | None = None, + mpris: bool = True, + offline: bool = False, + audio: bool = True, + spectrum_source: str = "auto", + ) -> None: + # ansi_color=True stops Textual from converting ANSI palette entries + # to RGB; the terminal then chooses the actual colours. + super().__init__(ansi_color=True) + self.register_theme(TERMINAL_THEME) + self.theme = TERMINAL_THEME.name + self.config = config.Config.load() + # 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 + self.mpris_enabled = self.config.mpris and mpris + if volume is not None: + self.config.volume = max(0, min(100, int(volume))) + + self.start_station_id = station_id or self.config.last_station + self.offline = offline + self.audio_enabled = audio + + self.stations: list[Station] = [] + self.view: list[Station] = [] + self.index = 0 + self.query = "" + self.current: Station | None = None + self.playing_state = "stopped" + self.volume = self.config.volume + self.visualizer = self.config.visualizer + self.tick_count = 0 + self.notice = "" + + self._notice_until = 0.0 + self._volume_before_mute = self.volume or 70 + self._last_history_fetch = 0.0 + self._history_refresh_seconds = 25.0 + self._stream_url = "" + self.artwork_url = "" + self.artwork_bytes: bytes | None = None + self._art_track_key = "" + self._http: httpx.AsyncClient | None = None + self._closed = False + self._tick_timer = None + + self.player = Player() + self.spectrum = SpectrumEngine(source_mode=spectrum_source) + self.mpris_actions: queue.Queue[str] = queue.Queue() + self.mpris = MprisService(self.mpris_actions, enabled=self.mpris_enabled) + + # ------------------------------------------------------------------ # + # layout + # ------------------------------------------------------------------ # + + def compose(self) -> ComposeResult: + yield HeaderBar() + with Horizontal(id="main"): + with Vertical(id="left"): + yield SearchInput() + yield StationList() + yield StagePanel() + yield SpectrumPanel(self.spectrum) + yield StatusBar() + + def on_mount(self) -> None: + config.ensure_dirs() + + if self.audio_enabled: + self.player.start() + self.player.set_volume(self.volume) + self.spectrum.start() + self.mpris.start() + + self._http = api.make_client() + + stations = api.load_cached_channels() + if not stations: + stations = api.parse_channels({"channels": fallback_channels()}) + self._set_stations(stations) + self._select_initial_station() + + if self.current is not None: + cached = api.load_cached_songs(self.current.id) + if cached: + self.current.history = cached + self.current.now_playing = cached[0] + self._update_mpris() + + self._tick_timer = self.set_interval(0.05, self._tick) + + if not self.offline: + self._refresh_channels() + if self.current is not None: + self._fetch_history(self.current.id) + + self.set_focus(None) + + if self.autoplay and self.current is not None and self.audio_enabled: + self._tune(self.current) + elif not self.audio_enabled: + self._set_notice("headless mode — audio disabled", 6) + else: + self._set_notice("press enter to tune in", 5) + + def on_unmount(self) -> None: + self._cleanup() + + # ------------------------------------------------------------------ # + # state helpers + # ------------------------------------------------------------------ # + + def _set_notice(self, message: str, seconds: float = 4.0) -> None: + self.notice = message + self._notice_until = time.monotonic() + seconds + + def _save_config(self) -> None: + self.config.volume = self.volume + if self.current is not None: + self.config.last_station = self.current.id + self.config.visualizer = self.visualizer + self.config.save() + + def _set_stations(self, stations: list[Station]) -> None: + if not stations: + return + existing = {station.id: station for station in self.stations} + for station in stations: + previous = existing.get(station.id) + if previous is not None: + station.history = previous.history + station.now_playing = previous.now_playing + 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) + + self.stations = stations + wanted = self.current.id if self.current is not None else self.start_station_id + self.current = next((s for s in stations if s.id == wanted), None) + if self.current is None: + self.current = stations[0] + self._rebuild_view(keep_id=wanted) + self.query_one(StationList).refresh() + self.query_one(StagePanel).refresh_stage() + self._update_artwork() + self.query_one(HeaderBar).refresh() + + def _select_initial_station(self) -> None: + if not self.view: + return + target = self.start_station_id + for index, station in enumerate(self.view): + if station.id == target: + self.index = index + self.current = station + break + else: + self.index = 0 + self.current = self.view[0] + self.config.last_station = self.current.id + self.query_one(StationList).refresh() + + def _select_station_id(self, station_id: str) -> None: + self._rebuild_view(keep_id=station_id) + for index, station in enumerate(self.view): + if station.id == station_id: + self.index = index + break + + def _rebuild_view(self, keep_id: str | None = None) -> None: + terms = self.query.casefold().split() + + def matches(station: Station) -> bool: + if not terms: + return True + haystack = ( + f"{station.title} {station.genre} {station.description} " + f"{station.dj} {station.id}" + ).casefold() + return all(term in haystack for term in terms) + + if keep_id is None and self.view and 0 <= self.index < len(self.view): + keep_id = self.view[self.index].id + + view = [station for station in self.stations if matches(station)] + view.sort(key=lambda station: (0 if station.favorite else 1, station.title.casefold())) + self.view = view + + if keep_id is not None: + for index, station in enumerate(view): + if station.id == keep_id: + self.index = index + break + else: + self.index = min(self.index, max(0, len(view) - 1)) + else: + self.index = min(self.index, max(0, len(view) - 1)) + + if self.current is None and view: + self.current = view[self.index] + + # ------------------------------------------------------------------ # + # playback + # ------------------------------------------------------------------ # + + def _tune(self, station: Station) -> None: + self.current = station + self.start_station_id = station.id + self.playing_state = "loading" + self._set_notice(f"tuning in to {station.title}…", 5) + self._save_config() + self._update_mpris() + self.query_one(StagePanel).refresh_stage() + self.query_one(HeaderBar).refresh() + + if not self.audio_enabled: + self.playing_state = "stopped" + self._set_notice("audio disabled (--no-audio)", 5) + return + + self._resolve_and_play(station.id) + self._fetch_history(station.id) + + @work(exclusive=True, group="stream", exit_on_error=False) + async def _resolve_and_play(self, station_id: str) -> None: + station = next((s for s in self.stations if s.id == station_id), None) + if station is None or self.current is None or self.current.id != station_id: + return + + if self.offline or self._http is None: + urls = station.synthetic_mirrors() + else: + try: + urls = await api.stream_candidates(self._http, station) + except Exception as exc: + self.playing_state = "error" + self._set_notice(f"could not resolve stream: {exc}", 8) + self._update_mpris() + self.query_one(StagePanel).refresh_stage() + return + + if self.current is None or self.current.id != station_id: + return + if not urls: + self.playing_state = "error" + self._set_notice("no stream mirrors available", 8) + self._update_mpris() + self.query_one(StagePanel).refresh_stage() + return + + direct = next((url for url in urls if api.is_direct_stream(url)), None) + self._stream_url = direct or urls[0] + if direct is not None: + self.spectrum.set_source(direct) + self.player.set_volume(self.volume) + self.player.play(urls) + + @work(exclusive=True, group="history", exit_on_error=False) + async def _fetch_history(self, station_id: str) -> None: + self._last_history_fetch = time.monotonic() + station = next((s for s in self.stations if s.id == station_id), None) + if station is None or self._http is None: + return + try: + songs = await api.fetch_history(self._http, station_id) + except Exception: + songs = api.load_cached_songs(station_id) + if not songs: + return + + previous = station.now_playing + station.history = songs + station.now_playing = songs[0] + changed = previous is None or previous.key() != songs[0].key() + if self.current is not None and self.current.id == station_id: + if changed: + self._set_notice( + f"♪ {songs[0].display_artist} — {songs[0].display_title}", 5 + ) + self._update_mpris() + self.query_one(StagePanel).refresh_stage() + self.query_one(HeaderBar).refresh() + + def _maybe_refresh_history(self) -> None: + if self.offline or self.current is None: + return + if self.playing_state not in ("playing", "buffering", "loading"): + return + if time.monotonic() - self._last_history_fetch >= self._history_refresh_seconds: + self._fetch_history(self.current.id) + + def _update_mpris(self) -> None: + station = self.current + if station is None: + return + song = station.now_playing + self.title = ( + f"SomaFM — {song.display_artist} · {song.display_title}" + if song is not None + else "SomaFM" + ) + if song is not None: + self.mpris.set_track( + station=station.title, + title=song.display_title, + artist=song.display_artist, + album=song.album, + art_url=song.art_url or station.image, + stream_url=self._stream_url, + track_key=f"{station.id}|{song.key()}", + ) + else: + self.mpris.set_track( + station=station.title, + title=station.last_playing or station.title, + artist=station.dj or "SomaFM", + album=station.description, + art_url=station.image, + stream_url=self._stream_url, + track_key=station.id, + ) + self.mpris.set_status(self._mpris_status()) + self.mpris.set_volume(self.volume / 100.0) + self._update_artwork() + + def _refresh_art(self) -> None: + with contextlib.suppress(NoMatches): + self.query_one(StagePanel).refresh_art() + + def _update_artwork(self) -> None: + """Pick the album art / station logo and fetch it when needed.""" + station = self.current + if station is None: + changed = bool(self.artwork_url or self.artwork_bytes) + self.artwork_url = "" + self.artwork_bytes = None + self._art_track_key = "" + if changed: + self._refresh_art() + return + + song = station.now_playing + track_key = f"{station.id}|{song.key() if song else ''}" + track_changed = track_key != self._art_track_key + self._art_track_key = track_key + + url = ( + song.art_url + if song is not None and song.art_url + else station.image + ) or "" + + if not artwork.supported() or not url: + changed = bool(self.artwork_url or self.artwork_bytes) + self.artwork_url = "" + self.artwork_bytes = None + if track_changed or changed: + self._refresh_art() + return + + if url != self.artwork_url: + self.artwork_url = url + self.artwork_bytes = artwork.load_cached(url) + self._refresh_art() + if self.artwork_bytes is None and not self.offline: + self._fetch_artwork(url) + elif track_changed: + # The image can stay, but the ANSI fallback changes per track. + self._refresh_art() + + @work(exclusive=True, group="artwork", exit_on_error=False) + async def _fetch_artwork(self, url: str) -> None: + if self._http is None or not artwork.supported(): + return + data = await artwork.fetch(self._http, url) + if self.artwork_url != url: + return + self.artwork_bytes = data + if data is None: + self._set_notice("artwork unavailable — using mosaic", 5) + self._refresh_art() + + def _mpris_status(self) -> str: + if self.playing_state == "paused": + return "Paused" + if self.playing_state in ("playing", "loading", "buffering"): + return "Playing" + return "Stopped" + + def _set_volume(self, value: int) -> None: + value = max(0, min(100, int(value))) + if value == self.volume: + return + self.volume = value + self.config.volume = value + if value > 0: + self._volume_before_mute = value + self.player.set_volume(value) + self.mpris.set_volume(value / 100.0) + self._save_config() + self.query_one(StatusBar).refresh() + self.query_one(HeaderBar).refresh() + + # ------------------------------------------------------------------ # + # event loops + # ------------------------------------------------------------------ # + + def _tick(self) -> None: + if self._closed: + return + try: + self._tick_body() + except NoMatches: + # A timer can fire in the tiny window while the screen is being + # unmounted; nothing to update any more. + return + + def _tick_body(self) -> None: + self.tick_count += 1 + self._drain_player() + self._drain_mpris() + + if self.notice and time.monotonic() >= self._notice_until: + self.notice = "" + self.query_one(StatusBar).refresh() + self.query_one(HeaderBar).refresh() + + frame = self.spectrum.snapshot() + self.query_one(SpectrumPanel).sync(self.visualizer, frame.label) + + if self.tick_count % 4 == 0: + self.query_one(StagePanel).refresh_stage() + self.query_one(HeaderBar).refresh() + if self.tick_count % 12 == 0: + self.query_one(StatusBar).refresh() + self.query_one(StationList).refresh() + self._maybe_refresh_history() + + def _drain_player(self) -> None: + while True: + try: + event = self.player.events.get_nowait() + except queue.Empty: + return + + if event.kind == "volume": + if event.volume is not None and event.volume != self.volume: + self.volume = event.volume + self.config.volume = event.volume + self.mpris.set_volume(self.volume / 100.0) + self._save_config() + continue + if event.kind == "notice": + if event.detail: + self._set_notice(event.detail, 6) + continue + + self.playing_state = event.state or self.playing_state + if event.url: + self._stream_url = event.url + + if event.state in ("playing", "loading", "buffering"): + self.mpris.set_status("Playing") + if event.url: + self.spectrum.set_source(event.url) + elif event.state == "paused": + self.mpris.set_status("Paused") + self.spectrum.set_source(None) + elif event.state in ("stopped", "error"): + self.mpris.set_status("Stopped") + self.spectrum.set_source(None) + + if event.detail: + self._set_notice(event.detail, 6) + self.query_one(StagePanel).refresh_stage() + + def _drain_mpris(self) -> None: + while True: + try: + action = self.mpris_actions.get_nowait() + except queue.Empty: + return + if action == "play": + if self.playing_state == "paused": + self.player.resume() + elif self.current is not None and self.playing_state in ("stopped", "error"): + self._tune(self.current) + elif self.current is None: + self.action_tune() + elif action == "pause": + self.player.pause() + elif action == "toggle": + self.action_toggle_play() + elif action == "stop": + self.action_stop() + elif action == "next": + self.action_next_station() + elif action == "prev": + self.action_prev_station() + elif action == "quit": + self.action_quit() + elif action.startswith("volume:"): + try: + value = int(action.split(":", 1)[1]) + except ValueError: + continue + self._set_volume(value) + + # ------------------------------------------------------------------ # + # workers + # ------------------------------------------------------------------ # + + @work(exclusive=True, group="channels", exit_on_error=False) + async def _refresh_channels(self) -> None: + if self.offline or self._http is None: + return + try: + stations = await api.fetch_channels(self._http) + except Exception: + if not self.stations: + self._set_stations( + api.parse_channels({"channels": fallback_channels()}) + ) + self._set_notice("offline — using the built-in station list", 8) + else: + self._set_notice("station refresh failed — using the cached list", 6) + return + before = self.current.id if self.current is not None else None + self._set_stations(stations) + if before is not None: + self._select_station_id(before) + self._set_notice(f"{len(stations)} stations", 3) + self.query_one(StationList).refresh() + + # ------------------------------------------------------------------ # + # actions + # ------------------------------------------------------------------ # + + def action_cursor_up(self) -> None: + if not self.view: + return + self.index = max(0, self.index - 1) + self.query_one(StationList).refresh() + + def action_cursor_down(self) -> None: + if not self.view: + return + self.index = min(len(self.view) - 1, self.index + 1) + self.query_one(StationList).refresh() + + def action_cursor_top(self) -> None: + if not self.view: + return + self.index = 0 + self.query_one(StationList).refresh() + + def action_cursor_bottom(self) -> None: + if not self.view: + return + self.index = len(self.view) - 1 + self.query_one(StationList).refresh() + + def action_tune(self) -> None: + if not self.view: + return + self.index = max(0, min(self.index, len(self.view) - 1)) + self._tune(self.view[self.index]) + + def action_toggle_play(self) -> None: + if self.current is None: + self.action_tune() + return + if self.playing_state in ("stopped", "error"): + self._tune(self.current) + elif not self.audio_enabled: + self._set_notice("audio disabled (--no-audio)", 4) + else: + self.player.toggle_pause() + + def action_stop(self) -> None: + self.player.stop() + self.spectrum.set_source(None) + self.playing_state = "stopped" + self.mpris.set_status("Stopped") + self._set_notice("stopped", 2) + self.query_one(StagePanel).refresh_stage() + self.query_one(HeaderBar).refresh() + + def _step_station(self, delta: int) -> None: + if not self.view: + return + self.index = (self.index + delta) % len(self.view) + self.query_one(StationList).refresh() + self._tune(self.view[self.index]) + + def action_next_station(self) -> None: + self._step_station(1) + + def action_prev_station(self) -> None: + self._step_station(-1) + + def action_volume_up(self) -> None: + self._set_volume(self.volume + VOLUME_STEP) + + def action_volume_down(self) -> None: + self._set_volume(self.volume - VOLUME_STEP) + + def action_mute(self) -> None: + if self.volume > 0: + self._volume_before_mute = self.volume + self._set_volume(0) + else: + self._set_volume(self._volume_before_mute or 70) + + def action_favorite(self) -> None: + station = self.view[self.index] if self.view else self.current + if station is None: + return + added = self.config.toggle_favorite(station.id) + station.favorite = added + self._save_config() + self._rebuild_view(keep_id=station.id) + self._set_notice(("★ added " if added else "☆ removed ") + station.title, 3) + self.query_one(StationList).refresh() + self.query_one(StagePanel).refresh_stage() + + def action_search(self) -> None: + search = self.query_one(SearchInput) + search.disabled = False + search.add_class("visible") + search.focus() + with contextlib.suppress(Exception): + search.cursor_position = len(search.value) + + def action_escape(self) -> None: + if self.query: + self._close_search(clear=True) + else: + self._close_search(clear=False) + + def _close_search(self, clear: bool = False) -> None: + search = self.query_one(SearchInput) + search.remove_class("visible") + if clear: + self.query = "" + search.value = "" + search.disabled = True + self.set_focus(None) + self._rebuild_view() + self.query_one(StationList).set_filter(self.query) + self.query_one(StationList).refresh() + self.query_one(StagePanel).refresh_stage() + + def action_visualizer(self) -> None: + modes = list(config.VISUALIZERS) + if self.visualizer in modes: + self.visualizer = modes[(modes.index(self.visualizer) + 1) % len(modes)] + else: + self.visualizer = "both" + self.config.visualizer = self.visualizer + self._save_config() + self._set_notice(f"visualiser: {self.visualizer}", 2) + self.query_one(SpectrumPanel).refresh() + + def action_refresh(self) -> None: + self._last_history_fetch = 0.0 + self._set_notice("refreshing…", 2) + if self.offline: + self._set_notice("offline mode — nothing to refresh", 3) + return + self.artwork_url = "" + self._refresh_channels() + if self.current is not None: + self._update_artwork() + self._fetch_history(self.current.id) + + def action_help(self) -> None: + self.push_screen(HelpScreen()) + + def action_quit(self) -> None: + self._cleanup() + self.exit() + + # ------------------------------------------------------------------ # + # input widget events + # ------------------------------------------------------------------ # + + def on_input_changed(self, event: Input.Changed) -> None: + if event.input.id != "search": + return + self.query = event.value + self._rebuild_view() + self.query_one(StationList).set_filter(self.query) + self.query_one(StationList).refresh() + + def on_input_submitted(self, event: Input.Submitted) -> None: + if event.input.id != "search": + return + station = self.view[self.index] if self.view else None + self._close_search(clear=True) + if station is not None: + self._tune(station) + + # ------------------------------------------------------------------ # + # shutdown + # ------------------------------------------------------------------ # + + def _cleanup(self) -> None: + if self._closed: + return + self._closed = True + if self._tick_timer is not None: + with contextlib.suppress(Exception): + self._tick_timer.stop() + self._save_config() + with contextlib.suppress(Exception): + self.mpris.stop() + with contextlib.suppress(Exception): + self.spectrum.stop() + with contextlib.suppress(Exception): + self.player.shutdown() + if self._http is not None: + with contextlib.suppress(RuntimeError): + asyncio.get_running_loop().create_task(self._http.aclose()) diff --git a/somafm_tui/artwork.py b/somafm_tui/artwork.py new file mode 100644 index 0000000..91870e6 --- /dev/null +++ b/somafm_tui/artwork.py @@ -0,0 +1,237 @@ +"""Terminal artwork support. + +The TUI draws a procedural ANSI mosaic when the terminal cannot display real +images. On terminals with a graphics protocol we fetch the SomaFM album art +(or channel logo) and render it directly, with the mosaic kept behind it: + +* Kitty -> Terminal Graphics Protocol (direct image placement) +* iTerm2 / WezTerm -> native iTerm2 inline images (OSC 1337) + +The protocols are implemented directly; there is no terminal capability probe +and no extra terminal input, so nothing leaks into the shell at exit. +Set ``SOMAFM_IMAGES=0`` to force the ANSI fallback, or ``SOMAFM_IMAGES=tgp`` / +``SOMAFM_IMAGES=iterm2`` to force a protocol. +""" + +from __future__ import annotations + +import hashlib +import os +import shutil +import subprocess +from collections.abc import Mapping +from io import BytesIO + +import httpx + +from . import config + +_OFF_VALUES = {"0", "off", "no", "none", "false"} +_PROTOCOL_VALUES = {"tgp", "iterm2"} + + +def detect_protocol(environ: Mapping[str, str] | None = None) -> str | None: + """Return ``"tgp"``, ``"iterm2"`` or ``None`` for the current terminal.""" + env = os.environ if environ is None else environ + if "NO_COLOR" in env: + return None + override = (env.get("SOMAFM_IMAGES") or "").strip().lower() + if override in _OFF_VALUES: + return None + if override in _PROTOCOL_VALUES: + return override + + term = (env.get("TERM") or "").lower() + program = (env.get("TERM_PROGRAM") or "").lower() + + if ( + env.get("KITTY_WINDOW_ID") + or env.get("KITTY_PID") + or "kitty" in term + or "kitty" in program + ): + # Kitty can receive TGP through tmux passthrough; we wrap the + # sequences for tmux. The mosaic remains behind the image if the + # multiplexer blocks graphics. + return "tgp" + + # Kitty inside tmux may not inherit TERM=xterm-kitty/KITTY_*. Ask tmux + # what the outer terminal is; TGP is wrapped for tmux passthrough. + if env.get("TMUX") and shutil.which("tmux"): + try: + result = subprocess.run( + ["tmux", "display-message", "-p", "#{client_termname}"], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + timeout=1.0, + check=False, + ) + if "kitty" in result.stdout.lower(): + return "tgp" + except (OSError, subprocess.SubprocessError): + pass + + # iTerm2/WezTerm inside tmux would need their own passthrough wrapper, so + # default to the mosaic there. + if ( + env.get("TMUX") + or env.get("ZELLIJ") + or env.get("STY") + or term.startswith("screen") + or term.startswith("tmux") + or program in {"tmux", "screen"} + ): + return None + + if ( + program in {"wezterm", "iterm.app"} + or env.get("WEZTERM_EXECUTABLE") + or env.get("ITERM_SESSION_ID") + ): + # WezTerm and iTerm2 both implement the iTerm2 inline-image protocol + # natively; WezTerm's TGP placeholders are less reliable inside Textual. + return "iterm2" + return None + + +PROTOCOL = detect_protocol() + + +def supported() -> bool: + return PROTOCOL in ("tgp", "iterm2") + + +def cell_size() -> tuple[int, int]: + """Terminal cell size in pixels (used to crop artwork to its cell box).""" + import sys + + width = os.environ.get("TEXTUAL_CELL_WIDTH", "") + height = os.environ.get("TEXTUAL_CELL_HEIGHT", "") + if width.isdigit() and height.isdigit() and int(width) > 0 and int(height) > 0: + return int(width), int(height) + + if sys.platform != "win32": + try: + import fcntl + import struct + import termios + + stdout = sys.__stdout__ + if stdout is not None and stdout.isatty(): + packed = fcntl.ioctl( + stdout.fileno(), termios.TIOCGWINSZ, b"\0" * 8 + ) + _, _, pixel_width, pixel_height = struct.unpack("HHHH", packed) + terminal = os.get_terminal_size(stdout.fileno()) + if ( + terminal.columns > 0 + and terminal.lines > 0 + and pixel_width > 0 + and pixel_height > 0 + ): + return ( + max(1, pixel_width // terminal.columns), + max(1, pixel_height // terminal.lines), + ) + except (OSError, ValueError): + pass + return 10, 20 + + +def _cache_path(url: str): + digest = hashlib.sha1(url.encode("utf-8")).hexdigest() + return config.CACHE_DIR / "artwork" / digest + + +def load_cached(url: str) -> bytes | None: + try: + data = _cache_path(url).read_bytes() + except OSError: + return None + return data or None + + +def save_cached(url: str, data: bytes) -> None: + try: + path = _cache_path(url) + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_name(path.name + ".tmp") + tmp.write_bytes(data) + os.replace(tmp, path) + except OSError: + pass + + +def diagnostics() -> dict[str, object]: + """Return a small dict describing image support for --diagnose-images.""" + import sys + from importlib.metadata import PackageNotFoundError, version + + def pkg_version(name: str) -> str: + try: + return version(name) + except PackageNotFoundError: + return "not installed" + + env = os.environ + return { + "tty": bool(sys.__stdout__ is not None and sys.__stdout__.isatty()), + "TERM": env.get("TERM", ""), + "TERM_PROGRAM": env.get("TERM_PROGRAM", ""), + "KITTY_WINDOW_ID": env.get("KITTY_WINDOW_ID", ""), + "WEZTERM_EXECUTABLE": env.get("WEZTERM_EXECUTABLE", ""), + "ITERM_SESSION_ID": env.get("ITERM_SESSION_ID", ""), + "TMUX": env.get("TMUX", ""), + "ZELLIJ": env.get("ZELLIJ", ""), + "SOMAFM_IMAGES": env.get("SOMAFM_IMAGES", ""), + "NO_COLOR": "set" if "NO_COLOR" in env else "", + "protocol": PROTOCOL, + "supported": supported(), + "cell_size": cell_size(), + "pillow": pkg_version("pillow"), + "cache": str(config.CACHE_DIR / "artwork"), + } + + +async def fetch(client: httpx.AsyncClient, url: str) -> bytes | None: + """Download and validate an image, caching the bytes on disk.""" + if not url.startswith(("http://", "https://")): + return None + try: + response = await client.get(url, headers={"Accept": "image/*,*/*"}) + response.raise_for_status() + data = response.content + if not data: + return None + from PIL import Image + + with Image.open(BytesIO(data)) as image: + image.verify() + save_cached(url, data) + return data + except (httpx.HTTPError, OSError, ValueError): + return None + + +def fit_image(data: bytes, width: int, height: int): + """Decode and center-crop an image to exactly ``width`` x ``height`` cells.""" + try: + from PIL import Image, ImageOps + + with Image.open(BytesIO(data)) as source: + image = source.convert("RGBA") + cell_width, cell_height = cell_size() + target = ( + max(1, width * cell_width), + max(1, height * cell_height), + ) + return ImageOps.fit( + image, + target, + method=Image.Resampling.LANCZOS, + centering=(0.5, 0.5), + ) + except Exception: + return None diff --git a/somafm_tui/config.py b/somafm_tui/config.py new file mode 100644 index 0000000..95d1fd7 --- /dev/null +++ b/somafm_tui/config.py @@ -0,0 +1,123 @@ +"""Paths, persistent settings and tiny cache helpers. + +Everything lives in the usual XDG places: + + ~/.config/somafm-tui/config.json favourites, volume, last station + ~/.cache/somafm-tui/channels.json last good copy of somafm.com/channels.json + ~/.cache/somafm-tui/playlists.json resolved .pls -> stream URLs + ~/.cache/somafm-tui/songs/.json last good copy of the song history +""" + +from __future__ import annotations + +import contextlib +import json +import os +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Any + +APP_NAME = "somafm-tui" + + +def _xdg(env_var: str, fallback: str) -> Path: + value = os.environ.get(env_var) + if value: + return Path(value).expanduser() + return Path.home() / fallback + + +CONFIG_DIR = _xdg("XDG_CONFIG_HOME", ".config") / APP_NAME +CACHE_DIR = _xdg("XDG_CACHE_HOME", ".cache") / APP_NAME +RUNTIME_DIR = Path(os.environ.get("XDG_RUNTIME_DIR") or "/tmp") / APP_NAME + +CONFIG_PATH = CONFIG_DIR / "config.json" +CHANNELS_CACHE = CACHE_DIR / "channels.json" +PLAYLIST_CACHE = CACHE_DIR / "playlists.json" +SONGS_CACHE_DIR = CACHE_DIR / "songs" + +VISUALIZERS = ("both", "bars", "wave") + + +def ensure_dirs() -> None: + for directory in (CONFIG_DIR, CACHE_DIR, SONGS_CACHE_DIR): + with contextlib.suppress(OSError): + directory.mkdir(parents=True, exist_ok=True) + + +def atomic_write_text(path: Path, text: str) -> None: + """Write ``text`` to ``path`` without ever leaving a half-written file.""" + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_name(f".{path.name}.tmp") + tmp.write_text(text, encoding="utf-8") + os.replace(tmp, path) + + +def read_json(path: Path) -> Any: + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + + +def write_json(path: Path, payload: Any) -> None: + with contextlib.suppress(OSError): + atomic_write_text(path, json.dumps(payload, indent=2, ensure_ascii=False)) + + +@dataclass(slots=True) +class Config: + favorites: list[str] = field(default_factory=list) + last_station: str = "groovesalad" + volume: int = 70 + visualizer: str = "both" + autoplay: bool = False + mpris: bool = True + + @classmethod + def load(cls) -> Config: + raw = read_json(CONFIG_PATH) + config = cls() + if not isinstance(raw, dict): + return config + + favorites = raw.get("favorites") + if isinstance(favorites, list): + config.favorites = [str(item) for item in favorites if isinstance(item, str)] + + last_station = raw.get("last_station") + if isinstance(last_station, str) and last_station: + config.last_station = last_station + + volume = raw.get("volume") + if isinstance(volume, (int, float)): + config.volume = max(0, min(100, int(volume))) + + visualizer = raw.get("visualizer") + if visualizer in VISUALIZERS: + config.visualizer = visualizer + + if isinstance(raw.get("autoplay"), bool): + config.autoplay = raw["autoplay"] + if isinstance(raw.get("mpris"), bool): + config.mpris = raw["mpris"] + return config + + def save(self) -> None: + try: + ensure_dirs() + atomic_write_text( + CONFIG_PATH, json.dumps(asdict(self), indent=2, ensure_ascii=False) + ) + except OSError: + pass + + def is_favorite(self, station_id: str) -> bool: + return station_id in self.favorites + + def toggle_favorite(self, station_id: str) -> bool: + if station_id in self.favorites: + self.favorites.remove(station_id) + return False + self.favorites.append(station_id) + return True diff --git a/somafm_tui/fallback.py b/somafm_tui/fallback.py new file mode 100644 index 0000000..9ffd824 --- /dev/null +++ b/somafm_tui/fallback.py @@ -0,0 +1,96 @@ +"""Bundled station list. + +Only used when somafm.com cannot be reached *and* no cache exists yet, so the +app is still useful on a plane. The live list from ``channels.json`` always +wins. Stream links are built from SomaFM's well-known ``.pls`` naming scheme +and are additionally resolved at runtime. +""" + +from __future__ import annotations + +from typing import Any + +# (id, title, genre, description) +_CHANNELS: tuple[tuple[str, str, str, str], ...] = ( + ("groovesalad", "Groove Salad", "ambient|electronic", "A nicely chilled plate of ambient/downtempo beats and grooves."), + ("dronezone", "Drone Zone", "ambient", "Served best chilled, safe with most medications. Atmospheric textures with minimal beats."), + ("deepspaceone", "Deep Space One", "ambient", "Deep ambient electronic, experimental and space music. For inner and outer space exploration."), + ("spacestation", "Space Station Soma", "electronic", "Tune in, turn on, space out. Spaced-out ambient and mid-tempo electronica."), + ("missioncontrol", "Mission Control", "ambient|electronic", "Celebrating NASA and Space Explorers everywhere."), + ("darkzone", "The Dark Zone", "ambient", "The darker side of deep ambient. Music for staring into the Abyss."), + ("dz2", "Drone Zone 2", "ambient", "A more eclectic alternative mix of atmospheric textures with minimal beats."), + ("synphaera", "Synphaera Radio", "ambient|electronic", "Featuring the music from an independent record label focused on modern electronic ambient and space music."), + ("secretagent", "Secret Agent", "lounge", "The soundtrack for your stylish, mysterious, dangerous life. For Spies and PIs too!"), + ("illstreet", "Illinois Street Lounge", "lounge", "Classic bachelor pad, playful exotica and vintage music of tomorrow."), + ("tikitime", "Tiki Time", "tiki|world", "Classic Tiki music and Vintage island rhythms to sip cocktails by."), + ("bossa", "Bossa Beyond", "bossanova|world", "Silky-smooth, laid-back Brazilian-style rhythms of Bossa Nova, Samba and beyond"), + ("suburbsofgoa", "Suburbs of Goa", "world", "Desi-influenced Asian world beats and beyond."), + ("thistle", "ThistleRadio", "celtic|world", "Exploring music from Celtic roots and branches"), + ("indiepop", "Indie Pop Rocks!", "alternative|rock", "New and classic favorite indie pop tracks."), + ("digitalis", "Digitalis", "electronic|alternative", "Digitally affected analog rock to calm the agitated heart."), + ("poptron", "PopTron", "alternative", "Electropop and indie dance rock with sparkle and pop."), + ("u80s", "Underground 80s", "alternative|electronic", "Early 80s UK Synthpop and a bit of New Wave."), + ("seventies", "Left Coast 70s", "70s|rock", "Mellow album rock from the Seventies. Yacht not required."), + ("reggae", "Heavyweight Reggae", "reggae", "Reggae, Ska, Rocksteady classic and deep tracks."), + ("bootliquor", "Boot Liquor", "americana", "Americana Roots music for Cowhands, Cowpokes and Cowtippers"), + ("folkfwd", "Folk Forward", "folk|alternative", "Indie Folk, Alt-folk and the occasional folk classics."), + ("lush", "Lush", "electronic", "Sensuous and mellow female vocals, many with an electronic influence."), + ("fluid", "Fluid", "electronic|hiphop", "Drown in the electronic sound of instrumental hiphop, future soul and liquid trap."), + ("beatblender", "Beat Blender", "electronic", "A late night blend of deep-house and downtempo chill."), + ("cliqhop", "cliqhop idm", "electronic", "Blips'n'beeps backed mostly w/beats. Intelligent Dance Music."), + ("thetrip", "The Trip", "electronic", "Progressive house / trance. Tip top tunes."), + ("defcon", "DEF CON Radio", "electronic|specials", "Music for Hacking. The DEF CON Year-Round Channel."), + ("dubstep", "Dub Step Beyond", "electronic", "Dubstep, Dub and Deep Bass. May damage speakers at high volume."), + ("vaporwaves", "Vaporwaves", "electronic", "All Vaporwave. All the time."), + ("sonicuniverse", "Sonic Universe", "jazz", "Transcending the world of jazz with eclectic, avant-garde takes on tradition."), + ("metal", "Metal Detector", "metal", "From black to doom, prog to sludge, thrash to post, stoner to crossover, punk to industrial."), + ("doomed", "Doomed", "ambient|industrial", "Where every day is Halloween: Dark industrial/ambient music for tortured souls."), + ("covers", "Covers", "eclectic", "Just covers. Songs you know by artists you don't. We've got you covered."), + ("brfm", "Black Rock FM", "eclectic", "From the Black Rock Desert playa to the world, year round!"), + ("7soul", "Seven Inch Soul", "oldies", "Vintage soul tracks from the original 45 RPM vinyl."), + ("insound", "The In-Sound", "oldies", "60s/70s Hipster Euro Pop where psychedelic melodies meets groovy vibes."), + ("sf1033", "SF 10-33", "ambient|news", "Ambient music mixed with the sounds of San Francisco public safety radio traffic."), + ("scanner", "SF Police Scanner", "live|news", "San Francisco Public Safety Scanner Feed"), + ("sfinsf", "SF in SF", "spoken", "Author readings and discussions from the science fiction, fantasy, horror, and genre literary fields."), + ("live", "SomaFM Live", "live|specials", "Special Live Events and rebroadcasts of past live events"), + ("specials", "SomaFM Specials", "specials", "Now featuring Afternoon Jazz, Wavepool, DubX, The Surf Report & More!"), + ("n5md", "n5MD Radio", "specials", "Emotional Experiments in Music: Ambient, modern composition, post-rock, & experimental electronic music"), + ("chillits", "Chillits Radio", "chill|live", "Celebrating 25 years of music, chilling and camping"), + ("groovesalad2", "Groove Salad 2", "ambient|electronic", "A different mix of nicely chilled ambient/downtempo beats and grooves."), + ("gsclassic", "Groove Salad Classic", "ambient|electronic", "The classic (early 2000s) version of a nicely chilled plate of ambient/downtempo beats and grooves."), + ("christmas", "Christmas Lounge", "holiday", "Chilled holiday grooves and classic winter lounge tracks."), + ("xmasrocks", "Christmas Rocks!", "holiday", "Have yourself an indie/alternative holiday season!"), + ("jollysoul", "Jolly Ol' Soul", "holiday", "Where we cut right to the soul of the season."), + ("deptstore", "Department Store Christmas", "holiday", "Holiday Elevator Music from a more innocent time."), + ("xmasinfrisko", "Xmas in Frisko", "holiday", "SomaFM's wacky and eclectic holiday mix. Not for the easily offended."), +) + + +def _playlists(channel_id: str) -> list[dict[str, str]]: + return [ + {"url": f"https://somafm.com/{channel_id}.pls", "format": "mp3", "quality": "highest"}, + {"url": f"https://somafm.com/{channel_id}130.pls", "format": "aac", "quality": "highest"}, + {"url": f"https://somafm.com/{channel_id}64.pls", "format": "aacp", "quality": "high"}, + {"url": f"https://somafm.com/{channel_id}32.pls", "format": "aacp", "quality": "low"}, + ] + + +def fallback_channels() -> list[dict[str, Any]]: + """Return a channels.json-shaped payload so the normal parser can eat it.""" + payload: list[dict[str, Any]] = [] + for channel_id, title, genre, description in _CHANNELS: + payload.append( + { + "id": channel_id, + "title": title, + "description": description, + "dj": "SomaFM", + "genre": genre, + "image": f"https://somafm.com/logos/256/{channel_id}256.png", + "largeimage": f"https://somafm.com/logos/512/{channel_id}512.png", + "playlists": _playlists(channel_id), + "listeners": "0", + "lastPlaying": "", + } + ) + return payload diff --git a/somafm_tui/mpris.py b/somafm_tui/mpris.py new file mode 100644 index 0000000..9e21ffa --- /dev/null +++ b/somafm_tui/mpris.py @@ -0,0 +1,420 @@ +"""MPRIS (org.mpris.MediaPlayer2) service. + +This is what makes GNOME/KDE media widgets, lock screens, ``playerctl``, +``media-control`` etc. show the station, the current artist/title/album and the +album art, and lets them play/pause, skip stations and change the volume. + +The service is optional: if dbus-next is missing or there is no session bus, +:attr:`MprisService.available` simply stays ``False`` and the TUI carries on. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import hashlib +import logging +import os +import queue +import threading +from typing import Any + +try: # pragma: no cover - exercised implicitly on systems without dbus + from dbus_next import PropertyAccess, Variant + from dbus_next.aio import MessageBus + from dbus_next.service import ServiceInterface, dbus_property, method + + DBUS_AVAILABLE = True +except Exception: # pragma: no cover + DBUS_AVAILABLE = False + + class PropertyAccess: # type: ignore[no-redef] + READ = "read" + READWRITE = "readwrite" + + class Variant: # type: ignore[no-redef] + def __init__(self, signature: str, value: Any) -> None: + self.signature = signature + self.value = value + + class ServiceInterface: # type: ignore[no-redef] + def __init__(self, name: str) -> None: + self.name = name + + def emit_properties_changed(self, *args: Any, **kwargs: Any) -> None: + return None + + def dbus_property(*args: Any, **kwargs: Any): # type: ignore[no-redef] + def decorator(function): + return function + + return decorator + + def method(*args: Any, **kwargs: Any): # type: ignore[no-redef] + def decorator(function): + return function + + return decorator + + +LOGGER = logging.getLogger(__name__) + + +def build_metadata( + *, + station: str, + title: str, + artist: str, + album: str = "", + art_url: str = "", + stream_url: str = "", + track_key: str = "", +) -> dict[str, Any]: + digest = hashlib.sha1( + (track_key or f"{station}|{artist}|{title}").encode("utf-8") + ).hexdigest()[:24] + metadata: dict[str, Any] = { + "mpris:trackid": Variant("o", f"/org/somafm/track/{digest}"), + "xesam:title": Variant("s", title or station or "SomaFM"), + "xesam:artist": Variant("as", [artist] if artist else ["SomaFM"]), + "xesam:comment": Variant("as", [f"SomaFM · {station}"] if station else ["SomaFM"]), + } + if album: + metadata["xesam:album"] = Variant("s", album) + if art_url.startswith(("http://", "https://", "file://")): + metadata["mpris:artUrl"] = Variant("s", art_url) + if stream_url.startswith(("http://", "https://")): + metadata["xesam:url"] = Variant("s", stream_url) + return metadata + + +class _RootInterface(ServiceInterface): + def __init__(self, service: "MprisService") -> None: + super().__init__("org.mpris.MediaPlayer2") + self._service = service + + @dbus_property(name="Identity", access=PropertyAccess.READ) + def Identity(self) -> "s": + return "SomaFM TUI" + + @dbus_property(name="DesktopEntry", access=PropertyAccess.READ) + def DesktopEntry(self) -> "s": + return "somafm-tui" + + @dbus_property(name="CanQuit", access=PropertyAccess.READ) + def CanQuit(self) -> "b": + return True + + @dbus_property(name="CanRaise", access=PropertyAccess.READ) + def CanRaise(self) -> "b": + return False + + @dbus_property(name="HasTrackList", access=PropertyAccess.READ) + def HasTrackList(self) -> "b": + return False + + @dbus_property(name="SupportedUriSchemes", access=PropertyAccess.READ) + def SupportedUriSchemes(self) -> "as": + return ["http", "https"] + + @dbus_property(name="SupportedMimeTypes", access=PropertyAccess.READ) + def SupportedMimeTypes(self) -> "as": + return ["audio/mpeg", "audio/aac", "audio/aacp", "audio/ogg"] + + @method(name="Raise") + def Raise(self): + return None + + @method(name="Quit") + def Quit(self): + self._service.actions.put("quit") + + +class _PlayerInterface(ServiceInterface): + def __init__(self, service: "MprisService") -> None: + super().__init__("org.mpris.MediaPlayer2.Player") + self._service = service + self._playback_status = "Stopped" + self._metadata: dict[str, Any] = {} + self._volume = 0.7 + + # ----- properties -------------------------------------------------- # + + @dbus_property(name="PlaybackStatus", access=PropertyAccess.READ) + def PlaybackStatus(self) -> "s": + return self._playback_status + + @dbus_property(name="LoopStatus", access=PropertyAccess.READWRITE) + def LoopStatus(self) -> "s": + return "None" + + @LoopStatus.setter + def LoopStatus(self, value: "s"): + return None + + @dbus_property(name="Rate", access=PropertyAccess.READ) + def Rate(self) -> "d": + return 1.0 + + @dbus_property(name="Shuffle", access=PropertyAccess.READWRITE) + def Shuffle(self) -> "b": + return False + + @Shuffle.setter + def Shuffle(self, value: "b"): + return None + + @dbus_property(name="Metadata", access=PropertyAccess.READ) + def Metadata(self) -> "a{sv}": + return self._metadata + + @dbus_property(name="Volume", access=PropertyAccess.READWRITE) + def Volume(self) -> "d": + return self._volume + + @Volume.setter + def Volume(self, value: "d"): + value = max(0.0, min(1.0, float(value))) + self._volume = value + self._service.actions.put(f"volume:{round(value * 100)}") + + @dbus_property(name="Position", access=PropertyAccess.READ) + def Position(self) -> "x": + return 0 + + @dbus_property(name="MinimumRate", access=PropertyAccess.READ) + def MinimumRate(self) -> "d": + return 1.0 + + @dbus_property(name="MaximumRate", access=PropertyAccess.READ) + def MaximumRate(self) -> "d": + return 1.0 + + @dbus_property(name="CanGoNext", access=PropertyAccess.READ) + def CanGoNext(self) -> "b": + return True + + @dbus_property(name="CanGoPrevious", access=PropertyAccess.READ) + def CanGoPrevious(self) -> "b": + return True + + @dbus_property(name="CanPlay", access=PropertyAccess.READ) + def CanPlay(self) -> "b": + return True + + @dbus_property(name="CanPause", access=PropertyAccess.READ) + def CanPause(self) -> "b": + return True + + @dbus_property(name="CanSeek", access=PropertyAccess.READ) + def CanSeek(self) -> "b": + return False + + @dbus_property(name="CanControl", access=PropertyAccess.READ) + def CanControl(self) -> "b": + return True + + # ----- methods ----------------------------------------------------- # + + @method(name="Play") + def Play(self): + self._service.actions.put("play") + + @method(name="Pause") + def Pause(self): + self._service.actions.put("pause") + + @method(name="PlayPause") + def PlayPause(self): + self._service.actions.put("toggle") + + @method(name="Stop") + def Stop(self): + self._service.actions.put("stop") + + @method(name="Next") + def Next(self): + self._service.actions.put("next") + + @method(name="Previous") + def Previous(self): + self._service.actions.put("prev") + + @method(name="Seek") + def Seek(self, offset: "x"): + return None + + @method(name="SetPosition") + def SetPosition(self, track_id: "o", position: "x"): + return None + + @method(name="OpenUri") + def OpenUri(self, uri: "s"): + return None + + +class MprisService: + def __init__( + self, actions: "queue.Queue[str]", enabled: bool = True + ) -> None: + self.actions = actions + self.enabled = bool(enabled) and DBUS_AVAILABLE + self.available = False + self.error = "" + + self._thread: threading.Thread | None = None + self._loop: asyncio.AbstractEventLoop | None = None + self._shutdown: asyncio.Event | None = None + self._bus: Any = None + self._root: _RootInterface | None = None + self._player: _PlayerInterface | None = None + + self._volume = 0.7 + self._status = "Stopped" + self._metadata: dict[str, Any] = {} + self.bus_name = ( + f"org.mpris.MediaPlayer2.somafmtui.instance{os.getpid()}" + ) + + # ------------------------------------------------------------------ # + # lifecycle + # ------------------------------------------------------------------ # + + def start(self) -> None: + if not self.enabled: + return + self._thread = threading.Thread( + target=self._thread_main, name="somafm-mpris", daemon=True + ) + self._thread.start() + + def _thread_main(self) -> None: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + self._loop = loop + try: + loop.run_until_complete(self._serve()) + except Exception as exc: + self.error = str(exc) or exc.__class__.__name__ + LOGGER.debug("MPRIS unavailable: %s", exc) + finally: + self.available = False + try: + pending = asyncio.all_tasks(loop) + for task in pending: + task.cancel() + if pending: + loop.run_until_complete( + asyncio.gather(*pending, return_exceptions=True) + ) + except Exception: + pass + loop.close() + self._loop = None + + async def _serve(self) -> None: + self._shutdown = asyncio.Event() + bus = await MessageBus().connect() + self._bus = bus + self._root = _RootInterface(self) + self._player = _PlayerInterface(self) + bus.export("/org/mpris/MediaPlayer2", self._root) + bus.export("/org/mpris/MediaPlayer2", self._player) + await bus.request_name(self.bus_name) + self.available = True + + # Push whatever we already know: the UI may have produced metadata + # before the bus name was acquired. + if self._metadata: + await self._apply_metadata(self._metadata) + await self._apply_status(self._status) + await self._apply_volume(self._volume) + + await self._shutdown.wait() + with contextlib.suppress(Exception): + bus.disconnect() + + def stop(self) -> None: + loop = self._loop + if loop is not None and not loop.is_closed(): + shutdown = self._shutdown + if shutdown is not None: + with contextlib.suppress(RuntimeError): + loop.call_soon_threadsafe(shutdown.set) + thread = self._thread + if thread is not None and thread.is_alive(): + thread.join(timeout=2.0) + + @property + def status_label(self) -> str: + if not self.enabled: + return "off" + if self.available: + return "on" + return "error" if self.error else "off" + + # ------------------------------------------------------------------ # + # updates from the app (thread safe) + # ------------------------------------------------------------------ # + + def _submit(self, factory, *args: Any) -> None: + if not self.available: + return + loop = self._loop + if loop is None or loop.is_closed(): + return + with contextlib.suppress(RuntimeError): + loop.call_soon_threadsafe(lambda: asyncio.ensure_future(factory(*args))) + + def set_status(self, status: str) -> None: + if status not in ("Playing", "Paused", "Stopped"): + return + self._status = status + self._submit(self._apply_status, status) + + def set_volume(self, ratio: float) -> None: + self._volume = max(0.0, min(1.0, float(ratio))) + self._submit(self._apply_volume, self._volume) + + def set_track( + self, + *, + station: str, + title: str, + artist: str, + album: str = "", + art_url: str = "", + stream_url: str = "", + track_key: str = "", + ) -> None: + self._metadata = build_metadata( + station=station, + title=title, + artist=artist, + album=album, + art_url=art_url, + stream_url=stream_url, + track_key=track_key, + ) + self._submit(self._apply_metadata, self._metadata) + + async def _apply_status(self, status: str) -> None: + player = self._player + if player is None or player._playback_status == status: + return + player._playback_status = status + player.emit_properties_changed({"PlaybackStatus": status}) + + async def _apply_volume(self, ratio: float) -> None: + player = self._player + if player is None: + return + player._volume = ratio + player.emit_properties_changed({"Volume": ratio}) + + async def _apply_metadata(self, metadata: dict[str, Any]) -> None: + player = self._player + if player is None: + return + player._metadata = metadata + player.emit_properties_changed({"Metadata": metadata}) diff --git a/somafm_tui/player.py b/somafm_tui/player.py new file mode 100644 index 0000000..a391855 --- /dev/null +++ b/somafm_tui/player.py @@ -0,0 +1,551 @@ +"""Audio playback. + +Primary backend is **mpv** driven over its JSON IPC socket (play/pause/volume/ +mirror failover/stop). If mpv is missing we fall back to **ffplay**, which can +still play but offers no live control beyond SIGSTOP/SIGCONT. + +The player runs in its own thread with a private asyncio loop. The UI talks to +it through :meth:`Player.play` / :meth:`Player.pause` / ... (thread safe) and +reads :attr:`Player.events` (a plain ``queue.Queue``) from its ticker. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import os +import queue +import shlex +import shutil +import signal +import subprocess +import threading +import time +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from .api import USER_AGENT +from .config import RUNTIME_DIR + + +@dataclass +class PlayerEvent: + kind: str = "state" # "state" | "volume" | "notice" + state: str = "" + detail: str = "" + volume: int | None = None + url: str = "" + + +class Player: + def __init__(self) -> None: + self.events: queue.Queue[PlayerEvent] = queue.Queue() + self.backend: str | None = None + self._loop: asyncio.AbstractEventLoop | None = None + self._thread: threading.Thread | None = None + self._ready = threading.Event() + self._stopping = threading.Event() + self._proc: subprocess.Popen[bytes] | None = None + self._reader: asyncio.StreamReader | None = None + self._writer: asyncio.StreamWriter | None = None + self._urls: list[str] = [] + self._url_index = 0 + self._volume = 70 + self._paused = False + self._has_file = False + self._state = "stopped" + self._sock_path: Path | None = None + self._ffplay_proc: subprocess.Popen[bytes] | None = None + self._retry_scheduled = False + + # ------------------------------------------------------------------ # + # setup + # ------------------------------------------------------------------ # + + @staticmethod + def detect_backend() -> str | None: + if shutil.which("mpv"): + return "mpv" + if shutil.which("ffplay"): + return "ffplay" + return None + + def start(self) -> None: + if self._thread is not None: + return + self.backend = self.detect_backend() + if self.backend is None: + self._state = "error" + self.events.put( + PlayerEvent( + kind="state", + state="error", + detail="no audio backend found — install mpv or ffmpeg/ffplay", + ) + ) + return + self._thread = threading.Thread( + target=self._thread_main, name="somafm-player", daemon=True + ) + self._thread.start() + self._ready.wait(10) + + def _thread_main(self) -> None: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + self._loop = loop + self._ready.set() + try: + loop.run_until_complete(self._run()) + except Exception as exc: # pragma: no cover - defensive + self.events.put( + PlayerEvent(kind="state", state="error", detail=f"player stopped: {exc}") + ) + finally: + try: + pending = asyncio.all_tasks(loop) + for task in pending: + task.cancel() + if pending: + loop.run_until_complete( + asyncio.gather(*pending, return_exceptions=True) + ) + except Exception: + pass + loop.close() + self._loop = None + + async def _run(self) -> None: + if self.backend == "mpv": + await self._run_mpv() + else: + await self._run_ffplay() + + # ------------------------------------------------------------------ # + # public, thread safe API + # ------------------------------------------------------------------ # + + def play(self, urls: list[str]) -> None: + if not urls: + self._set_state("error", "no playable stream found") + return + self._submit(self._do_play, list(urls)) + + def pause(self) -> None: + self._submit(self._do_pause) + + def resume(self) -> None: + self._submit(self._do_resume) + + def toggle_pause(self) -> None: + self._submit(self._do_toggle) + + def stop(self) -> None: + self._submit(self._do_stop) + + def set_volume(self, value: int) -> None: + self._submit(self._do_volume, int(value)) + + def shutdown(self, timeout: float = 3.0) -> None: + if self._stopping.is_set(): + return + self._stopping.set() + loop = self._loop + if loop is None or loop.is_closed(): + return + try: + future = asyncio.run_coroutine_threadsafe(self._do_shutdown(), loop) + future.result(timeout) + except Exception: + pass + + @property + def state(self) -> str: + return self._state + + # ------------------------------------------------------------------ # + # internals + # ------------------------------------------------------------------ # + + def _submit(self, factory: Callable[..., Any], *args: Any) -> None: + loop = self._loop + if loop is None or loop.is_closed() or self._stopping.is_set(): + return + with contextlib.suppress(RuntimeError): + loop.call_soon_threadsafe(lambda: asyncio.ensure_future(factory(*args))) + + def _set_state(self, state: str, detail: str = "", url: str = "") -> None: + self._state = state + self.events.put( + PlayerEvent(kind="state", state=state, detail=detail, url=url) + ) + + def _current_url(self) -> str: + if 0 <= self._url_index < len(self._urls): + return self._urls[self._url_index] + return "" + + # ----- commands ---------------------------------------------------- # + + async def _do_play(self, urls: list[str]) -> None: + self._urls = urls + self._url_index = 0 + self._paused = False + self._has_file = False + self._retry_scheduled = False + if self.backend == "mpv": + await self._mpv_load() + else: + await self._kill_ffplay() + + async def _do_pause(self) -> None: + if self.backend == "mpv": + await self._mpv_send(["set_property", "pause", True]) + elif self.backend == "ffplay" and self._ffplay_proc is not None: + self._ffplay_signal(signal.SIGSTOP) + self._paused = True + self._set_state("paused", "", self._current_url()) + + async def _do_resume(self) -> None: + if self.backend == "mpv": + await self._mpv_send(["set_property", "pause", False]) + elif self.backend == "ffplay" and self._ffplay_proc is not None: + self._ffplay_signal(signal.SIGCONT) + self._paused = False + self._set_state("playing", "", self._current_url()) + + async def _do_toggle(self) -> None: + if self._paused: + await self._do_resume() + else: + await self._do_pause() + + async def _do_stop(self) -> None: + self._urls = [] + self._url_index = 0 + self._paused = False + self._has_file = False + if self.backend == "mpv": + await self._mpv_send(["stop"]) + else: + await self._kill_ffplay() + self._set_state("stopped") + + async def _do_volume(self, value: int) -> None: + value = max(0, min(100, value)) + self._volume = value + if self.backend == "mpv": + await self._mpv_send(["set_property", "volume", value]) + self.events.put(PlayerEvent(kind="volume", volume=value)) + + async def _do_shutdown(self) -> None: + if self.backend == "mpv": + await self._mpv_send(["quit"]) + await self._kill_ffplay() + await self._stop_mpv() + + # ----- mpv --------------------------------------------------------- # + + async def _run_mpv(self) -> None: + while not self._stopping.is_set(): + try: + await self._start_mpv() + except Exception as exc: + if self._stopping.is_set(): + return + self._set_state("error", f"mpv failed to start: {exc}") + await asyncio.sleep(3.0) + continue + + try: + await self._read_mpv() + except (OSError, ValueError, ConnectionError, asyncio.IncompleteReadError): + pass + finally: + await self._stop_mpv() + + if self._stopping.is_set(): + return + self._set_state("buffering", "restarting audio engine…") + await asyncio.sleep(1.5) + + async def _start_mpv(self) -> None: + RUNTIME_DIR.mkdir(parents=True, exist_ok=True) + self._sock_path = RUNTIME_DIR / f"mpv-{os.getpid()}.sock" + with contextlib.suppress(OSError): + self._sock_path.unlink() + + command = [ + "mpv", + "--no-config", + "--no-video", + "--really-quiet", + "--no-terminal", + "--idle=yes", + "--force-window=no", + f"--input-ipc-server={self._sock_path}", + f"--volume={self._volume}", + "--cache=yes", + "--cache-secs=20", + "--demuxer-max-bytes=32MiB", + "--audio-client-name=somafm-tui", + f"--user-agent={USER_AGENT}", + "--network-timeout=20", + ] + extra = os.environ.get("SOMAFM_MPV_ARGS") + if extra: + command.extend(shlex.split(extra)) + + self._proc = subprocess.Popen( + command, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + + deadline = time.monotonic() + 6.0 + while True: + if self._stopping.is_set(): + raise RuntimeError("shutting down") + if self._proc.poll() is not None: + raise RuntimeError(f"mpv exited with code {self._proc.returncode}") + try: + self._reader, self._writer = await asyncio.open_unix_connection( + str(self._sock_path) + ) + break + except (FileNotFoundError, ConnectionRefusedError, OSError): + if time.monotonic() > deadline: + raise RuntimeError( + "timed out waiting for mpv IPC socket" + ) from None + await asyncio.sleep(0.05) + + await self._mpv_send(["observe_property", 1, "pause"]) + await self._mpv_send(["observe_property", 2, "volume"]) + if self._urls: + await self._mpv_load() + + async def _mpv_send(self, command: list[Any]) -> None: + writer = self._writer + if writer is None: + return + try: + writer.write((json.dumps({"command": command}) + "\n").encode("utf-8")) + await writer.drain() + except (OSError, RuntimeError): + pass + + async def _mpv_load(self) -> None: + url = self._current_url() + if not url: + return + self._has_file = False + await self._mpv_send(["loadfile", url, "replace"]) + await self._mpv_send(["set_property", "pause", False]) + self._paused = False + self._set_state("loading", "tuning in…", url) + + async def _read_mpv(self) -> None: + reader = self._reader + if reader is None: + return + while not self._stopping.is_set(): + line = await reader.readline() + if not line: + return + try: + message = json.loads(line) + except ValueError: + continue + if isinstance(message, dict): + self._handle_mpv_message(message) + + def _handle_mpv_message(self, message: dict[str, Any]) -> None: + event = message.get("event") + if event == "property-change": + name = message.get("name") + value = message.get("data") + if name == "pause": + self._paused = bool(value) + if self._has_file: + state = "paused" if self._paused else "playing" + self._set_state(state, "", self._current_url()) + elif name == "volume" and isinstance(value, (int, float)): + self._volume = max(0, min(100, round(value))) + self.events.put(PlayerEvent(kind="volume", volume=self._volume)) + return + + if event == "start-file": + self._has_file = True + self._set_state("loading", "buffering stream…", self._current_url()) + return + + if event == "file-loaded": + self._has_file = True + state = "paused" if self._paused else "playing" + self._set_state(state, "", self._current_url()) + return + + if event == "end-file": + self._has_file = False + reason = str(message.get("reason") or "") + if reason == "stop" or self._stopping.is_set(): + self._set_state("stopped") + elif reason in ("eof", "error"): + self._schedule_retry(reason) + return + + if event == "idle": + if not self._urls: + self._set_state("stopped") + return + + error = message.get("error") + if error and error != "success": + self.events.put(PlayerEvent(kind="notice", detail=str(error))) + + def _schedule_retry(self, reason: str) -> None: + if self._retry_scheduled or not self._urls or self._stopping.is_set(): + return + self._retry_scheduled = True + self._submit(self._retry_stream, reason) + + async def _retry_stream(self, reason: str) -> None: + self._retry_scheduled = False + if not self._urls or self._stopping.is_set(): + return + if self._url_index + 1 >= len(self._urls): + self._set_state( + "error", f"stream failed ({reason}) — press enter to retry" + ) + return + self._url_index += 1 + self._set_state( + "buffering", + f"reconnecting via mirror {self._url_index + 1}/{len(self._urls)}…", + self._current_url(), + ) + await asyncio.sleep(1.0) + if self.backend == "mpv": + await self._mpv_load() + else: + await self._kill_ffplay() + + async def _stop_mpv(self) -> None: + writer = self._writer + self._writer = None + self._reader = None + if writer is not None: + try: + writer.close() + await writer.wait_closed() + except Exception: + pass + + proc = self._proc + self._proc = None + if proc is not None: + try: + if proc.poll() is None: + proc.terminate() + try: + await asyncio.to_thread(proc.wait, 2) + except Exception: + proc.kill() + except Exception: + pass + + if self._sock_path is not None: + with contextlib.suppress(OSError): + self._sock_path.unlink() + self._sock_path = None + + # ----- ffplay fallback --------------------------------------------- # + + async def _run_ffplay(self) -> None: + while not self._stopping.is_set(): + if not self._urls: + if self._ffplay_proc is not None: + await self._kill_ffplay() + if self._state != "stopped": + self._set_state("stopped") + await asyncio.sleep(0.2) + continue + + if self._ffplay_proc is None: + self._start_ffplay() + elif self._ffplay_proc.poll() is not None: + self._ffplay_proc = None + if not self._paused and not self._stopping.is_set(): + if self._url_index + 1 < len(self._urls): + self._url_index += 1 + self._set_state( + "buffering", + f"reconnecting via mirror {self._url_index + 1}/{len(self._urls)}…", + self._current_url(), + ) + else: + self._set_state( + "error", "stream ended — press enter to retry" + ) + self._urls = [] + await asyncio.sleep(0.2) + + def _start_ffplay(self) -> None: + url = self._current_url() + if not url: + return + self._set_state("loading", "tuning in…", url) + command = [ + "ffplay", + "-nodisp", + "-autoexit", + "-loglevel", + "quiet", + "-volume", + str(self._volume), + url, + ] + try: + self._ffplay_proc = subprocess.Popen( + command, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + except OSError as exc: + self._set_state("error", f"ffplay failed: {exc}") + self._urls = [] + return + self._paused = False + self._set_state( + "playing", "ffplay fallback (volume changes apply on next tune-in)", url + ) + + def _ffplay_signal(self, sig: int) -> None: + proc = self._ffplay_proc + if proc is not None and proc.poll() is None: + with contextlib.suppress(OSError): + proc.send_signal(sig) + + async def _kill_ffplay(self) -> None: + proc = self._ffplay_proc + self._ffplay_proc = None + if proc is None: + return + try: + if proc.poll() is None: + proc.send_signal(signal.SIGCONT) + proc.terminate() + try: + await asyncio.to_thread(proc.wait, 1) + except Exception: + proc.kill() + except OSError: + pass diff --git a/somafm_tui/spectrum.py b/somafm_tui/spectrum.py new file mode 100644 index 0000000..5007401 --- /dev/null +++ b/somafm_tui/spectrum.py @@ -0,0 +1,534 @@ +"""Spectrum analyser. + +The audio you hear is played by mpv. The *synced* mode captures the real +PulseAudio/PipeWire monitor of the sink mpv is playing to (``parec``), so the +bars follow exactly what leaves the speakers (status ``SYNC``). + +If that is unavailable we fall back to a muted second ffmpeg decode of the +stream (status ``FFT``; it can drift relative to playback), and finally to a +synthesized show (``SIM``) so the UI stays alive. +""" + +from __future__ import annotations + +import contextlib +import math +import os +import select +import shutil +import subprocess +import threading +import time +from dataclasses import dataclass + +import numpy as np + +from .api import USER_AGENT + +SAMPLE_RATE = 44100 +FFT_SIZE = 2048 +READ_BYTES = 4096 # 2048 mono s16le samples +BAND_COUNT = 72 +WAVE_POINTS = 192 +MIN_FREQ = 35.0 +MAX_FREQ = 16000.0 +SOURCE_MODES = ("auto", "monitor", "stream", "sim") + + +@dataclass(frozen=True) +class Frame: + bands: tuple[float, ...] + peaks: tuple[float, ...] + wave: tuple[float, ...] + level: float + status: str # "monitor" | "live" | "sim" | "idle" | "off" + + @property + def is_live(self) -> bool: + return self.status in ("monitor", "live") + + @property + def label(self) -> str: + return { + "monitor": "SYNC", + "live": "FFT", + "sim": "SIM", + "idle": "IDLE", + "off": "OFF", + }.get(self.status, self.status.upper()) + + +class SpectrumEngine: + def __init__( + self, + bands: int = BAND_COUNT, + sample_rate: int = SAMPLE_RATE, + fft_size: int = FFT_SIZE, + wave_points: int = WAVE_POINTS, + ffmpeg_path: str | None = None, + source_mode: str = "auto", + ) -> None: + self.band_count = bands + self.sample_rate = sample_rate + self.fft_size = fft_size + self.wave_points = wave_points + self.source_mode = source_mode if source_mode in SOURCE_MODES else "auto" + self._ffmpeg = ffmpeg_path if ffmpeg_path is not None else shutil.which("ffmpeg") + self._parec = shutil.which("parec") + self._pactl = shutil.which("pactl") + + self._window = np.hanning(fft_size).astype(np.float32) + self._groups = self._make_groups() + self._history = np.zeros(fft_size, dtype=np.float32) + self._levels = np.zeros(bands, dtype=np.float32) + self._peaks = np.zeros(bands, dtype=np.float32) + self._wave = np.zeros(wave_points, dtype=np.float32) + self._level = 0.0 + self._gain = -24.0 + self._status = "idle" + + self._url: str | None = None + self._proc: subprocess.Popen[bytes] | None = None + self._capture_kind: str | None = None + self._monitor_failed = False + self._monitor_device_cache: str | None = None + self._lock = threading.Lock() + self._stop = threading.Event() + self._thread: threading.Thread | None = None + self._rng = np.random.default_rng() + self._sim_phase = 0.0 + + # ------------------------------------------------------------------ # + # public API + # ------------------------------------------------------------------ # + + @property + def ffmpeg_available(self) -> bool: + return bool(self._ffmpeg) + + @property + def monitor_available(self) -> bool: + return bool(self._parec) and not self._monitor_failed + + def start(self) -> None: + if self._thread is not None: + return + self._thread = threading.Thread( + target=self._run, name="somafm-spectrum", daemon=True + ) + self._thread.start() + + def set_source(self, url: str | None) -> None: + """Start analysing playback, or idle when ``url`` is ``None``. + + The URL is only a fallback source for the stream tap. A monitor + capture is tied to the actual audio output, so it deliberately keeps + running when the station changes underneath it. + """ + url = url or None + with self._lock: + changed = url != self._url + self._url = url + kind = self._capture_kind + if url is None: + self._status = "idle" + if changed and (url is None or kind != "monitor"): + self._kill_proc() + + def snapshot(self) -> Frame: + with self._lock: + return Frame( + bands=tuple(float(v) for v in self._levels), + peaks=tuple(float(v) for v in self._peaks), + wave=tuple(float(v) for v in self._wave), + level=float(self._level), + status=self._status, + ) + + def stop(self) -> None: + self._stop.set() + self._kill_proc() + thread = self._thread + if thread is not None and thread.is_alive(): + thread.join(timeout=2.0) + self._thread = None + with self._lock: + self._status = "off" + + # ------------------------------------------------------------------ # + # internals + # ------------------------------------------------------------------ # + + def _make_groups(self) -> list[tuple[int, int]]: + freqs = np.fft.rfftfreq(self.fft_size, 1.0 / self.sample_rate) + top = min(MAX_FREQ, self.sample_rate / 2 - 100) + edges = np.geomspace(MIN_FREQ, top, self.band_count + 1) + groups: list[tuple[int, int]] = [] + last = len(freqs) - 1 + for index in range(self.band_count): + lo = int(np.searchsorted(freqs, edges[index], side="left")) + hi = int(np.searchsorted(freqs, edges[index + 1], side="left")) + lo = min(max(lo, 0), last) + hi = min(max(hi, lo + 1), len(freqs)) + groups.append((lo, hi)) + return groups + + def _read_url(self) -> str | None: + with self._lock: + return self._url + + def _kill_proc(self) -> None: + with self._lock: + proc = self._proc + self._proc = None + self._capture_kind = None + if proc is not None: + with contextlib.suppress(OSError): + proc.kill() + with contextlib.suppress(Exception): + proc.wait(timeout=1) + + def _use_monitor(self) -> bool: + if self._monitor_failed: + return False + if self.source_mode == "sim": + return False + if self.source_mode == "stream": + return False + return bool(self._parec) + + def _monitor_device(self) -> str: + override = os.environ.get("SOMAFM_SPECTRUM_DEVICE", "").strip() + if override: + return override + if self._monitor_device_cache is not None: + return self._monitor_device_cache + if self._pactl: + try: + result = subprocess.run( + [self._pactl, "get-default-sink"], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + timeout=1.0, + check=False, + ) + sink = result.stdout.strip() + if result.returncode == 0 and sink: + self._monitor_device_cache = f"{sink}.monitor" + return self._monitor_device_cache + except (OSError, subprocess.SubprocessError): + pass + self._monitor_device_cache = "@DEFAULT_MONITOR@" + return self._monitor_device_cache + + def _monitor_command(self) -> list[str] | None: + if not self._parec: + return None + return [ + self._parec, + f"--device={self._monitor_device()}", + "--format=s16le", + "--rate=44100", + "--channels=1", + "--raw", + "--latency-msec=30", + "--client-name=somafm-tui", + ] + + def _spawn( + self, command: list[str], kind: str + ) -> subprocess.Popen[bytes] | None: + try: + proc = subprocess.Popen( + command, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + bufsize=0, + start_new_session=True, + ) + except (OSError, ValueError): + return None + with self._lock: + self._proc = proc + self._capture_kind = kind + self._status = "monitor" if kind == "monitor" else "live" + return proc + + def _stop_proc(self, proc: subprocess.Popen[bytes]) -> None: + with self._lock: + if self._proc is proc: + self._proc = None + self._capture_kind = None + with contextlib.suppress(OSError): + proc.kill() + with contextlib.suppress(Exception): + proc.wait(timeout=1) + + def _run(self) -> None: + while not self._stop.is_set(): + url = self._read_url() + if not url: + self._idle_frame() + self._stop.wait(0.05) + continue + + if self.source_mode == "sim": + self._simulate_for(url, 0.05) + continue + + if self._use_monitor(): + command = self._monitor_command() + proc = self._spawn(command, "monitor") if command else None + if proc is not None: + self._capture( + proc, url, stop_on_url_change=False, status="monitor" + ) + self._stop_proc(proc) + if self._stop.is_set(): + break + if self._read_url() is None: + continue + # The monitor device failed (or vanished). Use the stream + # tap for the rest of the session. + self._monitor_failed = True + with self._lock: + self._status = "sim" + continue + self._monitor_failed = True + + if not self._ffmpeg: + self._simulate_for(url, 0.05) + continue + + proc = self._spawn(self._command(url), "stream") + if proc is None: + with self._lock: + self._status = "sim" + self._simulate_for(url, 0.06, duration=5.0) + continue + self._capture(proc, url, stop_on_url_change=True, status="live") + self._stop_proc(proc) + + if self._stop.is_set(): + break + if self._read_url() == url: + with self._lock: + self._status = "sim" + self._simulate_for(url, 0.06, duration=12.0) + + def _command(self, url: str) -> list[str]: + return [ + self._ffmpeg or "ffmpeg", + "-hide_banner", + "-loglevel", + "error", + "-nostdin", + "-reconnect", + "1", + "-reconnect_streamed", + "1", + "-reconnect_at_eof", + "1", + "-reconnect_delay_max", + "5", + "-fflags", + "nobuffer", + "-flags", + "low_delay", + "-probesize", + "32768", + "-analyzeduration", + "0", + "-user_agent", + USER_AGENT, + "-i", + url, + "-map", + "0:a:0", + "-vn", + "-ac", + "1", + "-ar", + str(self.sample_rate), + "-f", + "s16le", + "-acodec", + "pcm_s16le", + "-", + ] + + def _capture( + self, + proc: subprocess.Popen[bytes], + url: str, + *, + stop_on_url_change: bool, + status: str, + ) -> bool: + """Read PCM until playback stops, the source changes, or the process dies.""" + stdout = proc.stdout + if stdout is None: + return False + fd = stdout.fileno() + got_data = False + buffer = bytearray() + while not self._stop.is_set(): + current = self._read_url() + if current is None: + break + if stop_on_url_change and current != url: + break + + # A monitor can go quiet if the device disappears; don't block the + # analyser thread forever. os.read returns whatever is available + # (instead of waiting for a full 4 KiB), which keeps latency low. + if status == "monitor" and os.name == "posix": + try: + ready, _, _ = select.select([fd], [], [], 2.0) + except (OSError, ValueError): + ready = [fd] + if not ready: + break + + try: + data = os.read(fd, READ_BYTES) + except OSError: + break + if not data: + break + buffer.extend(data) + # Process fixed-size blocks so attack/decay smoothing is tied to + # time (~46 ms per block), regardless of how the pipe fragments. + while len(buffer) >= READ_BYTES: + block = bytes(buffer[:READ_BYTES]) + del buffer[:READ_BYTES] + got_data = True + self.feed_pcm(block, status) + return got_data + + # ------------------------------------------------------------------ # + # signal processing + # ------------------------------------------------------------------ # + + def feed_pcm(self, data: bytes, status: str = "live") -> None: + """Feed raw little-endian mono s16 PCM (also used by the tests).""" + if not data or len(data) < 2: + return + usable = len(data) - (len(data) % 2) + try: + samples = np.frombuffer(data[:usable], dtype=" None: + size = self.fft_size + if samples.size >= size: + self._history[:] = samples[-size:] + else: + count = samples.size + self._history[:-count] = self._history[count:] + self._history[-count:] = samples + chunk = self._history + + spectrum = np.abs(np.fft.rfft(chunk * self._window)) + spectrum /= size * 0.5 + db = 20.0 * np.log10(spectrum + 1e-9) + values = np.array( + [db[lo:hi].max() for lo, hi in self._groups], dtype=np.float32 + ) + + peak_db = float(values.max()) + self._gain = max(peak_db, self._gain - 0.35) + floor = self._gain - 42.0 + norm = np.clip((values - floor) / 42.0, 0.0, 1.0) ** 1.4 + + rms = float(np.sqrt(np.mean(np.square(chunk)) + 1e-12)) + level = float(np.clip((20.0 * math.log10(rms + 1e-9) + 55.0) / 45.0, 0.0, 1.0)) + wave = self._downsample(chunk) + self._apply(norm.astype(np.float32), wave, level, status) + + def _downsample(self, samples: np.ndarray) -> np.ndarray: + target = self.wave_points + if samples.size < target: + padded = np.zeros(target, dtype=np.float32) + padded[-samples.size :] = samples + samples = padded + step = max(1, samples.size // target) + trimmed = samples[samples.size - step * target :] + try: + folded = trimmed.reshape(target, step).mean(axis=1) + except ValueError: + return np.zeros(target, dtype=np.float32) + peak = float(np.max(np.abs(folded))) + if peak > 1e-4: + folded = folded / peak + return np.clip(folded, -1.0, 1.0).astype(np.float32) + + def _apply( + self, values: np.ndarray, wave: np.ndarray, level: float, status: str + ) -> None: + with self._lock: + self._levels = np.maximum(values, self._levels * 0.65).astype(np.float32) + self._peaks = np.maximum(values, self._peaks - 0.04).astype(np.float32) + self._wave = (self._wave * 0.55 + wave * 0.45).astype(np.float32) + self._level = float(self._level * 0.5 + level * 0.5) + self._status = status + + def _idle_frame(self) -> None: + with self._lock: + self._levels = (self._levels * 0.86).astype(np.float32) + self._peaks = np.maximum(self._peaks - 0.02, 0.0).astype(np.float32) + self._wave = (self._wave * 0.8).astype(np.float32) + self._level *= 0.9 + self._status = "idle" + + # ------------------------------------------------------------------ # + # simulation + # ------------------------------------------------------------------ # + + def _simulate_for( + self, url: str, delay: float, duration: float | None = None + ) -> None: + deadline = None if duration is None else time.monotonic() + duration + while not self._stop.is_set(): + if self._read_url() != url: + return + self._simulate_step() + if self._stop.wait(delay): + return + if deadline is not None and time.monotonic() >= deadline: + return + + def _simulate_step(self) -> None: + self._sim_phase += 0.16 + t = self._sim_phase + index = np.linspace(0.0, 1.0, self.band_count, dtype=np.float32) + + humps = np.exp(-((index - 0.16) ** 2) / 0.020) * 0.58 + humps += np.exp(-((index - 0.46) ** 2) / 0.055) * 0.34 + humps += np.exp(-((index - 0.80) ** 2) / 0.130) * 0.20 + + motion = 0.55 + 0.45 * np.sin(t * 1.9 + index * 11.0) + noise = self._rng.random(self.band_count, dtype=np.float32) * 0.22 + beat_phase = (t * 0.55) % 1.0 + beat = math.exp(-(beat_phase / 0.16)) * 0.28 * np.exp(-(index**2) / 0.03) + + values = np.clip((humps * motion + noise + beat), 0.02, 1.0) + values = np.clip(values * (1.15 - 0.25 * index), 0.0, 1.0).astype(np.float32) + + points = np.linspace(0.0, 2.0 * math.pi * 7.0, self.wave_points) + wave = 0.42 * np.sin(points + t * 3.0) * (0.6 + 0.4 * math.sin(t * 0.7)) + wave += 0.20 * np.sin(points * 1.9 + t * 4.3) + wave += self._rng.normal(0.0, 0.05, self.wave_points) + wave = np.clip(wave, -1.0, 1.0).astype(np.float32) + + level = float( + min(1.0, 0.32 + 0.14 * math.sin(t * 1.7) + 0.25 * math.exp(-(beat_phase / 0.2))) + ) + self._apply(values, wave, level, "sim") diff --git a/somafm_tui/theme.tcss b/somafm_tui/theme.tcss new file mode 100644 index 0000000..42c2253 --- /dev/null +++ b/somafm_tui/theme.tcss @@ -0,0 +1,162 @@ +/* somafm-tui — stock terminal palette, transparent backgrounds. + Colours are ANSI names only; this file never paints a background. */ + +Screen { + color: ansi_default; + background: ansi_default; +} + +HeaderBar { + color: ansi_default; + height: 4; + padding: 0 1; + background: ansi_default; +} + +#main { + height: 3fr; + min-height: 10; + background: ansi_default; +} + +#left { + width: 38; + max-width: 50%; + min-width: 24; + height: 1fr; + background: ansi_default; +} + +SearchInput { + display: none; + height: 1; + border: none; + padding: 0 1; + background: ansi_default; + background-tint: transparent; + color: ansi_default; +} + +SearchInput.visible { + display: block; +} + +SearchInput:focus { + background: ansi_default; + background-tint: transparent; +} + +/* The input's cursor/selection ship with theme backgrounds; keep the + terminal background and use an underline instead. */ +SearchInput .input--cursor { + background: ansi_default; + color: ansi_default; + text-style: underline; +} + +SearchInput .input--selection { + background: ansi_default; + color: ansi_default; + text-style: underline; +} + +SearchInput .input--placeholder { + background: ansi_default; + color: ansi_default; + text-style: dim; +} + +SearchInput .input--suggestion { + background: ansi_default; + color: ansi_default; + text-style: dim; +} + +StationList { + color: ansi_default; + width: 1fr; + height: 1fr; + border: round ansi_bright_black; + border-title-color: ansi_default; + padding: 0 1; + background: ansi_default; +} + +StagePanel { + color: ansi_default; + layout: vertical; + width: 1fr; + border: round ansi_bright_black; + border-title-color: ansi_default; + padding: 0 1; + background: ansi_default; +} + +#stage_top { + height: 9; + max-height: 100%; + background: ansi_default; +} + +ArtPane { + color: ansi_default; + width: 20%; + min-width: 12; + max-width: 26; + height: 8; + margin: 0 2 0 0; + background: ansi_default; +} + +ArtPane > .art-image { + width: 100%; + height: 100%; + background: transparent; +} + +StageInfo { + color: ansi_default; + width: 1fr; + height: 1fr; + background: ansi_default; +} + +HistoryPane { + color: ansi_default; + width: 1fr; + height: 1fr; + background: ansi_default; +} + +SpectrumPanel { + color: ansi_default; + height: 2fr; + min-height: 6; + border: round ansi_bright_black; + border-title-color: ansi_default; + padding: 0 1; + background: ansi_default; +} + +StatusBar { + color: ansi_default; + height: 1; + padding: 0 1; + background: ansi_default; +} + +ModalScreen { + color: ansi_default; + align: center middle; + background: ansi_default; +} + +#help { + color: ansi_default; + width: 78; + height: auto; + border: round ansi_bright_black; + border-title-color: ansi_default; + padding: 1 2; + background: ansi_default; +} diff --git a/somafm_tui/widgets/__init__.py b/somafm_tui/widgets/__init__.py new file mode 100644 index 0000000..fe45433 --- /dev/null +++ b/somafm_tui/widgets/__init__.py @@ -0,0 +1 @@ +"""Custom Textual widgets for somafm-tui.""" diff --git a/somafm_tui/widgets/art.py b/somafm_tui/widgets/art.py new file mode 100644 index 0000000..9c49643 --- /dev/null +++ b/somafm_tui/widgets/art.py @@ -0,0 +1,375 @@ +"""ON AIR artwork: real terminal images over an always-present ANSI mosaic.""" + +from __future__ import annotations + +import base64 +import contextlib +import itertools +import os +import random +import sys +from io import BytesIO +from typing import TYPE_CHECKING, cast + +from rich.console import Group, RenderableType +from rich.control import Control +from rich.segment import ControlType, Segment +from rich.style import Style +from rich.text import Text +from textual.dom import NoScreen +from textual.geometry import Region +from textual.strip import Strip +from textual.widget import Widget + +from .. import artwork +from .common import art_lines + +_NULL_STYLE = Style() +_NULL_CONTROL = ((ControlType.CURSOR_FORWARD, 0),) + +if TYPE_CHECKING: + from ..app import SomaFMApp + + +_IMAGE_IDS = itertools.count(random.randrange(1, 2**31 - 1)) + + +def _tgp(data: str) -> str: + return f"\x1b_G{data}\x1b\\" + + +def _terminal_wrap(sequence: str) -> str: + """Wrap Kitty sequences for tmux DCS passthrough.""" + term = os.environ.get("TERM", "") + if os.environ.get("TMUX") or term.startswith(("screen", "tmux")): + return ( + "\x1bPtmux;\x1b" + + sequence.replace("\x1b", "\x1b\x1b") + + "\x1b\\" + ) + return sequence + + +class KittyImage(Widget): + """Direct Kitty graphics placement (no Unicode placeholders or probe).""" + + DEFAULT_CSS = """ + KittyImage { + width: 100%; + height: 100%; + } + """ + + can_focus = False + + def __init__(self, image) -> None: + super().__init__() + self._image = image + self._image_id: int | None = None + self._delete_ids: list[int] = [] + self._placed_key: tuple[int, int, int] | None = None + + @property + def image(self): + return self._image + + @image.setter + def image(self, value) -> None: + if self._image_id is not None and value is not self._image: + self._delete_ids.append(self._image_id) + self._image_id = None + self._image = value + self._placed_key = None + self.refresh(layout=True) + + def _sequences(self, width: int, height: int) -> list[str]: + key = (id(self._image), width, height) + if self._image_id is not None and self._placed_key == key: + return [] + + buffer = BytesIO() + self._image.save(buffer, format="PNG") + payload = base64.b64encode(buffer.getvalue()).decode("ascii") + chunks = [payload[i : i + 4096] for i in range(0, len(payload), 4096)] + + commands: list[str] = [] + for old_id in self._delete_ids: + commands.append(_terminal_wrap(_tgp(f"a=d,i={old_id},q=2"))) + self._delete_ids.clear() + + image_id = next(_IMAGE_IDS) + for index, chunk in enumerate(chunks): + more = 1 if index < len(chunks) - 1 else 0 + commands.append( + _terminal_wrap( + _tgp(f"a=t,i={image_id},f=100,m={more},q=2;{chunk}") + ) + ) + commands.append( + _terminal_wrap( + _tgp(f"a=p,i={image_id},c={width},r={height},C=1,q=2") + ) + ) + self._image_id = image_id + self._placed_key = key + return commands + + def cleanup(self) -> None: + ids = list(self._delete_ids) + if self._image_id is not None: + ids.append(self._image_id) + self._image_id = None + self._delete_ids.clear() + self._placed_key = None + stdout = sys.__stdout__ + if stdout is None: + return + for image_id in ids: + with contextlib.suppress(Exception): + stdout.write( + _terminal_wrap(_tgp(f"a=d,i={image_id},q=2")) + ) + with contextlib.suppress(Exception): + stdout.flush() + + def render_lines(self, crop: Region) -> list[Strip]: + empty = [Strip([]) for _ in range(max(0, crop.height))] + if self._image is None: + return empty + try: + if not self.screen.is_active: + return empty + except NoScreen: + return empty + + width = self.content_size.width + height = self.content_size.height + if width <= 0 or height <= 0: + return empty + + commands = self._sequences(width, height) + if not commands: + return empty + + try: + visible = self.screen.find_widget(self).visible_region + except NoScreen: + return empty + + segments: list[Segment] = [ + Segment( + Control.move_to(visible.x, visible.y).segment.text, + style=_NULL_STYLE, + ) + ] + for command in commands: + segments.append( + Segment(command, style=_NULL_STYLE, control=_NULL_CONTROL) + ) + segments.append( + Segment( + Control.move_to(visible.right, visible.bottom).segment.text, + style=_NULL_STYLE, + ) + ) + last = Strip(segments, cell_length=max(1, crop.width)) + return [Strip([]) for _ in range(max(0, crop.height - 1))] + [last] + + +class ITerm2Image(Widget): + """Inline image via OSC 1337, injected through Textual's render_lines(). + + WezTerm and iTerm2 both implement this natively. ``doNotMoveCursor=1`` + keeps the terminal cursor sane; the widget then restores it to the + bottom-right of the region. + """ + + DEFAULT_CSS = """ + ITerm2Image { + width: 100%; + height: 100%; + } + """ + + can_focus = False + + def __init__(self, image) -> None: + super().__init__() + self._image = image + self._cache_key: tuple[int, int, int] | None = None + self._cache_data = "" + + @property + def image(self): + return self._image + + @image.setter + def image(self, value) -> None: + self._image = value + self._cache_key = None + self._cache_data = "" + self.refresh(layout=True) + + def _sequence(self, width: int, height: int) -> str: + key = (id(self._image), width, height) + if key == self._cache_key: + return self._cache_data + buffer = BytesIO() + self._image.save(buffer, format="PNG") + payload = base64.b64encode(buffer.getvalue()).decode("ascii") + data = ( + "\x1b]1337;File=inline=1;" + f"width={width};height={height};" + f"preserveAspectRatio=0:{payload}\x07" + ) + self._cache_key = key + self._cache_data = data + return data + + def render_lines(self, crop: Region) -> list[Strip]: + empty = [Strip([]) for _ in range(max(0, crop.height))] + if self._image is None: + return empty + try: + if not self.screen.is_active: + return empty + except NoScreen: + return empty + + width = self.content_size.width + height = self.content_size.height + if width <= 0 or height <= 0: + return empty + + try: + visible = self.screen.find_widget(self).visible_region + except NoScreen: + return empty + + segments = [ + Segment( + Control.move_to(visible.x, visible.y).segment.text, + style=_NULL_STYLE, + ), + Segment( + self._sequence(width, height), + style=_NULL_STYLE, + control=_NULL_CONTROL, + ), + Segment( + Control.move_to(visible.right, visible.bottom).segment.text, + style=_NULL_STYLE, + ), + ] + last = Strip(segments, cell_length=max(1, crop.width)) + return [Strip([]) for _ in range(max(0, crop.height - 1))] + [last] + + +class ArtPane(Widget): + """Mosaic background plus an optional protocol-backed image overlay. + + The mosaic is always rendered by this widget. If the terminal image + protocol silently fails (or an image is transparent), the artwork area is + still not blank. + """ + + can_focus = False + + def __init__(self) -> None: + super().__init__() + self._image_widget: Widget | None = None + self._last_data: bytes | None = None + self._last_size: tuple[int, int] = (0, 0) + + # ------------------------------------------------------------------ # + # mosaic background + # ------------------------------------------------------------------ # + + def render(self) -> RenderableType: + app = cast("SomaFMApp", self.app) + station = app.current + if station is None: + return Text("") + width = max(1, self.size.width) + height = max(1, self.size.height) + song = station.now_playing + key = f"{station.id}|{song.artist if song else ''}|{song.title if song else ''}" + return Group(*art_lines(key, width, height)) + + # ------------------------------------------------------------------ # + # image overlay + # ------------------------------------------------------------------ # + + def on_mount(self) -> None: + self.sync_image() + + def on_resize(self, event) -> None: + self.refresh() + if self.display: + self.sync_image() + + def on_unmount(self) -> None: + self._drop_image_widget() + + def sync_image(self, *, force: bool = False) -> None: + # Keep the mosaic behind the image in sync with the current track. + self.refresh() + + app = cast("SomaFMApp", self.app) + station = app.current + data = app.artwork_bytes + width = max(1, self.size.width) + height = max(1, self.size.height) + + if station is not None and data and artwork.supported(): + if ( + not force + and self._image_widget is not None + and self._last_data is data + and self._last_size == (width, height) + ): + return + fitted = artwork.fit_image(data, width, height) + if fitted is not None: + self._show_image(fitted, data, width, height) + return + self._show_fallback() + + def _show_image(self, image, data: bytes, width: int, height: int) -> None: + if self._image_widget is None: + if artwork.PROTOCOL == "tgp": + widget_class = KittyImage + elif artwork.PROTOCOL == "iterm2": + widget_class = ITerm2Image + else: + self._show_fallback() + return + try: + widget = widget_class(image) + except Exception: + self._show_fallback() + return + widget.add_class("art-image") + self._image_widget = widget + self.mount(widget) + else: + self._image_widget.image = image + + self._last_data = data + self._last_size = (width, height) + + def _show_fallback(self) -> None: + self._drop_image_widget() + + def _drop_image_widget(self) -> None: + widget = self._image_widget + self._image_widget = None + self._last_data = None + self._last_size = (0, 0) + if widget is None: + return + with contextlib.suppress(Exception): + widget.cleanup() + with contextlib.suppress(Exception): + widget.image = None + widget.remove() diff --git a/somafm_tui/widgets/common.py b/somafm_tui/widgets/common.py new file mode 100644 index 0000000..22623c0 --- /dev/null +++ b/somafm_tui/widgets/common.py @@ -0,0 +1,212 @@ +"""Small rendering helpers shared by the widgets. + +Every colour in the UI is an ANSI palette name (``cyan``, ``bright_magenta``, +``dim`` …), never an RGB value. That way the TUI adopts whatever colours the +terminal is configured with, and nothing sets a background colour, so +transparent terminals stay transparent. +""" + +from __future__ import annotations + +import hashlib +import math +import os +import time +from datetime import datetime, timezone, tzinfo +from functools import lru_cache +from zoneinfo import ZoneInfo + +from rich.cells import cell_len +from rich.text import Text + +LOCALTIME_PATH = "/etc/localtime" +_UTC_NAMES = {"UTC", "UCT", "GMT", "ETC/UTC", "ETC/UCT", "ETC/GMT", "Z"} + +LEVELS = " ▁▂▃▄▅▆▇█" + +# Bottom-to-top spectrum colours, picked from the terminal's stock ANSI palette. +SPECTRUM_STOPS = ( + "cyan", + "blue", + "magenta", + "bright_magenta", + "yellow", + "bright_yellow", + "red", +) + +ART_STOPS = ("blue", "magenta", "bright_magenta", "yellow", "cyan", "green") + +STATE_STYLE = { + "playing": ("●", "green"), + "loading": ("◌", "yellow"), + "buffering": ("◌", "yellow"), + "paused": ("❚❚", "magenta"), + "stopped": ("○", "bright_black"), + "error": ("✕", "red"), +} + + +def truncate(text: str, width: int) -> str: + if width <= 0: + return "" + if cell_len(text) <= width: + return text + if width == 1: + return "…" + kept: list[str] = [] + used = 0 + for char in text: + char_width = cell_len(char) + if used + char_width > width - 1: + break + kept.append(char) + used += char_width + return "".join(kept) + "…" + + +def fit(text: str, width: int) -> str: + clipped = truncate(text, width) + return clipped + " " * max(0, width - cell_len(clipped)) + + +def compact_count(value: int) -> str: + if value <= 0: + return "—" + if value >= 1_000_000: + return f"{value / 1_000_000:.1f}M" + if value >= 1_000: + return f"{value / 1_000:.1f}k" + return str(value) + + +def time_ago(moment: datetime | None, now: datetime | None = None) -> str: + if moment is None: + return "" + now = now or datetime.now(timezone.utc) + if moment.tzinfo is None: + moment = moment.replace(tzinfo=timezone.utc) + seconds = max(0, int((now - moment).total_seconds())) + if seconds < 60: + return f"{seconds}s ago" + if seconds < 3600: + return f"{seconds // 60}m ago" + if seconds < 86_400: + return f"{seconds // 3600}h ago" + return f"{seconds // 86_400}d ago" + + +def _zone_from_file(path: str) -> tzinfo | None: + try: + with open(path, "rb") as handle: + return ZoneInfo.from_file(handle) + except (OSError, ValueError): + return None + + +@lru_cache(maxsize=1) +def local_zone() -> tzinfo: + """Return the user's local timezone. + + A nix-shell (or an FHS environment) can inherit a build-time ``TZ=UTC``, + which would otherwise make Python show UTC. Prefer an explicit non-UTC + ``TZ``, then the system's ``/etc/localtime`` (the host's configured zone), + and only then Python's idea of local time. + """ + tz_name = (os.environ.get("TZ") or "").strip() + if tz_name and tz_name.upper() not in _UTC_NAMES: + try: + return ZoneInfo(tz_name) + except Exception: + pass + + zone = _zone_from_file(LOCALTIME_PATH) + if zone is not None: + return zone + + if tz_name: + try: + return ZoneInfo(tz_name) + except Exception: + pass + return datetime.now().astimezone().tzinfo or timezone.utc + + +def _local(moment: datetime | None) -> datetime: + zone = local_zone() + if moment is None: + return datetime.now(zone) + if moment.tzinfo is None: + return moment.replace(tzinfo=zone) + return moment.astimezone(zone) + + +def clock_time(moment: datetime | None = None) -> str: + return _local(moment).strftime("%H:%M:%S") + + +def date_label(moment: datetime | None = None) -> str: + return _local(moment).strftime("%a %d %b") + + +def local_time(moment: datetime | None) -> str: + if moment is None: + return "--:--" + return _local(moment).strftime("%H:%M") + + +def gradient(stops: tuple[str, ...], position: float) -> str: + """Pick a colour from a stock palette (no RGB interpolation).""" + if not stops: + return "white" + position = max(0.0, min(1.0, position)) + index = min(len(stops) - 1, int(position * len(stops))) + return stops[index] + + +def marquee( + text: str, width: int, now: float | None = None, speed: float = 7.0, gap: int = 5 +) -> str: + if width <= 0: + return "" + if cell_len(text) <= width: + return text + now = time.monotonic() if now is None else now + cycle = text + " " * gap + length = max(1, cell_len(cycle)) + offset = int(now * speed) % length + return truncate(cycle[offset:] + cycle[:offset], width) + + +def state_glyph(state: str) -> tuple[str, str]: + return STATE_STYLE.get(state, STATE_STYLE["stopped"]) + + +def art_lines(key: str, width: int, height: int) -> list[Text]: + """Procedural album art as a solid ANSI-colour mosaic. + + Filled block glyphs are used (never cell backgrounds), so the cover is + opaque while everything around it stays transparent. + """ + width = max(1, width) + height = max(1, height) + seed = int(hashlib.sha1(key.encode("utf-8")).hexdigest()[:12], 16) + + lines: list[Text] = [] + for row in range(height): + line = Text(no_wrap=True) + ny = row / max(1, height - 1) + for column in range(width): + nx = column / max(1, width - 1) + wave = 0.5 + 0.5 * math.sin(nx * 2.9 + ny * 2.1 + (seed % 11)) + position = (nx * 0.55 + (1.0 - ny) * 0.30 + wave * 0.15) % 1.0 + colour = gradient(ART_STOPS, position) + # Mostly solid blocks with a little texture, like a tiny cover. + glyph = "█" if ((seed + column * 7 + row * 11) % 17) not in (0, 5) else "▓" + line.append(glyph, style=colour) + lines.append(line) + return lines + + +def rule(width: int, color: str = "bright_black", char: str = "─") -> Text: + return Text(char * max(0, width), style=color) diff --git a/somafm_tui/widgets/header.py b/somafm_tui/widgets/header.py new file mode 100644 index 0000000..cc9b48d --- /dev/null +++ b/somafm_tui/widgets/header.py @@ -0,0 +1,97 @@ +"""Header: logo, clock and a live ticker.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, cast + +from rich.cells import cell_len +from rich.console import Group, RenderableType +from rich.text import Text +from textual.widget import Widget + +from .common import ( + SPECTRUM_STOPS, + clock_time, + date_label, + gradient, + marquee, + state_glyph, +) + +if TYPE_CHECKING: + from ..app import SomaFMApp + +LOGO = ( + "▄▀▀▀ ▄▀▀▄ █▄ ▄█ ▄▀▀▄", + "▀▀▀▄ █ █ █ █ █ █▀▀█", + "▄▄▄▀ ▀▄▄▀ █ █ █ █", +) + +MIDDLE = ("SOMA FM RADIO", "LISTENER SUPPORTED", "SAN FRANCISCO") + + +class HeaderBar(Widget): + can_focus = False + + def render(self) -> RenderableType: + app = cast("SomaFMApp", self.app) + width = max(24, self.size.width) + height = max(1, self.size.height) + state = app.playing_state + glyph, colour = state_glyph(state) + + logo_width = max(len(row) for row in LOGO) + middle_width = max(len(row) for row in MIDDLE) + right = (clock_time(), date_label(), f"vol {app.volume}%") + right_width = max(len(item) for item in right) + + show_middle = width >= logo_width + middle_width + 8 + show_right = width >= logo_width + right_width + 8 + if show_middle and show_right: + show_middle = width >= logo_width + 4 + middle_width + 2 + right_width + + lines: list[Text] = [] + for index, row in enumerate(LOGO): + line = Text(no_wrap=True) + for column, char in enumerate(row): + position = ( + column / max(1, logo_width - 1) + index * 0.055 + ) % 1.0 + line.append( + char, style=f"bold {gradient(SPECTRUM_STOPS, position)}" + ) + if show_middle: + line.append(" " * 4) + line.append(MIDDLE[index], style="dim") + if show_right: + used = logo_width + if show_middle: + used += 4 + middle_width + pad = max(1, width - used - right_width) + line.append(" " * pad) + style = "bold" if index == 0 else "dim" + line.append(right[index], style=style) + lines.append(line) + + if height >= 4: + ticker = Text(no_wrap=True) + prefix = f"{glyph} {state.upper():<8} " + ticker.append(prefix, style=f"bold {colour}") + station = app.current + if app.notice: + text = app.notice + text_style = "italic yellow" + elif station is not None: + song = station.now_playing + if song is not None: + text = f"{station.title} · {song.display_artist} — {song.display_title}" + else: + text = f"{station.title} · {station.description}" + text_style = "" + else: + text = "Pick a station — press enter to tune in" + text_style = "italic dim" + ticker.append(marquee(text, max(4, width - cell_len(prefix))), style=text_style) + lines.append(ticker) + + return Group(*lines[:height]) diff --git a/somafm_tui/widgets/help.py b/somafm_tui/widgets/help.py new file mode 100644 index 0000000..88dbaf8 --- /dev/null +++ b/somafm_tui/widgets/help.py @@ -0,0 +1,74 @@ +"""Full-screen keyboard help.""" + +from __future__ import annotations + +import textwrap + +from rich.console import Group +from rich.table import Table +from rich.text import Text +from textual.app import ComposeResult +from textual.screen import ModalScreen +from textual.widgets import Static + +from .common import SPECTRUM_STOPS, gradient + +HELP_ROWS: tuple[tuple[str, str], ...] = ( + ("↑ / ↓ j / k", "move through the station list"), + ("enter", "tune in to the selected station"), + ("space", "play / pause the current station"), + ("s", "stop playback"), + ("] / [", "next / previous station (tunes immediately)"), + ("+ / -", "volume up / down"), + ("m", "mute / unmute"), + ("f", "favourite — favourites float to the top"), + ("/", "filter stations by title, genre or description"), + ("esc", "clear the filter / close overlays"), + ("v", "cycle visualiser: mirrored bars → bars → wave"), + ("r", "refresh station list and song history"), + ("g / G", "jump to the top / bottom of the list"), + ("? ", "this help"), + ("q", "quit"), +) + +ABOUT = ( + ( + "Stations and song history come from somafm.com " + "(channels.json + songs/.json)." + ), + ( + "The spectrum captures your actual PulseAudio/PipeWire output with " + "parec (SYNC); otherwise it falls back to a muted ffmpeg stream tap " + "(FFT, can drift) or a synthesized display (SIM)." + ), + ( + "The desktop picks us up over MPRIS as a " + "somafmtui media player — try `playerctl -l` / `playerctl metadata`." + ), +) + + +class HelpScreen(ModalScreen[None]): + def compose(self) -> ComposeResult: + table = Table.grid(padding=(0, 2), expand=False) + table.add_column(style="bold cyan", no_wrap=True, justify="right") + table.add_column(style="") + for key, description in HELP_ROWS: + table.add_row(key, description) + + 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(" keyboard & credits", style="dim") + + footer = Text() + for paragraph in ABOUT: + for line in textwrap.wrap(paragraph, width=70): + footer.append(line + "\n", style="dim") + + body = Group(header, Text(""), table, Text(""), footer) + yield Static(body, id="help") + + def on_key(self, event) -> None: + event.stop() + self.dismiss(None) diff --git a/somafm_tui/widgets/search.py b/somafm_tui/widgets/search.py new file mode 100644 index 0000000..20b164c --- /dev/null +++ b/somafm_tui/widgets/search.py @@ -0,0 +1,14 @@ +"""Filter input (hidden until you press /).""" + +from __future__ import annotations + +from textual.widgets import Input + + +class SearchInput(Input): + def __init__(self) -> None: + super().__init__( + placeholder="filter stations… (enter tune · esc close)", + id="search", + disabled=True, + ) diff --git a/somafm_tui/widgets/spectrum.py b/somafm_tui/widgets/spectrum.py new file mode 100644 index 0000000..2e22c29 --- /dev/null +++ b/somafm_tui/widgets/spectrum.py @@ -0,0 +1,202 @@ +"""The flashy bit: an ANSI-colour spectrum analyser / oscilloscope.""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import TYPE_CHECKING, cast + +from rich.console import Group, RenderableType +from rich.text import Text +from textual.widget import Widget + +from ..spectrum import Frame, SpectrumEngine +from .common import LEVELS, SPECTRUM_STOPS, gradient + +if TYPE_CHECKING: + from ..app import SomaFMApp + + +class SpectrumPanel(Widget): + can_focus = False + + def __init__(self, engine: SpectrumEngine) -> None: + super().__init__() + self.engine = engine + self.border_title = "SPECTRUM" + self._palettes: dict[tuple[int, int, bool], list[str]] = {} + + # Called from the app ticker so the border title tracks the analyser mode. + def sync(self, mode: str, status: str) -> None: + label = {"both": "MIRROR", "bars": "BARS", "wave": "WAVE"}.get( + mode, mode.upper() + ) + title = f"SPECTRUM · {label} · {status}" + if self.border_title != title: + self.border_title = title + self.refresh() + + def render(self) -> RenderableType: + app = cast("SomaFMApp", self.app) + frame = self.engine.snapshot() + width = max(8, self.size.width) + height = max(2, self.size.height) + mode = getattr(app, "visualizer", "both") + phase = int(getattr(app, "tick_count", 0) // 2) + dim = frame.status == "idle" + + lines: list[Text] + if mode == "wave": + lines = self._wave(frame, width, height, phase, dim) + elif mode == "bars" or height < 8: + lines = self._bars(frame, width, height, phase, dim) + else: + # Default mode: a full-height mirrored spectrum. The top half + # grows up from the centre line, the bottom half grows down from + # it (same values, flipped rows, dimmed like a reflection). + top_height = height // 2 + bottom_height = height - top_height + lines = self._bars(frame, width, top_height, phase, dim) + mirrored = self._bars( + frame, width, bottom_height, phase + 5, True, peaks=False + ) + mirrored.reverse() + lines.extend(mirrored) + return Group(*lines) + + # ------------------------------------------------------------------ # + # drawing + # ------------------------------------------------------------------ # + + def _row_palette(self, height: int, phase: int, dim: bool) -> list[str]: + key = (height, phase % 12, dim) + cached = self._palettes.get(key) + if cached is not None: + return cached + palette: list[str] = [] + for row in range(height): + position = (height - 1 - row) / max(1, height - 1) + colour = gradient(SPECTRUM_STOPS, position) + style = f"bold {colour}" if (row + phase) % 7 == 0 else colour + if dim: + style = f"dim {style}" + palette.append(style) + if len(self._palettes) > 96: + self._palettes.clear() + self._palettes[key] = palette + return palette + + @staticmethod + def _resample(values: Sequence[float], width: int) -> list[float]: + count = len(values) + if count == 0: + return [0.0] * width + if width <= 1 or count == 1: + return [float(values[0])] * width + last = count - 1 + output: list[float] = [] + for column in range(width): + position = column * last / (width - 1) + low = int(position) + high = min(low + 1, last) + fraction = position - low + output.append(values[low] * (1.0 - fraction) + values[high] * fraction) + return output + + @staticmethod + def _resample_mean(values: Sequence[float], width: int) -> list[float]: + count = len(values) + if count == 0: + return [0.0] * width + output: list[float] = [] + for column in range(width): + low = int(column * count / width) + high = max(low + 1, int((column + 1) * count / width)) + high = min(high, count) + chunk = values[low:high] + output.append(sum(chunk) / len(chunk) if chunk else 0.0) + return output + + def _bars( + self, + frame: Frame, + width: int, + height: int, + phase: int, + dim: bool, + peaks: bool = True, + ) -> list[Text]: + values = self._resample(frame.bands, width) + peaks = self._resample(frame.peaks, width) + palette = self._row_palette(height, phase, dim) + scale = height * 8.0 + + lines: list[Text] = [] + for row in range(height): + line = Text(no_wrap=True) + row_top = (height - row - 1) * 8.0 + style = palette[row] + for column in range(width): + units = min(values[column], 1.0) * scale - row_top + peak_units = min(peaks[column], 1.0) * scale - row_top + if units >= 8.0: + line.append("█", style=style) + elif units > 0.0: + level = max(1, min(7, int(units))) + line.append(LEVELS[level], style=style) + elif ( + peaks + and 0.0 <= peak_units < 8.0 + and peak_units > units + 1.0 + ): + line.append("▁", style="dim" if dim else "bold") + else: + line.append(" ", style="") + lines.append(line) + return lines + + def _wave( + self, frame: Frame, width: int, height: int, phase: int, dim: bool + ) -> list[Text]: + wave = self._resample_mean(frame.wave, width) + # The engine normalises each chunk, but smoothing and averaging make + # the trace sit around the middle. A gentle display gain lets the + # oscilloscope use the full height without amplifying silence into + # noise. + peak = max((abs(value) for value in wave), default=0.0) + if peak > 0.02: + gain = min(2.5, 0.9 / peak) + wave = [value * gain for value in wave] + palette = self._row_palette(height, phase, dim) + centre = (height - 1) / 2.0 + amplitude = max(0.75, (height - 1) / 2.0) + + grid = [[" " for _ in range(width)] for _ in range(height)] + previous_row: int | None = None + for column, value in enumerate(wave): + clamped = max(-1.0, min(1.0, float(value))) + position = centre - clamped * amplitude + row = max(0, min(height - 1, round(position))) + grid[row][column] = "█" + delta = position - row + if delta > 0.35 and row + 1 < height: + grid[row + 1][column] = "▄" + elif delta < -0.35 and row - 1 >= 0: + grid[row - 1][column] = "▀" + if previous_row is not None and previous_row != row: + low, high = sorted((previous_row, row)) + for between in range(low + 1, high): + if grid[between][column] == " ": + grid[between][column] = "│" + previous_row = row + + lines: list[Text] = [] + for row in range(height): + line = Text(no_wrap=True) + style = palette[row] + for char in grid[row]: + if char == " ": + line.append(" ", style="") + else: + line.append(char, style=style) + lines.append(line) + return lines diff --git a/somafm_tui/widgets/stage.py b/somafm_tui/widgets/stage.py new file mode 100644 index 0000000..d983f45 --- /dev/null +++ b/somafm_tui/widgets/stage.py @@ -0,0 +1,238 @@ +"""The stage: artwork, now playing details and song history.""" + +from __future__ import annotations + +import contextlib +from typing import TYPE_CHECKING, cast + +from rich.console import Group, RenderableType +from rich.text import Text +from textual.containers import Horizontal +from textual.css.query import NoMatches +from textual.widget import Widget + +from ..api import Song, Station +from .art import ArtPane +from .common import ( + compact_count, + local_time, + marquee, + state_glyph, + time_ago, + truncate, +) + +if TYPE_CHECKING: + from ..app import SomaFMApp + + +class StagePanel(Widget): + """Container for the artwork, track details and history.""" + + can_focus = False + + def __init__(self) -> None: + super().__init__() + self.border_title = "ON AIR" + + def compose(self): + with Horizontal(id="stage_top"): + yield ArtPane() + yield StageInfo() + yield HistoryPane() + + def on_mount(self) -> None: + self.refresh_art() + + def on_resize(self, event) -> None: + self.refresh_art() + + def refresh_stage(self) -> None: + """Refresh the text parts (details + history) without touching images.""" + with contextlib.suppress(NoMatches): + self.query_one(StageInfo).refresh() + self.query_one(HistoryPane).refresh() + + def refresh_art(self) -> None: + """Show/hide and update the artwork pane for the current layout/data.""" + with contextlib.suppress(NoMatches): + art = self.query_one(ArtPane) + visible = ( + self.app.current is not None + and self.size.width >= 56 + and self.size.height >= 8 + ) + force = visible and not art.display + if art.display != visible: + art.display = visible + if visible: + art.sync_image(force=force) + + +class StageInfo(Widget): + """Station heading, now-playing metadata and notices.""" + + can_focus = False + + def render(self) -> RenderableType: + app = cast("SomaFMApp", 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)) + + details = self._details(app, station, station.now_playing, width) + return Group(*details[:height]) + + def _splash(self, width: int, height: int) -> list[Text]: + lines = [ + Text("◤ SOMA FM ◢", style="bold magenta"), + Text("listener-supported radio · san francisco", 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"), + ] + return [ + Text(truncate(line.plain, width), style=line.style) + for line in lines[:height] + ] + + def _details( + self, + app: SomaFMApp, + station: Station, + song: Song | None, + width: int, + ) -> list[Text]: + width = max(8, width) + glyph, colour = state_glyph(app.playing_state) + + heading = Text(no_wrap=True) + heading.append( + truncate(station.title.upper(), max(8, width - 20)), + style="bold magenta", + ) + if station.favorite: + heading.append(" ★", style="bold yellow") + heading.append(" ") + heading.append( + f"{glyph} {app.playing_state.upper()}", style=f"bold {colour}" + ) + + description = Text(truncate(station.description, width), style="dim") + + label = Text(no_wrap=True) + label.append("NOW PLAYING", style="bold cyan") + if song is not None and song.played_at is not None: + label.append(f" · {time_ago(song.played_at)}", style="dim") + + if song is not None: + artist = Text(truncate(song.display_artist, width), style="bold yellow") + title = Text(marquee(song.display_title, width), style="bold") + album = Text(no_wrap=True) + if song.album: + album.append(truncate(song.album, max(8, width - 22)), style="dim") + if station.dj: + if song.album: + album.append(" · ", style="dim") + album.append(f"dj {truncate(station.dj, 18)}", style="dim") + if song.art_url: + album.append(" ▣", style="cyan") + else: + artist = Text("waiting for track data…", style="italic yellow") + fallback = station.last_playing or "tuning in" + title = Text(marquee(fallback, width), style="bold") + album = Text(truncate(station.genre_label, width), style="dim") + + meta = Text(no_wrap=True) + if station.listeners: + meta.append( + f"♪ {compact_count(station.listeners)} listeners", style="dim" + ) + meta.append(" · ", style="dim") + meta.append(station.format_label, style="dim") + meta.append(" · ", style="dim") + meta.append(station.genre_label, style="dim") + + notice = Text(truncate(app.notice, width), style="italic yellow") + + return [ + heading, + description, + Text(""), + label, + artist, + title, + album, + meta, + notice, + ] + + +class HistoryPane(Widget): + """Recently played songs.""" + + can_focus = False + + def render(self) -> RenderableType: + app = cast("SomaFMApp", self.app) + width = max(8, self.size.width) + height = max(1, self.size.height) + station = app.current + if station is None: + return Text("") + + song = station.now_playing + history = [ + entry + for entry in station.history + if not (song is not None and entry.key() == song.key()) + ] + if not history: + return Text("") + + lines: list[Text] = [] + if height >= 3: + lines.append(Text("")) + lines.append(self._header(width)) + for entry in history: + if len(lines) >= height: + break + lines.append(self._history_line(entry, width)) + elif height == 2: + lines.append(self._header(width)) + lines.append(self._history_line(history[0], width)) + else: + lines.append(self._history_line(history[0], width)) + return Group(*lines[:height]) + + @staticmethod + def _header(width: int) -> Text: + header = Text(no_wrap=True) + header.append("LAST PLAYED", style="bold cyan") + header.append(" ") + header.append("─" * max(0, width - 13), style="bright_black") + return header + + @staticmethod + def _history_line(song: Song, width: int) -> Text: + line = Text(no_wrap=True) + line.append(local_time(song.played_at), style="dim") + line.append(" ") + album_width = min(24, max(0, width // 3)) if width >= 56 else 0 + label_width = max( + 6, width - 5 - 2 - (album_width + 2 if album_width else 0) + ) + line.append( + truncate(f"{song.display_artist} — {song.display_title}", label_width), + style="", + ) + if album_width and song.album: + line.append(" ") + line.append(truncate(song.album, album_width), style="italic dim") + return line diff --git a/somafm_tui/widgets/station_list.py b/somafm_tui/widgets/station_list.py new file mode 100644 index 0000000..9d49feb --- /dev/null +++ b/somafm_tui/widgets/station_list.py @@ -0,0 +1,106 @@ +"""Station list with favourites, live markers and a filter.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, cast + +from rich.cells import cell_len +from rich.console import Group, RenderableType +from rich.text import Text +from textual.widget import Widget + +from .common import compact_count, truncate + +if TYPE_CHECKING: + from ..app import SomaFMApp + + +class StationList(Widget): + can_focus = False + + def __init__(self) -> None: + super().__init__() + self.border_title = "STATIONS" + + def set_filter(self, query: str) -> None: + safe_query = query.replace("[", "(").replace("]", ")") if query else "" + title = f"STATIONS · /{safe_query}" if safe_query else "STATIONS" + if self.border_title != title: + self.border_title = title + + def on_mouse_scroll_down(self, event) -> None: + cast("SomaFMApp", self.app).action_cursor_down() + + def on_mouse_scroll_up(self, event) -> None: + cast("SomaFMApp", self.app).action_cursor_up() + + def render(self) -> RenderableType: + app = cast("SomaFMApp", self.app) + width = max(12, self.size.width) + height = max(1, self.size.height) + view = app.view + + if not view: + message = "no stations match" if app.query else "loading stations…" + return Group(Text(truncate(message, width), style="dim italic")) + + first = 0 + if len(view) > height: + first = max(0, min(app.index - height // 2, len(view) - height)) + + lines: list[Text] = [] + for offset in range(min(height, len(view))): + index = first + offset + station = view[index] + selected = index == app.index + current = app.current is not None and app.current.id == station.id + playing = current and app.playing_state not in ("stopped", "error") + + marker = "▌" if selected else " " + star = "★" if station.favorite else " " + play = ( + ("‖" if app.playing_state == "paused" else "▶") if current else " " + ) + + count_text = "" + count_width = 0 + if width >= 26 and station.listeners > 0: + count_text = f"{compact_count(station.listeners)} ♪" + count_width = cell_len(count_text) + 1 + + prefix = marker + star + play + " " + available = width - cell_len(prefix) - count_width + title = truncate(station.title, max(4, available)) + gap = max(0, width - cell_len(prefix) - cell_len(title) - count_width) + + if selected: + base = "" + marker_style = "magenta" + title_style = "bold underline" + count_style = "bold" + elif current: + base = "" + marker_style = "cyan" + title_style = "cyan" + count_style = "dim" + else: + base = "" + marker_style = "" + title_style = "" + count_style = "dim" + + row = Text(no_wrap=True, style=base) + row.append(marker, style=marker_style) + row.append(star, style="bold yellow" if station.favorite else "") + row.append(play, style="cyan" if playing else "") + row.append(" ") + row.append(title, style=title_style) + row.append(" " * gap) + if count_text: + row.append(count_text, style=count_style) + used = cell_len(prefix) + cell_len(title) + gap + cell_len(count_text) + if used < width: + row.append(" " * (width - used)) + lines.append(row) + + return Group(*lines) diff --git a/somafm_tui/widgets/status.py b/somafm_tui/widgets/status.py new file mode 100644 index 0000000..2530394 --- /dev/null +++ b/somafm_tui/widgets/status.py @@ -0,0 +1,90 @@ +"""One-line status / key hint bar.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, cast + +from rich.cells import cell_len +from rich.console import RenderableType +from rich.text import Text +from textual.widget import Widget + +from .common import compact_count, state_glyph, truncate + +if TYPE_CHECKING: + from ..app import SomaFMApp + +HINTS_LONG = ( + "SPACE pause ↑↓ station ENTER tune +/- vol f ★ / filter " + "v visual r refresh ? help q quit" +) +HINTS_MEDIUM = ( + "space pause · ↑↓ station · enter tune · +/- vol · f ★ · / filter · ? help · q quit" +) +HINTS_SHORT = "↑↓ station · space play · ? help · q quit" + + +class StatusBar(Widget): + can_focus = False + + def render(self) -> RenderableType: + app = cast("SomaFMApp", self.app) + width = max(20, self.size.width) + glyph, colour = state_glyph(app.playing_state) + frame = app.spectrum.snapshot() + mpris = app.mpris.status_label + backend = app.player.backend or "no audio" + analyser = frame.label + + listeners = "" + if app.current is not None and app.current.listeners: + listeners = f"♪ {compact_count(app.current.listeners)} " + state = app.playing_state + volume = f"vol {app.volume:>3}%" + + # From most to least informative; pick the first combination that fits. + right_variants = ( + f"{state} {listeners}{volume} {backend} mpris {mpris} {analyser}", + f"{state} {volume} {analyser}", + f"{state} {volume}", + volume, + ) + options = ( + (HINTS_LONG, right_variants[0]), + (HINTS_LONG, right_variants[1]), + (HINTS_MEDIUM, right_variants[0]), + (HINTS_MEDIUM, right_variants[1]), + (HINTS_SHORT, right_variants[1]), + (HINTS_SHORT, right_variants[2]), + (HINTS_SHORT, right_variants[3]), + ) + + chosen: tuple[int, str, str] | None = None + for hint, right_text in options: + right_width = cell_len(glyph) + 1 + cell_len(right_text) + if cell_len(hint) + right_width + 2 <= width: + chosen = (right_width, right_text, hint) + break + + if chosen is None: + right_text = right_variants[-1] + right_width = cell_len(glyph) + 1 + cell_len(right_text) + if width <= right_width + 2: + line = Text(no_wrap=True) + line.append( + truncate(f"{glyph} {right_text}", width), style="dim" + ) + return line + chosen = (right_width, right_text, "") + + right_width, right_text, hint = chosen + left_width = max(0, width - right_width - 2) + left = app.notice or hint + left_style = "italic yellow" if app.notice else "dim" + + line = Text(no_wrap=True) + line.append(truncate(left, left_width), style=left_style) + line.append(" " * max(0, width - left_width - right_width)) + line.append(f"{glyph} ", style=f"bold {colour}") + line.append(right_text, style="dim") + return line