Add an on_progress callback for long wire transfers (SCREENSHOT, FSGET)

host/README.md

@@ -55,6 +55,16 @@ needs installing beyond Python 3.9+.
   and IFF ILBM, stdlib-only, no Pillow. `Amipilot.screenshot()` is
   the client entry point; the byte-exact decode/encode logic has its
   own unit tests against synthetic captures.
+- **Progress feedback for long transfers** (`WireClient.command()`'s
+  `on_progress` parameter, `amipilot.wire.OnProgress`) -- a real
+  `SCREENSHOT` over serial can take anywhere from seconds to several
+  minutes (see `userdocs/Wire-Protocol.md`'s own transfer-time table)
+  with zero feedback otherwise. `on_progress(bytes_so_far,
+  total_bytes)` is called as a response payload streams in, threaded
+  through `Amipilot.screenshot()`/`fs_get()`; `amipilot.stderr_progress()`
+  is a ready-made "print a progress line" callback for the common
+  case. Defaults to `None` everywhere -- zero behavior change when
+  unused.
 - **The MUI-ARexx bridge tier** (phase 0.5 -- `Amipilot.mui_command()`):
   `MUIREXX <app-base> [TIMEOUT=<n>] <command...>`'s host wrapper --
   sends `command` verbatim to a MUI application's own ARexx port and

host/amipilot/__init__.py

@@ -15,7 +15,7 @@
 from .model import Gadget, TreeParseError, Window
 from .screen import Screen, ScreenParseError
 from .screenshot import Screenshot, ScreenshotParseError
-from .wire import ProtocolMismatch, Reply, ServerInfo, WireClient, WireError
+from .wire import OnProgress, ProtocolMismatch, Reply, ServerInfo, WireClient, WireError, stderr_progress
 
 __all__ = [
     "ActionFailed",
@@ -30,6 +30,7 @@
     "MenuParseError",
     "MenuStrip",
     "NotFound",
+    "OnProgress",
     "ProtocolMismatch",
     "Reply",
     "Screen",
@@ -42,4 +43,5 @@
     "WireClient",
     "WireError",
     "Window",
+    "stderr_progress",
 ]

host/amipilot/client.py

@@ -29,7 +29,7 @@
 from .model import Window, parse_tree
 from .screen import Screen, parse_screens
 from .screenshot import Screenshot
-from .wire import RC_ERROR, RC_FAIL, RC_OK, RC_TIMEOUT, RC_WARN, Reply, ServerInfo, WireClient
+from .wire import RC_ERROR, RC_FAIL, RC_OK, RC_TIMEOUT, RC_WARN, OnProgress, Reply, ServerInfo, WireClient
 
 
 class AmipilotError(Exception):
@@ -374,8 +374,9 @@ def _run(
         *,
         allow: tuple[int, ...] = (RC_OK,),
         payload: bytes | None = None,
+        on_progress: OnProgress | None = None,
     ) -> Reply:
-        reply = self._wire.command(command, payload)
+        reply = self._wire.command(command, payload, on_progress=on_progress)
         if reply.rc not in allow:
             raise _ERROR_CLASSES.get(reply.rc, AmipilotError)(
                 reply.rc, command, reply.text.strip() or "(no message)"
@@ -748,16 +749,21 @@ def fs_delete(self, path: str) -> None:
         allowlist and NotFound rules as fs_list()."""
         self._run(f"FSDELETE {_quote(path)}")
 
-    def fs_get(self, path: str) -> bytes:
+    def fs_get(self, path: str, *, on_progress: OnProgress | None = None) -> bytes:
         """FSGET <path> -- reads a file's full contents back as raw
         bytes (may contain embedded NULs; the wire's length-prefixed
         framing carries them intact, unlike the other verbs' NUL-
         terminated text payloads). The server caps this at its own
         internal buffer size (server/src/fs.c's AMIP_FS_BUF_SIZE, a
         test-staging channel, not a file manager) and raises
         ActionFailed for anything larger. See fs_put() for the
-        opposite direction."""
-        return self._run(f"FSGET {_quote(path)}").payload
+        opposite direction.
+
+        `on_progress`, if given, is called as the payload streams in --
+        `amipilot.stderr_progress()` is a ready-made callback for the
+        common "print a progress line" case; see `WireClient.command()`
+        for the full contract."""
+        return self._run(f"FSGET {_quote(path)}", on_progress=on_progress).payload
 
     def fs_put(self, path: str, data: bytes, *, timeout: float = 30.0) -> None:
         """FSPUT <path> <byte-count> [TIMEOUT=<n>] -- writes `data` to
@@ -842,7 +848,13 @@ def screens(self) -> list[Screen]:
         rationale)."""
         return parse_screens(self._run("SCREENS").text)
 
-    def screenshot(self, *, screen: str | None = None, window: str | None = None) -> Screenshot:
+    def screenshot(
+        self,
+        *,
+        screen: str | None = None,
+        window: str | None = None,
+        on_progress: OnProgress | None = None,
+    ) -> Screenshot:
         """SCREENSHOT [SCREEN=<substring>] [WINDOW=<pattern>] -- raw
         bitmap capture, planar or Picasso96/RTG (phase 1.0,
         `amipilot.screenshot`'s own module docstring has the full
@@ -874,13 +886,20 @@ def screenshot(self, *, screen: str | None = None, window: str | None = None) ->
         (IFF ILBM) and a `.png` (`.save()`/`.to_ilbm()` raise
         ScreenshotParseError for a P96 truecolor/hicolor capture, which
         ILBM has no way to represent -- use `.to_png()`/`.to_rgb888()`
-        directly for those)."""
+        directly for those).
+
+        `on_progress`, if given, is called as the capture streams in --
+        a real capture can take anywhere from seconds to several
+        minutes over serial (server/README.md's transfer-time table),
+        with no other feedback otherwise. `amipilot.stderr_progress()`
+        is a ready-made callback for the common "print a progress
+        line" case; see `WireClient.command()` for the full contract."""
         parts = ["SCREENSHOT"]
         if screen is not None:
             parts.append(f"SCREEN={_quote(screen)}")
         if window is not None:
             parts.append(f"WINDOW={_quote(window)}")
-        reply = self._run(" ".join(parts))
+        reply = self._run(" ".join(parts), on_progress=on_progress)
         return Screenshot.parse(reply.payload)
 
     def wait_for_window(

host/amipilot/wire.py

@@ -25,11 +25,20 @@
 from __future__ import annotations
 
 import socket
+import sys
 from dataclasses import dataclass, field
+from typing import Callable
 
 #: The one protocol version this client speaks (WIRE.md "Versioning").
 PROTOCOL = 1
 
+#: `on_progress(bytes_so_far, total_bytes)`, invoked as a large payload
+#: (SCREENSHOT, FSGET, or any future binary-payload verb) streams in --
+#: see `WireClient.command()`/`_read_exact()`. `total_bytes` is always
+#: the full count declared in the response header (never 0 unless the
+#: payload itself is empty), so a callback can safely divide by it.
+OnProgress = Callable[[int, int], None]
+
 #: Maximum request line, INCLUDING the trailing '\n' terminator --
 #: server/WIRE.md's own cap (enforced server-side by AMIP_SER_LINE/
 #: AMIP_TCP_LINE, both 512, serial.c/tcp.c). Checked here so an
@@ -168,7 +177,13 @@ def close(self) -> None:
         if close is not None:
             close()
 
-    def command(self, line: str | bytes, payload: bytes | None = None) -> Reply:
+    def command(
+        self,
+        line: str | bytes,
+        payload: bytes | None = None,
+        *,
+        on_progress: OnProgress | None = None,
+    ) -> Reply:
         """Send one command line, return its Reply. The terminator is
         added here; passing a line containing one is an error.
 
@@ -180,7 +195,15 @@ def command(self, line: str | bytes, payload: bytes | None = None) -> Reply:
         method has no idea what the line means and does no such
         checking; it just sends `len(payload)` raw bytes right after
         the terminator, matching what the server's FSPUT handler reads
-        off the wire before even looking at the request further."""
+        off the wire before even looking at the request further.
+
+        `on_progress`, if given, is called as the RESPONSE payload
+        streams in (see `_read_exact()`) -- there is no equivalent
+        callback for sending `payload` itself, since every current
+        caller's request payload is small (FSPUT's own server-side
+        AMIP_FS_BUF_SIZE cap) while response payloads (SCREENSHOT,
+        FSGET) are the genuinely slow, multi-second-to-multi-minute
+        case this exists for."""
         if isinstance(line, str):
             line = line.encode("latin-1")
         if b"\n" in line or b"\r" in line:
@@ -202,7 +225,7 @@ def command(self, line: str | bytes, payload: bytes | None = None) -> Reply:
             raise WireError(f"malformed response header: {header!r}") from None
         if count < 0:
             raise WireError(f"negative byte count: {header!r}")
-        return Reply(rc, self._read_exact(count))
+        return Reply(rc, self._read_exact(count, on_progress))
 
     def handshake(self) -> ServerInfo:
         """VERSION exchange; raises ProtocolMismatch unless the server
@@ -250,8 +273,34 @@ def _read_line(self) -> bytes:
         line, self._buf = self._buf.split(b"\n", 1)
         return line
 
-    def _read_exact(self, count: int) -> bytes:
+    def _read_exact(self, count: int, on_progress: OnProgress | None = None) -> bytes:
+        if on_progress is not None:
+            on_progress(min(len(self._buf), count), count)
         while len(self._buf) < count:
             self._buf += self._recv()
+            if on_progress is not None:
+                on_progress(min(len(self._buf), count), count)
         data, self._buf = self._buf[:count], self._buf[count:]
         return data
+
+
+def stderr_progress(label: str = "") -> OnProgress:
+    """A ready-made `on_progress` callback (see `WireClient.command()`)
+    that prints a single self-overwriting `label: done/total bytes
+    (pct%)` line to stderr, ending with a newline once the transfer
+    completes -- the common case, so scripts don't all have to
+    hand-roll the same thing:
+
+        client.screenshot(on_progress=stderr_progress("screenshot"))
+
+    Callers wanting their own UI (a GUI progress bar, structured
+    logging, etc.) should pass their own callback instead; this one is
+    just the batteries-included default."""
+    prefix = f"{label}: " if label else ""
+
+    def _progress(done: int, total: int) -> None:
+        pct = f" ({done * 100 // total}%)" if total else ""
+        end = "\n" if done >= total else ""
+        print(f"\r{prefix}{done}/{total} bytes{pct}", end=end, file=sys.stderr, flush=True)
+
+    return _progress

host/tests/test_client.py

@@ -563,6 +563,14 @@ def test_fs_get_returns_raw_bytes_with_embedded_nul(self):
         c = client_with(b"RC 0 %d\n%s" % (len(payload), payload))
         self.assertEqual(c.fs_get("Work:build/data"), payload)
 
+    def test_fs_get_on_progress_reaches_final_full_count(self):
+        payload = b"hello\x00world"
+        c = client_with(b"RC 0 %d\n%s" % (len(payload), payload))
+        calls = []
+        result = c.fs_get("Work:build/data", on_progress=lambda d, n: calls.append((d, n)))
+        self.assertEqual(result, payload)
+        self.assertEqual(calls[-1], (len(payload), len(payload)))
+
     def test_fs_mkdir_and_delete_do_not_raise_on_ok(self):
         c = client_with(b"RC 0 0\n", b"RC 0 0\n")
         c.fs_mkdir("Work:newdir")
@@ -692,6 +700,13 @@ def test_screenshot_with_screen_and_window(self):
         c.screenshot(screen="Second", window="GadTools")
         self.assertEqual(c._wire._t.sent[0], b"SCREENSHOT SCREEN=Second WINDOW=GadTools\n")
 
+    def test_screenshot_on_progress_reaches_final_full_count(self):
+        payload = _fake_screenshot_capture()
+        c = client_with(b"RC 0 %d\n%s" % (len(payload), payload))
+        calls = []
+        c.screenshot(on_progress=lambda d, n: calls.append((d, n)))
+        self.assertEqual(calls[-1], (len(payload), len(payload)))
+
     def test_screenshot_no_match_raises_not_found(self):
         payload = b"no screen matched"
         c = client_with(b"RC 5 %d\n%s" % (len(payload), payload))

host/tests/test_wire.py

@@ -17,6 +17,7 @@
     WireClient,
     WireError,
     _SerialTransport,
+    stderr_progress,
 )
 
 
@@ -108,6 +109,72 @@ def test_command_over_max_line_rejected_locally(self):
             client.command(command)
 
 
+class ProgressCallback(unittest.TestCase):
+    def test_called_once_per_chunk_and_covers_header_over_read(self):
+        # chunk=3 means the header line and the start of the payload
+        # can arrive in the same recv() -- the callback must still
+        # report correctly from wherever the buffer already stands,
+        # not assume progress starts at 0.
+        t = FakeTransport(b"RC 0 5\nhello", chunk=3)
+        calls = []
+        reply = WireClient(t).command("GETTEXT W 1", on_progress=lambda d, n: calls.append((d, n)))
+        self.assertEqual(reply, Reply(0, b"hello"))
+        # Every call reports the true total (5) and non-decreasing progress.
+        self.assertTrue(all(total == 5 for _, total in calls))
+        self.assertEqual([d for d, _ in calls], sorted(d for d, _ in calls))
+        self.assertEqual(calls[-1], (5, 5))
+
+    def test_zero_length_payload_still_calls_once(self):
+        calls = []
+        WireClient(FakeTransport(b"RC 0 0\n")).command(
+            "CLICK W 1", on_progress=lambda d, n: calls.append((d, n))
+        )
+        self.assertEqual(calls, [(0, 0)])
+
+    def test_not_called_when_omitted(self):
+        # No callback, no behavior change from before this existed.
+        t = FakeTransport(b"RC 0 5\nhello", chunk=1)
+        reply = WireClient(t).command("GETTEXT W 1")
+        self.assertEqual(reply, Reply(0, b"hello"))
+
+    def test_default_none_does_not_raise(self):
+        WireClient(FakeTransport(b"RC 0 0\n")).command("CLICK W 1", on_progress=None)
+
+
+class StderrProgress(unittest.TestCase):
+    def test_prints_overwriting_lines_ending_in_newline(self):
+        import io
+
+        buf = io.StringIO()
+        progress = stderr_progress("shot")
+        real_stderr = sys.stderr
+        sys.stderr = buf
+        try:
+            progress(0, 10)
+            progress(5, 10)
+            progress(10, 10)
+        finally:
+            sys.stderr = real_stderr
+        output = buf.getvalue()
+        self.assertIn("shot: 0/10 bytes (0%)", output)
+        self.assertIn("shot: 10/10 bytes (100%)", output)
+        # Only the final, completed line ends with a real newline.
+        self.assertTrue(output.rstrip("\n").count("\n") == 0)
+        self.assertTrue(output.endswith("\n"))
+
+    def test_zero_total_does_not_divide_by_zero(self):
+        import io
+
+        buf = io.StringIO()
+        real_stderr = sys.stderr
+        sys.stderr = buf
+        try:
+            stderr_progress()(0, 0)
+        finally:
+            sys.stderr = real_stderr
+        self.assertIn("0/0 bytes", buf.getvalue())
+
+
 class Handshake(unittest.TestCase):
     PAYLOAD = (
         b"AMIPILOT 0.3 PROTOCOL 1\n"

userdocs/Wire-Protocol.md

@@ -387,6 +387,23 @@ all. These numbers exist so a serial-only setup (real hardware without
 a network card, or a Copperline config without `--serial tcp`) knows
 what to expect rather than guessing why a capture appears to hang.
 
+Since a capture can genuinely take anywhere from seconds to several
+minutes with the client otherwise giving zero feedback, `screenshot()`
+(and `fs_get()`, below) accept an optional `on_progress(bytes_so_far,
+total_bytes)` callback, invoked as the payload streams in —
+`amipilot.stderr_progress()` is a ready-made callback for the common
+"print a progress line" case:
+
+```python
+from amipilot import stderr_progress
+
+shot = client.screenshot(on_progress=stderr_progress("screenshot"))
+# stderr: "screenshot: 51200/524288 bytes (9%)" ... updated in place ...
+```
+
+Pass your own callback instead for a GUI progress bar or structured
+logging; omit it entirely for today's exact zero-overhead behavior.
+
 ## File API
 
 From 0.4, a connected session can list/stat/create/delete files and