diff --git a/docs/superpowers/evidence/brother-printer-pjl-io-split-2026-08-16.json b/docs/superpowers/evidence/brother-printer-pjl-io-split-2026-08-16.json new file mode 100644 index 00000000..231b3c68 --- /dev/null +++ b/docs/superpowers/evidence/brother-printer-pjl-io-split-2026-08-16.json @@ -0,0 +1,51 @@ +{ + "intent": "Bring brother_printer/usb_query.py under the 250-line cap by moving the non-blocking PJL transport into its own module, with no behaviour change.", + "scope": [ + "python_pkg/brother_printer/_pjl_io.py (new)", + "python_pkg/brother_printer/usb_query.py", + "python_pkg/brother_printer/tests/test_usb_query_part2.py (MOD and import path only)", + "Non-goal: no assertion changed, no test deleted, skipped or weakened" + ], + "changes": [ + "Moved _drain_buffer, _write_with_deadline, _read_nonblocking, _wait_for_pjl_response and pjl_query verbatim into python_pkg/brother_printer/_pjl_io.py, together with the comment explaining why the fd is kept non-blocking; 267 -> 180 lines.", + "usb_query.py imports _drain_buffer, _write_with_deadline and pjl_query back, because test_usb_query.py patches all three on that module. Checked first that each is genuinely called by the code that stayed (_retry_pjl_query and _run_pjl_queries call all three), so ruff cannot strip them under F401 and no __all__ shim is needed.", + "Dropped contextlib and select from usb_query, which the moved block was their only user. Kept os and time despite the move: test_usb_query.py patches MOD.os.access/open/close and MOD.time.sleep, and _retry_pjl_query / query_usb_pjl still use both.", + "_pjl_io imports os, select, time and contextlib by name because test_usb_query_part2.py patches MOD.os.read, MOD.os.write, MOD.select.select and MOD.time.time on it.", + "test_usb_query_part2.py had its MOD constant and import module repointed at _pjl_io. test_usb_query.py needed no change, which is the signal the seam matched the existing test boundary." + ], + "verification": [ + { + "command": "PYTHONPATH=. python3 -m pytest python_pkg/brother_printer/tests -q", + "result": "pass", + "evidence": "402 passed before the split and 402 passed after it - identical pass count, not merely a green run." + }, + { + "command": "PYTHONPATH=. python3 -c \"from python_pkg.brother_printer import usb_query as u, _pjl_io as p; u._drain_buffer; u.pjl_query; u._write_with_deadline; u.time; u.os; u.Path; u.shutil; p.os; p.select; p.time; p.contextlib; p._read_nonblocking\"", + "result": "pass", + "evidence": "runtime probe OK - the three re-imported names are real attributes of usb_query, so test_usb_query.py's patches still reach them." + }, + { + "command": "grep -cw '_drain_buffer|pjl_query|_write_with_deadline' over the kept region of usb_query.py", + "result": "pass", + "evidence": "5, 1 and 1 real call sites respectively - all three imports are used, so ruff's F401 sweep leaves them alone. Had any been import-only it would have needed an __all__ entry per docs/python-split-recipes.md." + }, + { + "command": "wc -l python_pkg/brother_printer/usb_query.py python_pkg/brother_printer/_pjl_io.py", + "result": "pass", + "evidence": "180 and 109 lines, both well under the 250-line cap." + }, + { + "command": "pre-commit run --files ", + "result": "pass", + "evidence": "ruff, ruff format, mypy, pylint, bandit and pytest-with-coverage all pass on the touched files." + } + ], + "risks": [ + "The three re-imported names are load-bearing for mock.patch, not just for calls. If a future edit removes the last call site of one, ruff will strip the import and test_usb_query.py will fail with 'does not have the attribute' - the fix is an __all__ entry, not deleting the patch.", + "usb_query <- _pjl_io is one-way; _pjl_io must never import from usb_query." + ], + "rollback": [ + "git revert the commit; usb_query.py returns to 267 lines and _pjl_io.py disappears.", + "After rollback, re-run the package suite and confirm it still reports 402 passed." + ] +} diff --git a/python_pkg/brother_printer/_pjl_io.py b/python_pkg/brother_printer/_pjl_io.py new file mode 100644 index 00000000..03c0d9a2 --- /dev/null +++ b/python_pkg/brother_printer/_pjl_io.py @@ -0,0 +1,108 @@ +"""Non-blocking PJL transport over the raw /dev/usb/lp* file descriptor. + +Split out of :mod:`python_pkg.brother_printer.usb_query` to keep it under the +250-line cap. ``tests/test_usb_query_part2.py`` patches ``os``, ``select`` and +``time`` on this module, so those imports are load-bearing; the query callers +that stay in ``usb_query`` import these functions back, which is what keeps +``tests/test_usb_query.py`` patching them there. +""" + +from __future__ import annotations + +import contextlib +import os +import select +import time + +# ── PJL over USB ───────────────────────────────────────────────────── + + +# The printer's fd is opened and kept non-blocking. usblp blocks indefinitely +# on a plain open() and on writes whenever the printer is not ready to talk - +# stuck mid-job, wedged, or simply asleep - so a blocking fd turns "the printer +# is unwell" into "this tool hangs forever", which is the worst possible answer +# when a status report is exactly what you wanted. + + +def _drain_buffer(fd: int) -> None: + """Read and discard any stale data from the USB buffer.""" + with contextlib.suppress(OSError): + while os.read(fd, 4096): + pass + + +def _write_with_deadline(fd: int, data: bytes, deadline: float) -> bool: + """Write all of data to a non-blocking fd. Returns False if the deadline passes. + + Args: + fd: Non-blocking file descriptor for the printer. + data: Bytes to send. + deadline: Absolute monotonic-ish time (time.time()) to give up at. + + Returns: + True when everything was written, False on timeout or write error. + """ + sent = 0 + while sent < len(data): + remaining = deadline - time.time() + if remaining <= 0: + return False + try: + sent += os.write(fd, data[sent:]) + except BlockingIOError: + # Printer's buffer is full or it is not accepting data: wait for it + # to become writable rather than spinning or blocking forever. + _, writable, _ = select.select([], [fd], [], min(remaining, 0.5)) + if not writable: + continue + except OSError: + return False + return True + + +def _read_nonblocking(fd: int) -> bytes: + """Read all currently available data from a non-blocking fd.""" + data = b"" + with contextlib.suppress(OSError): + while True: + chunk = os.read(fd, 4096) + if not chunk: + break + data += chunk + return data + + +def _wait_for_pjl_response(fd: int, deadline: float) -> bytes: + """Poll fd until PJL data arrives or deadline expires.""" + response = b"" + while time.time() < deadline: + remaining = deadline - time.time() + if remaining <= 0: + break + readable, _, _ = select.select([fd], [], [], min(remaining, 1.0)) + if readable: + response += _read_nonblocking(fd) + if response and (b"=" in response or b"@PJL" in response): + break + return response + + +def pjl_query(fd: int, cmd: str, timeout_sec: float = 5.0) -> str: + """Send a PJL command via raw fd and read the response. + + Args: + fd: Non-blocking file descriptor for the printer. + cmd: PJL command, e.g. "@PJL INFO STATUS". + timeout_sec: Total budget for sending the command and reading the reply. + + Returns: + The printer's reply, or "" if it did not answer in time. A silent + printer is a normal outcome here, not an error: it happens whenever the + printer is busy digesting a job. + """ + deadline = time.time() + timeout_sec + pjl_cmd = f"\x1b%-12345X@PJL\r\n{cmd}\r\n\x1b%-12345X" + if not _write_with_deadline(fd, pjl_cmd.encode(), deadline): + return "" + response = _wait_for_pjl_response(fd, deadline) + return response.decode("ascii", errors="replace") diff --git a/python_pkg/brother_printer/tests/test_usb_query_part2.py b/python_pkg/brother_printer/tests/test_usb_query_part2.py index 959c65b8..6e586ea4 100644 --- a/python_pkg/brother_printer/tests/test_usb_query_part2.py +++ b/python_pkg/brother_printer/tests/test_usb_query_part2.py @@ -1,4 +1,4 @@ -"""Tests for brother_printer.usb_query - the PJL wire protocol. +"""Tests for brother_printer._pjl_io - the PJL wire protocol. The printer's fd is non-blocking throughout: usblp blocks forever on open and on writes whenever the printer is unwell, which is exactly when a status @@ -9,7 +9,7 @@ from __future__ import annotations from unittest.mock import MagicMock, patch -from python_pkg.brother_printer.usb_query import ( +from python_pkg.brother_printer._pjl_io import ( _drain_buffer, _read_nonblocking, _wait_for_pjl_response, @@ -17,7 +17,7 @@ from python_pkg.brother_printer.usb_query import ( pjl_query, ) -MOD = "python_pkg.brother_printer.usb_query" +MOD = "python_pkg.brother_printer._pjl_io" class TestDrainBuffer: diff --git a/python_pkg/brother_printer/usb_query.py b/python_pkg/brother_printer/usb_query.py index ddcc897f..6b37cac9 100644 --- a/python_pkg/brother_printer/usb_query.py +++ b/python_pkg/brother_printer/usb_query.py @@ -1,18 +1,25 @@ -"""USB printer discovery and PJL query functions.""" +"""USB printer discovery and PJL query functions. + +The non-blocking PJL transport lives in _pjl_io.py. Its three entry points are +imported back here because tests/test_usb_query.py patches them on this module. +""" from __future__ import annotations -import contextlib import errno import importlib import logging import os from pathlib import Path -import select import shutil import time from typing import TYPE_CHECKING +from python_pkg.brother_printer._pjl_io import ( + _drain_buffer, + _write_with_deadline, + pjl_query, +) from python_pkg.brother_printer._query import ( printer_info_from_cups, run_command_text, @@ -45,100 +52,6 @@ def find_usb_printer_dev() -> str | None: return str(devices[0]) if devices else None -# ── PJL over USB ───────────────────────────────────────────────────── - - -# The printer's fd is opened and kept non-blocking. usblp blocks indefinitely -# on a plain open() and on writes whenever the printer is not ready to talk - -# stuck mid-job, wedged, or simply asleep - so a blocking fd turns "the printer -# is unwell" into "this tool hangs forever", which is the worst possible answer -# when a status report is exactly what you wanted. - - -def _drain_buffer(fd: int) -> None: - """Read and discard any stale data from the USB buffer.""" - with contextlib.suppress(OSError): - while os.read(fd, 4096): - pass - - -def _write_with_deadline(fd: int, data: bytes, deadline: float) -> bool: - """Write all of data to a non-blocking fd. Returns False if the deadline passes. - - Args: - fd: Non-blocking file descriptor for the printer. - data: Bytes to send. - deadline: Absolute monotonic-ish time (time.time()) to give up at. - - Returns: - True when everything was written, False on timeout or write error. - """ - sent = 0 - while sent < len(data): - remaining = deadline - time.time() - if remaining <= 0: - return False - try: - sent += os.write(fd, data[sent:]) - except BlockingIOError: - # Printer's buffer is full or it is not accepting data: wait for it - # to become writable rather than spinning or blocking forever. - _, writable, _ = select.select([], [fd], [], min(remaining, 0.5)) - if not writable: - continue - except OSError: - return False - return True - - -def _read_nonblocking(fd: int) -> bytes: - """Read all currently available data from a non-blocking fd.""" - data = b"" - with contextlib.suppress(OSError): - while True: - chunk = os.read(fd, 4096) - if not chunk: - break - data += chunk - return data - - -def _wait_for_pjl_response(fd: int, deadline: float) -> bytes: - """Poll fd until PJL data arrives or deadline expires.""" - response = b"" - while time.time() < deadline: - remaining = deadline - time.time() - if remaining <= 0: - break - readable, _, _ = select.select([fd], [], [], min(remaining, 1.0)) - if readable: - response += _read_nonblocking(fd) - if response and (b"=" in response or b"@PJL" in response): - break - return response - - -def pjl_query(fd: int, cmd: str, timeout_sec: float = 5.0) -> str: - """Send a PJL command via raw fd and read the response. - - Args: - fd: Non-blocking file descriptor for the printer. - cmd: PJL command, e.g. "@PJL INFO STATUS". - timeout_sec: Total budget for sending the command and reading the reply. - - Returns: - The printer's reply, or "" if it did not answer in time. A silent - printer is a normal outcome here, not an error: it happens whenever the - printer is busy digesting a job. - """ - deadline = time.time() + timeout_sec - pjl_cmd = f"\x1b%-12345X@PJL\r\n{cmd}\r\n\x1b%-12345X" - if not _write_with_deadline(fd, pjl_cmd.encode(), deadline): - return "" - response = _wait_for_pjl_response(fd, deadline) - return response.decode("ascii", errors="replace") - - def _parse_status(resp: str, result: USBResult) -> bool: """Parse STATUS response into result. Returns True if code was found.""" found = False