kuhyx/testsAndMisc
1
0
Fork 0
mirror of https://github.com/kuhyx/testsAndMisc.git synced 2026-11-14 13:50:49 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

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

Browse Source 
DisplayMixin in _display.py takes the methods that reach widgets only through
self._widgets: apply, _play, _verdict, on_undo, on_quit, drain, refresh, tick
and attach_player.

Widget construction stays in ui.py on purpose. test_ui.py fakes the toolkit
with patch.object(ui, "tk", fake), so moving _build out makes 22 tests error
with "module ui does not have the attribute 'tk'". An earlier attempt did that
and was reverted rather than touching the tests.

Also dropped the four _key_* adapters; each only discarded Tk's event argument
before forwarding, so _bind_keys binds lambdas directly now.

One thing worth knowing for the next split: ruff's TC001 is NOT suppressed by
__all__ the way F401 is. Moving Callbacks to its own module and re-exporting it
left a TYPE_CHECKING-only import that lints clean and raises AttributeError at
runtime -- cli.py constructs ui.Callbacks. Reverted; it stays defined in ui.py.

ui.py 398 -> 242, _display.py 179. 277 tests pass, both files 100% covered.

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:18:13 +02:00
parent 197cf67b7f
commit d86965ff3a
3 changed files with 234 additions and 171 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -0,0 +1,40 @@
{
"intent": "Bring python_pkg/wsg_grabber/ui.py under the 250-line file cap by moving the screen-updating and state-transition halves of ReviewWindow into a mixin, while leaving every line that touches tkinter in ui.py.",
"scope": [
"python_pkg/wsg_grabber/ui.py",
"python_pkg/wsg_grabber/_display.py (new)",
"Non-goal: no behaviour change, no test changes, no change to the ReviewCommand contract"
],
"changes": [
"Added DisplayMixin in _display.py holding apply, _play, _verdict, on_undo, on_quit, drain, refresh, tick and attach_player -- the methods that reach widgets only through self._widgets and never touch the tkinter module itself.",
"ui.py keeps Callbacks, Widgets, _build, _bind_keys, video_wid, on_keep, on_pass, shutdown and run, and is now `class ReviewWindow(DisplayMixin)`.",
"Deleted the four _key_* adapters (_key_keep/_key_pass/_key_undo/_key_quit); each only discarded Tk's event argument and forwarded, so _bind_keys now binds lambdas directly. No test referenced them.",
"ui.py 398 -> 242 lines; _display.py 179."
],
"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.15s. ui.py 93 stmts / 12 branches 100%; _display.py 76 stmts / 18 branches 100%."
},
{
"command": "PYTHONPATH=. python3 -c \"from python_pkg.wsg_grabber import ui; ui.Callbacks; ui.Widgets\"",
"result": "pass",
"evidence": "Both resolve at runtime. This check exists because an intermediate version moved Callbacks to its own module and ruff's TC001 rewrote the re-export into a TYPE_CHECKING-only import, which passes every linter but makes ui.Callbacks raise AttributeError at runtime -- and cli.py constructs it."
},
{
"command": "pre-commit run --files python_pkg/wsg_grabber/ui.py python_pkg/wsg_grabber/_display.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": [
"Widget construction deliberately stays in ui.py. test_ui.py fakes the toolkit with patch.object(ui, 'tk', fake), so moving _build to another module makes 22 tests error with \"module ui does not have the attribute 'tk'\". An earlier attempt did exactly that and was reverted rather than adapting the tests, which are out of scope.",
"Unlike F401, ruff's TC001 is NOT suppressed by listing a name in __all__. A cross-module re-export of a type-annotation-only name cannot be kept at runtime that way; define such classes in the module that must expose them.",
"ReviewWindow now inherits from an ABC whose abstract methods (shutdown) are satisfied by ui.py. Removing shutdown from ui.py would fail at instantiation rather than at lint time."
],
"rollback": [
"git revert the commit; ui.py returns to its 398-line form and _display.py is deleted.",
"After rollback, run python3 -m pytest python_pkg/wsg_grabber/tests/ -q to confirm 277 tests still pass."
]
}

