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


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