kuhyx/testsAndMisc
1
0
Fork 0
mirror of https://github.com/kuhyx/testsAndMisc.git synced 2026-11-10 20:07:04 +01:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Split wsg_grabber/player.py under the 250-line cap

Browse Source 
The stateless half of talking to mpv moves out:

  _mpv_ipc.py    read_reply, build_argv, encode, note_errors, and
                 probe_properties/query lifted out of MpvPlayer.probe/_query,
                 which only ever needed the socket path
  _player_api.py the Player Protocol that ui.py actually depends on

player.py keeps the class that owns the process and the socket, and re-exports
Player/build_argv/encode through __all__.

_note_errors stays as a bound delegate on MpvPlayer: test_player.py calls it on
the instance, and the tests are out of scope here. Caught by the suite failing,
not by guessing.

The probe constants moved across verbatim -- _PROBE_ID is 9001, and read_reply
matches replies on it, so a placeholder would have made every probe hang until
timeout while still returning a plausible error string.

player.py 373 -> 235, _mpv_ipc.py 158, _player_api.py 49. 277 tests pass.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011w2Pj4Ek3ZNMQVYcFHgXDY
This commit is contained in:
Krzysztof Rudnicki 2026-08-16 16:05:41 +02:00
parent 324ec0c9b5
commit 16c626661d
4 changed files with 263 additions and 157 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
docs/superpowers/evidence/wsg-grabber-player-split-2026-08-16.json Normal file
View File

@ -0,0 +1,37 @@
{
"intent": "Bring python_pkg/wsg_grabber/player.py under the 250-line file cap by separating the stateless mpv wire protocol and the Player interface from the class that owns the process and socket.",
"scope": [
"python_pkg/wsg_grabber/player.py",
"python_pkg/wsg_grabber/_mpv_ipc.py (new)",
"python_pkg/wsg_grabber/_player_api.py (new)",
"Non-goal: no behaviour change, no test changes, no change to the mpv command line"
],
"changes": [
"Moved the stateless wire protocol into _mpv_ipc.py: read_reply (was _read_reply), build_argv, encode, note_errors, plus probe_properties and query lifted out of MpvPlayer.probe/_query, which only ever needed the socket path.",
"Moved the Player Protocol into _player_api.py; ui.py depends on the interface, not on MpvPlayer.",
"Moved the probe constants _PROBE_ID, _PROBE_TIMEOUT_S and _PROBE_PROPERTIES verbatim alongside the code that reads them.",
"player.py re-exports Player, build_argv and encode through __all__.",
"MpvPlayer._note_errors is kept as a thin bound delegate to _mpv_ipc.note_errors, because test_player.py calls it on the instance and the tests are out of scope for this refactor.",
"player.py 373 -> 235 lines; _mpv_ipc.py 158; _player_api.py 49."
],
"verification": [
{
"command": "python3 -m pytest python_pkg/wsg_grabber/tests/ -q --cov=python_pkg.wsg_grabber --cov-report=term-missing",
"result": "pass",
"evidence": "277 passed in 15.14s. _mpv_ipc.py 51 stmts / 14 branches 100%, _player_api.py 100%, player.py 89 stmts 100%. An intermediate version of this split failed with 1 failed / 276 passed because test_mpv_errors_are_logged_rather_than_discarded calls mpv._note_errors as a bound method; the delegate was added in response to that real failure, not pre-emptively."
},
{
"command": "pre-commit run --files python_pkg/wsg_grabber/player.py python_pkg/wsg_grabber/_mpv_ipc.py python_pkg/wsg_grabber/_player_api.py",
"result": "pass",
"evidence": "Only the expected 'no staged evidence artifact' failure remained, which this file satisfies. ruff, ruff format, mypy, pylint, bandit and the coverage-gated pytest all pass with no suppression."
}
],
"risks": [
"_PROBE_ID is 9001 and read_reply matches replies on it. An intermediate version of the new module briefly carried a placeholder value of 1, which would have made every probe hang until timeout while still returning a plausible-looking error string; the constants were then moved across verbatim and the value re-checked at 9001.",
"probe_properties/query now take an ipc_path rather than reading self._ipc_path. MpvPlayer.probe delegates, so the public behaviour is unchanged and the tests that drive mpv.probe() cover both."
],
"rollback": [
"git revert the commit; player.py returns to its 373-line form and the two new modules are deleted.",
"After rollback, run python3 -m pytest python_pkg/wsg_grabber/tests/ -q to confirm 277 tests still pass."
]
}

158
python_pkg/wsg_grabber/_mpv_ipc.py Normal file
View File