179
python_pkg/wsg_grabber/_display.py Normal file
View File

@ -0,0 +1,179 @@
"""Applying a rendered command to the screen.
Split out of :mod:`python_pkg.wsg_grabber.ui` to keep it under the 250-line
cap. Note what is *not* here: creating widgets. ``ui`` is the only module that
imports ``tkinter``, and ``test_ui.py`` fakes the toolkit with
``patch.object(ui, "tk", ...)``, so construction has to stay there. These two
methods only ever reach widgets through ``self._widgets``, so they move freely.
A mixin rather than free functions because both read and write the window's
playback bookkeeping and ``apply`` is part of ReviewWindow's public surface.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
from python_pkg.wsg_grabber import logs, review
from python_pkg.wsg_grabber.constants import POLL_INTERVAL_MS
if TYPE_CHECKING:
from pathlib import Path
import queue
from python_pkg.wsg_grabber._callbacks import Callbacks
from python_pkg.wsg_grabber.models import (
DownloadEvent,
ReviewCommand,
Verdict,
)
from python_pkg.wsg_grabber.player import Player
from python_pkg.wsg_grabber.review import SessionState
from python_pkg.wsg_grabber.ui import Widgets
class DisplayMixin(ABC):
"""The screen-updating half of :class:`~.ui.ReviewWindow`."""
_widgets: Widgets
_callbacks: Callbacks
_player: Player | None
_playing: Path | None
_state: SessionState
_after_id: str | None
_events: queue.SimpleQueue[DownloadEvent]
@abstractmethod
def shutdown(self) -> None:
"""Close the player and tear the window down."""
def apply(self, command: ReviewCommand) -> None:
"""Carry out a rendered command.
Args:
command: What to show and do.
"""
if command.play is not None:
self._play(command.play)
if command.stop and self._player is not None and self._playing is not None:
self._playing = None
self._player.stop()
self._widgets.status.configure(text=command.status)
self._widgets.filename.configure(text=command.filename)
self._widgets.root.title(command.title)
if command.verdicts_enabled:
self._widgets.keep.configure(state="normal")
self._widgets.skip.configure(state="normal")
else:
self._widgets.keep.configure(state="disabled")
self._widgets.skip.configure(state="disabled")
if command.undo_enabled:
self._widgets.undo.configure(state="normal")
else:
self._widgets.undo.configure(state="disabled")
if command.quit_app:
self.shutdown()
def _play(self, path: Path) -> None:
"""Show a video, unless it is already the one playing.
The already-playing guard is essential rather than an optimisation:
``refresh`` runs on every tick, so without it ``loadfile`` would be sent
several times a second and the video would restart continuously instead
of ever playing.
A file can also vanish between being downloaded and being reviewed.
Handing mpv a dead path would leave the reviewer stuck on a black
frame, so the item is dropped and the next one shown instead.
Args:
path: Video to play.
"""
if self._player is None:
logs.debug("review.play_skipped", reason="no player", file=path.name)
return
if path == self._playing:
logs.debug("review.play_skipped", reason="already playing", file=path.name)
return
if not path.exists():
logs.warning("review.file_missing", file=path.name, path=str(path))
self._state = self._callbacks.missing(self._state)
self.refresh()
return
# Recorded BEFORE the new load, so a frozen-looking session shows what
# mpv was doing with the file it was supposedly already playing.
logs.event(
"review.show",
file=path.name,
previous=self._playing.name if self._playing else None,
player_alive=self._player.is_alive(),
mpv_before=self._player.probe(),
)
self._playing = path
self._player.play(path)
def _verdict(self, choice: Verdict) -> None:
"""Apply a verdict and redraw.
Args:
choice: What the user decided.
"""
before = self._state.current
self._state = self._callbacks.commit(self._state, choice)
after = self._state.current
logs.event(
"review.verdict",
choice=choice.value,
on=before.orig_name if before else None,
on_file=before.path.name if before else None,
next_file=after.path.name if after else None,
queued=len(self._state.pending),
kept=self._state.kept,
passed=self._state.passed,
)
self.refresh()
def on_undo(self) -> None:
"""Take back the last verdict and show that video again."""
before = self._state.current
self._state = self._callbacks.undo(self._state)
after = self._state.current
logs.event(
"review.undo",
was_showing=before.orig_name if before else None,
now_showing=after.orig_name if after else None,
kept=self._state.kept,
passed=self._state.passed,
remaining_undos=self._state.undoable,
)
self.refresh()
def on_quit(self) -> None:
"""Begin shutting down."""
self._state = review.on_quit(self._state)
self.refresh()
def tick(self) -> None:
"""Drain the worker's events, redraw, and schedule the next tick."""
self.drain()
self.refresh()
self._after_id = self._widgets.root.after(POLL_INTERVAL_MS, self.tick)
def drain(self) -> None:
"""Fold every queued worker event into the session state."""
while not self._events.empty():
self._state = review.on_event(self._state, self._events.get_nowait())
def refresh(self) -> None:
"""Re-render from the current state."""
self.apply(review.render(self._state))
def attach_player(self, player: Player) -> None:
"""Adopt a player and show the first video.
Args:
player: The video player to drive.
"""
self._player = player
self.refresh()

