gallery/tools/fr_capture.py

#!/usr/bin/env python3
"""
fr_capture.py — end-to-end 120 Hz scroll-jank capture for FR-line apps
(egtest / sltest family).

The jank_eval/jank_capture pair needs Slint's vendored PRES/VSYNC lines;
FR-line harnesses (sltest, egtest) log app-side frame gaps instead. This
tool runs the same disciplined capture pipeline and the same referee that
produced the sltest reference numbers (drops = present gaps > 1.8x the
median, within one process's log):

  fr_capture.py capture --app wow.gmp.egtest --flicks 5 [--tag egui031]

  1. device + build identity (apk sha256, versionName)      -> PROVENANCE
  2. force-stop, DELETE fr.log            (one process per file)
  3. cold launch, verify ResumedActivity
  4. scripted flicks: N fast swipes (350 ms, 1600 px) with 2 s settles,
     one slow drag (1200 ms) in the middle
  5. verify still ResumedActivity, force-stop
  6. pull the complete log (exactly one process)
  7. report: provenance + gap verdict + per-drop context
     (t, gap, neighbours — isolated vs chain) + render-cost stats

Exit: 0 = no drops, 1 = drops found / invalid capture.
fr_capture.py report FILE.log          re-analyze a pulled log
fr_capture.py --selftest               offline validation

Note (honesty): the gap clock is the APP's clock (swap-completion to
swap-completion), not the display slot grid. That is the same referee
used for all sltest A/B numbers, which keeps egtest-vs-sltest
apples-to-apples; jank_eval's slot grid remains the stricter referee
for builds that emit PRES/VSYNC.
"""
import argparse, hashlib, os, re, subprocess, sys, time

ADB_SERIAL = os.environ.get("ADB_SERIAL", "")
FR_RE = re.compile(r"^FR\s+([\d.]+)\s+gap=\s*([\d.]+)\s+cost=\s*([\d.]+)\s+drew=(\d)")

def adb(*args, timeout=120):
    cmd = ["adb"]
    if ADB_SERIAL: cmd += ["-s", ADB_SERIAL]
    cmd += list(args)
    return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)

def adb_shell(cmd, timeout=120):
    return adb("shell", cmd, timeout=timeout)

