testsAndMisc/python_pkg/wsg_grabber/player.py

"""Drives mpv as a subprocess embedded in a Tk frame, over its JSON IPC socket.

Not ``python-mpv``: that binding calls ``ctypes.util.find_library("mpv")`` at
import time and raises ``OSError`` wherever libmpv is absent, which includes
this repo's CI runner. It would pass locally and fail on push, and the usual
lazy-import escape trips ``PLC0415`` with no suppression available. A
subprocess needs only ``subprocess``, ``socket`` and ``json``.

``--wid`` is an mpv feature, so embedding behaves identically either way: mpv
reparents its own window onto the Tk frame and, per ``man mpv``, "will always be
resized to cover the parent window fully". That last part is why there is no
resize method here: setting ``geometry`` on an embedded mpv fights the
automatic sizing and leaves the video drawn at the wrong size.

``build_argv`` and ``encode`` are pure and tested directly. The rest is process
and socket handling: connecting, draining, and shutting down without leaving an
orphan. No reviewing decisions are made here.
"""

from __future__ import annotations

import socket
import subprocess
import threading
import time
from typing import TYPE_CHECKING

from python_pkg.wsg_grabber import logs

# Re-exported through __all__ so player.Player / player.build_argv /
# player.encode keep working for ui.py and test_player.py; the wire format and
# the protocol live in their own modules to keep this one under the 250-line cap.
from python_pkg.wsg_grabber._mpv_ipc import (
    build_argv,
    encode,
    note_errors,
    probe_properties,
)
from python_pkg.wsg_grabber._player_api import Player
from python_pkg.wsg_grabber.constants import IPC_READY_TIMEOUT_S

__all__ = ["MpvPlayer", "Player", "build_argv", "encode"]

if TYPE_CHECKING:
    from collections.abc import Sequence
    from pathlib import Path

_SOCKET_POLL_S = 0.05
_READ_CHUNK = 65536
_READ_TIMEOUT_S = 0.2
_READER_JOIN_S = 5.0


