gallery/tools/record_check.py

#!/usr/bin/env python3
"""Screen-recording jank verifier — the DISPLAY-side referee.

jank_capture.py judges the APP's self-report (present timestamps in its own
log). This tool records the ACTUAL display output with
`adb shell screenrecord` and measures the inter-frame gaps in the recording:

  * a clean 120 Hz scroll -> gaps ~8.3 ms, no gap > 1.25 x median
  * one dropped display frame during a drag -> one ~16.6 ms gap
  * at rest the app does not repaint, so gaps there are EXPECTED and are
    not judged (only the active drag windows are)

It answers the question an app log cannot: "did the display actually show
every frame?" — independent of any app-side instrumentation.

CAVEATS (printed in the report):
  * the screenrecord encoder itself can drop frames under extreme load
    (rare at 30 Mbps); a control run on a known-smooth app (e.g. Fossify)
    calibrates the floor.
  * screenrecord captures at the display refresh rate; it sees what SF
    composited, i.e. what was on glass.
  * swipes are scripted `adb input` (mid-screen) unless --finger is used
    (then the user drags by hand during the recording window).

usage:
  record_check.py record --app wow.gmp.sltest [--seconds 12] [--drags 4,2,4]
                         [--finger] [--outdir /tmp/jank] [--tag X]
  record_check.py analyze FILE.mp4 [--drags 4,2,4] [--active S]
  record_check.py --selftest

exit: 0 = CLEAN (no dropped frames during active windows),
      1 = JANK (>=1 clear drop) / error,  2 = SUSPECT (only 1.25-1.8x gaps)
"""
import argparse, glob, os, re, shutil, subprocess, sys, time

ADB_SERIAL = os.environ.get("ADB_SERIAL", os.environ.get("ANDROID_SERIAL", ""))
REC_PATH = "/sdcard/jankrec.mp4"


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 ffprobe_frames(mp4):
    """Return [(pts_ms, ...)] for every video frame, in decode order."""
    r = subprocess.run(
        ["ffprobe", "-v", "error", "-select_streams", "v:0",
         "-show_entries", "frame=pts_time", "-of", "csv=p=0", mp4],
        capture_output=True, text=True, timeout=120)
    if r.returncode != 0:
        sys.exit(f"ffprobe failed: {r.stderr.strip()[:200]}")
    pts = []
    for line in r.stdout.splitlines():
        line = line.strip().split(",")[0].strip()
        if line and line != "N/A":
            try:
                pts.append(float(line) * 1000.0)
            except ValueError:
                pass
    return pts