# ------------------------------------------------------------------ analysis
def analyze(lines, label="log"):
    frames = []
    for ln in lines:
        m = FR_RE.match(ln.strip())
        if m:
            frames.append((float(m.group(1)), float(m.group(2)),
                           float(m.group(3)), int(m.group(4))))
    if len(frames) < 30:
        print(f"  {label}: only {len(frames)} FR lines — nothing to judge");
        return None
    # drop the first frame (gap=0) and process-suspend artifacts
    gaps = [(t, g, c) for (t, g, c, d) in frames[1:] if 0 < g < 200]
    if len(gaps) < 20:
        print(f"  {label}: only {len(gaps)} usable gaps — nothing to judge");
        return None
    gs = sorted(g for (_, g, _) in gaps)
    med = gs[len(gs) // 2]
    p90 = gs[int(len(gs) * 0.90)]
    p99 = gs[int(len(gs) * 0.99)]
    mx = gs[-1]
    thr = 1.8 * med
    drops = [(t, g) for (t, g, _) in gaps if g > thr]
    cs = sorted(c for (_, _, c) in gaps)
    return {
        "n": len(gaps), "med": med, "p90": p90, "p99": p99, "max": mx,
        "thr": thr, "drops": drops,
        "fps": 1000.0 / med,
        "cost_med": cs[len(cs) // 2], "cost_p99": cs[int(len(cs) * 0.99)],
        "cost_max": cs[-1],
        "frames": frames,
    }

def print_report(res, title=""):
    if res is None:
        return 1
    drops = res["drops"]
    pct = 100.0 * len(drops) / res["n"]
    print(f"\n=== JANK (gap referee, {title}) ===")
    print(f"frames={res['n']}  median gap={res['med']:.2f} ms (~{res['fps']:.0f} fps)")
    print(f"p90={res['p90']:.2f}  p99={res['p99']:.2f}  max={res['max']:.2f}  "
          f"threshold={res['thr']:.2f} ms (1.8x med)")
    print(f"render cost: med={res['cost_med']:.2f} ms  p99={res['cost_p99']:.2f}  "
          f"max={res['cost_max']:.2f}")
    print(f"DROPS: {len(drops)} of {res['n']}  ({pct:.2f}%)")
    if drops:
        print(f"{'t(ms)':>10} {'gap(ms)':>9}  context (prev, next gap)")
        shown = 0
        for i, (t, g, c, d) in enumerate(res["frames"][1:], start=1):
            if g > res["thr"]:
                p = res["frames"][i - 1][1] if i >= 1 else 0.0
                n = res["frames"][i + 1][1] if i + 1 < len(res["frames"]) else 0.0
                kind = "chain" if (p > res["thr"] or n > res["thr"]) else "isolated"
                print(f"{t:10.1f} {g:9.2f}  prev={p:7.2f} next={n:7.2f}  {kind}")
                shown += 1
                if shown >= 25:
                    print(f"... {len(drops) - 25} more")
                    break
    return 0 if not drops else 1

# -------------------------------------------------------------------- capture
def capture(app, flicks, tag="", manual=0):
    ok = True
    r = adb_shell("getprop ro.boot.cpu; getprop ro.product.model")
    print("=== CAPTURE ===")
    print(f"device: serial={ADB_SERIAL or '(default)'} model={r.stdout.strip()!r}")

    # build identity
    path = adb_shell(f"pm path {app}").stdout.strip().replace("package:", "").strip()
    sha = adb_shell(f"sha256sum {path}").stdout.split()[0] if path else "?"
    ver = adb_shell(f"dumpsys package {app} | grep versionName | head -1").stdout.strip()
    print(f"app: {app}  {ver}  apk sha256={sha[:16]}…  tag={tag or '-'}")
    if not path:
        print("INVALID: package not installed"); return 1

    # Three log conventions: egtest/harness apps write
    # /sdcard/Android/data/<app>/files/fr.log; tttest (Tauri) writes
    # .../files/Documents/fr.log (Tauri's document_dir); the gallery
    # mirrors its FR lines to /sdcard/Download/.gallery_perf.log.
    log = f"/sdcard/Android/data/{app}/files/fr.log"
    alts = [
        f"/sdcard/Android/data/{app}/files/Documents/fr.log",
        "/sdcard/Download/.gallery_perf.log",
    ]

    adb_shell(f"am force-stop {app}"); time.sleep(1.5)
    adb_shell(f"rm -f {log} " + " ".join(alts)); time.sleep(0.5)
    # NativeActivity apps (slint/egui harnesses) vs real activities
    # (Tauri → .MainActivity): try the specific entry, fall back to the
    # package's default LAUNCHER activity via monkey.
    adb_shell(f"am start -n {app}/android.app.NativeActivity")
    time.sleep(3)
    r = adb_shell("dumpsys activity activities | grep ResumedActivity")
    if app not in r.stdout:
        adb_shell(f"monkey -p {app} -c android.intent.category.LAUNCHER 1")
        time.sleep(3)
        r = adb_shell("dumpsys activity activities | grep ResumedActivity")
    if app not in r.stdout:
        print("INVALID: app not ResumedActivity after cold launch"); return 1
    print("launched (ResumedActivity verified)")

    if manual > 0:
        # THE USER'S FINGER is the input: no scripted swipes. Just watch
        # them scroll for `manual` seconds and measure what they made
        # the renderer do. (adb `input swipe` injects at ~20 Hz —
        # bursty, not representative of a real touch stream.)
        print(f"!!! MANUAL: scroll the app normally for {manual}s — GO !!!")
        for _ in range(manual):
            time.sleep(1)
            sys.stderr.write(".")
        print("")
        time.sleep(2.0)
    else:
        # scripted drags
        H = 2856
        flick = f"input swipe 640 {int(H*0.77)} 640 {int(H*0.21)} 350"
        slow = f"input swipe 640 {int(H*0.77)} 640 {int(H*0.21)} 1200"
        for i in range(flicks):
            if i == flicks // 2:
                adb_shell(slow); time.sleep(3.0)
            else:
                adb_shell(flick); time.sleep(2.0)
            if i % 2 == 0:  # flip direction to cover up-scrolls too
                adb_shell("input swipe 640 600 640 2200 350"); time.sleep(2.0)
        time.sleep(2.0)

    r = adb_shell("dumpsys activity activities | grep -E 'ResumedActivity|topResumedActivity'")
    adb_shell(f"am force-stop {app}"); time.sleep(1.0)
    if app not in r.stdout:
        print("WARNING: app not resumed right before stop (backgrounded?)")

    # Pick whichever log the app actually wrote.
    for alt in alts:
        if adb_shell(f"[ -f {alt} ] && [ $(wc -c < {alt}) -gt 0 ] && echo y").stdout.strip() == "y":
            log = alt
            break
    dest = os.path.join(os.path.dirname(os.path.abspath(__file__)), f".fr_{app.replace('.','_')}.log")
    r = adb("pull", log, dest, timeout=300)
    if r.returncode != 0:
        print(f"INVALID: log pull failed: {r.stderr.strip()}"); return 1
    with open(dest) as f:
        lines = f.readlines()
    print(f"pulled {len(lines)} lines -> {dest}")
    res = analyze(lines, label="capture")
    rc = print_report(res, title=tag or app)
    return rc

# -------------------------------------------------------------------- selftest
def selftest():
    # synthetic: 1200 frames at 8.33 ms, 4 injected drops (two chains)
    lines, t, prev = [], 0.0, None
    import random; random.seed(7)
    for i in range(1200):
        gap = 8.33 + random.uniform(-0.15, 0.15)
        if i in (300, 301, 700, 1100):
            gap = 25.0
        t += gap
        lines.append(f"FR {t:11.2f} gap={gap:7.2f} cost={2.0:6.2f} drew=1")
    lines.insert(0, "FR        0.00 gap=   0.00 cost=180.00 drew=1")
    res = analyze(lines)
    assert res is not None
    assert res["n"] == 1200, res["n"]
    assert len(res["drops"]) == 4, res["drops"]
    assert abs(res["med"] - 8.33) < 0.05
    # no-drop case
    lines2, t = [], 0.0
    for i in range(1200):
        gap = 8.33 + random.uniform(-0.1, 0.1)
        t += gap
        lines2.append(f"FR {t:11.2f} gap={gap:7.2f} cost=2.00 drew=1")
    res2 = analyze(lines2)
    assert len(res2["drops"]) == 0
    # too-small case
    assert analyze(["FR 0 0 0 1"] * 5) is None
    print("selftest OK (drop detection, clean case, short-log case)")
    return 0

def main():
    ap = argparse.ArgumentParser()
    sub = ap.add_subparsers(dest="cmd")
    c = sub.add_parser("capture")
    c.add_argument("--app", required=True)
    c.add_argument("--flicks", type=int, default=6)
    c.add_argument("--tag", default="")
    c.add_argument("--manual", type=int, default=0,
                   help="seconds to wait for the USER to scroll (no scripted swipes)")
    r = sub.add_parser("report")
    r.add_argument("file")
    ap.add_argument("--selftest", action="store_true")
    a = ap.parse_args()
    if a.selftest:
        return selftest()
    if a.cmd == "capture":
        return capture(a.app, a.flicks, a.tag, manual=a.manual)
    if a.cmd == "report":
        with open(a.file) as f:
            return print_report(analyze(f.readlines()), title=a.file)
    ap.print_help()
    return 2

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