path: root/tools/ui_flow.py

#!/usr/bin/env python3
# Author:  cirakg & claude ai
# Summary: Drive the app in the headless simulator with scripted input and check a screenshot of every screen.
"""Scripted UI flow for the UI-test build (README 17.4).

Usage (inside the dev container, with the simulator running and the
UI-test build launched by docker/scripts/screenshots.sh):
    ui_flow.py --device fr70 --device-dir DIR --log APP.log --window ID
               --out build/screenshots/fr70 --golden test/golden/fr70
               [--scenario main|extra] [--approve]

Input goes to the simulator window with xdotool: clicks on the device
drawing's buttons (positions from DIR/simulator.json) and taps and swipes on
the screen. Watches without a button for an action use touch instead (Venu
Sq 2 has no UP/DOWN and no select button).

The app (UI-test build, source/util/TestMode.mc) starts with fake entries,
the PIN 1234 and a frozen clock, and prints "DIAG shown <screen>" after
drawing each screen, so every step waits for the screen it expects instead
of sleeping. Menus are native Menu2 views that print nothing; for those the
script waits a fixed time.

The flow covers: the results of an Aegis import and of the Add OTP, Add
code and Manage entry forms, all sent through the settings (the fixture
sets the forms, as the phone would), the two entries those forms added (a
TOTP whose code the golden image shows, and a Code 128 card whose
screenshot must decode), the Manage form's change to the locked Bank entry
waiting for the PIN and confirmed from the menu (watches with buttons), the
list, a TOTP entry, a HOTP entry and its next code, the
PIN screen, a wrong PIN, a correct PIN opening a locked entry, auto-lock
closing it, the menu, the About screen (watches with buttons; touch-only
menus have no focus to move), the lockout after 5 wrong PINs, the barcodes
whose screenshots must decode with zxing-cpp (README 12.4), the Steam and
Library entries, and the text screens the menu opens (settings results,
entry IDs, clock status, About on touch-only watches, the benchmark),
scrolled with the buttons or a swipe. The glance with the favourite's code
is a separate launch (--shot glance_code, from screenshots.sh): the
simulator doesn't return to the glance when the app exits.

--scenario extra drives the second UI-test build (uitest-extra.jungle,
Fixtures.loadExtra): a "Require PIN" from the Manage form waiting while the
watch is locked; UPC-A and Code 39 cards that must decode; an EAN-13 with a
wrong check digit (the code error screen); a QR whose encoding progress must
start low and only go up; a 6-digit PIN from before the 4-digit rule,
unlocked with OK and replaced (new PIN, a repeat that differs, the repeat);
the waiting lock applied at that PIN; auto-lock closing a locked code;
Change PIN; Reset PIN and creating a PIN. Its glance (no favourite code) is
checked as glance.png.

Menus are reached by label: the debug build logs the menu's items ("DIAG
menu items=..."). Watches with buttons move the focus; touch-only watches
(Venu Sq 2, Venu X1) have none, so the item is tapped where that watch's
menu draws it (TOUCH_MENU_ROWS), after slow drags whose distance is
measured on the screen.

Screens whose content changes from run to run (the encoding progress, the
benchmark's timings) are saved for review only, never compared.
Each screenshot is compared with test/golden/<device>/<screen>.png like
sim_screenshot.py does (or saved there with --approve).

Exit: 0 all screens match, 1 a screen differs or is missing a golden image,
2 the flow failed (a screen didn't appear in time).
"""

from __future__ import annotations

import argparse
import itertools
import json
import re
import subprocess
import sys
import time
from dataclasses import dataclass
from pathlib import Path

import sim_screenshot as shots
import zxingcpp
from PIL import Image, ImageChops, ImageStat

