From 900f8eaa2dcfb4b4b1f4769fd8bfd5237df5cdf1 Mon Sep 17 00:00:00 2001 From: Maxim Averyanov Date: Wed, 26 Aug 2026 23:15:19 +0200 Subject: [PATCH 1/2] Make Stop stop, and let the phone be held sideways MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit **Stop did nothing, and the reason is that there was nothing left to stop.** Every wait inside a stroke happens on the phone - that is the entire point of the injector - so writing a stroke costs the host almost nothing. The loop therefore handed the whole drawing to the device's pipe in a fraction of a second, finished, and sat there while the phone drew on out of a buffer that nothing on this end can reach. `should_continue` was being polled faithfully between strokes; by the time anyone could press the button there were no strokes left to skip. The host now refuses to run more than a quarter of a second ahead of the device: it counts the milliseconds it has queued and waits for the injector to catch up when it goes over. Stopping is then immediate to a person, the backlog being at most that. It costs a round trip per 250 ms of drawing - about eleven of them across a three-second drawing, which is lost in the noise. Two tests hold it down: one that the whole drawing is not queued at once, and one that stopping leaves the rest unsent. **A phone held sideways was squeezed into a portrait frame.** `wm size` reports the orientation the phone was manufactured in and does not change when it is turned, so it cannot be asked. The captured frame knows, being a picture of the screen as it is now, so the engine compares the two: if the frame is the other way round from the natural size, so is the phone. That correction now feeds the placement overlay and the drawing itself, not just the preview - a drawing placed from the natural size lands rotated on a phone in landscape. The window reshapes its phone from every frame whose proportions differ from the last, so turning the device mid-session is followed rather than ignored. Also: the device dropdown showed "Looking for a device…" over a device that was connected and mirroring, because setting SelectedIndex on a ComboBox whose items were added moments earlier does not reliably take. Selecting the item does. Co-Authored-By: Claude Opus 5 --- mthread/device.py | 23 ++++++++++-- mthread/injector.py | 16 ++++++--- mthread_draw/server.py | 28 +++++++++++++-- tests/test_device.py | 75 +++++++++++++++++++++++++++++++++++++++- winui/Engine.cs | 6 ++-- winui/MainWindow.xaml.cs | 33 +++++++++++++++--- 6 files changed, 164 insertions(+), 17 deletions(-) diff --git a/mthread/device.py b/mthread/device.py index c07e119..9871694 100644 --- a/mthread/device.py +++ b/mthread/device.py @@ -83,6 +83,12 @@ def find_devices(adb_path: str | None = None) -> list[DeviceInfo]: return list_devices(adb) +#: How many milliseconds of drawing the host may queue on the device before +#: waiting for it to catch up. Small enough that stopping is immediate, large +#: enough that the round trips it costs are lost in the noise. +LOOKAHEAD_MS = 250.0 + + class Device: """A connected Android device. @@ -353,6 +359,7 @@ def clamp(point): sent = 0 pen = None + queued = 0.0 with TouchInjector(self) as injector: for index, path in enumerate(paths, start=1): if should_continue is not None and not should_continue(): @@ -364,12 +371,24 @@ def clamp(point): if pen is not None and pacing.lift_ms: # The hand has to get there. Longer gaps take longer. travel = math.dist(pen, points[0]) / max(pacing.travel_speed, 1.0) * 1000.0 - injector.pause(pacing.lift_ms + travel) + queued += injector.pause(pacing.lift_ms + travel) - injector.stroke(points, pacing, rng) + queued += injector.stroke(points, pacing, rng) pen = points[-1] sent += 1 + # Do not run further ahead than this. Writing a stroke costs + # almost nothing - every wait in it happens on the phone - so + # without a bound the whole drawing lands in the device's pipe + # in a fraction of a second. Stop then has nothing left to + # cancel: the host has finished and the phone is still drawing + # from a buffer nobody can reach. Waiting for the device to + # catch up keeps the backlog to a quarter of a second, which is + # what makes stopping feel immediate. + if queued >= LOOKAHEAD_MS: + injector.sync() + queued = 0.0 + if progress is not None and index % 8 == 0: progress(index, len(paths)) diff --git a/mthread/injector.py b/mthread/injector.py index db5efed..e18262d 100644 --- a/mthread/injector.py +++ b/mthread/injector.py @@ -205,10 +205,15 @@ def _write(self, lines) -> None: self.process.stdin.flush() def stroke(self, points, pacing: Pacing | None = None, - rng: random.Random | None = None) -> None: - """Draw one stroke, waiting between points as *pacing* says to.""" + rng: random.Random | None = None) -> float: + """Draw one stroke, waiting between points as *pacing* says to. + + Returns how many milliseconds of work were handed to the device, which + the caller needs in order to know how far ahead of the drawing it has + run. Writing is nearly free; the waiting all happens on the phone. + """ if len(points) < 2: - return + return 0.0 pacing = pacing or Pacing() rng = rng or random.Random() delays = pace_stroke(points, pacing, rng) @@ -234,11 +239,14 @@ def stroke(self, points, pacing: Pacing | None = None, with self._lock: self._write(lines) + return float(sum(delays)) - def pause(self, millis: float) -> None: + def pause(self, millis: float) -> float: if millis > 0.05: with self._lock: self._write([f"S {millis:.2f}"]) + return float(millis) + return 0.0 def sync(self, timeout: float = 600.0) -> None: """Block until everything sent so far has been injected. diff --git a/mthread_draw/server.py b/mthread_draw/server.py index a603698..c9405d7 100644 --- a/mthread_draw/server.py +++ b/mthread_draw/server.py @@ -64,6 +64,9 @@ def __init__(self, write): self.mirror: ScreenMirror | None = None self.mirror_thread: threading.Thread | None = None self.mirror_stop: threading.Event | None = None + #: The size of the last frame captured, which is the only thing that + #: knows which way up the phone is being held. + self.frame_size: tuple[int, int] | None = None # ---------------------------------------------------------------- talking @@ -138,7 +141,10 @@ def _mirror_loop(self) -> None: slot = 1 - slot path = Path(tempfile.gettempdir()) / f"mthread_draw_frame_{slot}.jpg" path.write_bytes(jpeg) - self.event("frame", path=str(path)) + with Image.open(path) as frame: + self.frame_size = frame.size + self.event("frame", path=str(path), + width=self.frame_size[0], height=self.frame_size[1]) def _stop_mirror(self) -> None: if self.mirror_stop is not None: @@ -181,7 +187,7 @@ def _overlay(self, margin: float) -> str | None: if self.device is None or not self.paths: return None - width, height = self.device.screen_size + width, height = self.screen_now() placed = fit_to_screen(self.paths, width, height, margin=margin) scale = OVERLAY_WIDTH / width @@ -209,7 +215,7 @@ def op_draw(self, margin: float = 0.06, speed: float = 1.0, if not self.paths: raise MThreadError("Load an image and take a preview first.") - width, height = self.device.screen_size + width, height = self.screen_now() placed = fit_to_screen(self.paths, width, height, margin=margin) self.cancel.clear() self.drawing.set() @@ -231,6 +237,22 @@ def op_stop(self) -> dict: self.cancel.set() return {"stopping": True} + def screen_now(self) -> tuple[int, int]: + """The screen as it is being held, not as it was manufactured. + + ``wm size`` reports the natural orientation and does not change when the + phone is turned, so a drawing placed from it lands rotated on a phone in + landscape, and the preview is squashed into a portrait frame. The + captured frame is the one thing that knows: if it is the other way round + from the natural size, so is the phone. + """ + width, height = self.device.screen_size + if self.frame_size: + frame_width, frame_height = self.frame_size + if (frame_width > frame_height) != (width > height): + return height, width + return width, height + def _require_device(self) -> None: if self.device is None: raise MThreadError("No device is connected.") diff --git a/tests/test_device.py b/tests/test_device.py index 064f790..15fe10b 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -228,9 +228,12 @@ def __exit__(self, *exc): def stroke(self, points, pacing=None, rng=None): started.append(("stroke", len(points))) + # The real one answers with the milliseconds it queued, which is + # how the caller knows when to wait for the device. + return 0.0 def pause(self, millis): - pass + return 0.0 def sync(self, timeout=600.0): pass @@ -240,6 +243,76 @@ def sync(self, timeout=600.0): self.assertEqual(started[0], device) self.assertEqual(device.scripts, []) + def test_the_injector_waits_rather_than_queueing_the_whole_drawing(self): + """Stop can only work if the host has not already sent everything. + + Every wait in a stroke happens on the phone, so writing one costs + nothing and an unbounded loop hands over the entire drawing in a + fraction of a second. The host then finishes while the device is still + drawing out of a buffer nothing can reach, and Stop does nothing at all. + """ + device = make_device(raw=False) + events = [] + + class Stub: + def __init__(self, owner, jar=None): + pass + + def __enter__(self): + return self + + def __exit__(self, *exc): + return None + + def stroke(self, points, pacing=None, rng=None): + events.append("stroke") + return 300.0 # more than the look-ahead budget, on its own + + def pause(self, millis): + return 0.0 + + def sync(self, timeout=600.0): + events.append("sync") + + paths = [[(1, 1), (2, 2)], [(3, 3), (4, 4)], [(5, 5), (6, 6)]] + with unittest.mock.patch("mthread.device.TouchInjector", Stub): + device.draw_paths(paths) + + # A wait after each stroke, because each one exceeds the budget alone. + self.assertEqual(events, ["stroke", "sync", "stroke", "sync", + "stroke", "sync", "sync"]) + + def test_stopping_leaves_the_rest_of_the_drawing_unsent(self): + device = make_device(raw=False) + drawn = [] + + class Stub: + def __init__(self, owner, jar=None): + pass + + def __enter__(self): + return self + + def __exit__(self, *exc): + return None + + def stroke(self, points, pacing=None, rng=None): + drawn.append(points[0]) + return 300.0 + + def pause(self, millis): + return 0.0 + + def sync(self, timeout=600.0): + pass + + paths = [[(index, index), (index + 1, index + 1)] for index in range(10)] + with unittest.mock.patch("mthread.device.TouchInjector", Stub): + sent = device.draw_paths(paths, should_continue=lambda: len(drawn) < 3) + + self.assertEqual(sent, 3) + self.assertEqual(len(drawn), 3) + def test_a_refusing_injector_drops_through_to_input(self): """Nothing guarantees app_process will run a jar for us. The slow path always works, so an unavailable injector must not be fatal.""" diff --git a/winui/Engine.cs b/winui/Engine.cs index 726d798..fcc66e5 100644 --- a/winui/Engine.cs +++ b/winui/Engine.cs @@ -28,7 +28,7 @@ public sealed class Engine : IAsyncDisposable public event Action? StatusChanged; public event Action? ProgressChanged; public event Action? Failed; - public event Action? FrameReady; + public event Action? FrameReady; public event Action? MirrorLost; private Engine(Process process) @@ -180,7 +180,9 @@ private void Dispatch(string name, JsonNode message) message["total"]?.GetValue() ?? 1); break; case "frame": - FrameReady?.Invoke(message["path"]?.GetValue() ?? string.Empty); + FrameReady?.Invoke(message["path"]?.GetValue() ?? string.Empty, + message["width"]?.GetValue() ?? 0, + message["height"]?.GetValue() ?? 0); break; case "mirror_lost": MirrorLost?.Invoke(message["error"]?.GetValue() ?? string.Empty); diff --git a/winui/MainWindow.xaml.cs b/winui/MainWindow.xaml.cs index dbbf522..429b6ae 100644 --- a/winui/MainWindow.xaml.cs +++ b/winui/MainWindow.xaml.cs @@ -74,7 +74,8 @@ private async Task StartEngineAsync() _engine.Failed += text => Dispatch(() => StatusText.Text = text); _engine.ProgressChanged += (done, total) => Dispatch(() => Progress.Value = total > 0 ? 100.0 * done / total : 0); - _engine.FrameReady += path => Dispatch(() => ShowFrame(path)); + _engine.FrameReady += (path, width, height) => + Dispatch(() => ShowFrame(path, width, height)); _engine.MirrorLost += text => Dispatch(() => StatusText.Text = $"The live view stopped: {text}"); @@ -150,7 +151,11 @@ private async Task RefreshDevicesAsync() if (DeviceBox.Items.Count > 0) { - DeviceBox.SelectedIndex = 0; + // By item, not by index. Setting SelectedIndex on a ComboBox + // whose items were added moments earlier does not always take, + // and the box then shows "Looking for a device…" over a device + // that is connected and mirroring - which reads as a failure. + DeviceBox.SelectedItem = DeviceBox.Items[0]; StatusText.Text = $"{DeviceBox.Items.Count} device(s) found."; // With one device there is nothing to choose, and the live view @@ -220,16 +225,34 @@ private async void OnConnect(object sender, RoutedEventArgs args) } /// Give the on-screen phone the proportions of the real one. + /// + /// Called again for every frame whose shape differs from the last, because + /// a phone that is turned reports the same `wm size` as before. Without it + /// a landscape screen is squeezed into the portrait frame it was given at + /// connect, and everything on it is drawn narrow. + /// private void ShapeToDevice(int width, int height) { + if (width <= 0 || height <= 0) + { + return; + } var scale = 780.0 / Math.Max(width, height); // The frame's padding is its bezel, and it is outside the screen area. - PhoneFrame.Width = Math.Round(width * scale) + 18; - PhoneFrame.Height = Math.Round(height * scale) + 18; + var frameWidth = Math.Round(width * scale) + 18; + var frameHeight = Math.Round(height * scale) + 18; + if (Math.Abs(PhoneFrame.Width - frameWidth) < 0.5 + && Math.Abs(PhoneFrame.Height - frameHeight) < 0.5) + { + return; + } + PhoneFrame.Width = frameWidth; + PhoneFrame.Height = frameHeight; } - private void ShowFrame(string path) + private void ShowFrame(string path, int width, int height) { + ShapeToDevice(width, height); try { // Read the bytes rather than pointing a BitmapImage at the file: the From 9d6486a343d57a335b152b99b1161fd7d8150514 Mon Sep 17 00:00:00 2001 From: Maxim Averyanov Date: Wed, 26 Aug 2026 23:30:53 +0200 Subject: [PATCH 2/2] Put the drawing where you want it, and record something that travels **Placing by hand.** The drawing was always fitted to the screen with a fixed margin, and the margin was a slider nobody wanted. Now it is dragged: pull it about over the live view, wheel to resize, Shift and wheel to turn, double-click to fit again. Placement is held in fractions of the screen and in degrees rather than pixels, so it survives the phone being turned and means the same thing on a different device; a scale of one is "as large as it goes with a margin", whatever the drawing and whatever the phone, so the number a person sees is a multiplier on something recognisable. Rotation turns about the drawing's own middle and fits the rotated bounding box - without that a drawing turned forty-five degrees quietly grows past the edges of the screen. The front end coalesces the gestures rather than asking for an overlay per mouse report. **Recordings that replay somewhere else.** The old format was faithful and useless anywhere but the phone it came from, twice over: the numbers are digitizer coordinates, whose range has little to do with any display, and replaying them means writing to /dev/input, which every recent Pixel denies to the shell. So a recording could not be replayed on a different phone, and on a current one it could not be replayed at all. mthread.gestures decodes those events into strokes of (time, x, y) with the coordinates as fractions of the screen, and Device.play_gestures scales them to whatever screen it is handed and plays them through the injector. A fraction means the same thing everywhere, so the file travels. Nine tests cover the decoding: one finger, two fingers, a finger that is held rather than moved, a panel whose axes are swapped, and the three ways a file can be the wrong version. The window gets Record, Play, Open and Save alongside the drawing controls. Known and not yet chased: replaying a two-second recording takes longer than two seconds. Some of that is the injector starting and stopping, which is a fixed cost either way, but it has not been measured properly and the timing should be faithful before anyone relies on it for a regression pass. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 14 ++ mthread/device.py | 62 ++++++++- mthread/gestures.py | 234 +++++++++++++++++++++++++++++++ mthread/injector.py | 20 +++ mthread_draw/geometry.py | 84 +++++++++++- mthread_draw/server.py | 115 ++++++++++++++-- tests/test_gestures.py | 127 +++++++++++++++++ winui/MainWindow.xaml | 31 ++++- winui/MainWindow.xaml.cs | 287 +++++++++++++++++++++++++++++++++++++-- 9 files changed, 948 insertions(+), 26 deletions(-) create mode 100644 mthread/gestures.py create mode 100644 tests/test_gestures.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 0228c49..533e940 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,20 @@ ## Unreleased +- **Place the drawing by hand.** Drag it over the live view, wheel to resize, + Shift and wheel to turn, double-click to fit again. `Placement` holds the + position in fractions of the screen and degrees, so it survives the phone + being turned; a scale of one means "as large as it goes with a margin", + whatever the drawing and whatever the phone. +- **Recordings that replay on a different phone.** `mthread.gestures` decodes + raw touch events into strokes of `(time, x, y)` with the coordinates as + fractions of the screen, and `Device.play_gestures` scales them to whatever + screen it is given. The old format stored digitizer coordinates - which is + why replaying one elsewhere was refused - and replayed through `/dev/input`, + which every recent Pixel denies to the shell, so it could not be replayed on + the phone that made it either. The Windows front end records, plays, opens + and saves. + - **A live view of the device screen**, in its own shape, with the strokes laid over exactly where they will land. `mthread.mirror.ScreenMirror` runs `com.mthread.Mirror` from the same jar as the injector: the phone scales and diff --git a/mthread/device.py b/mthread/device.py index 9871694..9bd2fab 100644 --- a/mthread/device.py +++ b/mthread/device.py @@ -12,7 +12,7 @@ from typing import Iterable, Sequence from .adb import find_adb, popen_adb, run_adb -from .errors import DeviceNotConnectedError, TouchDeviceNotFoundError +from .errors import DeviceNotConnectedError, MThreadError, TouchDeviceNotFoundError from .hand import HandSettings, simulate from .injector import InjectorUnavailableError, Pacing, TouchInjector from .touch import TouchDevice, build_stroke_events, parse_getevent_pl, pick_touchscreen @@ -111,6 +111,7 @@ def __init__(self, serial: str | None = None, adb_path: str | None = None): self._touch_device: TouchDevice | None = None self._screen_size: tuple[int, int] | None = None self._supports_raw: bool | None = None + self._model: str | None = None #: Set when the injector refused to start and the slow path was used. self.injector_error = False @@ -165,6 +166,17 @@ def screenshot(self, local_path: str) -> str: handle.write(proc.stdout) return local_path + @property + def model(self) -> str: + """What the phone calls itself, for labelling a recording.""" + if self._model is None: + try: + self._model = self.shell("getprop", "ro.product.model", + check=False).stdout.strip() + except MThreadError: + self._model = "" + return self._model + def set_pointer_location(self, enabled: bool) -> None: """Toggle the developer-option touch overlay, handy for calibration.""" self.shell("settings", "put", "system", "pointer_location", "1" if enabled else "0", check=False) @@ -398,6 +410,54 @@ def clamp(point): progress(len(paths), len(paths)) return sent + def play_gestures(self, session, *, speed: float = 1.0, repeat: int = 1, + progress=None, should_continue=None) -> int: + """Replay a :class:`~mthread.gestures.GestureSession` on this device. + + The recording holds fractions of a screen rather than pixels, so it is + scaled to whatever screen this device has - which is the whole reason + the format exists. Playback goes through the injector, so it works on + devices that refuse raw input, which is where the old raw-event replay + stopped being usable at all. + """ + if speed <= 0: + raise ValueError("speed must be positive") + + width, height = self.screen_size + strokes = session.to_pixels(width, height) + played = 0 + queued = 0.0 + + with TouchInjector(self) as injector: + for turn in range(max(1, repeat)): + previous_end = None + for index, (stroke, points) in enumerate(zip(session.strokes, strokes), start=1): + if should_continue is not None and not should_continue(): + return played + if len(points) < 2: + continue + + # The gap between one finger lifting and the next landing is + # as much a part of a recording as the strokes themselves. + if previous_end is not None: + gap = (stroke.start - previous_end) * 1000.0 / speed + queued += injector.pause(max(0.0, gap)) + + delays = [(later[0] - earlier[0]) * 1000.0 / speed + for earlier, later in zip(stroke.points, stroke.points[1:])] + queued += injector.timed_stroke(points, delays) + previous_end = stroke.end + played += 1 + + if queued >= LOOKAHEAD_MS: + injector.sync() + queued = 0.0 + if progress is not None: + progress(index + turn * len(strokes), + len(strokes) * max(1, repeat)) + injector.sync() + return played + def _draw_paths_raw( self, paths, diff --git a/mthread/gestures.py b/mthread/gestures.py new file mode 100644 index 0000000..2ce478d --- /dev/null +++ b/mthread/gestures.py @@ -0,0 +1,234 @@ +"""Recordings that can be replayed on a phone other than the one they came from. + +The original recording format stores raw kernel events, which is faithful and +useless anywhere else. Two things are wrong with it: + +- The numbers are digitizer coordinates, and a digitizer's range has little to + do with the display's. A recording made on a 4096-step panel replayed on a + 1080-pixel one lands nowhere near where it was made, which is why replaying + one elsewhere was refused rather than attempted. +- Replaying it means writing to ``/dev/input``, which every recent Pixel denies + to the shell. So the raw format cannot even be replayed on the phone that + produced it. + +This module decodes those events into gestures - strokes of ``(time, x, y)`` +with the coordinates as fractions of the screen - and plays them back through +whichever path the device allows, the injector included. A fraction means the +same thing on any screen, so a recording travels. + + session = GestureSession.from_events(recorder.stop().events, device.touch_device) + session.save("login.json") + ... + device.play_gestures(GestureSession.load("login.json")) +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path + +__all__ = ["GESTURE_FORMAT_VERSION", "GestureSession", "Stroke"] + +#: Version 1 was the raw-event format in :mod:`mthread.session`; this is the +#: portable one and reading it is what tells the two apart. +GESTURE_FORMAT_VERSION = 2 + +# Multi-touch protocol B, which is what every modern touchscreen speaks. +EV_SYN, EV_ABS, EV_KEY = 0x00, 0x03, 0x01 +SYN_REPORT = 0x00 +ABS_MT_SLOT = 0x2F +ABS_MT_POSITION_X, ABS_MT_POSITION_Y = 0x35, 0x36 +ABS_MT_TRACKING_ID = 0x39 +ABS_X, ABS_Y = 0x00, 0x01 +BTN_TOUCH = 0x14A + + +@dataclass +class Stroke: + """One finger, from the moment it lands to the moment it lifts.""" + + #: ``(seconds from the start of the recording, x, y)``, x and y in 0..1. + points: list[tuple[float, float, float]] = field(default_factory=list) + + @property + def start(self) -> float: + return self.points[0][0] if self.points else 0.0 + + @property + def end(self) -> float: + return self.points[-1][0] if self.points else 0.0 + + def as_list(self) -> list: + return [[round(t, 4), round(x, 5), round(y, 5)] for t, x, y in self.points] + + @classmethod + def from_list(cls, rows) -> "Stroke": + return cls(points=[(float(t), float(x), float(y)) for t, x, y in rows]) + + +@dataclass +class GestureSession: + """A recording in screen fractions, replayable anywhere.""" + + strokes: list[Stroke] = field(default_factory=list) + screen_size: tuple[int, int] | None = None + device_model: str = "" + device_serial: str = "" + created_at: str = "" + note: str = "" + version: int = GESTURE_FORMAT_VERSION + + def __post_init__(self): + if not self.created_at: + self.created_at = datetime.now(timezone.utc).isoformat(timespec="seconds") + + @property + def duration(self) -> float: + return max((stroke.end for stroke in self.strokes), default=0.0) + + @property + def point_count(self) -> int: + return sum(len(stroke.points) for stroke in self.strokes) + + # ------------------------------------------------------------- recording + + @classmethod + def from_events(cls, events, touch, **fields) -> "GestureSession": + """Decode raw input events into strokes. + + Args: + events: :class:`~mthread.session.InputEvent` objects, in order. + touch: The :class:`~mthread.touch.TouchDevice` they came from, + whose axis ranges are what makes the numbers meaningful. + """ + if touch is None or not touch.is_touchscreen: + raise ValueError("the recording cannot be converted without the digitizer's ranges") + + x_min, x_max = touch.x_range + y_min, y_max = touch.y_range + x_span = max(x_max - x_min, 1) + y_span = max(y_max - y_min, 1) + + # One in-flight stroke per slot. Protocol B addresses a slot, changes + # some of its axes, and commits with SYN_REPORT; a value not mentioned + # since the last report has not changed, which is why the last position + # is carried forward rather than reset. + slot = 0 + live: dict[int, Stroke] = {} + position: dict[int, list[int | None]] = {} + finished: list[Stroke] = [] + pending_lift: list[int] = [] + touching = True # single-touch panels announce this with BTN_TOUCH + + def commit(t: float) -> None: + for index, stroke in list(live.items()): + raw = position.get(index) + if not raw or raw[0] is None or raw[1] is None: + continue + x = (raw[0] - x_min) / x_span + y = (raw[1] - y_min) / y_span + if touch.swap_xy: + x, y = y, x + point = (t, min(1.0, max(0.0, x)), min(1.0, max(0.0, y))) + # A finger reporting the same place twice is not a new sample. + if stroke.points and stroke.points[-1][1:] == point[1:]: + continue + stroke.points.append(point) + + for event in events: + if event.type == EV_ABS: + if event.code == ABS_MT_SLOT: + slot = event.value + elif event.code == ABS_MT_TRACKING_ID: + if event.value < 0: + pending_lift.append(slot) + else: + live[slot] = Stroke() + position.setdefault(slot, [None, None]) + elif event.code in (ABS_MT_POSITION_X, ABS_X): + position.setdefault(slot, [None, None])[0] = event.value + if event.code == ABS_X and slot not in live and touching: + live.setdefault(slot, Stroke()) + elif event.code in (ABS_MT_POSITION_Y, ABS_Y): + position.setdefault(slot, [None, None])[1] = event.value + elif event.type == EV_KEY and event.code == BTN_TOUCH: + touching = bool(event.value) + if touching: + live.setdefault(slot, Stroke()) + position.setdefault(slot, [None, None]) + else: + pending_lift.append(slot) + elif event.type == EV_SYN and event.code == SYN_REPORT: + commit(event.t) + for index in pending_lift: + stroke = live.pop(index, None) + if stroke is not None and len(stroke.points) >= 1: + finished.append(stroke) + pending_lift.clear() + + for stroke in live.values(): + if stroke.points: + finished.append(stroke) + + finished.sort(key=lambda stroke: stroke.start) + return cls(strokes=finished, **fields) + + # ------------------------------------------------------------- replaying + + def to_pixels(self, width: int, height: int) -> list[list[tuple[int, int]]]: + """The strokes as pixel paths on a screen of this size.""" + return [[(round(x * (width - 1)), round(y * (height - 1))) + for _, x, y in stroke.points] for stroke in self.strokes] + + # ------------------------------------------------------------------ file + + def to_dict(self) -> dict: + return { + "version": self.version, + "kind": "gestures", + "created_at": self.created_at, + "device_model": self.device_model, + "device_serial": self.device_serial, + "screen_size": list(self.screen_size) if self.screen_size else None, + "note": self.note, + "duration": round(self.duration, 4), + "stroke_count": len(self.strokes), + "point_count": self.point_count, + "strokes": [stroke.as_list() for stroke in self.strokes], + } + + @classmethod + def from_dict(cls, data: dict) -> "GestureSession": + version = int(data.get("version", 0)) + if version < GESTURE_FORMAT_VERSION: + raise ValueError( + "This is a raw-event recording from an older version. Those hold digitizer " + "coordinates and can only be replayed on the phone that made them; record " + "it again to get a portable one." + ) + if version > GESTURE_FORMAT_VERSION: + raise ValueError( + f"This recording uses format version {version}, but this build only " + f"understands up to {GESTURE_FORMAT_VERSION}. Update mthread to open it." + ) + screen = data.get("screen_size") + return cls( + strokes=[Stroke.from_list(rows) for rows in data.get("strokes", [])], + screen_size=tuple(screen) if screen else None, + device_model=data.get("device_model", ""), + device_serial=data.get("device_serial", ""), + created_at=data.get("created_at", ""), + note=data.get("note", ""), + version=version, + ) + + def save(self, path: str | Path) -> Path: + path = Path(path) + path.write_text(json.dumps(self.to_dict()), encoding="utf-8") + return path + + @classmethod + def load(cls, path: str | Path) -> "GestureSession": + return cls.from_dict(json.loads(Path(path).read_text(encoding="utf-8"))) diff --git a/mthread/injector.py b/mthread/injector.py index e18262d..e182061 100644 --- a/mthread/injector.py +++ b/mthread/injector.py @@ -241,6 +241,26 @@ def stroke(self, points, pacing: Pacing | None = None, self._write(lines) return float(sum(delays)) + def timed_stroke(self, points, delays) -> float: + """Draw a stroke whose waits are given rather than computed. + + Replay needs the timing that was recorded, not the timing a pacing model + would invent, so this takes the gaps as they were measured. + """ + if len(points) < 2: + return 0.0 + lines = [f"D {points[0][0]:.1f} {points[0][1]:.1f}"] + total = 0.0 + for (x, y), delay in zip(points[1:], list(delays) + [0.0] * len(points)): + if delay > 0.05: + lines.append(f"S {delay:.2f}") + total += delay + lines.append(f"M {x:.1f} {y:.1f}") + lines.append(f"U {points[-1][0]:.1f} {points[-1][1]:.1f}") + with self._lock: + self._write(lines) + return total + def pause(self, millis: float) -> float: if millis > 0.05: with self._lock: diff --git a/mthread_draw/geometry.py b/mthread_draw/geometry.py index 0de5b0b..6ba2bde 100644 --- a/mthread_draw/geometry.py +++ b/mthread_draw/geometry.py @@ -5,10 +5,11 @@ from __future__ import annotations -from dataclasses import dataclass +import math +from dataclasses import dataclass, replace from typing import Sequence -__all__ = ["CanvasView", "place_paths"] +__all__ = ["CanvasView", "Placement", "fit_to_screen", "place_on_screen", "place_paths"] @dataclass(frozen=True) @@ -83,6 +84,85 @@ def place_paths( return placed +@dataclass(frozen=True) +class Placement: + """Where a drawing goes on the screen, and how big and how tilted. + + Held in fractions of the screen rather than pixels so that it survives the + phone being turned, and so the same numbers mean the same thing on a + different device. + + Attributes: + centre: Where the middle of the drawing lands, 0 to 1 across the screen. + scale: Multiplier on the size it would have if fitted to the screen, so + 1.0 is "as large as it goes with a margin". + rotation: Degrees clockwise. + """ + + centre: tuple[float, float] = (0.5, 0.5) + scale: float = 1.0 + rotation: float = 0.0 + + def moved(self, dx: float, dy: float) -> "Placement": + return replace(self, centre=(self.centre[0] + dx, self.centre[1] + dy)) + + def zoomed(self, factor: float, *, low: float = 0.05, high: float = 6.0) -> "Placement": + return replace(self, scale=max(low, min(high, self.scale * factor))) + + def turned(self, degrees: float) -> "Placement": + return replace(self, rotation=(self.rotation + degrees) % 360.0) + + +def place_on_screen(paths, width: int, height: int, placement: Placement, + *, margin: float = 0.06): + """Put *paths* on a screen of *width* x *height* as *placement* says. + + The base size is the one :func:`fit_to_screen` would choose, so a scale of + one means the same thing whatever the drawing and whatever the phone, and + the slider a person turns is a multiplier on something recognisable. + """ + points = [point for path in paths for point in path] + if not points: + return [] + + min_x = min(x for x, _ in points) + max_x = max(x for x, _ in points) + min_y = min(y for _, y in points) + max_y = max(y for _, y in points) + source_width = max(max_x - min_x, 1) + source_height = max(max_y - min_y, 1) + source_centre = ((min_x + max_x) / 2.0, (min_y + max_y) / 2.0) + + # Rotation happens about the drawing's own middle, so turning it does not + # also throw it across the screen. + angle = math.radians(placement.rotation) + cos, sin = math.cos(angle), math.sin(angle) + + # Fit the rotated bounding box, not the upright one; a drawing turned + # forty-five degrees needs more room than it did before, and without this it + # would quietly grow past the edges of the screen. + turned_width = abs(source_width * cos) + abs(source_height * sin) + turned_height = abs(source_width * sin) + abs(source_height * cos) + + pad_x, pad_y = width * margin, height * margin + base = min((width - 2 * pad_x) / turned_width, (height - 2 * pad_y) / turned_height) + scale = base * placement.scale + + target_x = placement.centre[0] * width + target_y = placement.centre[1] * height + + placed = [] + for path in paths: + line = [] + for x, y in path: + ox, oy = x - source_centre[0], y - source_centre[1] + rx = ox * cos - oy * sin + ry = ox * sin + oy * cos + line.append((round(target_x + rx * scale), round(target_y + ry * scale))) + placed.append(line) + return placed + + def fit_to_screen(paths, width: int, height: int, *, margin: float = 0.06): """Scale paths to fill the device screen, keeping shape and a clear margin. diff --git a/mthread_draw/server.py b/mthread_draw/server.py index c9405d7..54e2536 100644 --- a/mthread_draw/server.py +++ b/mthread_draw/server.py @@ -40,10 +40,12 @@ from PIL import Image, ImageDraw from mthread import Device, VectorizeSettings, Vectorizer, find_devices, simulate +from mthread.gestures import GestureSession from mthread.mirror import ScreenMirror +from mthread.recorder import Recorder from mthread.errors import MThreadError -from .geometry import fit_to_screen +from .geometry import Placement, place_on_screen #: Width of the placement overlay. It is stretched over the mirror by the #: front end, so it only has to be fine enough to read, not full resolution. @@ -67,6 +69,11 @@ def __init__(self, write): #: The size of the last frame captured, which is the only thing that #: knows which way up the phone is being held. self.frame_size: tuple[int, int] | None = None + #: Where the drawing sits. Starts fitted to the screen; the front end + #: moves, scales and turns it from there. + self.placement = Placement() + self.recorder: Recorder | None = None + self.session: GestureSession | None = None # ---------------------------------------------------------------- talking @@ -159,6 +166,30 @@ def op_load_image(self, path: str) -> dict: image = self.vectorizer.load_image(path) return {"width": int(image.shape[1]), "height": int(image.shape[0])} + def op_place(self, dx: float = 0.0, dy: float = 0.0, zoom: float = 1.0, + turn: float = 0.0, reset: bool = False) -> dict: + """Nudge the drawing about, and redraw the overlay that shows where. + + Relative rather than absolute, because the front end is describing a + gesture - a drag of so far, a wheel notch - and relative keeps the two + ends from disagreeing about what the current position is. + """ + if reset: + self.placement = Placement() + else: + if dx or dy: + self.placement = self.placement.moved(dx, dy) + if zoom and zoom != 1.0: + self.placement = self.placement.zoomed(zoom) + if turn: + self.placement = self.placement.turned(turn) + return { + "overlay": self._overlay(), + "centre": list(self.placement.centre), + "scale": round(self.placement.scale, 3), + "rotation": round(self.placement.rotation, 1), + } + def op_preview(self, sensitivity: float = 5, detail: float = 7, method: str = "canny", margin: float = 0.06) -> dict: started = time.perf_counter() @@ -170,13 +201,13 @@ def op_preview(self, sensitivity: float = 5, detail: float = 7, preview.save(path) return { "path": str(path), - "overlay": self._overlay(margin), + "overlay": self._overlay(), "strokes": len(paths), "points": sum(len(p) for p in paths), "seconds": round(time.perf_counter() - started, 2), } - def _overlay(self, margin: float) -> str | None: + def _overlay(self) -> str | None: """The strokes where they will actually land, on a transparent ground. A preview beside the phone tells you what will be drawn; a preview lying @@ -188,7 +219,7 @@ def _overlay(self, margin: float) -> str | None: return None width, height = self.screen_now() - placed = fit_to_screen(self.paths, width, height, margin=margin) + placed = place_on_screen(self.paths, width, height, self.placement) scale = OVERLAY_WIDTH / width image = Image.new("RGBA", (OVERLAY_WIDTH, round(height * scale)), (0, 0, 0, 0)) @@ -209,14 +240,14 @@ def op_estimate(self, speed: float = 1.0, human: float = 0.0) -> dict: paths = simulate(self.paths, human, seed=0) if human > 0 else self.paths return {"seconds": round(self.device.estimate_duration(paths, speed=speed), 1)} - def op_draw(self, margin: float = 0.06, speed: float = 1.0, - human: float = 0.0) -> dict: + def op_draw(self, speed: float = 1.0, human: float = 0.0, + margin: float = 0.06) -> dict: self._require_device() if not self.paths: raise MThreadError("Load an image and take a preview first.") width, height = self.screen_now() - placed = fit_to_screen(self.paths, width, height, margin=margin) + placed = place_on_screen(self.paths, width, height, self.placement) self.cancel.clear() self.drawing.set() self.status(f"Drawing {len(placed)} strokes") @@ -233,6 +264,73 @@ def op_draw(self, margin: float = 0.06, speed: float = 1.0, self.drawing.clear() return {"strokes": drawn, "stopped": self.cancel.is_set()} + # ---------------------------------------------------- recording gestures + + def op_record_start(self) -> dict: + self._require_device() + if self.recorder is not None and self.recorder.is_recording: + raise MThreadError("Already recording.") + self.recorder = Recorder(self.device) + self.recorder.start() + self.status("Recording. Do something on the phone.") + return {"recording": True} + + def op_record_stop(self) -> dict: + if self.recorder is None: + raise MThreadError("Nothing is being recorded.") + raw = self.recorder.stop() + self.recorder = None + + # The digitizer's ranges are what turn its numbers into fractions of a + # screen, and they are only knowable here, on the phone that recorded. + self.session = GestureSession.from_events( + raw.events, self.device.touch_device, + screen_size=self.device.screen_size, + device_serial=self.device.serial, + device_model=self.device.model, + ) + return self._session_summary() + + def op_session_save(self, path: str) -> dict: + if self.session is None: + raise MThreadError("There is no recording to save.") + self.session.save(path) + return {"path": path, **self._session_summary()} + + def op_session_open(self, path: str) -> dict: + self.session = GestureSession.load(path) + return {"path": path, **self._session_summary()} + + def op_play(self, speed: float = 1.0, repeat: int = 1) -> dict: + self._require_device() + if self.session is None or not self.session.strokes: + raise MThreadError("Open or record something first.") + + self.cancel.clear() + self.drawing.set() + self.status(f"Replaying {len(self.session.strokes)} strokes") + try: + played = self.device.play_gestures( + self.session, speed=speed, repeat=repeat, + progress=lambda done, total: self.event("progress", done=done, total=total), + should_continue=lambda: not self.cancel.is_set(), + ) + finally: + self.drawing.clear() + return {"strokes": played, "stopped": self.cancel.is_set()} + + def _session_summary(self) -> dict: + session = self.session + if session is None: + return {"strokes": 0, "points": 0, "seconds": 0.0} + return { + "strokes": len(session.strokes), + "points": session.point_count, + "seconds": round(session.duration, 2), + "recorded_on": session.device_model or session.device_serial, + "recorded_size": list(session.screen_size) if session.screen_size else None, + } + def op_stop(self) -> dict: self.cancel.set() return {"stopping": True} @@ -259,7 +357,8 @@ def _require_device(self) -> None: #: Operations that may take a while, and so must not block the reader. -SLOW = {"preview", "draw", "screenshot", "connect", "mirror"} +SLOW = {"preview", "draw", "screenshot", "connect", "mirror", "place", + "play", "record_stop", "session_open", "session_save"} def serve(stdin=None, stdout=None) -> int: diff --git a/tests/test_gestures.py b/tests/test_gestures.py new file mode 100644 index 0000000..2a88651 --- /dev/null +++ b/tests/test_gestures.py @@ -0,0 +1,127 @@ +"""Decoding raw touch events into recordings that travel.""" + +import json +import tempfile +import unittest +from pathlib import Path + +from mthread.gestures import GESTURE_FORMAT_VERSION, GestureSession, Stroke +from mthread.session import InputEvent +from mthread.touch import TouchDevice + + +def panel(**fields) -> TouchDevice: + """A digitizer whose range is deliberately not the display's.""" + return TouchDevice(path="/dev/input/event2", name="touch", + x_range=(0, 4095), y_range=(0, 4095), has_slot=True, **fields) + + +def events(rows) -> list[InputEvent]: + return [InputEvent(t=t, device="/dev/input/event2", type=etype, code=code, value=value) + for t, etype, code, value in rows] + + +class DecodingTests(unittest.TestCase): + + def test_one_finger_becomes_one_stroke_in_screen_fractions(self): + session = GestureSession.from_events(events([ + (0.00, 0x03, 0x39, 5), # tracking id: a finger arrives + (0.00, 0x03, 0x35, 0), # x at the very left + (0.00, 0x03, 0x36, 0), # y at the very top + (0.00, 0x00, 0x00, 0), # SYN_REPORT + (0.05, 0x03, 0x35, 4095), # to the far corner + (0.05, 0x03, 0x36, 4095), + (0.05, 0x00, 0x00, 0), + (0.10, 0x03, 0x39, -1), # lifted + (0.10, 0x00, 0x00, 0), + ]), panel()) + + self.assertEqual(len(session.strokes), 1) + self.assertEqual(session.strokes[0].points, + [(0.0, 0.0, 0.0), (0.05, 1.0, 1.0)]) + + def test_the_same_recording_lands_on_any_screen(self): + """The point of the format: a fraction means the same thing everywhere. + + One maps to the last addressable pixel rather than to the width, so a + touch recorded at the very edge stays on the screen instead of landing + one pixel past it. + """ + session = GestureSession(strokes=[Stroke([(0.0, 0.0, 0.5), (0.1, 1.0, 0.5)])]) + self.assertEqual(session.to_pixels(1000, 2000), [[(0, 1000), (999, 1000)]]) + self.assertEqual(session.to_pixels(400, 800), [[(0, 400), (399, 400)]]) + + def test_two_fingers_are_two_strokes(self): + session = GestureSession.from_events(events([ + (0.0, 0x03, 0x2F, 0), (0.0, 0x03, 0x39, 1), + (0.0, 0x03, 0x35, 1000), (0.0, 0x03, 0x36, 1000), + (0.0, 0x03, 0x2F, 1), (0.0, 0x03, 0x39, 2), + (0.0, 0x03, 0x35, 3000), (0.0, 0x03, 0x36, 3000), + (0.0, 0x00, 0x00, 0), + (0.1, 0x03, 0x2F, 0), (0.1, 0x03, 0x35, 1200), + (0.1, 0x03, 0x2F, 1), (0.1, 0x03, 0x35, 2800), + (0.1, 0x00, 0x00, 0), + (0.2, 0x03, 0x2F, 0), (0.2, 0x03, 0x39, -1), + (0.2, 0x03, 0x2F, 1), (0.2, 0x03, 0x39, -1), + (0.2, 0x00, 0x00, 0), + ]), panel()) + + self.assertEqual(len(session.strokes), 2) + self.assertEqual([len(stroke.points) for stroke in session.strokes], [2, 2]) + + def test_a_finger_that_does_not_move_reports_once(self): + """A held finger repeats its position every frame; that is not motion.""" + rows = [(0.0, 0x03, 0x39, 7), (0.0, 0x03, 0x35, 100), (0.0, 0x03, 0x36, 100), + (0.0, 0x00, 0x00, 0)] + for frame in range(1, 6): + rows.append((frame / 60.0, 0x00, 0x00, 0)) + rows += [(0.2, 0x03, 0x39, -1), (0.2, 0x00, 0x00, 0)] + + session = GestureSession.from_events(events(rows), panel()) + self.assertEqual(len(session.strokes[0].points), 1) + + def test_a_swapped_panel_is_untangled_on_the_way_in(self): + session = GestureSession.from_events(events([ + (0.0, 0x03, 0x39, 1), (0.0, 0x03, 0x35, 0), (0.0, 0x03, 0x36, 4095), + (0.0, 0x00, 0x00, 0), (0.1, 0x03, 0x39, -1), (0.1, 0x00, 0x00, 0), + ]), panel(swap_xy=True)) + self.assertEqual(session.strokes[0].points[0][1:], (1.0, 0.0)) + + def test_a_recording_without_the_digitizer_ranges_is_refused(self): + with self.assertRaises(ValueError): + GestureSession.from_events(events([]), TouchDevice(path="/dev/input/event2")) + + +class FileTests(unittest.TestCase): + + def test_it_round_trips(self): + session = GestureSession(strokes=[Stroke([(0.0, 0.1, 0.2), (0.5, 0.3, 0.4)])], + screen_size=(1080, 1920), device_model="Pixel 8 Pro") + with tempfile.TemporaryDirectory() as folder: + path = session.save(Path(folder) / "one.json") + back = GestureSession.load(path) + self.assertEqual(back.device_model, "Pixel 8 Pro") + self.assertEqual(back.screen_size, (1080, 1920)) + self.assertEqual(back.strokes[0].points, session.strokes[0].points) + self.assertAlmostEqual(back.duration, 0.5) + + def test_an_old_raw_recording_says_what_to_do_about_it(self): + """Version 1 files hold digitizer coordinates and cannot be made portable.""" + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "old.json" + path.write_text(json.dumps({"version": 1, "events": []}), encoding="utf-8") + with self.assertRaises(ValueError) as caught: + GestureSession.load(path) + self.assertIn("record it again", str(caught.exception).lower()) + + def test_a_newer_file_is_refused_rather_than_half_read(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "future.json" + path.write_text(json.dumps({"version": GESTURE_FORMAT_VERSION + 1}), + encoding="utf-8") + with self.assertRaises(ValueError): + GestureSession.load(path) + + +if __name__ == "__main__": + unittest.main() diff --git a/winui/MainWindow.xaml b/winui/MainWindow.xaml index 5425a66..79fd174 100644 --- a/winui/MainWindow.xaml +++ b/winui/MainWindow.xaml @@ -93,6 +93,27 @@ Click="OnDraw" IsEnabled="False"/>