#!/usr/bin/env python3 """STEP-113 on-device PBF Fast Path verifier. Does not build, install or alter APKs.""" import argparse import base64 import datetime import hashlib import json import os from pathlib import Path import re import subprocess import sys import time ROOT = Path(__file__).resolve().parents[2] ARTIFACTS = ROOT / "artifacts/maplibre-particle/android-validation" APK = ARTIFACTS / "MapLibreAndroidTestApp-validation-opengl-debug-x86_64.apk" BUILD_INFO = ARTIFACTS / "build-info.txt" PACKAGE = "org.maplibre.android.testapp" ACTIVITY = f"{PACKAGE}/{PACKAGE}.activity.particle.ParticleCoreLayerValidationActivity" SCENARIOS = ("STATIC_RC", "PBF63", "PARTICLE_OFF", "PARTIAL_RCRR", "SOURCE_CHANGE", "STATIC_RC_NO_TIDE", "RCRR_C_NARROW_NO_TIDE", "RCRR_C_WIDE_NO_TIDE", "RCRR_R_DYNAMIC_NO_TIDE", "RCR_PARTICLE_ONLY_NO_TIDE", "RCRRR_PARTICLE_ONLY_NO_TIDE", "RCRR_FULL_CONTROL_NO_TIDE") NO_TIDE_SCENARIOS = {"STATIC_RC_NO_TIDE", "RCRR_C_NARROW_NO_TIDE", "RCRR_C_WIDE_NO_TIDE", "RCRR_R_DYNAMIC_NO_TIDE", "RCR_PARTICLE_ONLY_NO_TIDE", "RCRRR_PARTICLE_ONLY_NO_TIDE", "RCRR_FULL_CONTROL_NO_TIDE"} LOG_ENTRY = re.compile(r"^[A-Z]/STEP113Gate\(\s*(\d+)\):\s*(\{.+\})\s*$") def call(args, *, check=True): result = subprocess.run(args, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) if check and result.returncode: raise RuntimeError(f"command returned {result.returncode}: {' '.join(args)}\n" f"{result.stderr[-1200:]}\n{result.stdout[-1200:]}") return result.stdout.strip() def sha256(path): h = hashlib.sha256() with open(path, "rb") as f: for block in iter(lambda: f.read(1048576), b""): h.update(block) return h.hexdigest() def inspect(apk, adb, serial): if not apk.is_file() or not BUILD_INFO.is_file(): raise RuntimeError("缺少 Validation APK/build-info.txt，必须先由用户手工运行标准构建脚本") info = dict(line.split("=", 1) for line in BUILD_INFO.read_text(encoding="utf-8").splitlines() if "=" in line) source_head = call(["git", "-C", str(ROOT), "rev-parse", "HEAD"]) tracked_changes = call(["git", "-C", str(ROOT), "status", "--porcelain", "--untracked-files=no"]) if tracked_changes: raise RuntimeError("SOURCE_NOT_LOCKED: tracked源码有未提交变更；RDC须先独立审计并scoped提交，用户再构建最新HEAD") if info.get("commit") != source_head: raise RuntimeError(f"STALE_APK: build-info commit={info.get('commit')}, " f"current HEAD={source_head}; 禁止用旧APK验收本轮修复") local_sha = sha256(apk) if local_sha != info.get("sha256"): raise RuntimeError(f"APK_SHA_MISMATCH: actual={local_sha}, build-info={info.get('sha256')}") if info.get("abi") != "x86_64": raise RuntimeError(f"unexpected ABI: {info.get('abi')}") device = lambda *a: call([adb, "-s", serial, *a]) abi = device("shell", "getprop", "ro.product.cpu.abi") if abi != "x86_64": raise RuntimeError(f"DEVICE_ABI_MISMATCH: {abi}") paths = device("shell", "pm", "path", PACKAGE).splitlines() if not paths or not paths[0].startswith("package:"): raise RuntimeError(f"TEST_APP_NOT_INSTALLED: {PACKAGE}") remote = paths[0].split("package:", 1)[1].strip() installed_sha = device("shell", "sha256sum", remote).split()[0] if installed_sha != local_sha: raise RuntimeError(f"STALE_INSTALLED_APK: installed={installed_sha} local={local_sha}; " "用户须核实后手工 adb install -r，不能用旧版本验收") return info, local_sha, source_head, device def extract(logcat, pid, expected): chunks = {} for line in logcat.splitlines(): found = LOG_ENTRY.match(line.strip()) if not found or int(found.group(1)) != pid: continue try: item = json.loads(found.group(2)) if item.get("encoding") != "base64-utf8" or item.get("kind") not in ("summary", "blocked"): continue ident = (item["exportId"], item["kind"]) record = chunks.setdefault(ident, {"count": item["chunkCount"], "items": {}}) if record["count"] == item["chunkCount"]: record["items"][int(item["chunkIndex"])] = item["payload"] except (KeyError, ValueError, TypeError): continue result = [] for (ident, kind), record in chunks.items(): if len(record["items"]) != record["count"]: continue try: packed = "".join(record["items"][i] for i in range(record["count"])) sample = json.loads(base64.b64decode(packed, validate=True).decode("utf-8")) if kind == "blocked": sample["blockedReason"] = sample.get("reason") or "SCENE_BLOCKED" sample["sceneReady"] = False # Keep revoked/blocked summaries as negative evidence too; an # earlier Fast counter must not conceal a later Annotation guard. if sample.get("scenario") == expected: result.append((int(ident.split("-", 1)[0]), sample)) except (KeyError, ValueError, TypeError, UnicodeError): continue return [x[1] for x in sorted(result, key=lambda item: item[0])] def rcrr_segments_valid(sample, scenario): """Verify exact R/C boundaries in the new isolated, 64-layer PBF Test App cases.""" retained = sample.get("retained", []) ranges = [(entry.get("begin"), entry.get("end")) for entry in retained] expected = { "RCRR_C_NARROW_NO_TIDE": [(0, 32), (33, 63), (63, 64)], "RCRR_FULL_CONTROL_NO_TIDE": [(0, 32), (33, 63), (63, 64)], "RCR_PARTICLE_ONLY_NO_TIDE": [(0, 32), (33, 64)], "RCRRR_PARTICLE_ONLY_NO_TIDE": [(0, 32), (33, 38), (38, 63), (63, 64)], "RCRR_C_WIDE_NO_TIDE": [(0, 2), (38, 63), (63, 64)], "RCRR_R_DYNAMIC_NO_TIDE": [(0, 33), (38, 63), (63, 64)], } expected_r = expected.get(scenario) if expected_r is None: return False # The actual Native segment ranges must prove that the single C consists # only of Particle for the normal CPU candidates. An R-only list cannot # prove the C type/range; fail closed if the Native list is missing. boundaries = sample.get("segmentRanges") if not isinstance(boundaries, list): return False expected_segments = [{"strategy": "RETAINED", "begin": expected_r[0][0], "end": expected_r[0][1]}, {"strategy": "CONTINUOUS", "begin": expected_r[0][1], "end": expected_r[1][0]}] expected_segments.extend({"strategy": "RETAINED", "begin": a, "end": b} for a, b in expected_r[1:]) if boundaries != expected_segments: return False # Cache-disabled Full control may omit cache diagnostics but must still # supply the exact actual R/C segment list above. return (not ranges or ranges == expected_r) if scenario == ( "RCRR_FULL_CONTROL_NO_TIDE") else ranges == expected_r def scene_parity_valid(sample, scenario): """Fail closed on the *declared* base style, not just a loose layer count. Original five STEP-113 scenarios require all 63 original base layers. STEP-118 explicitly names a user-approved test-copy exclusion of the orphaned tide-spots layer: 63 remote originals -> 62 copied basics, plus the two known weather layers (and no empty Annotation point layer). It is not equivalent to the unmodified original 63-layer test. """ if not sample.get("planValid"): return False if scenario not in NO_TIDE_SCENARIOS: return sample.get("basePreserved") is True and not sample.get("tideLayerExcluded", False) return ( sample.get("basePreserved") is False and sample.get("tideLayerExcluded") is True and sample.get("originalBaseLayerCount") == 63 and sample.get("effectiveBaseLayerCount") == 62 and sample.get("excludedLayerId") == "tide-spots" and sample.get("excludedLayerSource") == "tide" and sample.get("excludedLayerCount") == 1 and sample.get("actualLayers") == 64 and sample.get("commands", {}).get("emptyBuiltinAnnotationLayerRemoved") is True and sample.get("commands", {}).get("annotationSourcePreserved") is True and all(source.get("sourceId") != "tide" for source in sample.get("sources", [])) and sample.get("actualLayers") == 64 and ( (scenario == "STATIC_RC_NO_TIDE" and len(sample.get("retained", [])) == 1 and sample["retained"][0].get("begin") == 0 and sample["retained"][0].get("end") == 62) or (scenario != "STATIC_RC_NO_TIDE" and rcrr_segments_valid(sample, scenario)) ) ) def run(): parser = argparse.ArgumentParser(description="Check actual STEP-113 PBF on-device Fast frames") parser.add_argument("--serial", default="192.168.200.139:5555") parser.add_argument("--adb", default=str(Path.home() / "Library/Android/sdk/platform-tools/adb")) parser.add_argument("--apk", default=str(APK)) parser.add_argument("--scenario", choices=SCENARIOS, default="STATIC_RC") parser.add_argument("--seconds", type=int, default=75) parser.add_argument("--preflight-only", action="store_true") args = parser.parse_args() if args.seconds < 75 and not args.preflight_only: parser.error("原STEP-113场景至少75秒；摘要与完整120帧导出仍须分别验收") if not Path(args.adb).is_file(): raise RuntimeError(f"adb not found: {args.adb}") info, local_sha, head, device = inspect(Path(args.apk), args.adb, args.serial) print(f"APK_SOURCE_HEAD={head}") print(f"APK_SHA256={local_sha}") print(f"DEVICE={args.serial} ABI=x86_64 INSTALLED_SHA_MATCH=True") if args.preflight_only: print("PREFLIGHT_PASS（仅证明同版本APK已安装，未测试Fast）") return 0 stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") output = ARTIFACTS / f"step113-retest-{stamp}" / args.scenario.lower() output.mkdir(parents=True, exist_ok=False) command = ["shell", "am", "start", "-S", "-n", ACTIVITY, "--es", "step113_scenario", args.scenario] print(f"START_SCENARIO={args.scenario}") output.joinpath("launch.txt").write_text(device(*command) + "\n", encoding="utf-8") process_id = None for _ in range(12): time.sleep(1) pid_text = call([args.adb, "-s", args.serial, "shell", "pidof", PACKAGE], check=False) if pid_text.strip().split(): process_id = int(pid_text.strip().split()[0]) break if process_id is None: raise RuntimeError("APP_PROCESS_NOT_RUNNING after launch") print(f"APP_PID={process_id}") # PID acquisition can take less than 12 seconds. Never subtract the # maximum poll duration from the actual scenario observation period. observation_started = time.monotonic() time.sleep(args.seconds) observation_elapsed = time.monotonic() - observation_started raw_log = device("logcat", "-d", "-v", "brief", "-s", "STEP113Gate:I", "*:S") output.joinpath("logcat.txt").write_text(raw_log + "\n", encoding="utf-8") all_samples = extract(raw_log, process_id, args.scenario) samples = [s for s in all_samples if s.get("sceneReady") and not s.get("blockedReason")] if not samples: status = "SCENE_BLOCKED_OR_REVOKED" if any(s.get("blockedReason") for s in all_samples) else "NO_READY_SUMMARY" fast_delta = full_delta = 0 else: start, end = samples[0], samples[-1] fast_delta = end["fastRendered"] - start["fastRendered"] full_delta = end["fullRendered"] - start["fullRendered"] if len(samples) < 2: status = "INSUFFICIENT_SAMPLES" elif any(s.get("blockedReason") for s in all_samples): status = "SCENE_BLOCKED_OR_REVOKED" elif any((s.get("sceneGeneration"), s.get("lifecycle")) != (start.get("sceneGeneration"), start.get("lifecycle")) for s in samples): status = "SUMMARY_LIFECYCLE_CHANGED" elif any(not scene_parity_valid(s, args.scenario) for s in samples): status = "INVALID_PLAN_OR_BASE_PARITY" elif any(len(s.get("gateFailureCounts", [])) != 24 for s in samples): status = "INCOMPLETE_GATE_SUMMARY" elif any(s["decisionMismatch"] != 0 for s in samples): status = "DECISION_MISMATCH" elif any(b["fastRendered"] < a["fastRendered"] or b["fullRendered"] < a["fullRendered"] for a, b in zip(samples, samples[1:])): status = "COUNTER_RESET" elif fast_delta > 0: status = "FAST_OBSERVED" else: status = "NO_FAST_OBSERVED" report = { "schema": "step113-fast-path-retest", "sourceCommit": head, "apkSha256": local_sha, "installedShaMatched": True, "device": args.serial, "appPid": process_id, "scenario": args.scenario, "sourceReport": ( "STEP-118 user-approved TEST-APP-ONLY 62-layer derivative; " "original 63-layer PBF Style remains unmodified; not STEP-114 smoke" if args.scenario in NO_TIDE_SCENARIOS else "STEP-113 original 63-layer PBF test scene; not STEP-114 smoke" ), "durationSecondsRequested": args.seconds, "durationSecondsObservedAfterPid": observation_elapsed, "completeSummarySamples": len(samples), "blockedSummarySamples": [s for s in all_samples if s.get("blockedReason")], "status": status, "fastDelta": fast_delta, "fullDelta": full_delta, "initialSummary": samples[0] if samples else None, "finalSummary": samples[-1] if samples else None, } output.joinpath("result.json").write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") print(f"STATUS={status} FastDelta={fast_delta} FullDelta={full_delta} Samples={len(samples)}") print(f"EVIDENCE={output / 'result.json'}") # Fail closed: no Fast observed is a meaningful negative result, not PASS. if status == "FAST_OBSERVED": print("FAST_OBSERVED仅表示真实Fast计数增长；缓存/像素/安全回退仍须逐帧验收。") return 0 return 2 if __name__ == "__main__": try: sys.exit(run()) except (OSError, RuntimeError, ValueError, KeyError) as exc: print(f"STEP113_RETEST_BLOCKED: {exc}", file=sys.stderr) sys.exit(3)