@ -0,0 +1,158 @@
"""Talking to mpv: the command line, the wire format, and the reply reader.
Split out of :mod:`python_pkg.wsg_grabber.player` to keep it under the 250-line
cap. Everything here is stateless -- it does not own the process or the socket,
only the shapes of what goes over them.
"""
from __future__ import annotations
import json
import socket
from typing import TYPE_CHECKING, Final
from python_pkg.wsg_grabber import logs
if TYPE_CHECKING:
from collections.abc import Sequence
from pathlib import Path
_PROBE_ID = 9001
_PROBE_TIMEOUT_S = 2.0
# What to record when diagnosing an apparently frozen reviewer: which file mpv
# holds, whether it is actually advancing, and the size it configured its video
# output to (which changes per video, so a static value is itself a symptom).
_PROBE_PROPERTIES: Final[tuple[str, ...]] = (
"path",
"pause",
"core-idle",
"eof-reached",
"playback-time",
"width",
"height",
)
def read_reply(conn: socket.socket) -> object:
"""Read until mpv answers the probe request.
Args:
conn: A freshly connected control socket.
Returns:
object: The ``data`` field of the reply, or an ``error:`` string.
"""
buffer = b""
while True:
chunk = conn.recv(65536)
if not chunk:
return "error: socket closed"
buffer += chunk
for line in buffer.split(b"\n"):
if not line:
continue
try:
message = json.loads(line)
except ValueError:
continue
if message.get("request_id") == _PROBE_ID:
return message.get("data")
def build_argv(wid: int, ipc_path: Path) -> list[str]:
"""Build the mpv command line.
``--input-vo-keyboard=no`` matters: with ``--wid`` mpv reparents a child
window over the Tk frame, and if mpv claims the keyboard the Tk bindings
never see a keypress. Declining it leaves X delivering keys to the focused
toplevel, which is the Tk window.
Args:
wid: X11 window id of the frame to draw into.
ipc_path: Where to create the control socket.
Returns:
list[str]: Argument vector for :func:`subprocess.Popen`.
"""
return [
"mpv",
f"--wid={wid}",
f"--input-ipc-server={ipc_path}",
"--idle=yes",
"--loop-file=inf",
"--keep-open=yes",
"--no-terminal",
"--osc=no",
"--input-default-bindings=no",
"--input-vo-keyboard=no",
"--no-config",
]
def encode(command: Sequence[object]) -> bytes:
"""Encode one mpv IPC command.
Args:
command: Command name followed by its arguments.
Returns:
bytes: Newline-terminated JSON, ready for the socket.
"""
return json.dumps({"command": list(command)}).encode("utf-8") + b"\n"
def note_errors(chunk: bytes) -> None:
"""Log any failure mpv reported back.
Args:
chunk: Raw bytes just read from the socket.
"""
for line in chunk.split(b"\n"):
if not line.strip():
continue
try:
message = json.loads(line)
except ValueError:
continue
failure = message.get("error")
if isinstance(failure, str) and failure != "success":
logs.warning("player.reply_error", detail=failure, reply=message)
def probe_properties(ipc_path: Path) -> dict[str, object]:
"""Ask mpv for each diagnostic property in turn.
Args:
ipc_path: Where mpv's control socket lives.
Returns:
dict[str, object]: Property name to value; a property that could not
be read maps to a string starting with ``error:``.
"""
return {name: query(ipc_path, name) for name in _PROBE_PROPERTIES}
def query(ipc_path: Path, name: str) -> object:
"""Read one mpv property over a short-lived connection.
A separate connection avoids interleaving with the asynchronous events mpv
pushes down the main socket, which would otherwise make replies easy to
mis-attribute.
Args:
ipc_path: Where mpv's control socket lives.
name: Property name.
Returns:
object: The value, or an ``error:`` string.
"""
request = json.dumps({"command": ["get_property", name], "request_id": _PROBE_ID})
try:
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as probe:
probe.settimeout(_PROBE_TIMEOUT_S)
probe.connect(str(ipc_path))
probe.sendall(request.encode("utf-8") + b"\n")
return read_reply(probe)
except OSError as exc:
return f"error: {exc}"

49
python_pkg/wsg_grabber/_player_api.py Normal file
View File