EXIT_OK, EXIT_DIFFERENT, EXIT_FLOW = 0, 1, 2
SCREEN_TIMEOUT_S = 20.0
AUTO_LOCK_TIMEOUT_S = 25.0  # the UI-test build locks after 10 s without input
AUTO_LOCK_IDLE_S = 13.0  # idle time after which that lock has surely happened
IMPORT_TIMEOUT_S = 90.0  # the 100-entry fixture import runs on a timer after start-up
IMPORT_MEMORY_BUDGET = 0.70  # README 18: peak memory at most 70 % of the watch-app memory
MENU_SETTLE_S = 1.5  # Menu2 has no "shown" line
SETTLE_S = 0.4  # after a "shown" line, before the screenshot
CLICK_PAUSE_S = 0.25
HOLD_S = 1.6  # long enough for a hold (menu) on every device
FIXTURE_PIN = "1234"  # source/app/Fixtures.mc
WRONG_PIN = "0000"
FAILURES_UNTIL_WAIT = 5  # source/storage/Lockout.mc: the 5th failure starts a delay
SELECTOR_DIGITS = 10  # source/ui/PinView.mc: selector positions 0-9 (no OK; the 4th digit submits)
GYM_CARD = "5901234123457"  # source/app/Fixtures.mc: EAN-13 with its check digit
MEMBER_QR = (  # source/app/Fixtures.mc: a version 6 QR
    "https://example.com/member/fake-0042?name=Alice%20Example&card=0000111122223333&valid=2027-12-31&level=gold"
)
PARCEL_DM = "PARCEL-FAKE-0042-7731"  # source/app/Fixtures.mc: Data Matrix
TICKET_PDF417 = "Boarding pass FAKE 0042 seat 12A"  # source/app/Fixtures.mc: PDF417
RAIL_AZTEC = "RAIL-FAKE-0042 Zone 1-3"  # source/app/Fixtures.mc: Aztec
LOCKER_CODE128 = "FAKE-0042"  # source/app/Fixtures.mc: the Add code form's Code 128 card
# The Add OTP form's TOTP ("Hello!" demo key) at TestMode.NOW is 309848, next
# 489290 (RFC 6238 computed in Python); the golden image otp_form shows them.
QR_TIMEOUT_S = 30.0  # the QR is encoded in timer steps the first time
LEGACY_PIN = "123456"  # source/app/Fixtures.mc FIXTURE_LEGACY_PIN: a PIN from before the 4-digit rule
NEW_PIN, OTHER_PIN, CHANGED_PIN, CREATED_PIN = "5678", "5679", "2468", "1357"
SELECTOR_LEGACY = 11  # the selector adds "OK" (position 10) for a legacy PIN
KEY_OK = -2  # source/ui/PinView.mc: the keypad's OK key (legacy PINs only)
MODE_UNLOCK, MODE_CREATE, MODE_CHANGE = 0, 1, 2  # source/ui/PinView.mc
# source/app/Fixtures.mc loadExtra: the UPC-A 036000291452, which zxing-cpp
# reports as the EAN-13 it is a subset of, and the Code 39 "FAKE-39" with the
# mod 43 check character the app adds ("L"), which zxing-cpp reports as text.
SHOP_UPCA = "0036000291452"
BADGE_CODE39 = "FAKE-39L"
BIG_QR_TEXT = "Fake event ticket 0042 for the UI test." + " Row 12, seat 34, gate b, valid all day." * 4
PROGRESS_FIRST_MAX = 30  # the encoding progress must start below this (percent)
PROGRESS_MIN_STEPS = 3  # and show at least this many different values
BENCH_TIMEOUT_S = 30.0
# Touch-only menus (no focus): the centre of the first item and the item
# pitch, in screen pixels, at the top of a Menu2 with a title, as these
# watches draw it (measured from the simulator).
TOUCH_MENU_ROWS = {"venusq2": (100, 100), "venux1": (184, 140)}
DRAG_STEPS = 30  # a slow drag: the list follows the finger without a fling
DRAG_HOLD_S = 0.6
MAX_MENU_DRAGS = 16


class FlowError(Exception):
    """The app didn't reach an expected screen."""


@dataclass(frozen=True)
class Key:
    x: int
    y: int
    behavior: str
    hold: bool


def barcodes_on(screen: Image.Image) -> list[str]:
    """The texts of every barcode zxing-cpp finds on a screenshot."""
    return [r.text for r in zxingcpp.read_barcodes(screen)]


