initial

.gitignore

@@ -0,0 +1,2 @@
+__pycache__
+*.egg-info

README.md

@@ -0,0 +1,84 @@
+# asot-player
+
+A terminal player for the **ASOT (A State of Trance — Armin van Buuren)** weekly mix stream from [miroppb.com/ASOT](https://miroppb.com/asot). It plays live MP3 streams with `mpv`, shows a small multi-line ANSI UI (episode number, current song, date, progress bar), and resumes the episode you left off on next launch.
+
+```
+ASOT 1298                      PLAYING
+    Armin van Buuren - A State of Trance (2026)         07 Feb 2026
+[██████████████████████░░░░░░] 00:12:34 / 05:28:00   80% ▓▓▓▓░░
+space play/pause   ←/→ ±15s   ↑/↓ volume   m mute   n next  p prev  q quit
+```
+
+## Features
+
+- **mpv backend** — playback, seeking, pause/mute/volume are all driven over mpv's JSON IPC socket.
+- **Live song/track info without downloading** — the ID3 header of each episode is probed with a `Range: bytes=0-1048576` HTTP request. miroppb embeds per-track `TIT2` frames (plus `CHAP`, `TPE2`, `TALB`, `TYER`, `TXXX`) inside the first MiB, so the "song if known" line usually updates within seconds.
+- **Non-standard writer support** — the parser tolerates the quirks of miroppb's tagger: frames separated by junk gaps, frame sizes declared one byte short, and UTF-16 payloads with repeated BOM pairs (see `asot_player/id3.py`).
+- **Resumes on reopen** — episode, URL, position, length, pause/mute/volume are persisted to `~/.config/asot-player/state.json` (or `$XDG_STATE_HOME/asot-player/`, overridable via `--state-dir` / `ASOT_STATE_DIR`). Re-opening picks up the saved episode.
+- **Auto-advance** — when an episode ends (`eof`), the next episode in the feed starts automatically; at the end of the feed it reports and exits.
+- **Minimal UI** — a few fixed ANSI lines exactly terminal-width in both colour modes; no full-screen / ncurses.
+
+## Installation
+
+Requires Python ≥ 3.10 and the system [mpv](https://mpv.io/) binary (no Python dependencies):
+
+```sh
+# Arch            sudo pacman -S mpv
+# Debian/Ubuntu   sudo apt-get install -y mpv
+# Fedora          sudo dnf install mpv
+# macOS           brew --cask install mpv
+
+uv sync              # builds and installs this package (dev venv)
+asot                 # ready to go
+```
+
+If mpv is installed somewhere outside PATH: `asot --mpv-binary /path/to/mpv`.
+
+## Usage
+
+```sh
+asot                          # latest episode (or resume saved position)
+asot --ep 1298                # start a specific episode
+asot --fresh                  # ignore saved state, start the latest episode
+asot --width 78               # force terminal width (tests, narrow terminals)
+asot --no-color               # disable ANSI colour
+asot --fake                   # no mpv / no stream: exercises UI + state only
+asot --fake-speed 1           # fake engine: seconds of "playback" per tick
+asot --state-dir /tmp/x       # custom state directory (also: $ASOT_STATE_DIR)
+```
+
+**Keys** (single lines, no full-screen mode):
+
+| Key | Action |
+|-----|--------|
+| `space` / `t` | play / pause |
+| `m` | mute toggle |
+| `↑` / `↓` | volume up / down (0.1 steps) |
+| `→` / `←` | forward / backward 15 s |
+| `n` / `p` | next / previous episode |
+| `q` / `Ctrl-C` | quit |
+
+## How it works
+
+- **Feed**: `miroppb.com/rss` is parsed for `<item>` titles matching "Episode NNN" and enclosure URLs.
+- **Metadata**: `fetch_meta(url)` sends `Range: bytes=0-1048576`, parses the ID3 header from the returned bytes (`asot_player/id3.py`). Empty/failed probes just mean "no track info" — playback never blocks on metadata.
+- **Engine**: an mpv process per session with a private Unix socket; status is polled from `playback-state`, `pause`, `mute`, `volume`, `playback-time`, `length`.
+- **State**: small JSON file rewritten whenever position/length/mute/pause/volume change or every 3 s, so reopening (even minutes/hours later) resumes the right episode.
+
+## Development / testing
+
+```sh
+uv run python -m unittest tests.test_asot -v
+# ~35 tests, sub-second (all network mocked; ID3 tests include a real 1 MiB
+# episode head saved at /tmp/ep1298_head.bin when available)
+
+# UI/state smoke test without mpv or a live stream:
+timeout 60 uv run asot --fake --fake-speed 1 \
+    --state-dir /tmp/stF --ep 1298 --width 78
+```
+
+Package layout is flat (no `src/`): `asot_player/{main,engine,feed,id3,state,ui}.py`, entry point `asot = asot_player.main:main`.
+
+## License
+
+MIT.

asot_player/__init__.py

@@ -0,0 +1,3 @@
+"""asot-player — a minimal terminal player for A State of Trance (miroppb.com/ASOT)."""
+
+__version__ = "0.1.0"

asot_player/engine.py

@@ -0,0 +1,350 @@
+"""Drive an external ``mpv`` process from Python over its JSON IPC unix socket.
+
+The whole playback backend is a single mpv subprocess spawned with
+``--terminal=no --really-quiet --input-ipc-server=<socket>`` and controlled
+purely through JSON lines. No Python dependency beyond the system mpv binary;
+status (position / length / state), seeking, pause, mute and volume all come
+from the official mpv client API.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import json
+import os
+import shutil
+import select
+import socket
+import subprocess
+import tempfile
+import time
+from dataclasses import dataclass
+from pathlib import Path
+
+
+class EngineError(RuntimeError):
+    """The playback backend failed at runtime."""
+
+
+class MpvNotFoundError(EngineError):
+    pass
+
+
+def find_mpv(binary: str | None = None) -> str:
+    """Locate an executable mpv.  Raises MpvNotFoundError with install hints."""
+    candidates: list[Path] = []
+    if binary:
+        candidates.append(Path(binary).expanduser())
+    p = shutil.which("mpv")
+    if p:
+        candidates.append(Path(p))
+    for d in (Path.home() / ".local" / "bin", Path("/usr/local/bin"), Path("/opt")):
+        if d.exists():
+            for c in sorted(d.glob("mpv")):
+                if c.is_file():
+                    candidates.append(c)
+    for c in candidates:
+        try:
+            if c.is_file() and os.access(c, os.X_OK):
+                return str(c)
+        except OSError:
+            continue
+    raise MpvNotFoundError(
+        "could not find an 'mpv' binary — please install mpv then retry:\n"
+        "  Arch Linux:            sudo pacman -S mpv\n"
+        "  Debian / Ubuntu:       sudo apt-get install -y mpv\n"
+        "  Fedora:                sudo dnf install mpv\n"
+        "  macOS (Homebrew):      brew install --cask mpv\n"
+        "\n(or run the player with: asot --mpv-binary /path/to/mpv)"
+    )
+
+
+@dataclass(frozen=True)
+class Status:
+    """Snapshot of playback state."""
+
+    position: float   # seconds since episode start
+    length: float     # total length in seconds; -1.0 if unknown
+    state: str        # playing / paused / buffering / eof / stop
+    paused: bool
+    muted: bool
+    volume: float     # 0.0 – 1.0
+
+
+class MpEngine:
+    """An mpv process controlled through its JSON IPC unix socket."""
+
+    def __init__(self, binary: str | None = None, no_config: bool = False,
+                 sock_path: str | None = None):
+        self.bin = find_mpv(binary)
+        self.no_config = no_config
+        self.sock_path = sock_path or os.path.join(
+            tempfile.gettempdir(), f".asot-ipc-{os.getpid()}.sock"
+        )
+        self.proc: subprocess.Popen | None = None
+        self._sock: socket.socket | None = None
+        self._cid = 0
+        self.url = ""
+        self.last_event: dict = {}
+
+    # ----------------------------------------------------------------------
+
+    def alive(self) -> bool:
+        return self.proc is not None and self.proc.poll() is None
+
+    def play(self, url: str, seek_to: float = 0.0, muted: bool = False,
+             volume: float = 0.8):
+        """(Re)start playback of ``url``; resume at ``seek_to`` seconds if set."""
+        self.stop()
+        try:
+            os.unlink(self.sock_path)
+        except OSError:
+            pass
+
+        args = [
+            self.bin,
+            "--no-video",
+            "--terminal=no",  # fully disable mpv terminal use (including keys)
+            "--really-quiet",
+            f"--input-ipc-server={self.sock_path}",
+            "--loop=0",
+        ]
+        if self.no_config:
+            args.append("--no-config")
+        self.url = url
+        self.proc = subprocess.Popen(
+            args,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            stdin=subprocess.DEVNULL,
+        )
+        self._connect()
+        try:
+            self.set_mute(muted)
+            self.set_volume(volume)
+            self._wait_ready()
+        except EngineError:
+            self.stop()
+            raise
+        if seek_to > 1.0:
+            try:
+                self.seek(seek_to)
+            except EngineError:
+                pass
+
+    def _connect(self):
+        """Open the IPC socket (mpv creates it at startup)."""
+        self._sock = None
+        deadline = time.time() + 15.0
+        while time.time() < deadline:
+            if self.proc is not None and self.proc.poll() is not None:
+                rc = self.proc.returncode
+                raise EngineError(
+                    f"mpv exited before playback could start "
+                    f"(return code {rc}) — the stream may be unavailable."
+                )
+            try:
+                self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+                self._sock.connect(self.sock_path)
+                return
+            except OSError:
+                pass
+            time.sleep(0.2)
+        self.stop()
+        raise EngineError("timed out waiting for mpv to create its IPC socket")
+
+    def _wait_ready(self):
+        """Block until playback begins (or fails)."""
+        deadline = time.time() + 18.0
+        while time.time() < deadline:
+            state = self._prop("playback-state")
+            if isinstance(state, str) and state in ("playing", "buffering", "paused", "eof"):
+                return
+            if self.proc is not None and self.proc.poll() is not None:
+                rc = self.proc.returncode
+                raise EngineError(
+                    f"mpv failed to load '{self.url}' (return code {rc}) — "
+                    "check your network or the stream URL."
+                )
+            time.sleep(0.35)
+        raise EngineError("timed out waiting for mpv to become ready")
+
+    # ----------------------------------------------------------------------
+
+    def status(self) -> Status:
+        state = self._prop("playback-state")
+        paused = self._prop("pause")
+        muted = self._prop("mute")
+        volume = self._prop("volume")
+        pos = self._prop("playback-time")
+        length = self._prop("length")
+
+        def _num(x) -> float:
+            if x is None or isinstance(x, bool):
+                return -1.0
+            try:
+                return float(x)
+            except (TypeError, ValueError):
+                return -1.0
+
+        def _bool(x) -> bool:
+            if isinstance(x, bool):
+                return x
+            try:
+                return bool(float(x))
+            except (TypeError, ValueError):
+                return False
+
+        state_s = state if isinstance(state, str) else "stop"
+        return Status(
+            position=_num(pos),
+            length=_num(length),
+            state=state_s,
+            paused=_bool(paused),
+            muted=_bool(muted),
+            volume=_num(volume),
+        )
+
+    def seek(self, seconds: float):
+        """Seek to ``seconds`` from episode start (absolute)."""
+        self._cmd(["seek", round(max(0.0, min(seconds, 9_999_999.0)), 3), "absolute"])
+
+    def set_pause(self, paused: bool):
+        self._cmd(["set_property", "pause", paused])
+
+    def set_mute(self, muted: bool):
+        self._cmd(["set_property", "mute", muted])
+
+    def set_volume(self, volume: float):
+        self._cmd(["set_property", "volume", max(0.0, min(1.0, float(volume)))])
+
+    # ----------------------------------------------------------------------
+
+    def _prop(self, name: str):
+        return self._cmd(["get_property", name])
+
+    def _cmd(self, cmd, timeout: float = 4.0):
+        """Send one JSON command; return its ``data`` (raises on failure)."""
+        if not self.alive():
+            raise EngineError("mpv process is dead")
+        self._cid += 1
+        cmd_id = self._cid
+        self._sock.sendall((json.dumps({"command": cmd, "command_id": cmd_id}) + "\n").encode())
+        deadline = time.time() + timeout
+        while True:
+            remaining = max(0.01, deadline - time.time())
+            rlist, _, _ = select.select([self._sock], [], remaining)
+            if rlist:
+                try:
+                    data = self._sock.recv(65536)
+                    for lin in data.split(b"\n"):
+                        if not lin.strip():
+                            continue
+                        try:
+                            obj = json.loads(lin.decode())
+                        except json.JSONDecodeError:
+                            continue
+                        if obj.get("event") is not None:
+                            self.last_event = obj
+                        else:
+                            if obj.get("command_id") == cmd_id:
+                                return obj.get("data")
+                except (ConnectionResetError, OSError):
+                    raise EngineError("mpv IPC socket closed unexpectedly")
+            if time.time() >= deadline:
+                if self.proc and self.proc.poll() is None:
+                    self.proc.kill()
+                    self.proc.wait()
+                raise EngineError(f"no reply from mpv ({timeout:.1f}s timeout)")
+
+    def stop(self):
+        try:
+            self._cmd(["set_property", "stop", True], timeout=1.5)
+        except Exception:
+            pass
+        try:
+            if self.proc:
+                self.proc.terminate()
+                self.proc.wait(timeout=3.0)
+        except (AttributeError, subprocess.SubprocessError):
+            if self.proc is not None:
+                self.proc.kill()
+                self.proc.wait()
+        try:
+            if self._sock:
+                self._sock.close()
+        except OSError:
+            pass
+        try:
+            os.unlink(self.sock_path)
+        except OSError:
+            pass
+        self.proc = None
+        self._sock = None
+
+
+# ---------------------------------------------------------------------------
+# Fake engine: lets the full player (UI + state + key loop) run without mpv
+# or a live stream.  Used for tests and `asot --fake`.
+# ---------------------------------------------------------------------------
+
+class FakeEngine:
+    def __init__(self, url: str, length: float = 75 * 60.0, speed: float = 1.0):
+        self.url = url
+        self._length = float(length)
+        self._speed = speed
+        self._pos = 0.0
+        self.paused = False
+        self.muted = False
+        self.volume = 0.8
+        self.state = "stop"
+        self.played_url = None
+        self.seek_count = 0
+
+    def play(self, url, seek_to=0.0, muted=False, volume=0.8):
+        self.url = url
+        self.played_url = url
+        self._pos = max(0.0, min(seek_to, self._length))
+        self.paused = False
+        self.muted = bool(muted)
+        self.volume = max(0.0, min(1.0, float(volume)))
+        self.state = "buffering" if self._length < 120 else "playing"
+
+    def status(self) -> Status:
+        if not self.paused and self._length > 0:
+            self._pos += self._speed
+            if self._pos >= self._length:
+                self.state = "eof"
+                self._pos = self._length
+            elif self.played_url == self.url and self._pos <= 1.5:
+                self.state = "buffering"
+            else:
+                self.state = "playing"
+        return Status(
+            position=self._pos,
+            length=self._length,
+            state=self.state,
+            paused=self.paused,
+            muted=self.muted,
+            volume=self.volume,
+        )
+
+    def seek(self, pos_seconds):
+        self.seek_count += 1
+        self._pos = max(0.0, min(pos_seconds, self._length))
+
+    def set_pause(self, paused):
+        self.paused = paused
+
+    def set_mute(self, muted):
+        self.muted = muted
+
+    def set_volume(self, volume):
+        self.volume = max(0.0, min(1.0, float(volume)))
+
+    def stop(self):
+        self.state = "stop"
+        self.played_url = None
+
+    def alive(self) -> bool:
+        return True

asot_player/feed.py

@@ -0,0 +1,97 @@
+"""Fetch and parse the miroppb.com/ASOT RSS feed."""
+
+from __future__ import annotations
+
+import re
+import urllib.request
+import xml.etree.ElementTree as ET
+from dataclasses import dataclass
+from datetime import datetime
+from email.utils import parsedate_to_datetime
+
+BASE = "https://miroppb.com"
+FEED_URL = BASE + "/rss"
+USER_AGENT = (
+    "asot-player/0.1 (a terminal ASOT player; reads episodes for playback) "
+    "https://github.com/asot-player"
+)
+
+_EP_RE = re.compile(r"Episode\s+(\d+)", re.IGNORECASE)
+
+
+@dataclass
+class Episode:
+    number: int
+    url: str
+    pubdate: datetime | None = None
+    title: str = ""
+
+    @property
+    def date_str(self) -> str:
+        return self.pubdate.strftime("%d %b %Y") if self.pubdate else ""
+
+
+def _ua_request(url: str, headers: dict[str, str] | None = None):
+    req = urllib.request.Request(
+        url,
+        headers={**{"User-Agent": USER_AGENT}, **(headers or {})},
+        method="GET",
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=20) as r:
+            return r.status, r.headers, r.read()
+    except Exception as e:  # noqa: BLE001
+        raise ConnectionError(f"fetch failed: {url} ({e})") from e
+
+
+def _parse_date(raw: str) -> datetime | None:
+    if not raw:
+        return None
+    try:
+        return parsedate_to_datetime(raw.strip())
+    except Exception:
+        return None
+
+
+def fetch_episodes(limit: int | None = None) -> list[Episode]:
+    """Return episodes newest-first from the miroppb RSS feed."""
+    _, _, body = _ua_request(FEED_URL)
+    root = ET.fromstring(body)
+    eps: list[Episode] = []
+    for item in root.iter("item"):
+        title = item.findtext("title", "") or ""
+        m = _EP_RE.search(title)
+        if not m:
+            continue
+        num = int(m.group(1))
+        enc = item.find("enclosure")
+        url = (enc.get("url") if enc is not None else "") or f"{BASE}/ASOT/{num}"
+        num_from_url = re.search(r"ASOT_Ep_(\d+)\.mp3$", url)
+        if num_from_url:
+            num = int(num_from_url.group(1))
+        eps.append(
+            Episode(
+                number=num,
+                url=url,
+                pubdate=_parse_date(item.findtext("pubDate", "")),
+                title=title,
+            )
+        )
+    # de-dupe on number (feed can repeat items after a rebuild)
+    seen: dict[int, Episode] = {}
+    for ep in eps:
+        if ep.number not in seen:
+            seen[ep.number] = ep
+    out = sorted(seen.values(), key=lambda e: -e.number)
+    return out[:limit] if limit else out
+
+
+def episode_by_number(episodes: list[Episode], number: int) -> Episode | None:
+    for e in episodes:
+        if e.number == number:
+            return e
+    return None
+
+
+def latest_number(episodes: list[Episode]) -> int:
+    return episodes[0].number

asot_player/id3.py

@@ -0,0 +1,246 @@
+"""Lightweight ID3 tag reader for the episode MP3s.
+
+Reads the first ~1 MiB of a file via an HTTP range request so we can know
+track/album metadata without downloading the full (~150 MB) episode. This
+powers the "song if known" line: miroppb's files carry ID3v2 tags (TPE2 /
+TALB / TIT2 per track / TYER / TXXX), plus CHAP chapter frames that embed
+full text frames (a per-chapter TIT2 inside its declared payload).
+
+The writer used by miroppb is non-standard in three ways this parser tolerates:
+
+  * frames are not always contiguous — junk gaps exist between them, so we
+    search for frame headers anywhere in the tag rather than walking strictly;
+  * every frame declares its payload size exactly one byte too short (the last
+    content byte falls on the boundary of the next frame); the true content
+    length is always ``size + 1``, which is all we need — no frame-boundary
+    hunting is required;
+  * text-frame payloads begin with junk bytes (0x00 0x01) followed by a BOM
+    then UTF-16BE text; TYER and COMM-language fields are NUL-padded plain
+    ASCII, and the writer sometimes repeats the BOM before the text (COMM).
+
+Only text frames are materialised; artwork (APIC), chapter frames and other
+frames are skipped from memory as we scan.
+"""
+
+from __future__ import annotations
+
+import urllib.request
+from dataclasses import dataclass
+
+_RANGE_LIMIT = 1048576  # 1 MiB: comfortably covers the whole ID3 tag
+_MAX_FRAME_SIZE = 1_048_576
+_BOM_SCAN = 32  # any BOM, if present, sits within the first bytes of the payload
+
+
+@dataclass
+class Meta:
+    title: str = ""
+    artist: str = ""
+    album: str = ""
+    year: str = ""
+    comment: str = ""
+    titles: tuple = ()  # all TIT2 values, in order (track list)
+
+    @property
+    def display(self) -> str:
+        """The "song if known" line.  Falls back to artist/album when the
+        file carries a per-track TIT2 list (typical ASOT episodes)."""
+        unique = list(dict.fromkeys(t for t in self.titles if t))
+        artist = self.artist.strip()
+        album = self.album.strip()
+        year = self.year.strip()
+        if len(unique) == 1:
+            t = unique[0]
+            return f"{artist} - {t}" if artist and t else t
+        if artist and album:
+            return f"{artist} - {album} ({year})" if year else f"{artist} - {album}"
+        if artist:
+            return artist
+        if len(unique) > 1:
+            # no artist info, but several tracks: show album/year if any
+            if album:
+                return album + (f" ({year})" if year else "")
+            return unique[0]
+        return self.title or ""
+
+
+def _v2_3_size(b: bytes) -> int:
+    """ID3v2.3 size field: 4 bytes, 7 usable bits each, big-endian."""
+    n = 0
+    for i, byte in enumerate(b[:4]):
+        n |= (byte & 0x7F) << (21 - 7 * i)
+    return n
+
+
+def _v2_4_size(b: bytes) -> int:
+    """ID3v2.4+ size field: 4 bytes, 6 usable bits each, big-endian."""
+    n = 0
+    for i, byte in enumerate(b):
+        n |= (byte & 0x3F) << ((3 - i) * 6)
+    return n
+
+
+def _ascii_run(b: bytes, min_len: int = 3) -> str | None:
+    """Longest run of printable ASCII in ``b`` (>= min_len chars) or None.
+
+    The writer's ASCII payloads (e.g. TYER: NUL padding followed by '2026')
+    only decode via this byte-level scan."""
+    best_start, best_len = 0, 0
+    start = 0
+    for i, ch in enumerate(b):
+        if 0x20 <= ch <= 0x7E:
+            continue
+        run_len = i - start
+        if run_len > best_len:
+            best_start, best_len = start, run_len
+        start = i + 1
+    run_len = len(b) - start
+    if run_len > best_len:
+        best_start, best_len = start, run_len
+    if best_len >= min_len:
+        return bytes(b[best_start: best_start + best_len]).decode("latin-1")
+    return None
+
+
+def _is_bad_char(c: str, allow_nul: bool) -> bool:
+    """Junk characters that mean a decoded payload is binary garbage, not text."""
+    if c == "\ufffd" or c == "\ufeff":  # replacement char / ZWSP (mid-payload BOMs are junk outside TXXX values)
+        # TXXX values legitimately contain the writer's embedded BOM byte
+        if allow_nul and c == "\ufeff":
+            return False
+        return True
+    if c == "\x00" and not allow_nul:
+        return True
+    if "\x01" <= c <= "\x1f" and c not in "\t\n\r":
+        return True
+    return False
+
+
+def _clean(text: str, allow_nul: bool) -> bool:
+    if not text:
+        return False
+    if any(_is_bad_char(c, allow_nul) for c in text):
+        return False
+    printable = sum(ch.isprintable() or ch in " \t\n\r" for ch in text)
+    return printable / len(text) >= 0.5
+
+
+def _decode_payload(payload: bytes, allow_nul: bool = False) -> str | None:
+    """Decode an ID3 text-frame payload (caller passes the full content:
+    declared size + 1), or None if the bytes are binary garbage rather than
+    text.
+
+    Text frames are junk(0x00 0x01) + BOM + UTF-16BE text.  The writer may
+    repeat the BOM before the text (COMM), so try every BOM occurrence within
+    the first bytes and accept the first decode that looks like real text."""
+    if not payload:
+        return None
+    boms: list[int] = []
+    for pat in (b"\xff\xfe", b"\xfe\xff"):
+        off = payload.find(pat)
+        while off != -1 and off <= _BOM_SCAN:
+            boms.append(off)
+            off = payload.find(pat, off + 1)
+    if boms:
+        for k in sorted(set(boms)):
+            body = payload[k:]
+            try:
+                text = body.decode("utf-16")
+            except (UnicodeDecodeError, LookupError):
+                continue
+            text = text.lstrip("\ufeff").rstrip("\x00").strip()
+            if _clean(text, allow_nul):
+                return text
+    # No BOM decode: TYER-style NUL-padded ASCII (or LE pairs without BOM).
+    n = min(len(payload), 64)
+    zeros = sum(payload[i] == 0 for i in range(1, n, 2)) / max(1, (n - 1) // 2 + 1)
+    text = None
+    try:
+        if len(payload) % 2 == 0 and zeros > 0.4:
+            text = payload.decode("utf-16-le")
+        else:
+            text = payload.decode("utf-8")
+    except (UnicodeDecodeError, LookupError):
+        pass
+    if text is not None:
+        text = text.rstrip("\x00").strip()
+        if _clean(text, allow_nul):
+            return text
+    return _ascii_run(payload)
+
+
+def parse_id3(data: bytes) -> Meta:
+    """Parse a buffer starting with an ID3v2 tag header; returns a :class:`Meta`.
+
+    Empty :class:`Meta` when there is no valid header (or no decodable frames)."""
+    meta = Meta()
+    if not data.startswith(b"ID3"):
+        return meta
+    major = data[3]
+    if not (3 <= major <= 5):
+        return meta
+    size_fn = _v2_3_size if major == 3 else _v2_4_size
+    # ten-byte tag header, original tag size in bytes 6..9.
+    tag_size = size_fn(data[6:10])
+    end = min(10 + tag_size, len(data))
+
+    out: dict[str, list] = {}
+    for fid in (b"TIT2", b"TTUN", b"TPE1", b"TPE2", b"TALB", b"TYER", b"TXXX", b"COMM"):
+        allow_nul = fid == b"TXXX"  # key/value separator is a literal NUL
+        off = 10
+        while True:
+            off = data.find(fid, off)
+            if not (10 <= off < end - 4):
+                break
+            fsz = size_fn(data[off + 4: off + 8])
+            if not (1 <= fsz <= _MAX_FRAME_SIZE):
+                off += 1
+                continue
+            pstart = off + (9 if major == 3 else 10)
+            # writer invariant: declared size is exactly one byte short of the
+            # real content — read all of it, clamped to the buffer.
+            plen = min(fsz + 1, len(data) - pstart)
+            t = _decode_payload(data[pstart:pstart + plen], allow_nul=allow_nul)
+            if t is not None and t.strip():
+                out.setdefault(fid.decode(), []).append(t)
+            off += 1
+
+    meta.titles = tuple(out.get("TIT2", []))
+    meta.title = next((t for t in out.get("TIT2", []) if t), "")
+    artists = out.get("TPE2") or out.get("TPE1") or []
+    meta.artist = (artists[0] if artists else "").strip()
+    meta.album = (out.get("TALB") or [""])[0]
+    meta.year = (out.get("TYER") or [""])[0]
+    if out.get("TXXX"):
+        raw = out["TXXX"][0]
+        # TXXX: <key>\0<data>; drop the key part if present
+        meta.comment = raw.split("\x00", 1)[-1] if "\x00" in raw else raw
+        meta.comment = meta.comment.lstrip("\ufeff").strip()
+    elif out.get("COMM"):
+        meta.comment = out["COMM"][0].strip()
+    return meta
+
+
+def fetch_meta(url: str) -> Meta:
+    """Fetch the opening tag bytes of ``url`` and parse them.
+
+    Returns an empty :class:`Meta` when the file has no ID3v2 header or the
+    request keeps failing — callers fall back to "no track info"."""
+
+    def _request() -> bytes:
+        req = urllib.request.Request(
+            url,
+            headers={"User-Agent": "asot-player/0.1 (metadata probe)",
+                     "Range": f"bytes=0-{_RANGE_LIMIT}"},
+        )
+        with urllib.request.urlopen(req, timeout=30) as r:
+            return r.read(_RANGE_LIMIT + 1)
+
+    for _ in range(3):  # retries: upstream / proxy connections are flaky
+        try:
+            raw = _request()
+        except Exception:
+            continue
+        if len(raw) > 128:
+            return parse_id3(raw)
+    return Meta()

asot_player/main.py

@@ -0,0 +1,382 @@
+"""asot — A State of Trance player for your terminal (mpv-driven)."""
+
+from __future__ import annotations
+
+import argparse
+import os
+import re
+import select
+import sys
+import time
+from pathlib import Path
+
+from asot_player.feed import Episode, fetch_episodes
+from asot_player.id3 import Meta, fetch_meta
+from asot_player.engine import EngineError, FakeEngine, MpvNotFoundError, MpEngine
+from asot_player.state import State
+from asot_player.ui import UI, color_safe, terminal_width, fmt_time
+
+try:  # POSIX key handling
+    import termios
+    import tty
+except ImportError:
+    termios = None
+    tty = None
+
+
+# ---------------------------------------------------------------------------
+# key handling (non-full-screen; cbreak mode, restored on exit)
+# ---------------------------------------------------------------------------
+
+class Keyboard:
+    def __init__(self, fd):
+        self.fd = fd
+        self.oldattr = termios.tcgetattr(fd)
+        tty.setcbreak(fd)
+
+    def drain(self, max_keys=8):
+        keys = []
+        for _ in range(max_keys):
+            if not select.select([self.fd], [], [], 0.06)[0]:
+                break
+            b = os.read(self.fd, 1)
+            ch = b.decode("latin1")
+            if ch != "\x1b":
+                keys.append(ch)
+                continue
+            # possible arrow sequence: ESC [ A|B|C|D
+            if not select.select([self.fd], [], [], 0.12)[0]:
+                continue
+            c2 = os.read(self.fd, 1).decode("latin1")
+            if c2 != "[":
+                continue
+            if not select.select([self.fd], [], [], 0.08)[0]:
+                continue
+            c3 = os.read(self.fd, 1).decode("latin1")
+            if c3 in "ABCD":
+                keys.append({"A": "up", "B": "down", "C": "right",
+                             "D": "left"}.get(c3))
+        return keys
+
+    def stop(self):
+        try:
+            termios.tcsetattr(self.fd, termios.TCSU, self.oldattr)
+        except OSError:
+            pass
+
+
+# ---------------------------------------------------------------------------
+# episode resolution helpers
+# ---------------------------------------------------------------------------
+
+def construct_url(base_url: str, number: int) -> str:
+    """Build a probable episode URL when the feed has no such episode."""
+    idx = base_url.rfind("Ep_")
+    if idx < 0:
+        return f"https://bfgc2.miroppb.com/asot/rss/ASOT_Ep_{number}.mp3"
+    prefix = base_url[:idx]
+    suffix = base_url[idx + 3:]
+    if suffix.lower().endswith(".mp3"):
+        return prefix + str(number) + ".mp3"
+    return prefix + suffix
+
+
+def resolve_start(eps, state, number):
+    """Pick which episode to start and whether we resume a saved position.
+
+    Returns (start_ep, resume, resume_position).
+    """
+    if number is not None:
+        found = next((e for e in eps if e.number == number), None)
+        if found is None:
+            return Episode(number=number,
+                           url=construct_url(eps[0].url, number)), False, 0.0
+        start = found
+    elif state is not None and state.ep > 0 and state.url:
+        start = next((e for e in eps if e.number == state.ep
+                      and e.url == state.url), None) or eps[0]
+    else:
+        return eps[0], False, 0.0
+
+    resume = (state is not None
+              and state.ep == start.number
+              and state.url == start.url
+              and state.position > 1.0)
+    if not resume:
+        return start, False, 0.0
+    if state.length > 5.0:
+        pos = min(state.position, state.length - 5.0)
+    else:
+        pos = 0.0  # very short file / fake file — start clean
+    return start, True, max(0.0, float(pos))
+
+
+# ---------------------------------------------------------------------------
+# player core
+# ---------------------------------------------------------------------------
+
+class Player:
+    def __init__(self, eps, engine, ui, state):
+        self.eps = eps
+        self.ep = None
+        self.meta = Meta()
+        self.engine = engine
+        self.ui = ui
+        self.finished = False
+        self.msg_override = ""
+        self.msg_until = 0.0
+        self.state = state
+
+    def _ep_by(self, target):
+        for e in self.eps:
+            if e.number == target:
+                return e
+        return None
+
+    def advance(self, direction):
+        """Jump to the next/previous episode.  Returns True on success."""
+        cur = self.ep.number
+        nxt = self._ep_by(cur + direction)
+        if nxt is None:
+            try:
+                idx = next(i for i, e in enumerate(self.eps) if e.number == cur)
+            except StopIteration:
+                return False
+            j = idx - direction
+            if 0 <= j < len(self.eps):
+                nxt = self.eps[j]
+        if nxt is None:
+            return False
+        kind = "next episode" if direction > 0 else "previous episode"
+        self._load(nxt, kind)
+        return True
+
+    def _load(self, ep, reason):
+        """(Re)start playback of `ep` (fresh start from zero)."""
+        self.ep = ep
+        try:
+            self.meta = fetch_meta(ep.url)
+        except Exception:
+            self.meta = Meta()
+        if not self.engine.alive():
+            # first load of this process — restore saved volume / mute
+            muted = bool(self.state.muted)
+            volume = max(0.0, min(1.0, float(self.state.volume)))
+        else:
+            s = None
+            try:
+                s = self.engine.status()
+            except EngineError:
+                s = None
+            muted = bool(s.muted) if s is not None else False
+            volume = max(0.0, min(1.0, float(s.volume))) if s is not None else 0.8
+
+        self.state.ep = ep.number
+        self.state.url = ep.url
+        self.state.position = 0.0
+        try:
+            self.engine.play(ep.url, seek_to=0.0, muted=muted, volume=volume)
+        except EngineError as e:
+            print(f"[asot] {reason}: {e}", file=sys.stderr)
+            sys.exit(2)
+        self.msg_override = f"{reason.capitalize()} \u2192 ASOT {ep.number:04d}"
+        self.msg_until = time.time() + 3.0
+
+    def _status(self):
+        try:
+            return self.engine.status()
+        except EngineError as e:
+            print(f"[asot] engine error: {e}", file=sys.stderr)
+            sys.exit(2)
+
+    def handle_key(self, key):
+        st = self._status()
+        if key in ("q", "\x03"):             # q or Ctrl-C (cbreak mode)
+            self.finished = True
+        elif key in (" ", "t"):              # play / pause
+            self._engine_cmd(lambda: self.engine.set_pause(not st.paused))
+        elif key in ("m", "M"):              # mute
+            self._engine_cmd(lambda: self.engine.set_mute(not st.muted))
+        elif key == "up":                    # volume +
+            self._engine_cmd(lambda: self.engine.set_volume(min(1.0, st.volume + 0.1)))
+        elif key == "down":                  # volume -
+            self._engine_cmd(lambda: self.engine.set_volume(max(0.0, st.volume - 0.1)))
+        elif key == "right":                 # forward 15s
+            self._engine_cmd(lambda: self.engine.seek(st.position + 15.0))
+        elif key == "left":                  # back 15s
+            self._engine_cmd(lambda: self.engine.seek(st.position - 15.0))
+        elif key == "n":
+            self.advance(+1)
+        elif key == "p":
+            self.advance(-1)
+
+    def _engine_cmd(self, fn):
+        try:
+            fn()
+        except EngineError:
+            pass
+
+    def render_once(self, st):
+        now = time.time()
+        if self.msg_override and now < self.msg_until:
+            status, kind = self.msg_override, "info"
+        elif st.paused:
+            status, kind = "PAUSED", "pause"
+        elif st.state in ("stop", "buffering") and st.position < 1.0:
+            status, kind = "Connecting…", "buffer"
+        else:
+            status, kind = "PLAYING", "ok"
+        song = self.meta.display
+        date = self.ep.date_str if self.ep is not None else ""
+        out = self.ui.render(
+            self.ep.number if self.ep else 0,
+            song, date, status, kind,
+            st.position, st.length, st.volume, st.paused)
+        sys.stdout.write(out)
+        sys.stdout.flush()
+
+    def record(self, st):
+        """Merge latest status into state and persist to disk."""
+        self.state.position = round(st.position, 1)
+        self.state.length = st.length
+        self.state.muted = st.muted
+        self.state.volume = st.volume
+        self.state.paused = st.paused
+        self.state.save()
+
+
+# ---------------------------------------------------------------------------
+# entry point
+# ---------------------------------------------------------------------------
+
+def main():
+    ap = argparse.ArgumentParser(
+        prog="asot",
+        description="A State of Trance player for your terminal (driven by mpv). "
+                    "Resumes the saved episode when re-opened.")
+    ap.add_argument("--ep", type=int,
+                    help="start with a specific episode number (default: latest)")
+    ap.add_argument("--fresh", action="store_true",
+                    help="ignore saved state and play the latest episode")
+    ap.add_argument("--mpv-binary", default=None,
+                    help="path to an mpv binary (overrides PATH lookup)")
+    ap.add_argument("--state-dir", default=None,
+                    help="directory for state.json (default: ~/.config/asot-player)")
+    ap.add_argument("--fake", action="store_true",
+                    help="run the fake engine (no mpv, no stream) — UI/state testing")
+    ap.add_argument("--fake-speed", type=float, default=1.0,
+                    help="fake engine speed: seconds of episode played per tick (default 1)")
+    ap.add_argument("--no-color", action="store_true", help="disable ANSI colours")
+    ap.add_argument("--width", type=int,
+                    help="force terminal width (useful for tests / narrow terminals)")
+    args = ap.parse_args()
+
+    if args.state_dir:
+        os.environ["ASOT_STATE_DIR"] = str(Path(args.state_dir).expanduser())
+
+    print("[asot] fetching episodes…", file=sys.stderr)
+    try:
+        eps = fetch_episodes()
+    except Exception as e:
+        print(f"[asot] could not fetch episode list: {e}", file=sys.stderr)
+        sys.exit(2)
+    if not eps:
+        print("[asot] no episodes found in the feed", file=sys.stderr)
+        sys.exit(2)
+
+    state = State.load() if not args.fresh else None
+    start_ep, resume, resume_pos = resolve_start(eps, state, args.ep)
+
+    if args.fake:
+        # Short fake track; --fake-speed controls seconds-per-tick of progress.
+        engine = FakeEngine(start_ep.url, length=3.0, speed=args.fake_speed)
+    else:
+        try:
+            engine = MpEngine(binary=args.mpv_binary)
+        except MpvNotFoundError as e:
+            print(str(e), file=sys.stderr)
+            sys.exit(2)
+
+    ui = UI(color_safe(), args.width or terminal_width(), 40)
+    kb = None
+    try:
+        stdin_tty = sys.stdin.isatty()
+    except AttributeError:
+        stdin_tty = False
+    if termios is not None and tty is not None and stdin_tty:
+        kb = Keyboard(sys.stdin.fileno())
+
+    p = Player(eps, engine, ui, state or State())
+    if resume:
+        msg = f"[asot] starting ASOT {start_ep.number:04d} " \
+              f"(resuming at {fmt_time(resume_pos)})"
+    else:
+        msg = f"[asot] starting ASOT {start_ep.number:04d}"
+    print(msg, file=sys.stderr)
+
+    try:
+        p._load(start_ep, reason="Starting")
+        if resume and resume_pos > 1.0:
+            engine.seek(resume_pos)
+
+        tick = 0
+        last_save = time.time()
+        while not p.finished and tick < 50_000:
+            t0 = time.time()
+
+            # keys (applied immediately)
+            keys = kb.drain() if kb is not None else []
+            for k in keys:
+                if p.finished:
+                    break
+                p.handle_key(k)
+
+            st = p._status()
+
+            # eof → next episode (auto-advance)
+            if st.state == "eof":
+                if p.advance(1):
+                    continue
+                p.msg_override = f"Finished — end of feed at ASOT {p.ep.number:04d}"
+                p.msg_until = time.time() + 6.0
+                p.render_once(st)
+                p.finished = True
+                break
+
+            p.render_once(st)
+
+            changed = (abs(st.position - p.state.position) > 1.0
+                       or st.muted != p.state.muted
+                       or st.paused != p.state.paused)
+            if changed or time.time() - last_save >= 3.0:
+                p.record(st)
+                last_save = time.time()
+
+            elapsed = time.time() - t0
+            sleep_for = 0.25 if keys else 0.9
+            rem = sleep_for - elapsed
+            if rem > 0.1:
+                time.sleep(min(rem, 0.3))
+            tick += 1
+        sys.stdout.write("\033[H")
+    except EngineError as e:
+        print(f"[asot] playback error: {e}", file=sys.stderr)
+    finally:
+        try:
+            if kb is not None:
+                kb.stop()
+        except Exception:
+            pass
+        try:
+            engine.stop()
+        except Exception:
+            pass
+        try:
+            p.state.save()
+        except Exception:
+            pass
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

asot_player/state.py

@@ -0,0 +1,86 @@
+"""Persistent player state so a reopened session picks up where we left off."""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+@dataclass
+class State:
+    ep: int = 0
+    url: str = ""
+    position: float = 0.0
+    length: float = 0.0
+    paused: bool = False
+    muted: bool = False
+    volume: float = 0.8
+    updated_at: str = ""
+
+    def save(self) -> None:
+        payload = {**asdict(self), "updated_at": _now_iso()}
+        tmpdir = self.state_dir()
+        os.makedirs(tmpdir, exist_ok=True)
+        tmp = Path(os.path.join(tmpdir, ".state.dmp"))
+        tmp.write_text(json.dumps(payload, indent=2))
+        tmp.replace(self.state_file())
+
+    @classmethod
+    def load(cls) -> State | None:
+        path = Path(cls.state_file())
+        if not path.exists():
+            return None
+        try:
+            raw = json.loads(path.read_text())
+        except Exception:
+            return None
+        try:
+            return cls.from_raw(raw)
+        except (TypeError, ValueError, KeyError):
+            return None
+
+    @classmethod
+    def from_raw(cls, raw: dict) -> State:
+        ep = int(raw.get("ep", 0))
+        pos = float(raw.get("position", 0.0))
+        vol = max(0.0, min(1.0, float(raw.get("volume", 0.8))))
+        return cls(
+            ep=ep,
+            url=str(raw.get("url", "")),
+            position=pos,
+            length=float(raw.get("length", 0.0)),
+            paused=bool(raw.get("paused", False)),
+            muted=bool(raw.get("muted", False)),
+            volume=vol,
+            updated_at=str(raw.get("updated_at", "")),
+        )
+
+    @classmethod
+    def state_dir(cls) -> Path:
+        env = os.environ.get("ASOT_STATE_DIR")
+        if env:
+            return Path(env).expanduser()
+        xdg = os.environ.get("XDG_STATE_HOME")
+        if xdg:
+            return Path(xdg).expanduser() / "asot-player"
+        return Path.home() / ".config" / "asot-player"
+
+    @classmethod
+    def state_file(cls) -> Path:
+        return cls.state_dir() / "state.json"
+
+    @classmethod
+    def clear(cls) -> bool:
+        path = Path(cls.state_file())
+        if path.exists():
+            path.unlink()
+            return True
+        return False
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat(sep="T", timespec="seconds")

asot_player/ui.py

@@ -0,0 +1,167 @@
+"""ANSI rendering for the minimal display."""
+
+from __future__ import annotations
+
+import os
+import re
+import shutil
+import sys
+
+_ESCAPES = re.compile(r"\x1b\[[0-9;]*m")
+
+
+def color_safe() -> bool:
+    """ANSI is enabled only for a colouring-capable terminal."""
+    try:
+        tty_ok = sys.stdout.isatty()
+    except AttributeError:
+        return False
+    return tty_ok and not os.environ.get("NO_COLOR")
+
+
+def terminal_width(default: int = 72) -> int:
+    cols = shutil.get_terminal_size().columns
+    return max(48, cols) if cols > 30 else default
+
+
+def visible_len(s: str) -> int:
+    """Length of ``s`` without ANSI escapes — what the user actually sees."""
+    return len(_ESCAPES.sub("", s))
+
+
+def fmt_time(seconds: float | None) -> str:
+    if seconds is None or seconds < 0:
+        return "--:--:--"
+    total = int(seconds + 0.5)
+    h, rem = divmod(total, 3600)
+    m, s = divmod(rem, 60)
+    return f"{h:02d}:{m:02d}:{s:02d}"
+
+
+class C:  # noqa: N801
+    """ANSI colour tokens; inert when disabled."""
+
+    def __init__(self, on: bool = True):
+        self.on = on
+        self.reset = "\033[0m" if on else ""
+        self.dim = "\033[90m" if on else ""
+        self.cyan = "\033[1;96m" if on else ""
+        self.green = "\033[92m" if on else ""
+        self.amber = "\033[93m" if on else ""
+        self.red = "\033[91m" if on else ""
+        self.white = "\033[37m" if on else ""
+        self.dark = "\033[38;5;6m" if on else ""
+        self.vol = "\033[94m" if on else ""
+
+
+class UI:
+    def __init__(self, color: bool, width: int, bar_width: int):
+        self.c = C(color)
+        self.width = max(52, width)  # 52 keeps the bar row fitting with its time + volume
+        self.bar_w = max(16, min(bar_width, self.width - 34))
+
+    def _status(self, text: str, kind: str) -> str:
+        if kind == "ok":
+            return f"{self.c.green}{text}{self.c.reset}"
+        if kind == "buffer":
+            return f"{self.c.amber}{text}{self.c.reset}"
+        if kind == "error":
+            return f"{self.c.red}{text}{self.c.reset}"
+        if kind == "pause":
+            return f"{self.c.white}{text}{self.c.reset}"
+        return f"{self.c.dim}{text}{self.c.reset}"
+
+    def episode_line(self, number: int, status: str, kind: str) -> str:
+        left = f"ASOT {number:04d}"
+        max_status = self.width - len(left) - 1
+        status = status[:max_status] if len(status) > max_status else status
+        gap = self.width - len(left) - len(status)  # >= 1 by construction
+        return f"{self.c.cyan}{left}{self.c.reset}{' ' * gap}{self._status(status, kind)}"
+
+    def song_line(self, song: str, date: str) -> str:
+        prefix = "    "
+        d_raw = (date or "").strip()
+        s_raw = (song or "").rstrip()
+        if not s_raw or not s_raw.strip():
+            s = f"{self.c.dim}(no track info){self.c.reset}"
+        else:
+            # reserve one column for the separator space before the date
+            budget = max(0, self.width - len(prefix) - len(d_raw) - 1)
+            if len(s_raw) > budget:
+                s_raw = s_raw[:budget].rstrip()
+            s = f"{self.c.white}{s_raw}{self.c.reset}"
+        gap = self.width - len(prefix) - visible_len(s) - len(d_raw)  # >= 1
+        d = f"{self.c.amber}{d_raw}{self.c.reset}" if d_raw else ""
+        return prefix + s + " " * gap + d
+
+    def bar_line(self, position, length, volume, paused):
+        """Player bar row.  Exactly ``width`` characters in both colour modes."""
+        w = self.bar_w
+        filled = 0
+        if length and length > 0 and position is not None and 0 <= position <= length:
+            filled = int(round((position / length) * w))
+        on = "\u2588" * filled
+        off = "\u2591" * (w - filled)
+        bar = f"[{self.c.green}{on}{self.c.reset}{self.c.dark}{off}{self.c.reset}]"
+
+        pos_s = fmt_time(position)
+        tot_s = fmt_time(length)
+        vol_int = int(round(volume * 100))
+        meter = ("█" * int(round(volume * 6)) + "░" * (6 - int(round(volume * 6))))
+        vol_part = f"{vol_int:03d}% {meter}"
+        right_time = f"{pos_s} / {tot_s}" if length and length > 0 else f"{pos_s} / ???"
+
+        # gap is computed in *visible* characters, and includes the literal
+        # single space between the time and the volume readout.
+        gap = max(1, self.width - visible_len(bar) - len(right_time) - 1 - len(vol_part))
+        return (f"{bar}{' ' * gap}{self.c.white}{right_time}{self.c.reset} "
+                f"{self.c.vol}{vol_part}{self.c.reset}")
+
+    def hint_line(self) -> str:
+        text = "space play/pause   ←/→ ±15s   ↑/↓ volume   m mute   n next  p prev  q quit"
+        if len(text) > self.width:
+            text = text[:self.width]
+        elif len(text) < self.width:
+            text += " " * (self.width - len(text))
+        return f"{self.c.dim}{text}{self.c.reset}"
+
+    def render(
+        self,
+        number: int,
+        song: str,
+        date: str,
+        status: str,
+        kind: str,
+        position,
+        length,
+        volume: float,
+        paused: bool,
+    ) -> str:
+        lines = [
+            self.episode_line(number, status, kind),
+            self.song_line(song, date),
+            self.bar_line(position, length, volume, paused),
+            self.hint_line(),
+        ]
+        return "\033[H" + "\n".join(lines) + "\n"
+
+    def plain_render(
+        self,
+        number: int,
+        song: str,
+        date: str,
+        status: str,
+        kind: str,
+        position,
+        length,
+        volume: float,
+        paused: bool,
+    ) -> str:
+        """Same layout without ANSI (non-tty / tests)."""
+        lines = [
+            self.episode_line(number, status, kind),
+            self.song_line(song, date),
+            self.bar_line(position, length, volume, paused),
+            self.hint_line(),
+        ]
+        return "\n".join(lines) + "\n\n"

pyproject.toml

@@ -0,0 +1,17 @@
+[build-system]
+requires = ["setuptools>=65"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "asot-player"
+version = "0.1.0"
+description = "A State of Trance (Armin van Buuren) player for your terminal, powered by mpv"
+readme = "README.md"
+license = { text = "MIT" }
+requires-python = ">=3.10"
+dependencies = []
+scripts = { asot = "asot_player.main:main" }
+
+[tool.setuptools.packages.find]
+where = ["."]
+include = ["asot_player*"]

tests/test_asot.py

@@ -0,0 +1,456 @@
+"""Unit tests for asot_player (stdlib unittest only)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import Mock
+
+sys.path.insert(0, os.path.abspath(os.path.dirname(__file__)) + "/..")
+
+import asot_player.main as main_mod
+from asot_player.feed import Episode
+from asot_player.id3 import Meta, parse_id3
+from asot_player.engine import FakeEngine
+from asot_player.main import Player, resolve_start
+from asot_player.state import State
+from asot_player.ui import UI, visible_len
+
+FIXTURE = os.environ.get("ASOT_FIXTURE", "/tmp/ep1298_head.bin")
+
+
+# --------------------------------------------------------------------------
+# id3 helpers (writer layout: junk 0x00 0x01 + BOM + UTF-16BE text; every
+# declared frame size is exactly one byte short of its content)
+# --------------------------------------------------------------------------
+
+def _v23_size(n: int) -> bytes:
+    out = bytearray(4)
+    for i in range(4):
+        out[i] = (n >> (21 - 7 * i)) & 0x7F
+    return bytes(out)
+
+
+def _text_content(text: str) -> bytes:
+    """Writer's canonical text-frame payload: junk 0x00 0x01, an FF FE BOM byte
+    pair, then the text in UTF-16LE byte order (the real writer's layout)."""
+    return b"\x00\x01\xff\xfe" + text.encode("utf-16-le")
+
+
+def _frame(fid: bytes, content: bytes) -> bytes:
+    """One frame honouring the writer invariant: declared size = len - 1."""
+    return fid + _v23_size(len(content) - 1) + b"\x00" + content
+
+
+def build_tag(frames: list[bytes]) -> bytes:
+    """A 10-byte-header ID3v2.3 tag (as emitted by the real writer: ID3 + ver
+    + flags + padding byte + 4-byte size field, first frame at offset 10) with
+    exactly these frames (junk gaps may be inserted as bytes between them)."""
+    payload = b"".join(frames)
+    return b"ID3\x03\x00\x00" + _v23_size(len(payload)) + payload
+
+
+class Id3SyntheticTests(unittest.TestCase):
+    def test_synthetic_basic(self):
+        frames = [
+            _frame(b"TIT2", _text_content("Main Title")),
+            _frame(b"TPE2", _text_content("Some Artist")),
+            _frame(b"TALB", _text_content("Some Album")),
+            _frame(b"TYER", b"\x00\x00\x00" + b"1999"),
+            _frame(b"TXXX", _text_content("COMMENT\x00micro")),
+        ]
+        m = parse_id3(build_tag(frames))
+        self.assertEqual(m.titles, ("Main Title",))
+        self.assertEqual(m.title, "Main Title")
+        self.assertEqual(m.artist, "Some Artist")
+        self.assertEqual(m.album, "Some Album")
+        self.assertEqual(m.year, "1999")
+        self.assertEqual(m.comment, "micro")
+
+    def test_tyer_ascii(self):
+        # TYER: NUL-padded plain ASCII; content is fsz+1 bytes.
+        frames = [
+            b"TIT2" + _v23_size(0) + b"\x00" + _text_content("x"),
+            _frame(b"TYER", b"\x00\x00\x00\x00" + b"2026"),
+        ]
+        m = parse_id3(build_tag(frames))
+        self.assertEqual(m.year, "2026")
+
+    def test_txxx_with_repeated_bom(self):
+        # Writer repeats the BOM inside TXXX values (see real episode data);
+        # key and value are both junk+BOM+UTF-16LE, like every real frame.
+        content = (b"\x00\x01\xff\xfe"
+                   + "DESC".encode("utf-16-le") + b"\x00"
+                   + b"\xff\xfe"
+                   + "microsite".encode("utf-16-le"))
+        m = parse_id3(build_tag([_frame(b"TXXX", content)]))
+        self.assertIn("microsite", m.comment)
+
+    def test_comm_fallback_with_repeated_bom(self):
+        # No TXXX: comment must come from COMM; its text follows a second BOM.
+        content = (b"\x00\x01eng\xff\xfe\x00\x00\xff\xfe"
+                   + "MiroPPB.com".encode("utf-16-le"))
+        m = parse_id3(build_tag([_frame(b"COMM", content)]))
+        self.assertEqual(m.comment, "MiroPPB.com")
+
+    def test_ghost_empty_frame(self):
+        # junk+BOM only (declared size 3) -> decodes to nothing, not a title.
+        frames = [
+            _frame(b"TIT2", b"\x00\x01\xff\xfe"),
+            _frame(b"TIT2", _text_content("Real")),
+        ]
+        m = parse_id3(build_tag(frames))
+        self.assertEqual(m.titles, ("Real",), f"titles={m.titles!r}")
+
+    def test_odd_utf16_body_does_not_crash(self):
+        frames = [_frame(b"TIT2", b"\x00\x01\xff\xfe\x41")]  # odd-length body
+        m = parse_id3(build_tag(frames))
+        self.assertEqual(m.title, "")
+
+    def test_no_header_or_bad_major(self):
+        self.assertEqual(parse_id3(b""), Meta())
+        self.assertEqual(parse_id3(b"NOTAG").title, "")
+        m = parse_id3(b"ID3\x02\x00\x00\x00\x00" + b"\x00" * 4)
+        self.assertEqual(m.title, "")
+
+    def test_frames_scanned_among_junk_gaps(self):
+        frames = [
+            b"\x00" * 5,
+            _frame(b"TPE2", _text_content("Gap Artist")),
+            bytes([0xaa] * 37),
+            _frame(b"TIT2", _text_content("Later Title")),
+        ]
+        m = parse_id3(build_tag(frames))
+        self.assertIn("Later Title", m.titles)
+        self.assertEqual(m.artist, "Gap Artist")
+
+    def test_display_fallbacks(self):
+        m = Meta(titles=("A Song",), artist="The Band")
+        self.assertEqual(m.display, "The Band - A Song")
+        m = Meta(titles=("One", "Two"), artist="The Band", album="Records", year="2026")
+        self.assertEqual(m.display, "The Band - Records (2026)")
+        self.assertEqual(Meta(artist="Solo").display, "Solo")
+        self.assertEqual(Meta().display, "")
+        self.assertEqual(Meta(titles=("X", "X", ""), artist="A").display, "A - X")
+        # no artist, several tracks -> album/year fallback
+        self.assertEqual(Meta(titles=("A", "B"), album="Alb").display, "Alb")
+
+    def test_real_head_fixture(self):
+        """Parse the 1 MiB real fixture of ASOT_Ep_1298 (saved from miroppb)."""
+        if not os.path.isfile(FIXTURE):
+            self.skipTest("fixture not available in this environment")
+        with open(FIXTURE, "rb") as f:
+            m = parse_id3(f.read())
+        self.assertGreaterEqual(len(m.titles), 35)
+        self.assertEqual(m.artist, "Armin van Buuren")
+        self.assertEqual(m.album, "A State of Trance")
+        self.assertEqual(m.year, "2026")
+        self.assertIn("MiroPPB.com", m.comment or "")
+        self.assertTrue("Episode 1298" in " ".join(m.titles))
+
+
+# --------------------------------------------------------------------------
+# resolve_start (main.py)
+# --------------------------------------------------------------------------
+
+def fake_eps():
+    return [
+        Episode(number=1301, url="https://miroppb.com/b/rss/ASOT_Ep_1301.mp3"),
+        Episode(number=1300, url="https://miroppb.com/b/rss/ASOT_Ep_1300.mp3"),
+        Episode(number=1299, url="https://miroppb.com/b/rss/ASOT_Ep_1299.mp3"),
+        Episode(number=1298, url="https://miroppb.com/b/rss/ASOT_Ep_1298.mp3"),
+    ]
+
+
+class ResolveStartTests(unittest.TestCase):
+    def setUp(self):
+        self.eps = fake_eps()
+
+    def test_explicit_episode_found(self):
+        ep, resume, pos = resolve_start(self.eps, None, 1299)
+        self.assertEqual(ep.number, 1299)
+        self.assertFalse(resume)
+        self.assertEqual(pos, 0.0)
+
+    def test_explicit_episode_missing_constructed(self):
+        ep, resume, pos = resolve_start(self.eps, None, 1999)
+        self.assertEqual(ep.number, 1999)
+        self.assertFalse(resume)
+        self.assertIn("1999", ep.url)
+        self.assertTrue(ep.url.endswith(".mp3"), f"bad URL: {ep.url!r}")
+
+    def test_state_resumes_matching_episode(self):
+        st = State()
+        st.ep = 1298
+        st.url = self.eps[-1].url
+        st.position = 28.0
+        st.length = 30.0
+        ep, resume, pos = resolve_start(self.eps, st, None)
+        self.assertEqual(ep.number, 1298)
+        self.assertTrue(resume)
+        # clamped to length - 5 so we don't seek into the last seconds
+        self.assertEqual(pos, 25.0)
+
+    def test_state_short_length_starts_clean(self):
+        # length <= 5 s (fake-engine size): marked as resuming but position
+        # is reset to 0 so playback starts clean.
+        st = State()
+        st.ep = 1298
+        st.url = self.eps[-1].url
+        st.position = 2.0
+        st.length = 3.0
+        ep, resume, pos = resolve_start(self.eps, st, None)
+        self.assertEqual(ep.number, 1298)
+        self.assertTrue(resume)
+        self.assertEqual(pos, 0.0)
+
+    def test_state_not_in_feed_falls_back(self):
+        st = State()
+        st.ep = 500  # not present (feed was rebuilt without it)
+        ep, resume, _ = resolve_start(self.eps, st, None)
+        self.assertEqual(ep.number, self.eps[0].number)
+        self.assertFalse(resume)
+
+    def test_position_zero_no_resume(self):
+        st = State()
+        st.ep = 1298
+        st.url = self.eps[-1].url
+        st.position = 0.0
+        _, resume, _ = resolve_start(self.eps, st, None)
+        self.assertFalse(resume)
+
+
+# --------------------------------------------------------------------------
+# Player.advance (direction + edge behaviour)
+# --------------------------------------------------------------------------
+
+class PlayerAdvanceTests(unittest.TestCase):
+    def setUp(self):
+        # patch must span the whole test body, not just setUp: _load() runs
+        # inside test methods, so the module attribute has to stay mocked.
+        self._orig_fetch_meta = main_mod.fetch_meta
+        main_mod.fetch_meta = Mock(return_value=Meta())
+        self.episodes = [
+            Episode(number=1301, url="https://x/1301.mp3"),
+            Episode(number=1300, url="https://x/1300.mp3"),
+            Episode(number=1299, url="https://x/1299.mp3"),
+            Episode(number=1298, url="https://x/1298.mp3"),
+        ]
+        self.p = Player(eps=self.episodes, engine=FakeEngine("u", 3.0),
+                        ui=UI(False, 56, 40), state=State())
+
+    def tearDown(self):
+        main_mod.fetch_meta = self._orig_fetch_meta
+        del self._orig_fetch_meta
+
+    def load(self, number):
+        for e in self.p.eps:
+            if e.number == number:
+                self.p._load(e, "test")
+                break
+        return self.p
+
+    def test_next_and_prev_by_number(self):
+        self.load(1299)
+        self.assertTrue(self.p.advance(+1))
+        self.assertEqual(self.p.ep.number, 1300)
+        self.assertTrue(self.p.advance(-1))
+        self.assertEqual(self.p.ep.number, 1299)
+        self.assertTrue(self.p.advance(-1))
+        self.assertEqual(self.p.ep.number, 1298)
+        self.assertTrue(self.p.advance(+1))
+        self.assertEqual(self.p.ep.number, 1299)
+
+    def test_edge_end_of_feed(self):
+        self.load(1301)
+        self.assertFalse(self.p.advance(+1))
+        self.assertEqual(self.p.ep.number, 1301)
+        self.load(1298)
+        self.assertFalse(self.p.advance(-1))
+        self.assertEqual(self.p.ep.number, 1298)
+
+    def test_msg_override_on_advance(self):
+        self.load(1299)
+        self.p.advance(+1)
+        self.assertIn("ASOT 1300", self.p.msg_override)
+        self.assertTrue("next" in self.p.msg_override.lower())
+
+
+# --------------------------------------------------------------------------
+# FakeEngine behaviour
+# --------------------------------------------------------------------------
+
+class FakeEngineTests(unittest.TestCase):
+    @staticmethod
+    def make(speed=1.0, length=3.0):
+        return FakeEngine("u", length=length, speed=speed)
+
+    def test_position_advances_per_status(self):
+        e = self.make(speed=2.5, length=10.0)
+        e.play("u")
+        self.assertEqual(e.status().position, 2.5)
+        self.assertEqual(e.status().position, 5.0)
+
+    def test_eof_reached_and_clamped(self):
+        e = self.make(length=3.0)
+        e.play("u")
+        e.status()  # 1
+        e.status()  # 2
+        s = e.status()
+        self.assertEqual(s.state, "eof")
+        self.assertEqual(s.position, 3.0)
+
+    def test_buffering_then_playing(self):
+        e = self.make(length=10.0)
+        e.play("u")
+        self.assertIn(e.status().state, ("buffering",))
+        self.assertEqual(e.status().state, "playing")
+
+    def test_seeks_clamp(self):
+        # speed 0 -> status() does not advance the position, keeps asserts exact
+        e = self.make(length=5.0, speed=0.0)
+        e.play("u")
+        e.seek(999.0)
+        self.assertEqual(e.status().position, 5.0)
+        e.seek(-10.0)
+        self.assertEqual(e.status().position, 0.0)
+        e.seek(2.0)
+        self.assertEqual(e.status().position, 2.0)
+        self.assertEqual(e.seek_count, 3)
+
+    def test_volume_clamps(self):
+        e = self.make()
+        e.play("u")
+        e.set_volume(1.4)
+        self.assertEqual(e.status().volume, 1.0)
+        e.set_volume(-0.2)
+        self.assertEqual(e.status().volume, 0.0)
+        e.set_volume(0.75)
+        self.assertAlmostEqual(e.status().volume, 0.75)
+
+    def test_seek_to_on_play(self):
+        # speed 0 -> no implicit advance between play() and status()
+        e = self.make(length=10.0, speed=0.0)
+        e.play("u", seek_to=4.0)
+        self.assertEqual(e.status().position, 4.0)
+        e.play("u", seek_to=99.0)  # clamped to length
+        self.assertEqual(e.status().position, 10.0)
+
+    def test_stop_resets_url_tracking(self):
+        e = self.make()
+        e.play("u")
+        e.stop()
+        self.assertEqual(e.state, "stop")
+        self.assertIsNone(e.played_url)
+
+
+# --------------------------------------------------------------------------
+# State: round-trip / corrupt file / env dir
+# --------------------------------------------------------------------------
+
+class StateTests(unittest.TestCase):
+    def setUp(self):
+        self._dir = tempfile.TemporaryDirectory()
+        self._saved = os.environ.pop("ASOT_STATE_DIR", None)
+        os.environ["ASOT_STATE_DIR"] = self._dir.name
+
+    def tearDown(self):
+        del os.environ["ASOT_STATE_DIR"]
+        if self._saved is not None:
+            os.environ["ASOT_STATE_DIR"] = self._saved
+        self._dir.cleanup()
+
+    def test_roundtrip(self):
+        st = State(ep=1298, url="https://x/1.mp3", position=21.5, length=100.0,
+                   paused=False, muted=True, volume=0.42)
+        st.save()
+        loaded = State.load()
+        self.assertIsNotNone(loaded)
+        self.assertEqual(loaded.ep, 1298)
+        self.assertEqual(loaded.url, "https://x/1.mp3")
+        self.assertAlmostEqual(loaded.position, 21.5)
+        self.assertTrue(loaded.muted)
+        self.assertAlmostEqual(loaded.volume, 0.42)
+        self.assertTrue(loaded.updated_at)
+
+    def test_load_corrupt_file_returns_none(self):
+        State().save()
+        State.state_file().write_text("not json at all {{{{", encoding="utf-8")
+        self.assertIsNone(State.load())
+
+    def test_load_empty_file_returns_none(self):
+        State().save()
+        State.state_file().write_text("", encoding="utf-8")
+        self.assertIsNone(State.load())
+
+    def test_load_missing_returns_none(self):
+        self.assertIsNone(State.load())
+
+    def test_from_raw_defaults_and_clamps(self):
+        st = State.from_raw({"ep": 42, "volume": 3.0, "position": 5,
+                             "length": "1.0"})
+        self.assertEqual(st.ep, 42)
+        self.assertEqual(st.volume, 1.0)
+        self.assertEqual(st.position, 5.0)
+        self.assertEqual(st.length, 1.0)
+
+    def test_clear(self):
+        State().save()
+        self.assertTrue(os.path.exists(State.state_file()))
+        self.assertTrue(State.clear())
+        self.assertFalse(os.path.exists(State.state_file()))
+        self.assertFalse(State.clear())
+
+
+# --------------------------------------------------------------------------
+# UI: every line is exactly `width` visible characters in both colour modes
+# --------------------------------------------------------------------------
+
+def rendered_lines(ui, song, date, status, kind, position, length, volume):
+    out = ui.render(1298, song, date, status, kind, position, length, volume, False)
+    assert out.startswith("\x1b[H")
+    return out[len("\x1b[H"):].split("\n")  # last element is "" (trailing newline)
+
+
+class UITests(unittest.TestCase):
+    @staticmethod
+    def widths_and_colors():
+        for w in (52, 53, 78, 120):
+            for color in (True, False):
+                yield UI(color, w, 40), w
+
+    def test_line_lengths_exact(self):
+        for ui, w in self.widths_and_colors():
+            lines = [ui.episode_line(1298, "PLAYING", "ok"),
+                     ui.episode_line(1298, "Finished — end of feed at ASOT 1301", "info"),
+                     ui.song_line("", ""),
+                     ui.song_line("Armin van Buuren - A State of Trance (2026) Extra Long",
+                                  "07 Feb 2026"),
+                     ui.bar_line(None, None, 0.5, False),
+                     ui.bar_line(0.0, 0.0, 0.5, False),
+                     ui.bar_line(1234.5, 7200.0, 0.8, True),
+                     ui.hint_line()]
+            for ln in lines:
+                self.assertEqual(visible_len(ln), w, f"visible len != width: {ln!r}")
+
+    def test_rendered_block_lengths_exact(self):
+        for ui, w in self.widths_and_colors():
+            body = rendered_lines(ui, "Armin - A Long Song Title That Keeps Going",
+                                  "07 Feb 2026", "Buffering", "buffer",
+                                  333.5, 7200.0, 0.85)
+            for ln in body[:-1]:
+                self.assertEqual(visible_len(ln), w, f"visible len != width: {ln!r}")
+
+    def test_plain_mode_no_escapes(self):
+        ui = UI(False, 80, 40)
+        out = ui.plain_render(9, "s", "d", "ok", "ok", 1.0, 10.0, 0.5, False)
+        self.assertNotIn("\x1b", out)
+        lines = [l for l in out.splitlines() if l]
+        self.assertEqual(len(lines), 4)
+        self.assertTrue(all(len(l) == 80 for l in lines))
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

uv.lock

@@ -0,0 +1,8 @@
+version = 1
+revision = 5
+requires-python = ">=3.10"
+
+[[package]]
+name = "asot-player"
+version = "0.1.0"
+source = { editable = "." }