asot_player/asot_player/engine.py

"""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


# Property names vary across mpv releases ("position" vs "playback-time",
# "length" vs "duration"), and newer builds (>= ~0.41) have no state property
# at all.  Candidates are tried in order once per process start; the first one
# that answers with a usable value wins.
_PROP_CANDIDATES: dict[str, tuple[str, ...]] = {
    "state": ("playback-state", "state"),
    "pos":   ("position", "playback-time", "time-pos"),
    "dur":   ("length", "duration"),
}


def _as_float(x) -> float:
    if x is None or isinstance(x, bool):
        return -1.0
    try:
        return float(x)
    except (TypeError, ValueError):
        return -1.0


def _as_bool(x) -> bool:
    if isinstance(x, bool):
        return x
    try:
        return bool(float(x))
    except (TypeError, ValueError):
        return False


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.mpv_log = os.path.join(
            tempfile.gettempdir(), f".asot-ipc-{os.getpid()}.log"
        )
        self._errlog: "io.TextIOWrapper | None" = None
        self.proc: subprocess.Popen | None = None
        self._sock: socket.socket | None = None
        self._cid = 0
        self.url = ""
        self.last_event: dict = {}
        self._prop_names: dict[str, str] = {}
        self._adv: tuple[float, float] | None = None

    # ----------------------------------------------------------------------

    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",
            url,
        ]
        if self.no_config:
            args.append("--no-config")
        self.url = url
        # Keep mpv's own stderr for diagnostics; referenced from the error
        # messages when the stream cannot be started.
        self._errlog = open(self.mpv_log, "w")
        self.proc = subprocess.Popen(
            args,
            stdout=subprocess.DEVNULL,
            stderr=self._errlog,
            stdin=subprocess.DEVNULL,
        )
        self._connect()
        self._errlog.flush()
        # Fresh mpv process → re-discover which property names this build
        # speaks (newer mpv dropped "playback-state" and friends).
        self._prop_names = {}
        self._adv = None
        for logical, cands in _PROP_CANDIDATES.items():
            self._prop_names[logical] = self._discover_property(logical, cands)
        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 "
                    f"(see {self.mpv_log})."
                )
            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).  While waiting, re-probe
        property names for this mpv build: newer builds only report some of
        them once the media metadata has loaded, so the one-shot probe at
        connect time can miss."""
        deadline = time.time() + 18.0
        while time.time() < deadline:
            # Re-discover any property we still haven't found; candidates are
            # retried until one answers or the deadline passes.
            for logical, cands in _PROP_CANDIDATES.items():
                if self._prop_names.get(logical) is None:
                    self._prop_names[logical] = self._discover_property(
                        logical, cands)
            state_name = self._prop_names.get("state")
            state = self._get(state_name) if state_name else None
            if isinstance(state, str) and state in ("playing", "buffering",
                                                   "paused", "eof"):
                return
            dur_name = self._prop_names.get("dur")
            if dur_name:
                d = _as_float(self._get(dur_name))
                if d > 60.0:
                    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}) — "
                    f"check your network or the stream URL "
                    f"(see {self.mpv_log})."
                )
            time.sleep(0.35)
        raise EngineError("timed out waiting for mpv to become ready")

    # ----------------------------------------------------------------------

    def status(self) -> Status:
        """Snapshot of playback state."""
        state_name = self._prop_names.get("state")
        state = self._get(state_name) if state_name else None
        paused = _as_bool(self._get("pause"))
        muted = _as_bool(self._get("mute"))
        volume = _as_float(self._get("volume"))
        # mpv reports "volume" as a 0-100 percentage; normalize to 0-1.
        if volume > 0.0:
            volume /= 100.0
        pos_name = self._prop_names.get("pos")
        dur_name = self._prop_names.get("dur")
        pos = _as_float(self._get(pos_name)) if pos_name else -1.0
        length = _as_float(self._get(dur_name)) if dur_name else -1.0

        state_s = state if (isinstance(state, str) and state) \
                    else self._derived_state(paused, pos, length)
        return Status(
            position=pos,
            length=length,
            state=state_s,
            paused=paused,
            muted=muted,
            volume=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):
        """Set volume; mpv's ``volume`` property is a 0-100 percentage."""
        pct = max(0.0, min(1.0, float(volume))) * 100.0
        self._cmd(["set_property", "volume", round(pct, 2)])

    # ----------------------------------------------------------------------
    # low-level helpers
    # ----------------------------------------------------------------------

    def _derived_state(self, paused: bool, pos: float, length: float) -> str:
        """Infer playback state for mpv builds without a state property
        (e.g. newer mpv >= 0.41).  Position normally advances ~1:1 with real
        time while playing; a position that stays flat for a long stretch means
        buffering or end of stream."""
        now = time.time()
        if paused:
            return "paused"
        if pos < 0.0:
            return "buffering" if length > 0.0 else "playing"
        if length > 10.0 and pos >= length - 2.0:
            return "eof"
        if self._adv is None:
            self._adv = (pos, now)
            return "playing"
        last_pos, last_t = self._adv
        advanced = pos - last_pos
        if advanced > 0.5:               # real progress → update reference
            self._adv = (pos, now)
            return "playing"
        stall = now - last_t             # seconds without real progress
        if stall > 8.0:
            return "eof"                 # no progress for a while → end of stream
        if stall > 2.0:
            return "buffering"
        return "playing"

    def _get(self, name: str):
        """Get a property by name; None when this mpv build lacks it."""
        try:
            return self._cmd(["get_property", name])
        except EngineError as e:
            if "not found" in str(e) or "unavailable" in str(e):
                return None
            raise

    def _discover_property(self, logical: str,
                           candidates: tuple[str, ...]) -> str | None:
        """Try ``candidates`` in order; return the first property this mpv
        build reports a usable value for (None if none work)."""
        for cand in candidates:
            d = self._get(cand)
            if logical == "state":
                if isinstance(d, str) and d:
                    return cand
            else:
                if _as_float(d) != -1.0:
                    return cand
        return None

    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:
                            # Unsolicited event — remember it; keep waiting.
                            self.last_event = obj
                        else:
                            # Ack.  Older mpv acks carry command/command_id/
                            # type; newer builds (>= ~0.41) use
                            # request_id/data/error instead.  Both shapes
                            # carry the result under "data" and no "event".
                            error = obj.get("error")
                            if error and error != "success":
                                raise EngineError(
                                    f"mpv rejected {cmd!r}: {error}")
                            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(["stop"], 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
        try:
            if self._errlog is not None:
                self._errlog.close()
        except OSError:
            pass
        self._errlog = None
        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