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 c07e119..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 @@ -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. @@ -105,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 @@ -159,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) @@ -353,6 +371,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 +383,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)) @@ -379,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 db5efed..e182061 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,34 @@ def stroke(self, points, pacing: Pacing | None = None, with self._lock: 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) -> 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/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 a603698..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. @@ -64,6 +66,14 @@ 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 + #: 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 @@ -138,7 +148,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: @@ -153,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() @@ -164,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 @@ -181,8 +218,8 @@ def _overlay(self, margin: float) -> str | None: if self.device is None or not self.paths: return None - width, height = self.device.screen_size - placed = fit_to_screen(self.paths, width, height, margin=margin) + width, height = self.screen_now() + 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)) @@ -203,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.device.screen_size - placed = fit_to_screen(self.paths, width, height, margin=margin) + width, height = self.screen_now() + placed = place_on_screen(self.paths, width, height, self.placement) self.cancel.clear() self.drawing.set() self.status(f"Drawing {len(placed)} strokes") @@ -227,17 +264,101 @@ 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} + 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.") #: 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_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/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/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 b/winui/MainWindow.xaml index 5425a66..79fd174 100644 --- a/winui/MainWindow.xaml +++ b/winui/MainWindow.xaml @@ -93,6 +93,27 @@ Click="OnDraw" IsEnabled="False"/>