class MpvPlayer:
    """An mpv process drawing into a Tk frame."""

    def __init__(self, wid: int, ipc_path: Path) -> None:
        """Start mpv and connect to its control socket.

        Args:
            wid: X11 window id of the frame to draw into.
            ipc_path: Where mpv should create its control socket.

        Raises:
            TimeoutError: If mpv does not create the socket in time, which
                means it failed to start and there is nothing to talk to.
            OSError: If the socket exists but refuses the connection, e.g. a
                stale file left by a previous run.
        """
        ipc_path.unlink(missing_ok=True)
        ipc_path.parent.mkdir(parents=True, exist_ok=True)
        self._ipc_path = ipc_path
        argv = build_argv(wid, ipc_path)
        logs.event("player.start", wid=wid, socket=str(ipc_path), argv=argv)
        self._process = subprocess.Popen(argv)
        try:
            self._socket = self._connect()
        except BaseException:
            # Reaping matters as much as killing: an unwaited child stays a
            # zombie, and the caller is about to abandon this object entirely.
            self._process.kill()
            self._process.wait(timeout=IPC_READY_TIMEOUT_S)
            raise
        self._stopping = threading.Event()
        self._reader = threading.Thread(
            target=self._drain,
            name="mpv-ipc-reader",
            daemon=False,
        )
        self._reader.start()
        logs.event("player.ready", pid=self._process.pid)

    def _drain(self) -> None:
        """Continuously read and discard what mpv sends us.

        This is not optional bookkeeping, it is what keeps the player working.
        mpv pushes an asynchronous event stream plus a reply per command down
        the same socket. A client that only ever writes lets that backlog fill
        the kernel buffer; mpv's writer for this connection then blocks, and it
        stops reading our commands. The symptom is nasty because nothing
        errors: ``sendall`` still succeeds, mpv stays alive and keeps playing
        whatever it had, so the reviewer advances through filenames while the
        picture never changes. Measured before this existed: commands were
        silently ignored from roughly the thirty-fifth video onwards, while the
        very same command sent on a fresh connection worked instantly.

        Replies carrying an error are logged, which is also how a file mpv
        cannot decode becomes visible instead of just looking stuck.
        """
        while not self._stopping.is_set():
            try:
                chunk = self._socket.recv(_READ_CHUNK)
            except TimeoutError:
                continue
            except OSError:
                return
            if not chunk:
                return
            self._note_errors(chunk)

    def _note_errors(self, chunk: bytes) -> None:
        """Log any failure mpv reported back.

        Delegates to :func:`._mpv_ipc.note_errors`; kept as a bound method
        because the reader thread and the tests both reach it that way.

        Args:
            chunk: Raw bytes just read from the socket.
        """
        note_errors(chunk)

    def _connect(self) -> socket.socket:
        """Wait for mpv's socket to appear and connect to it.

        Returns:
            socket.socket: Connected unix socket.

        Raises:
            TimeoutError: If the socket never appears.
            OSError: If connecting to it fails.
        """
        deadline = time.monotonic() + IPC_READY_TIMEOUT_S
        while not self._ipc_path.exists():
            if time.monotonic() >= deadline:
                # __init__ owns killing and reaping the process on every
                # failure path; doing it here too would just double up.
                msg = f"mpv did not create {self._ipc_path} within the timeout"
                raise TimeoutError(msg)
            time.sleep(_SOCKET_POLL_S)
        control = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        try:
            control.connect(str(self._ipc_path))
        except BaseException:
            control.close()
            raise
        # so the reader thread can notice the stop flag rather than
        # blocking in recv() forever at shutdown
        control.settimeout(_READ_TIMEOUT_S)
        return control

    def _send(self, command: Sequence[object]) -> None:
        """Write a command to mpv.

        A failure here is logged rather than raised -- mpv may simply have
        exited, and is_alive() reports that to the caller -- but it is never
        silently discarded, because a dead socket makes a working reviewer and
        a frozen one look identical from the outside.

        Args:
            command: Command name followed by its arguments.
        """
        try:
            self._socket.sendall(encode(command))
        except OSError as exc:
            logs.error(
                "player.send_failed",
                command=list(command),
                detail=str(exc),
                alive=self.is_alive(),
            )
            return
        logs.debug("player.command", command=list(command))

    def play(self, path: Path) -> None:
        """Show *path*, replacing whatever is on screen.

        Args:
            path: Video file to play.
        """
        logs.event("player.load", file=path.name, path=str(path), exists=path.exists())
        self._send(["loadfile", str(path), "replace"])

    def stop(self) -> None:
        """Blank the video area."""
        self._send(["stop"])

    def probe(self) -> dict[str, object]:
        """Ask mpv what it is actually doing.

        This exists for after-the-fact diagnosis. If the reviewer ever appears
        to freeze on one video while the filenames advance, the recorded answer
        settles in one line whether mpv was handed the new file and was
        decoding it, or whether the problem is upstream of mpv.

        Returns:
            dict[str, object]: Property name to value; a property that could
            not be read maps to a string starting with ``error:``.
        """
        return probe_properties(self._ipc_path)

    def is_alive(self) -> bool:
        """Report whether mpv is still running.

        Returns:
            bool: True while the process lives.
        """
        return self._process.poll() is None

    def close(self) -> None:
        """Ask mpv to quit, then make sure it has.

        Leaving an orphaned mpv behind would keep a window on screen after the
        reviewer exits, so a process that ignores ``quit`` is killed.
        """
        logs.event("player.close", alive=self.is_alive())
        self._stopping.set()
        self._send(["quit"])
        try:
            self._process.wait(timeout=IPC_READY_TIMEOUT_S)
        except subprocess.TimeoutExpired:
            self._process.kill()
            self._process.wait(timeout=IPC_READY_TIMEOUT_S)
        self._reader.join(timeout=_READER_JOIN_S)
        self._socket.close()
        self._ipc_path.unlink(missing_ok=True)