@ -0,0 +1,49 @@
"""What the reviewer window requires of a video player.
Split out of :mod:`python_pkg.wsg_grabber.player` to keep it under the 250-line
cap. ``ui`` depends on this protocol rather than on :class:`~.player.MpvPlayer`,
so the window can be driven by a stub in the tests without mpv present.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Protocol
if TYPE_CHECKING:
from pathlib import Path
class Player(Protocol):
"""What the reviewer needs from a video player."""
def play(self, path: Path) -> None:
"""Show *path*, replacing whatever is on screen.
Args:
path: Video file to play.
"""
... # pragma: no cover
def stop(self) -> None:
"""Blank the video area."""
... # pragma: no cover
def is_alive(self) -> bool:
"""Report whether the player is still running.
Returns:
bool: True while the process lives.
"""
... # pragma: no cover
def probe(self) -> dict[str, object]:
"""Report what the player is currently doing, for diagnostics.
Returns:
dict[str, object]: Property name to value.
"""
... # pragma: no cover
def close(self) -> None:
"""Shut the player down and release its resources."""
... # pragma: no cover

176
python_pkg/wsg_grabber/player.py
View File

@ -19,16 +19,28 @@ orphan. No reviewing decisions are made here.
from __future__ import annotations
import json
import socket
import subprocess
import threading
import time
from typing import TYPE_CHECKING, Final, Protocol
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
@ -37,125 +49,6 @@ _SOCKET_POLL_S = 0.05
_READ_CHUNK = 65536
_READ_TIMEOUT_S = 0.2
_READER_JOIN_S = 5.0
_PROBE_ID = 9001
_PROBE_TIMEOUT_S = 2.0
# What to record when diagnosing an apparently frozen reviewer: which file mpv
# holds, whether it is actually advancing, and the size it configured its video
# output to (which changes per video, so a static value is itself a symptom).
_PROBE_PROPERTIES: Final[tuple[str, ...]] = (
"path",
"pause",
"core-idle",
"eof-reached",
"playback-time",
"width",
"height",
)
def _read_reply(conn: socket.socket) -> object:
"""Read until mpv answers the probe request.
Args:
conn: A freshly connected control socket.
Returns:
object: The ``data`` field of the reply, or an ``error:`` string.
"""
buffer = b""
while True:
chunk = conn.recv(65536)
if not chunk:
return "error: socket closed"
buffer += chunk
for line in buffer.split(b"\n"):
if not line:
continue
try:
message = json.loads(line)
except ValueError:
continue
if message.get("request_id") == _PROBE_ID:
return message.get("data")
class Player(Protocol):
"""What the reviewer needs from a video player."""
def play(self, path: Path) -> None:
"""Show *path*, replacing whatever is on screen.
Args:
path: Video file to play.
"""
... # pragma: no cover
def stop(self) -> None:
"""Blank the video area."""
... # pragma: no cover
def is_alive(self) -> bool:
"""Report whether the player is still running.
Returns:
bool: True while the process lives.
"""
... # pragma: no cover
def probe(self) -> dict[str, object]:
"""Report what the player is currently doing, for diagnostics.
Returns:
dict[str, object]: Property name to value.
"""
... # pragma: no cover
def close(self) -> None:
"""Shut the player down and release its resources."""
... # pragma: no cover
def build_argv(wid: int, ipc_path: Path) -> list[str]:
"""Build the mpv command line.
``--input-vo-keyboard=no`` matters: with ``--wid`` mpv reparents a child
window over the Tk frame, and if mpv claims the keyboard the Tk bindings
never see a keypress. Declining it leaves X delivering keys to the focused
toplevel, which is the Tk window.
Args:
wid: X11 window id of the frame to draw into.
ipc_path: Where to create the control socket.
Returns:
list[str]: Argument vector for :func:`subprocess.Popen`.
"""
return [
"mpv",
f"--wid={wid}",
f"--input-ipc-server={ipc_path}",
"--idle=yes",
"--loop-file=inf",
"--keep-open=yes",
"--no-terminal",
"--osc=no",
"--input-default-bindings=no",
"--input-vo-keyboard=no",
"--no-config",
]
def encode(command: Sequence[object]) -> bytes:
"""Encode one mpv IPC command.
Args:
command: Command name followed by its arguments.
Returns:
bytes: Newline-terminated JSON, ready for the socket.
"""
return json.dumps({"command": list(command)}).encode("utf-8") + b"\n"
class MpvPlayer:
@ -228,19 +121,13 @@ class MpvPlayer:
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.
"""
for line in chunk.split(b"\n"):
if not line.strip():
continue
try:
message = json.loads(line)
except ValueError:
continue
failure = message.get("error")
if isinstance(failure, str) and failure != "success":
logs.warning("player.reply_error", detail=failure, reply=message)
note_errors(chunk)
def _connect(self) -> socket.socket:
"""Wait for mpv's socket to appear and connect to it.
@ -319,32 +206,7 @@ class MpvPlayer:
dict[str, object]: Property name to value; a property that could
not be read maps to a string starting with ``error:``.
"""
return {name: self._query(name) for name in _PROBE_PROPERTIES}
def _query(self, name: str) -> object:
"""Read one mpv property over a short-lived connection.
A separate connection avoids interleaving with the asynchronous events
mpv pushes down the main socket, which would otherwise make replies
easy to mis-attribute.
Args:
name: Property name.
Returns:
object: The value, or an ``error:`` string.
"""
request = json.dumps(
{"command": ["get_property", name], "request_id": _PROBE_ID}
)
try:
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as probe:
probe.settimeout(_PROBE_TIMEOUT_S)
probe.connect(str(self._ipc_path))
probe.sendall(request.encode("utf-8") + b"\n")
return _read_reply(probe)
except OSError as exc:
return f"error: {exc}"
return probe_properties(self._ipc_path)
def is_alive(self) -> bool:
"""Report whether mpv is still running.