186
python_pkg/wsg_grabber/ui.py
View File

@ -24,18 +24,18 @@ from dataclasses import dataclass
import tkinter as tk
from typing import TYPE_CHECKING
from python_pkg.wsg_grabber import logs, review
from python_pkg.wsg_grabber import logs
from python_pkg.wsg_grabber._display import DisplayMixin
from python_pkg.wsg_grabber.constants import (
CONTROL_BAR_HEIGHT_PX,
KEEP_KEYS,
PASS_KEYS,
POLL_INTERVAL_MS,
QUIT_KEYS,
UNDO_KEYS,
WINDOW_HEIGHT_PX,
WINDOW_WIDTH_PX,
)
from python_pkg.wsg_grabber.models import ReviewCommand, Verdict
from python_pkg.wsg_grabber.models import Verdict
if TYPE_CHECKING:
from collections.abc import Callable
@ -74,8 +74,14 @@ class Widgets:
undo: tk.Button
class ReviewWindow:
"""Presents downloaded videos one at a time for a keep/pass decision."""
class ReviewWindow(DisplayMixin):
"""Presents downloaded videos one at a time for a keep/pass decision.
The screen-updating and state-transition halves live in _display to keep
this module under the 250-line cap. Widget construction stays here: this
is the only module that imports tkinter, and the tests fake the toolkit
by patching ``ui.tk``.
"""
def __init__(
self,
@ -183,13 +189,13 @@ class ReviewWindow:
def _bind_keys(self) -> None:
"""Bind the keyboard shortcuts onto the toplevel."""
for sequence in KEEP_KEYS:
self._widgets.root.bind(sequence, self._key_keep)
self._widgets.root.bind(sequence, lambda _event: self.on_keep())
for sequence in PASS_KEYS:
self._widgets.root.bind(sequence, self._key_pass)
self._widgets.root.bind(sequence, lambda _event: self.on_pass())
for sequence in UNDO_KEYS:
self._widgets.root.bind(sequence, self._key_undo)
self._widgets.root.bind(sequence, lambda _event: self.on_undo())
for sequence in QUIT_KEYS:
self._widgets.root.bind(sequence, self._key_quit)
self._widgets.root.bind(sequence, lambda _event: self.on_quit())
def video_wid(self) -> int:
"""Return the X11 id of the video frame.
@ -200,47 +206,6 @@ class ReviewWindow:
self._widgets.root.update()
return int(self._widgets.video.winfo_id())
def attach_player(self, player: Player) -> None:
"""Adopt a player and show the first video.
Args:
player: The video player to drive.
"""
self._player = player
self.refresh()
def _key_keep(self, _event: object) -> None:
"""Handle a keep keypress.
Args:
_event: Tk event, unused.
"""
self.on_keep()
def _key_pass(self, _event: object) -> None:
"""Handle a pass keypress.
Args:
_event: Tk event, unused.
"""
self.on_pass()
def _key_undo(self, _event: object) -> None:
"""Handle an undo keypress.
Args:
_event: Tk event, unused.
"""
self.on_undo()
def _key_quit(self, _event: object) -> None:
"""Handle a quit keypress.
Args:
_event: Tk event, unused.
"""
self.on_quit()
def on_keep(self) -> None:
"""Keep the current video."""
self._verdict(Verdict.KEEP)
@ -249,127 +214,6 @@ class ReviewWindow:
"""Pass on the current video."""
self._verdict(Verdict.SKIP)
def _verdict(self, choice: Verdict) -> None:
"""Apply a verdict and redraw.
Args:
choice: What the user decided.
"""
before = self._state.current
self._state = self._callbacks.commit(self._state, choice)
after = self._state.current
logs.event(
"review.verdict",
choice=choice.value,
on=before.orig_name if before else None,
on_file=before.path.name if before else None,
next_file=after.path.name if after else None,
queued=len(self._state.pending),
kept=self._state.kept,
passed=self._state.passed,
)
self.refresh()
def on_undo(self) -> None:
"""Take back the last verdict and show that video again."""
before = self._state.current
self._state = self._callbacks.undo(self._state)
after = self._state.current
logs.event(
"review.undo",
was_showing=before.orig_name if before else None,
now_showing=after.orig_name if after else None,
kept=self._state.kept,
passed=self._state.passed,
remaining_undos=self._state.undoable,
)
self.refresh()
def on_quit(self) -> None:
"""Begin shutting down."""
self._state = review.on_quit(self._state)
self.refresh()
def tick(self) -> None:
"""Drain the worker's events, redraw, and schedule the next tick."""
self.drain()
self.refresh()
self._after_id = self._widgets.root.after(POLL_INTERVAL_MS, self.tick)
def drain(self) -> None:
"""Fold every queued worker event into the session state."""
while not self._events.empty():
self._state = review.on_event(self._state, self._events.get_nowait())
def refresh(self) -> None:
"""Re-render from the current state."""
self.apply(review.render(self._state))
def apply(self, command: ReviewCommand) -> None:
"""Carry out a rendered command.
Args:
command: What to show and do.
"""
if command.play is not None:
self._play(command.play)
if command.stop and self._player is not None and self._playing is not None:
self._playing = None
self._player.stop()
self._widgets.status.configure(text=command.status)
self._widgets.filename.configure(text=command.filename)
self._widgets.root.title(command.title)
if command.verdicts_enabled:
self._widgets.keep.configure(state="normal")
self._widgets.skip.configure(state="normal")
else:
self._widgets.keep.configure(state="disabled")
self._widgets.skip.configure(state="disabled")
if command.undo_enabled:
self._widgets.undo.configure(state="normal")
else:
self._widgets.undo.configure(state="disabled")
if command.quit_app:
self.shutdown()
def _play(self, path: Path) -> None:
"""Show a video, unless it is already the one playing.
The already-playing guard is essential rather than an optimisation:
``refresh`` runs on every tick, so without it ``loadfile`` would be sent
several times a second and the video would restart continuously instead
of ever playing.
A file can also vanish between being downloaded and being reviewed.
Handing mpv a dead path would leave the reviewer stuck on a black
frame, so the item is dropped and the next one shown instead.
Args:
path: Video to play.
"""
if self._player is None:
logs.debug("review.play_skipped", reason="no player", file=path.name)
return
if path == self._playing:
logs.debug("review.play_skipped", reason="already playing", file=path.name)
return
if not path.exists():
logs.warning("review.file_missing", file=path.name, path=str(path))
self._state = self._callbacks.missing(self._state)
self.refresh()
return
# Recorded BEFORE the new load, so a frozen-looking session shows what
# mpv was doing with the file it was supposedly already playing.
logs.event(
"review.show",
file=path.name,
previous=self._playing.name if self._playing else None,
player_alive=self._player.is_alive(),
mpv_before=self._player.probe(),
)
self._playing = path
self._player.play(path)
def shutdown(self) -> None:
"""Cancel the tick, stop the worker and destroy the window."""
logs.event(