def analyze_gaps(pts, active_windows, label="", mp4=None):
    """Pure gap analysis. Returns (verdict_code, stats) — 0 clean, 2 suspect,
    1 jank. Shared by the mp4 path and the self-test.

    IMPORTANT: the median (the refresh-period reference) is computed from
    ACTIVE-WINDOW gaps only. Rest periods produce idle frames at irregular
    intervals (20-100 ms) that would otherwise pollute the median and create
    false 'drops'."""
    gaps = [pts[i + 1] - pts[i] for i in range(len(pts) - 1)]

    def in_active(t):
        return any(a <= t <= b for a, b in active_windows)

    active_gaps = [g for i, g in enumerate(gaps) if in_active(pts[i])]
    if len(active_gaps) < 30:
        active_gaps = gaps  # not enough active data — fall back to all
    gaps_s = sorted(active_gaps)
    med = gaps_s[len(gaps_s) // 2]
    p90 = gaps_s[int(len(gaps_s) * 0.9)]
    mx = gaps_s[-1]
    print(f"\n[RECORD ANALYSIS] {label or (os.path.basename(mp4) if mp4 else 'pts')}")
    print(f"  frames={len(pts)} span={pts[-1] - pts[0]:.0f} ms")
    print(f"  active-window gap median={med:.2f} ms  p90={p90:.2f} ms  max={mx:.1f} ms")
    inferred_hz = 1000.0 / med
    print(f"  (implied refresh ≈ {inferred_hz:.0f} Hz — gaps are judged RELATIVE to the median)")

    susp, drops = [], []
    for i, g in enumerate(gaps):
        t = pts[i]
        if not in_active(t):
            continue  # rest periods legitimately have long gaps
        if g > 1.8 * med:
            drops.append((t, g))
        elif g > 1.25 * med:
            susp.append((t, g))
    for t, g in drops[:20]:
        print(f"  DROP  at t={t:.0f} ms: gap {g:.1f} ms  ({g / med:.2f}x)")
    for t, g in susp[:20]:
        print(f"  SUSP  at t={t:.0f} ms: gap {g:.1f} ms  ({g / med:.2f}x)")
    print(f"\n  active-window gaps >1.8x med: {len(drops)}   (1.25-1.8x: {len(susp)})")
    print("  note: the encoder itself can drop frames under extreme load;")
    print("        a control recording of a known-smooth app calibrates the floor.")
    if drops:
        print("  VERDICT: JANK — the display missed frames during active scrolling")
        return 1
    if susp:
        print("  VERDICT: SUSPECT — no clear drops, but near-miss gaps in the active window")
        return 2
    print("  VERDICT: CLEAN — every frame was displayed during the active windows")
    return 0


def analyze(mp4, active_windows, label=""):
    pts = ffprobe_frames(mp4)
    if len(pts) < 20:
        sys.exit(f"only {len(pts)} frames in {mp4} — recording too short/empty")
    return analyze_gaps(pts, active_windows, label, mp4=mp4)


def active_windows_for_drags(drag_events, rec_start_wall):
    """Map drag events (wall-clock start/end seconds) onto the recording
    timeline (ms). The recording starts `rec_start_wall` in wall time but
    the first frame appears ~0.3-1.5 s later (encoder warm-up), so we pad:
    [start - 1500, end + 500]. Padding errs toward judging MORE (never
    misses a jank frame; at worst judges a few pre-drag idle frames)."""
    wins = []
    for ds, de in drag_events:
        a = (ds - rec_start_wall) * 1000.0 - 1500.0
        b = (de - rec_start_wall) * 1000.0 + 500.0
        wins.append((max(a, 0.0), b))
    return wins


def do_record(app, seconds, profile, outdir, tag, finger):
    os.makedirs(outdir, exist_ok=True)
    r = adb("shell", "getprop", "ro.build.version.release")
    print(f"[0] device {ADB_SERIAL or '(default)'} android={r.stdout.strip() or '?'}")

    # wake + unlock-ish sanity (we cannot unlock a secure screen)
    adb("shell", "input", "keyevent", "KEYCODE_WAKEUP")
    time.sleep(1)

    print("[1] force-stop + launch")
    adb("shell", "am", "force-stop", app)
    adb("shell", "am", "start", "-n", f"{app}/android.app.NativeActivity")
    time.sleep(8)
    r = adb("shell", "pidof", app)
    if not r.stdout.strip():
        sys.exit("app did not start")
    r = adb("shell", "dumpsys", "activity", "activities")
    if f"ResumedActivity" not in r.stdout and "mResumedActivity" not in r.stdout:
        # newer builds print differently; just warn
        print("  warn: could not confirm ResumedActivity — check the screen is on the app")

    # clear old recording
    adb("shell", "rm", "-f", REC_PATH)

    print(f"[2] start screenrecord ({seconds}s, 30 Mbps)")
    t_rec = time.time()
    # background on the device:
    rec_proc = subprocess.Popen(
        ["adb"] + (["-s", ADB_SERIAL] if ADB_SERIAL else []) +
        ["shell", f"screenrecord --time-limit {seconds} --bit-rate 30000000 {REC_PATH}"],
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    time.sleep(2)  # let the recorder settle

    print(f"[3] {'USER: drag by hand now' if finger else 'scripted drags ' + profile}")
    drag_events = []  # (start_wall, end_wall)
    if finger:
        # wait for the user
        try:
            input("   ... press ENTER when the dragging is done ...")
        except EOFError:
            time.sleep(6)
        drag_events.append((t_rec + 2.0, time.time()))
    else:
        parts = [int(x) for x in profile.split(",")]
        i = 0
        while i < len(parts):
            d = parts[i]
            time.sleep(1.0)  # settle before each drag
            # mid-screen drags (edge swipes trigger system gestures)
            s = time.time()
            p = adb("shell", "input", "swipe", "640", "2000", "640", "700", str(d * 1000))
            drag_events.append((s, time.time()))
            i += 1
            if i < len(parts):
                time.sleep(parts[i])
                i += 1
    print(f"    dragging done at +{time.time() - t_rec:.1f}s")
    # wait for the recorder to finish (it runs for `seconds` total)
    rec_proc.wait(timeout=seconds + 30)

    print("[4] pull recording")
    name = f"rec_{tag or int(time.time())}.mp4"
    dest = os.path.join(outdir, name)
    r = adb("pull", REC_PATH, dest, timeout=300)
    if not os.path.exists(dest) or os.path.getsize(dest) < 10000:
        sys.exit("recording pull failed")
    adb("shell", "rm", "-f", REC_PATH)
    print(f"    {dest} ({os.path.getsize(dest)} bytes)")

    wins = active_windows_for_drags(drag_events, t_rec)
    print(f"  active windows (rec ms): {[(int(a), int(b)) for a, b in wins]}")
    return analyze(dest, wins, label=name)


def selftest():
    """Two halves:
    1. PURE gap logic on synthetic pts streams: perfect 120 Hz -> CLEAN;
       one dropped frame in the active window -> JANK; a drop OUTSIDE the
       active window (rest period) -> CLEAN.
    2. End-to-end mp4 path: real ffmpeg 120 fps encode -> ffprobe ->
       analyze() -> CLEAN (verifies the parse pipeline)."""
    import tempfile
    P = 1000.0 / 120
    def stream(n, drop_slots=(), t0=0.0):
        pts = []
        for i in range(n):
            if i in drop_slots:
                continue
            pts.append(t0 + i * P)
        return pts

    active = [(0.0, 10.0 * 1000)]
    r1 = analyze_gaps(stream(600), active, label="st-perfect")
    r2 = analyze_gaps(stream(600, drop_slots={300}), active, label="st-drop-in")
    # drop at t=11s, active window ends at 10s -> must be ignored
    r3 = analyze_gaps(stream(1320, drop_slots={1320 - 2}), [(0.0, 10000.0)], label="st-drop-out")

    mp4 = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False).name
    subprocess.run(
        ["ffmpeg", "-y", "-v", "error", "-f", "lavfi",
         "-i", "testsrc=duration=3:rate=120:size=320x240", mp4], check=True)
    pts = ffprobe_frames(mp4)
    r4 = analyze_gaps(pts, [(0.0, 10.0 * 1000)], label="st-mp4")
    os.unlink(mp4)

    print("self-test:", "PASS" if (r1 == 0 and r2 == 1 and r3 == 0 and r4 == 0)
          else f"FAIL (r1={r1} r2={r2} r3={r3} r4={r4})")
    return 0 if (r1 == 0 and r2 == 1 and r3 == 0 and r4 == 0) else 1


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = ap.add_subparsers(dest="cmd")
    r = sub.add_parser("record", help="record + analyze on device")
    r.add_argument("--app", required=True)
    r.add_argument("--seconds", type=int, default=14)
    r.add_argument("--drags", default="4,2,4", help="drag,rest,dur,rest,dur (seconds)")
    r.add_argument("--outdir", default="/tmp/jank")
    r.add_argument("--tag", default="")
    r.add_argument("--finger", action="store_true",
                   help="user drags by hand; press ENTER when done")
    a = sub.add_parser("analyze", help="analyze an existing mp4")
    a.add_argument("mp4")
    a.add_argument("--active", default="", help="comma list of start,end (ms); default: all")
    ap.add_argument("--selftest", action="store_true")
    args = ap.parse_args()

    if args.selftest:
        sys.exit(selftest())
    if args.cmd == "record":
        sys.exit(do_record(args.app, args.seconds, args.drags, args.outdir, args.tag, args.finger))
    if args.cmd == "analyze":
        if args.active:
            wins = []
            for part in args.active.split(";"):
                s, e = part.split(",")
                wins.append((float(s), float(e)))
        else:
            wins = [(0.0, 1e12)]
        sys.exit(analyze(args.mp4, wins, label=args.mp4))
    ap.print_help()
    sys.exit(1)


if __name__ == "__main__":
    main()