#!/usr/bin/env python3 """STEP-119 / RCRR CPU 对照。只启动已安装测试场景并读取真机CPU/Native计数，不编译/修改24 Gate。""" import argparse import json import statistics import subprocess import sys import time from datetime import datetime from pathlib import Path import verify_step113_fast_path as verifier BENCH_SCENARIOS = ( "RCRR_FULL_CONTROL_NO_TIDE", "RCRR_C_NARROW_NO_TIDE", "RCR_PARTICLE_ONLY_NO_TIDE", "RCRRR_PARTICLE_ONLY_NO_TIDE", ) def parse_proc_stat(raw: str): """Linux /proc/PID/stat: comm may contain whitespace or ')'; indices are post-comm.""" close = raw.rfind(")") if close <= 0 or not raw.startswith(raw.split(" ", 1)[0] + " ("): raise ValueError("invalid /proc/PID/stat header") rest = raw[close + 2:].split() if len(rest) < 20: raise ValueError("truncated /proc/PID/stat") # Fields 14/15 utime/stime and field 22 starttime; rest starts at field 3. return int(rest[11]) + int(rest[12]), int(rest[19]) def collect_result(samples, cpu_start, cpu_end, clock_ticks, elapsed): native_start, native_end = samples fast_delta = native_end["fastRendered"] - native_start["fastRendered"] full_delta = native_end["fullRendered"] - native_start["fullRendered"] interval = (native_end["sampleElapsedMs"] - native_start["sampleElapsedMs"]) / 1000 if interval <= 0 or fast_delta < 0 or full_delta < 0: raise ValueError("invalid Native window/counters") cpu_sec = (cpu_end["ticks"] - cpu_start["ticks"]) / clock_ticks if cpu_sec < 0: raise ValueError("CPU counter reset") fps = (fast_delta + full_delta) / interval cpu_cores = cpu_sec / elapsed return { "cpuSeconds": cpu_sec, "cpuCoreEquivalent": cpu_cores, "cpuPercentOfSingleCore": 100 * cpu_cores, "nativeWindowSeconds": interval, "nativeWindowFullFrames": full_delta, "nativeWindowFastFrames": fast_delta, "nativeObservedFps": fps, "fastFraction": fast_delta / (fast_delta + full_delta) if fast_delta + full_delta else None, "estimatedCpuMsPerNativeFrame": 1000 * cpu_cores / fps if fps else None, "caveat": ("CPU时间为整个App进程，含UI、日志和其他线程；" "帧数由邻近原生摘要窗口得出；CPU毫秒/帧为近似值，不是原生逐帧计时。"), } def run(args): adb = args.adb or str(Path.home() / "Library/Android/sdk/platform-tools/adb") info, sha, head, _ = verifier.inspect(verifier.APK, adb, args.serial) def device(*argv): return verifier.call([adb, "-s", args.serial, *argv]) expected_full = args.scenario == "RCRR_FULL_CONTROL_NO_TIDE" clock_ticks = int(device("shell", "getconf", "CLK_TCK")) if clock_ticks <= 0: raise RuntimeError("invalid CLOCK_TICKS") stamp = datetime.now().strftime("%Y%m%d_%H%M%S") out = verifier.ARTIFACTS / "step119-rcrr-cpu" / f"{stamp}-{args.scenario.lower()}" out.mkdir(parents=True, exist_ok=False) launch = device("shell", "am", "start", "-S", "-n", verifier.ACTIVITY, "--es", "step113_scenario", args.scenario) (out / "launch.txt").write_text(launch + "\n", encoding="utf-8") pid = None for _ in range(12): time.sleep(1) pid_tokens = verifier.call([adb, "-s", args.serial, "shell", "pidof", verifier.PACKAGE], check=False).split() if pid_tokens: pid = int(pid_tokens[0]) break if pid is None: raise RuntimeError("APP_PID_MISSING") print(f"SCENARIO={args.scenario} PID={pid} SHA_MATCHED={sha} SOURCE={head}", flush=True) time.sleep(args.warmup) warm_log = device("logcat", "-d", "-v", "brief", "-s", "STEP113Gate:I", "*:S") warm_samples = verifier.extract(warm_log, pid, args.scenario) if not any(s.get("sceneReady") and not s.get("blockedReason") for s in warm_samples): (out / "warmup_logcat.txt").write_text(warm_log, encoding="utf-8") raise RuntimeError("scene not ready after warmup (saved log)") def cpu_sample(): # Send one shell command string. Separate sh/-c args are not safely # quoted by all ADB versions and previously blocked the first run. raw = device("shell", f"cat /proc/{pid}/stat; cat /proc/uptime") lines = [line for line in raw.splitlines() if line.strip()] if len(lines) != 2: raise RuntimeError(f"unexpected /proc sample: {raw[:120]}") ticks, born = parse_proc_stat(lines[0]) uptime = float(lines[1].split()[0]) return {"ticks": ticks, "startTimeTicks": born, "deviceElapsedMs": uptime * 1000, "hostMonotonic": time.monotonic()} (out / "warmup_logcat.txt").write_text(warm_log + "\n", encoding="utf-8") observations = [cpu_sample()] deadline = observations[0]["hostMonotonic"] + args.seconds while True: remaining = deadline - time.monotonic() if remaining <= 0: break time.sleep(min(args.interval, remaining)) sample = cpu_sample() if sample["startTimeTicks"] != observations[0]["startTimeTicks"]: raise RuntimeError("process PID reused/restarted: cannot compare") observations.append(sample) first_cpu, last_cpu = observations[0], observations[-1] elapsed = last_cpu["hostMonotonic"] - first_cpu["hostMonotonic"] raw_log = device("logcat", "-d", "-v", "brief", "-s", "STEP113Gate:I", "*:S") (out / "logcat.txt").write_text(raw_log + "\n", encoding="utf-8") # Preserve early summaries when Android logcat rotates during sampling. # Extract combines records by exportId so repeated chunks do not count twice. native = verifier.extract(warm_log + "\n" + raw_log, pid, args.scenario) ready = [s for s in native if s.get("sceneReady") and not s.get("blockedReason") and isinstance(s.get("sampleElapsedMs"), (float, int))] def closest(device_time): if not ready: raise RuntimeError("no timestamped Native summary") return min(ready, key=lambda s: abs(s["sampleElapsedMs"] - device_time)) a, b = closest(first_cpu["deviceElapsedMs"]), closest(last_cpu["deviceElapsedMs"]) offsets = [abs(a["sampleElapsedMs"] - first_cpu["deviceElapsedMs"]), abs(b["sampleElapsedMs"] - last_cpu["deviceElapsedMs"])] parity = all(verifier.scene_parity_valid(s, args.scenario) for s in (a, b)) same_lifecycle = (a.get("sceneGeneration"), a.get("lifecycle")) == ( b.get("sceneGeneration"), b.get("lifecycle")) control_match = all(s.get("commands", {}).get("forceFullControl") == expected_full for s in (a, b)) metrics = collect_result((a, b), first_cpu, last_cpu, clock_ticks, elapsed) loads = [] for old, new in zip(observations, observations[1:]): dt = new["hostMonotonic"] - old["hostMonotonic"] if dt > 0: loads.append((new["ticks"] - old["ticks"]) / clock_ticks / dt) metrics["medianCpuCoreEquivalentPerInterval"] = statistics.median(loads) if loads else None status = ( "INVALID_PARITY_OR_LIFECYCLE" if not parity or not same_lifecycle or not control_match else "SUMMARY_ALIGNMENT_LIMIT" if max(offsets) > 9000 else "DECISION_MISMATCH" if b.get("decisionMismatch") != 0 else "FULL_CONTROL_OBSERVED" if expected_full and metrics["nativeWindowFullFrames"] > 0 and metrics["nativeWindowFastFrames"] == 0 else "FAST_DOMINANT" if not expected_full and metrics["fastFraction"] is not None and metrics["fastFraction"] >= 0.95 else "MIXED_OR_NO_FAST" if not expected_full else "NO_EXPECTED_FRAMES" ) data = { "schema": "step119-rcrr-cpu-benchmark-v1", "status": status, "sourceCommit": head, "apkSha256": sha, "device": args.serial, "pid": pid, "scenario": args.scenario, "fullControl": expected_full, "durationSecondsRequested": args.seconds, "durationSecondsObserved": elapsed, "warmupSeconds": args.warmup, "clockTicksPerSecond": clock_ticks, "nativeCount": len(ready), "nativeCpuBoundaryOffsetMs": offsets, "sameLifecycle": same_lifecycle, "parityValid": parity, "controlMatched": control_match, "nativeStart": a, "nativeEnd": b, "cpuObservations": observations, "metrics": metrics, "performanceQualification": ("进程级CPU测量仅可与同APK/相同Style相机负载的对照比较；" "应取Full-Fast-Full或交叉顺序重复，排除热机、缓存热度和帧率不同造成的误判；" "不能用此数据宣称GPU时间、省电或逐帧CPU精确耗时。"), } (out / "result.json").write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps({"status": status, "scenario": args.scenario, "metrics": metrics, "evidence": str(out / "result.json")}, ensure_ascii=False), flush=True) return 0 if status in ("FAST_DOMINANT", "FULL_CONTROL_OBSERVED") else 2 def main(): parser = argparse.ArgumentParser() parser.add_argument("--serial", default="192.168.200.139:5555") parser.add_argument("--adb", default=None) parser.add_argument("--scenario", choices=BENCH_SCENARIOS, required=True) parser.add_argument("--seconds", type=int, default=90) parser.add_argument("--warmup", type=int, default=35) parser.add_argument("--interval", type=float, default=6.0) args = parser.parse_args() if args.seconds < 75 or args.warmup < 20 or args.interval < 1: parser.error("观察至少75秒、Warmup至少20秒、采样间隔至少1秒") try: return run(args) except (RuntimeError, OSError, ValueError, KeyError) as e: print(f"STEP119_CPU_BENCH_BLOCKED: {e}", file=sys.stderr) return 3 if __name__ == "__main__": sys.exit(main())