def vertical_shift(before: Image.Image, after: Image.Image) -> int:
    """How many pixels the content moved up from `before` to `after` (a
    scrolled list; negative: down): the shift whose overlapping rows differ
    least, compared in the middle half of the screen."""
    a, b = before.convert("L"), after.convert("L")
    width, height = a.size
    best, best_score = 0, float("inf")
    top, bottom = height // 4, height * 3 // 4
    for shift in range(-(height // 4), height // 4 + 1):
        moved = ImageChops.difference(a.crop((0, top + shift, width, bottom + shift)), b.crop((0, top, width, bottom)))
        score = ImageStat.Stat(moved).mean[0]
        if score < best_score:
            best, best_score = shift, score
    return best


def load_keys(device_dir: Path) -> dict[str, Key]:
    """Button centres in device-drawing coordinates, by button id (enter, up, down, esc, menu)."""
    sim = json.loads((device_dir / "simulator.json").read_text(encoding="utf-8"))
    keys: dict[str, Key] = {}
    for key in sim.get("keys", []):
        loc = key["location"]
        keys[str(key["id"])] = Key(
            int(loc["x"]) + int(loc["width"]) // 2,
            int(loc["y"]) + int(loc["height"]) // 2,
            str(key.get("behavior", "")),
            bool(key.get("isHold", False)),
        )
    return keys


class AppLog:
    """The app's console output, read as it grows."""

    def __init__(self, path: Path) -> None:
        self.path = path
        self.mark = 0

    def lines(self) -> list[str]:
        try:
            return self.path.read_text(encoding="utf-8", errors="replace").splitlines()
        except OSError:
            return []

    def set_mark(self) -> None:
        """Later waits only look at lines written after this point."""
        self.mark = len(self.lines())

    def wait_for(self, pattern: str, timeout: float) -> re.Match[str]:
        regex = re.compile(pattern)
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            for line in self.lines()[self.mark :]:
                match = regex.search(line)
                if match:
                    return match
            time.sleep(0.05)
        raise FlowError(f"timed out after {timeout:.0f} s waiting for /{pattern}/")

    def latest(self, pattern: str) -> re.Match[str] | None:
        regex = re.compile(pattern)
        found = None
        for line in self.lines():
            match = regex.search(line)
            if match:
                found = match
        return found


class Simulator:
    """Input to and screenshots of the simulator window."""

    def __init__(self, window: str, device_dir: Path, raw_dir: Path) -> None:
        self.window = window
        self.device_dir = device_dir
        self.display = shots.load_display(device_dir)
        self.keys = load_keys(device_dir)
        self.raw_dir = raw_dir
        self.offset = self._find_offset()

    def capture(self, name: str) -> Path:
        path = self.raw_dir / f"{name}-window.png"
        path.parent.mkdir(parents=True, exist_ok=True)
        subprocess.run(["import", "-window", self.window, str(path)], check=True)
        return path

    def _find_offset(self) -> tuple[int, int]:
        raw = shots.open_image(self.capture("offset"))
        device = shots.open_image(self.device_dir / self.display.image)
        dx, dy, score = shots.find_offset(raw, shots.bezel_only(device, self.display))
        if score > shots.MAX_BEZEL_DIFF:
            raise FlowError(f"device drawing not found in the simulator window (difference {score:.1f})")
        return dx, dy

    def _xdotool(self, *args: str) -> None:
        subprocess.run(["xdotool", *args], check=True)

    def _move(self, wx: int, wy: int) -> None:
        self._xdotool("mousemove", "--window", self.window, str(wx), str(wy))

    def _to_window(self, dx: int, dy: int) -> tuple[int, int]:
        """Device-drawing coordinates to window coordinates."""
        return self.offset[0] + dx, self.offset[1] + dy

    def has(self, key_id: str, behavior: str | None = None) -> bool:
        key = self.keys.get(key_id)
        return key is not None and (behavior is None or key.behavior == behavior)

    def press(self, key_id: str, hold: bool = False) -> None:
        key = self.keys[key_id]
        self._move(*self._to_window(key.x, key.y))
        if hold:
            self._xdotool("mousedown", "1")
            time.sleep(HOLD_S)
            self._xdotool("mouseup", "1")
        else:
            self._xdotool("click", "1")
        time.sleep(CLICK_PAUSE_S)

    def tap(self, sx: int, sy: int) -> None:
        """Tap at screen coordinates (0, 0 = top left of the watch screen)."""
        self._move(*self._to_window(self.display.x + sx, self.display.y + sy))
        self._xdotool("click", "1")
        time.sleep(CLICK_PAUSE_S)

    def long_press(self, sx: int, sy: int) -> None:
        self._move(*self._to_window(self.display.x + sx, self.display.y + sy))
        self._xdotool("mousedown", "1")
        time.sleep(HOLD_S)
        self._xdotool("mouseup", "1")
        time.sleep(CLICK_PAUSE_S)

    def drag(self, sx: int, sy0: int, sy1: int) -> None:
        """A slow vertical drag that stops before lifting the finger, so a
        list moves with it and doesn't fling on."""
        self._move(*self._to_window(self.display.x + sx, self.display.y + sy0))
        self._xdotool("mousedown", "1")
        for i in range(1, DRAG_STEPS + 1):
            self._move(*self._to_window(self.display.x + sx, self.display.y + sy0 + (sy1 - sy0) * i // DRAG_STEPS))
            time.sleep(0.03)
        time.sleep(DRAG_HOLD_S)
        self._xdotool("mouseup", "1")
        time.sleep(CLICK_PAUSE_S)

    def swipe(self, sx0: int, sy0: int, sx1: int, sy1: int) -> None:
        self._move(*self._to_window(self.display.x + sx0, self.display.y + sy0))
        self._xdotool("mousedown", "1")
        steps = 8
        for i in range(1, steps + 1):
            x = sx0 + (sx1 - sx0) * i // steps
            y = sy0 + (sy1 - sy0) * i // steps
            self._move(*self._to_window(self.display.x + x, self.display.y + y))
            time.sleep(0.02)
        self._xdotool("mouseup", "1")
        time.sleep(CLICK_PAUSE_S * 2)


class Flow:
    """The abstract actions of README 13.4, mapped to this device's buttons or touch screen."""

    def __init__(self, sim: Simulator, log: AppLog, out: Path, golden: Path, approve: bool) -> None:
        self.sim = sim
        self.log = log
        self.out = out
        self.golden = golden
        self.approve = approve
        self.results: dict[str, int] = {}
        self.selector = 0  # the PIN digit selector's position (button-only watches)
        touch = log.wait_for(r"DIAG device touch=(\d)", SCREEN_TIMEOUT_S)
        self.touch = touch.group(1) == "1"

    # -- navigation ---------------------------------------------------------

    def _w(self) -> int:
        return self.sim.display.width

    def _h(self) -> int:
        return self.sim.display.height

    def next(self) -> None:
        if self.sim.has("down"):
            self.sim.press("down")
        else:
            self.sim.swipe(self._w() // 2, self._h() * 3 // 4, self._w() // 2, self._h() // 4)

    def prev(self) -> None:
        if self.sim.has("up"):
            self.sim.press("up")
        else:
            self.sim.swipe(self._w() // 2, self._h() // 4, self._w() // 2, self._h() * 3 // 4)

    def back(self) -> None:
        self.sim.press("esc")

    def menu(self) -> None:
        """Hold MENU, or long-press the screen on watches without one (Venu X1)."""
        if self.sim.has("menu"):
            self.sim.press("menu", hold=True)
        else:
            self.sim.long_press(self._w() // 2, self._h() // 2)

    def open_focused_row(self) -> None:
        if self.sim.has("enter", "onSelect"):
            self.sim.press("enter")
            return
        focus = self.log.latest(r"DIAG list_focus y=(\d+)")
        if focus is None:
            raise FlowError("the list didn't report its focused row")
        self.sim.tap(self._w() // 2, int(focus.group(1)))

    def select(self) -> None:
        """START, or a tap in the middle of the screen."""
        if self.sim.has("enter", "onSelect"):
            self.sim.press("enter")
        else:
            self.sim.tap(self._w() // 2, self._h() // 2)

    def enter_pin(self, pin: str, legacy: bool = False, ok: bool | None = None) -> None:
        """Type a PIN; the 4th digit submits it. A legacy PIN (any length)
        is submitted with OK (unless ok is False: more digits follow), after
        which the screen's selector is back at 0."""
        submit = legacy if ok is None else ok
        if self.touch:
            keys = self.log.latest(r"DIAG pin_keys (\S+)")
            if keys is None:
                raise FlowError("the PIN screen didn't report its keypad")
            centres = {}
            for item in keys.group(1).split(";"):
                value, xy = item.split("=")
                x, y = xy.split(",")
                centres[int(value)] = (int(x), int(y))
            for digit in pin:
                self.sim.tap(*centres[int(digit)])
            if submit:
                if KEY_OK not in centres:
                    raise FlowError("the legacy PIN keypad has no OK key")
                self.sim.tap(*centres[KEY_OK])
            return
        positions = SELECTOR_LEGACY if legacy else SELECTOR_DIGITS
        for position in [int(d) for d in pin] + ([SELECTOR_DIGITS] if submit else []):
            steps = (position - self.selector) % positions
            for _ in range(steps):
                self.sim.press("down")
            self.selector = position
            self.sim.press("enter")
        if submit:
            self.selector = 0  # the new-PIN screen has no OK position

    def expect_pin(self, mode: int, digits: int = 0, repeat: bool = False, busy: bool = False) -> None:
        """Wait until the PIN screen shows this state (PinView MODE_*: 0
        unlock, 1 create, 2 change): every key press redraws the screen, so
        "shown pin" alone can come before the last digit arrived."""
        state = f"mode={mode} digits={digits} repeat={int(repeat)} busy={int(busy)}"
        self.log.wait_for(rf"DIAG pin_state {state}\s*$", SCREEN_TIMEOUT_S)
        time.sleep(SETTLE_S)

    def menu_items(self) -> list[str]:
        """The open menu's item labels, from its debug log line."""
        items = self.log.latest(r"DIAG menu items=(.*?)\s*$")
        if items is None:
            raise FlowError("the menu didn't report its items")
        return items.group(1).split("|")

    def open_menu_item(self, label: str) -> None:
        """Open the menu and select the item `label`."""
        self.act()
        self.menu()
        self.log.wait_for(r"DIAG menu items=", SCREEN_TIMEOUT_S)
        time.sleep(MENU_SETTLE_S)
        items = self.menu_items()
        if label not in items:
            raise FlowError(f"the menu has no item {label!r}: {items}")
        index = items.index(label)
        self.act()
        if self.sim.has("down"):
            for _ in range(index):
                self.next()
            time.sleep(MENU_SETTLE_S)
            self.select()
            return
        self.tap_touch_menu_item(index)

    def tap_touch_menu_item(self, index: int) -> None:
        """Tap item `index` of a menu without focus, dragging it into view
        first; each drag's real distance is measured on the screen."""
        name = self.sim.device_dir.name
        if name not in TOUCH_MENU_ROWS:
            raise FlowError(f"no menu layout for the touch-only watch {name} in TOUCH_MENU_ROWS")
        first, pitch = TOUCH_MENU_ROWS[name]
        x, h = self._w() // 2, self._h()
        # A new menu doesn't always start at the top: drag down to it first.
        for _ in range(MAX_MENU_DRAGS):
            before = self.screen("menu-before")
            self.sim.drag(x, h // 4, h // 4 + h // 5)
            time.sleep(MENU_SETTLE_S)
            if vertical_shift(before, self.screen("menu-after")) == 0:
                break
        y = first + index * pitch
        for _ in range(MAX_MENU_DRAGS):
            if y <= h - pitch // 2:
                break
            before = self.screen("menu-before")
            distance = min(y - h // 2, h // 5)
            self.sim.drag(x, h * 3 // 4, h * 3 // 4 - distance)
            time.sleep(MENU_SETTLE_S)
            shift = vertical_shift(before, self.screen("menu-after"))
            if shift <= 0:
                break  # the end of the menu
            y -= shift
        if not 0 < y < h:
            raise FlowError(f"menu item {index} can't be brought into view")
        self.sim.tap(x, y)

    def screen(self, name: str) -> Image.Image:
        """The current screen, uncompared (saved under raw/)."""
        return shots.extract_screen(self.sim.capture(name), self.sim.device_dir, self.out / "raw" / f"{name}.png")

    def mark_after(self, pattern: str, timeout: float = SCREEN_TIMEOUT_S) -> None:
        """Wait for a line matching `pattern`; later waits look only after it."""
        self.log.wait_for(pattern, timeout)
        regex = re.compile(pattern)
        lines = self.log.lines()
        for index in range(len(lines) - 1, self.log.mark - 1, -1):
            if regex.search(lines[index]):
                self.log.mark = index + 1
                return

    # -- checking -----------------------------------------------------------

    def expect(self, screen: str, timeout: float = SCREEN_TIMEOUT_S) -> None:
        self.log.wait_for(rf"DIAG shown {re.escape(screen)}\s*$", timeout)
        time.sleep(SETTLE_S)

    def opened_pin_screen(self) -> None:
        """A new PIN screen starts with its selector at 0."""
        self.selector = 0

    def act(self) -> None:
        """Call before an input whose result `expect` should wait for."""
        self.log.set_mark()

    def shot(self, name: str, expect_code: str | None = None, review: bool = False) -> None:
        """Screenshot as `name` and compare it with its golden image; with
        review, only save it (screens that differ from run to run)."""
        raw = self.sim.capture(name)
        out = self.out / f"{name}.png"
        label = f"{self.sim.device_dir.name}/{name}"
        try:
            screen = shots.extract_screen(raw, self.sim.device_dir, out)
        except shots.ScreenshotError as err:
            print(f"[{label}] {err}")
            self.results[name] = EXIT_DIFFERENT
            return
        if review:
            print(f"[{label}] saved for review only: {out}")
            return
        if expect_code is not None:
            # README 12.4 check 3: the code on the screenshot must decode.
            read = barcodes_on(screen)
            if expect_code not in read:
                print(f"[{label}] the barcode didn't decode to {expect_code!r}: read {read}")
                self.results[name] = EXIT_DIFFERENT
                return
            print(f"[{label}] barcode decodes to {expect_code!r}")
        golden = self.golden / f"{name}.png"
        if self.approve:
            shots.save_png(screen, golden, f"Approved simulator screen '{name}' of {self.sim.device_dir.name}")
            print(f"[{label}] approved as {golden}")
            self.results[name] = EXIT_OK
            return
        self.results[name] = shots.check_golden(screen, label, out, golden)


def flow_import(f: Flow) -> None:
    """The fixture's 100-entry Aegis import through the settings form opens its
    results; its peak memory must stay within the budget."""
    f.expect("results", IMPORT_TIMEOUT_S)
    stats = f.log.latest(r"DIAG import entries=(\d+) wall_ms=(\d+) mem_peak=(\d+) mem_total=(\d+)")
    if stats is None:
        raise FlowError("the import didn't report its statistics")
    entries, wall_ms, peak, total = (int(v) for v in stats.groups())
    fraction = peak / total
    print(f"[{f.sim.device_dir.name}] import of {entries} entries: {wall_ms} ms, peak memory {fraction:.1%}")
    if fraction > IMPORT_MEMORY_BUDGET:
        raise FlowError(f"import peak memory {fraction:.1%} is over {IMPORT_MEMORY_BUDGET:.0%}")
    f.shot("import_results")
    f.act()
    f.back()
    f.expect("list")


def flow_forms(f: Flow) -> None:
    """The entries the Add code and Add OTP forms added end the list: the
    list wraps, so UP from the first row reaches them."""
    f.prev()  # the Code 128 card (added last)
    f.act()
    f.open_focused_row()
    f.expect("code")
    f.shot("code_form", expect_code=LOCKER_CODE128)
    f.act()
    f.back()
    f.expect("list")
    f.prev()  # the TOTP from Add OTP
    f.act()
    f.open_focused_row()
    f.expect("otp")
    f.shot("otp_form")
    f.act()
    f.back()
    f.expect("list")
    f.next()
    f.next()  # back to the first row (GitHub)
    time.sleep(SETTLE_S)


def flow_entries(f: Flow) -> None:
    """The list, a TOTP entry, a HOTP entry and its next code."""
    f.shot("list")
    f.act()
    f.open_focused_row()  # GitHub, TOTP
    f.expect("otp")
    f.shot("otp")
    f.act()
    f.back()
    f.expect("list")
    for _ in range(3):  # Google, Bank, VPN
        f.next()
    f.act()
    f.open_focused_row()  # VPN, HOTP
    f.expect("otp_hotp")
    f.shot("otp_hotp")
    f.act()
    f.select()  # next HOTP code
    f.expect("otp_hotp")
    f.shot("otp_hotp_next")
    f.act()
    f.back()
    f.expect("list")


def flow_locked_entry(f: Flow) -> None:
    """Bank is locked: PIN screen, a wrong PIN, the right PIN, then auto-lock."""
    f.prev()  # from VPN back to Bank
    f.act()
    f.open_focused_row()
    f.expect("pin")
    f.opened_pin_screen()
    f.shot("pin")
    f.act()
    f.enter_pin(WRONG_PIN)
    f.expect("pin_error")
    f.shot("pin_error")
    f.act()
    f.enter_pin(FIXTURE_PIN)
    f.expect("otp")
    f.shot("otp_locked")
    # Auto-lock closes the locked entry.
    f.act()
    f.expect("list", AUTO_LOCK_TIMEOUT_S)


def flow_swipes(f: Flow, screen: str) -> None:
    """On touch screens a vertical swipe scrolls `screen` once, whatever the
    watch does with page behaviour; checked through the debug logs so no
    image is involved. A swipe up must report one scroll; on the list the
    focus must move and a swipe back must return it."""
    if not f.touch:
        return
    pattern = r"DIAG list_move focus=(\d+)" if screen == "list" else r"DIAG info_scroll first=(\d+)"
    regex = re.compile(pattern)
    x, top, bottom = f._w() // 2, f._h() // 4, f._h() * 3 // 4
    seen: list[str] = []
    for start, end in ((bottom, top), (top, bottom)):  # swipe up: next; down: previous
        f.act()
        f.sim.swipe(x, start, x, end)
        f.log.wait_for(pattern, SCREEN_TIMEOUT_S)
        time.sleep(SETTLE_S)
        found = [m.group(1) for line in f.log.lines()[f.log.mark :] if (m := regex.search(line))]
        if len(found) != 1:
            raise FlowError(f"one swipe on {screen} scrolled {len(found)} times, not once")
        seen.append(found[0])
    if screen == "list" and seen[0] == seen[1]:
        raise FlowError("a swipe on the list didn't move the focus")


def flow_menu(f: Flow) -> None:
    """The menu, and About on watches whose menu has a focus to move."""
    flow_swipes(f, "list")
    f.menu()
    time.sleep(MENU_SETTLE_S)
    f.shot("menu")
    f.act()
    if f.sim.has("down"):
        # Locked with a PIN and a phone change waiting: Phone changes,
        # Change PIN, Reset PIN, Settings results, Entry IDs, Clock status,
        # About.
        for _ in range(6):
            f.next()
        time.sleep(MENU_SETTLE_S)
        f.act()
        f.select()
        f.expect("about")
        f.shot("about")
        flow_swipes(f, "about")
        f.act()
    # Touch-only watches show the menu as a freely scrolling list without
    # focus, so there is no reliable "select About"; just close it.
    f.back()
    f.expect("list")


def flow_queue(f: Flow) -> None:
    """The Manage form's edit of the locked Bank entry waits for the PIN:
    menu, Phone changes, Confirm all, PIN, results. Touch-only
    menus have no focus to select with, so those watches skip it."""
    if not f.sim.has("down"):
        return
    f.menu()
    time.sleep(MENU_SETTLE_S)
    f.select()  # Phone changes (the first item while one waits)
    time.sleep(MENU_SETTLE_S)
    f.shot("queue")
    f.act()
    f.select()  # Confirm all
    f.expect("pin")
    f.opened_pin_screen()
    f.act()
    f.enter_pin(FIXTURE_PIN)
    f.expect("results")
    f.shot("queue_results")
    f.act()
    f.back()
    f.expect("list")
    # The PIN unlocked the session; the lockout flow needs it locked again.
    time.sleep(AUTO_LOCK_IDLE_S)


def flow_lockout(f: Flow) -> None:
    """Five wrong PINs start the 30 s wait."""
    f.act()
    f.open_focused_row()
    f.expect("pin")
    f.opened_pin_screen()
    for attempt in range(1, FAILURES_UNTIL_WAIT + 1):
        f.act()
        f.enter_pin(WRONG_PIN)
        f.expect("pin_wait" if attempt == FAILURES_UNTIL_WAIT else "pin_error")
    f.shot("pin_wait")


def flow_code(f: Flow) -> None:
    """The EAN-13 gym card (index 6), the QR (7), Data Matrix (8), PDF417 (9)
    and Aztec (10) cards: their screenshots must decode."""
    f.act()
    f.back()  # leave the waiting PIN screen
    f.expect("list")
    for _ in range(4):  # Bank (2) to the gym card (6)
        f.next()
    f.act()
    f.open_focused_row()
    f.expect("code")
    f.shot("code", expect_code=GYM_CARD)
    # The QR card (index 7) is encoded on the watch as a job, then shown.
    f.act()
    f.back()
    f.expect("list")
    f.next()
    f.act()
    f.open_focused_row()
    f.expect("code", QR_TIMEOUT_S)
    encoded = f.log.latest(r"DIAG matrix_encoded type=\d+ size=(\S+) ms=(\d+)")
    if encoded is not None:
        print(f"[{f.sim.device_dir.name}] QR {encoded.group(1)} encoded in {encoded.group(2)} ms")
    f.shot("code_qr", expect_code=MEMBER_QR)
    # Data Matrix (8), PDF417 (9) and Aztec (10), also encoded as jobs.
    for name, expected in (("code_datamatrix", PARCEL_DM), ("code_pdf417", TICKET_PDF417), ("code_aztec", RAIL_AZTEC)):
        f.act()
        f.back()
        f.expect("list")
        f.next()
        f.act()
        f.open_focused_row()
        f.expect("code", QR_TIMEOUT_S)
        f.shot(name, expect_code=expected)


def flow_details(f: Flow) -> None:
    """From the Aztec card (index 10): the Steam (4) and Library (5) entries."""
    f.act()
    f.back()
    f.expect("list")
    for _ in range(6):
        f.prev()
    for name in ("otp_steam", "otp_library"):
        f.act()
        f.open_focused_row()
        f.expect("otp")
        f.shot(name)
        f.act()
        f.back()
        f.expect("list")
        f.next()


def flow_info_screens(f: Flow) -> None:
    """The text screens the menu opens, each scrolled with a swipe on touch
    screens; the entry IDs (110 entries) also one page down. Touch-only
    watches get About here (the others in flow_menu). The benchmark's
    timings vary, so its screens are kept for review only."""
    screens = [
        ("Settings results", "results", "menu_results"),
        ("Entry IDs", "ids", "entry_ids"),
        ("Clock status", "clock", "clock_status"),
    ]
    if not f.sim.has("down"):
        screens.append(("About", "about", "about"))
    for label, screen, name in screens:
        f.open_menu_item(label)
        f.expect(screen)
        f.shot(name)
        flow_swipes(f, screen)
        if screen == "ids":
            f.act()
            f.next()
            f.log.wait_for(r"DIAG info_scroll first=[1-9]", SCREEN_TIMEOUT_S)
            time.sleep(SETTLE_S)
            f.shot("entry_ids_next")
        f.act()
        f.back()
        f.expect("list")
    f.open_menu_item("Debug: benchmark")
    f.expect("bench_running")
    f.shot("bench_running", review=True)
    f.expect("bench", BENCH_TIMEOUT_S)
    f.shot("bench", review=True)
    f.act()
    f.back()
    f.expect("list")


def run_flow(f: Flow) -> None:
    """The scripted session; see the module docstring."""
    flow_import(f)
    flow_forms(f)
    flow_entries(f)
    flow_locked_entry(f)
    flow_menu(f)
    flow_queue(f)
    flow_lockout(f)
    flow_code(f)
    flow_details(f)
    flow_info_screens(f)


def open_row(f: Flow, moves: int, screen: str, timeout: float = SCREEN_TIMEOUT_S) -> None:
    """Move the list focus by `moves` rows (negative: up) and open that row."""
    for _ in range(abs(moves)):
        if moves > 0:
            f.next()
        else:
            f.prev()
    f.act()
    f.open_focused_row()
    f.expect(screen, timeout)


def close_to_list(f: Flow) -> None:
    f.act()
    f.back()
    f.expect("list")


def extra_codes(f: Flow) -> None:
    """From GitHub (0): UPC-A (2) and Code 39 (3) must decode, the EAN-13
    with a wrong check digit (4) shows the code error, and the large QR (5)
    shows its encoding progress, which must start low and only go up."""
    open_row(f, 2, "code")
    f.shot("code_upca", expect_code=SHOP_UPCA)
    close_to_list(f)
    open_row(f, 1, "code")
    f.shot("code_39", expect_code=BADGE_CODE39)
    close_to_list(f)
    open_row(f, 1, "code_error")
    f.shot("code_error")
    close_to_list(f)
    f.next()
    f.act()
    f.open_focused_row()
    f.log.wait_for(r"DIAG code_progress percent=", SCREEN_TIMEOUT_S)
    f.shot("code_progress", review=True)
    f.expect("code", QR_TIMEOUT_S)
    check_progress(f)
    f.shot("code_qr_large", expect_code=BIG_QR_TEXT)
    close_to_list(f)


def check_progress(f: Flow) -> None:
    """The encoding progress since the last mark starts low, never goes
    back, and moves through several values."""
    regex = re.compile(r"DIAG code_progress percent=(\d+)")
    seen = [int(m.group(1)) for line in f.log.lines()[f.log.mark :] if (m := regex.search(line))]
    steps = sorted(set(seen))
    print(f"[{f.sim.device_dir.name}] encoding progress: {' '.join(str(p) for p in steps)} %")
    if not seen or seen[0] >= PROGRESS_FIRST_MAX:
        raise FlowError(f"the encoding progress started at {seen[:1]}, not below {PROGRESS_FIRST_MAX} %")
    if any(b < a for a, b in itertools.pairwise(seen)):
        raise FlowError(f"the encoding progress went back: {seen}")
    if len(steps) < PROGRESS_MIN_STEPS:
        raise FlowError(f"the encoding progress showed only {steps}")


def extra_legacy_pin(f: Flow) -> None:
    """From the large QR (5): Bank (1) asks for the legacy 6-digit PIN with
    OK; it unlocks, applies the waiting "Require PIN", then asks for a new
    4-digit PIN: a repeat that differs, then the same PIN twice opens Bank."""
    open_row(f, -4, "pin")
    f.opened_pin_screen()
    f.shot("pin_legacy")
    f.act()
    f.enter_pin(LEGACY_PIN[:-1], legacy=True, ok=False)
    f.expect_pin(MODE_UNLOCK, digits=len(LEGACY_PIN) - 1)
    f.shot("pin_legacy_typed")
    f.act()
    f.enter_pin(LEGACY_PIN[-1], legacy=True)
    f.log.wait_for(r"DIAG locks_applied count=1\b", SCREEN_TIMEOUT_S)
    f.expect_pin(MODE_CHANGE)
    f.shot("pin_new_forced")
    f.act()
    f.enter_pin(NEW_PIN)
    f.expect_pin(MODE_CHANGE, repeat=True)
    f.shot("pin_repeat")
    f.act()
    f.enter_pin(OTHER_PIN)
    f.expect("pin_error")
    f.shot("pin_differ")
    f.act()
    f.enter_pin(NEW_PIN)
    f.expect_pin(MODE_CHANGE, repeat=True)
    f.act()
    f.enter_pin(NEW_PIN)
    f.expect("otp")
    close_to_list(f)
    f.shot("list_locked")


def extra_auto_lock_code(f: Flow) -> None:
    """The Shop card (2) now requires the PIN; the session is unlocked, so
    it opens directly, and auto-lock closes it back to the list."""
    open_row(f, 1, "code")
    f.act()
    f.expect("list", AUTO_LOCK_TIMEOUT_S)


def extra_change_pin(f: Flow) -> None:
    """Locked: Change PIN asks for the current PIN, then a new one twice.
    After "Lock now", the new PIN opens Bank (1)."""
    f.open_menu_item("Change PIN")
    f.expect_pin(MODE_UNLOCK)
    f.opened_pin_screen()
    f.act()
    f.enter_pin(NEW_PIN)
    f.expect_pin(MODE_CHANGE)
    f.opened_pin_screen()
    f.shot("pin_change")
    f.act()
    f.enter_pin(CHANGED_PIN)
    f.expect_pin(MODE_CHANGE, repeat=True)
    f.act()
    f.enter_pin(CHANGED_PIN)
    f.expect("list")
    f.open_menu_item("Lock now")  # changing the PIN left the watch unlocked
    f.expect("list")
    open_row(f, -1, "pin")
    f.opened_pin_screen()
    f.act()
    f.enter_pin(CHANGED_PIN)
    f.expect("otp")
    close_to_list(f)


def extra_reset_pin(f: Flow) -> None:
    """Reset PIN: the confirmation, then creating a new PIN (entered twice)."""
    f.open_menu_item("Reset PIN")
    time.sleep(MENU_SETTLE_S)
    f.shot("reset_confirm")
    f.act()
    confirm(f)
    f.expect_pin(MODE_CREATE)
    f.opened_pin_screen()
    f.shot("pin_create")
    f.act()
    f.enter_pin(CREATED_PIN)
    f.expect_pin(MODE_CREATE, repeat=True)
    f.act()
    f.enter_pin(CREATED_PIN)
    f.expect("list")
    f.shot("list_reset")


def confirm(f: Flow) -> None:
    """Answer yes in a native confirmation: UP to "Yes" and START with
    buttons, a tap on the right half ("Confirm") on touch-only watches."""
    if f.sim.has("up"):
        f.sim.press("up")
        time.sleep(MENU_SETTLE_S)
        f.sim.press("enter")
    else:
        f.sim.tap(f._w() * 3 // 4, f._h() * 5 // 8)


def run_extra_flow(f: Flow) -> None:
    """The extra scenario's session; see the module docstring."""
    f.expect("results", IMPORT_TIMEOUT_S)  # the Manage form's "Require PIN" waits for the PIN
    f.shot("lock_queued")
    close_to_list(f)
    f.shot("list_extra")
    extra_codes(f)
    extra_legacy_pin(f)
    extra_auto_lock_code(f)
    extra_change_pin(f)
    extra_reset_pin(f)


def find_key(data: object, key: str) -> object:
    """The first value stored under `key` anywhere in a JSON structure, or None."""
    if isinstance(data, dict):
        if key in data:
            return data[key]
        data = list(data.values())
    if isinstance(data, list):
        for item in data:
            found = find_key(item, key)
            if found is not None:
                return found
    return None


def open_from_glance(window: str, device_dir: Path, raw_dir: Path) -> None:
    """Open the full app from its glance, as found in the simulator: a tap on
    it on touch screens; on button-only watches DOWN (which selects the
    glance) and then START. START alone does nothing in either case."""
    sim = Simulator(window, device_dir, raw_dir)
    config = json.loads((device_dir / "simulator.json").read_text(encoding="utf-8"))
    area = find_key(config.get("glance", {}), "contentArea")
    if find_key(config, "isTouch") is True and isinstance(area, dict):
        sim.tap(int(area["x"]) + int(area["width"]) // 2, int(area["y"]) + int(area["height"]) // 2)
    else:
        sim.press("down")
        time.sleep(1.0)
        sim.press("enter")


def decode_screen(window: str, device_dir: Path, out: Path, text: str) -> int:
    """Screenshot the simulator; EXIT_OK when a barcode on it reads `text`."""
    name = device_dir.name
    try:
        sim = Simulator(window, device_dir, out / "raw")
        screen = shots.extract_screen(sim.capture("decode"), device_dir, out / "decode.png")
    except (FlowError, shots.ScreenshotError, subprocess.CalledProcessError, OSError) as err:
        print(f"[{name}] no screenshot to decode: {err}")
        return EXIT_FLOW
    read = barcodes_on(screen)
    if text not in read:
        print(f"[{name}] the barcode on the screen didn't decode to {text!r}: read {read}")
        return EXIT_DIFFERENT
    print(f"[{name}] the barcode on the screen decodes to {text!r}")
    return EXIT_OK


def open_app(window: str, device_dir: Path, out: Path) -> int:
    """--open-from-glance: EXIT_OK once the input was sent."""
    try:
        open_from_glance(window, device_dir, out / "raw")
    except (FlowError, subprocess.CalledProcessError, OSError) as err:
        print(f"[{device_dir.name}] could not open the app from its glance: {err}")
        return EXIT_FLOW
    return EXIT_OK


def single_shot(args: argparse.Namespace) -> int:
    """Screenshot the simulator as args.shot and compare it with its golden image (or approve it)."""
    window, device_dir, out, golden, name = args.window, args.device_dir, args.out, args.golden, args.shot
    label = f"{device_dir.name}/{name}"
    try:
        sim = Simulator(window, device_dir, out / "raw")
        screen = shots.extract_screen(sim.capture(name), device_dir, out / f"{name}.png")
    except (FlowError, shots.ScreenshotError, subprocess.CalledProcessError, OSError) as err:
        print(f"[{label}] {err}")
        return EXIT_FLOW
    if args.approve:
        shots.save_png(screen, golden / f"{name}.png", f"Approved simulator screen '{name}' of {device_dir.name}")
        print(f"[{label}] approved")
        return EXIT_OK
    return shots.check_golden(screen, label, out / f"{name}.png", golden / f"{name}.png")


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Scripted UI flow with screenshot checks.")
    parser.add_argument("--device-dir", type=Path, required=True)
    parser.add_argument("--window", required=True, help="X window id of the simulator")
    parser.add_argument("--out", type=Path, required=True, help="directory for the screenshots")
    parser.add_argument("--log", type=Path, help="the app's console log (monkeydo output)")
    parser.add_argument("--golden", type=Path, help="directory of the approved images")
    parser.add_argument("--approve", action="store_true")
    parser.add_argument("--scenario", choices=["main", "extra"], default="main", help="which UI-test build runs")
    parser.add_argument("--open-from-glance", action="store_true", help="only open the app from its glance")
    parser.add_argument("--decode", metavar="TEXT", help="only check that the screen shows a barcode reading TEXT")
    parser.add_argument("--shot", metavar="NAME", help="only screenshot the current screen as NAME (needs --golden)")
    args = parser.parse_args(argv)
    name = args.device_dir.name
    if args.decode is not None:
        return decode_screen(args.window, args.device_dir, args.out, args.decode)
    if args.shot is not None:
        if args.golden is None:
            parser.error("--shot needs --golden")
        return single_shot(args)
    if args.open_from_glance:
        return open_app(args.window, args.device_dir, args.out)
    if args.log is None or args.golden is None:
        parser.error("--log and --golden are required for the flow")
    try:
        sim = Simulator(args.window, args.device_dir, args.out / "raw")
        flow = Flow(sim, AppLog(args.log), args.out, args.golden, args.approve)
        (run_extra_flow if args.scenario == "extra" else run_flow)(flow)
    except (FlowError, subprocess.CalledProcessError, OSError) as err:
        print(f"[{name}] UI flow failed: {err}")
        return EXIT_FLOW
    failed = sorted(screen for screen, code in flow.results.items() if code != EXIT_OK)
    if failed:
        print(f"[{name}] {len(failed)} of {len(flow.results)} screens need attention: {', '.join(failed)}")
        return EXIT_DIFFERENT
    print(f"[{name}] all {len(flow.results)} screens match")
    return EXIT_OK


if __name__ == "__main__":
    sys.exit(main())