303 lines
14 KiB
Python
Executable File
303 lines
14 KiB
Python
Executable File
#!/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)
|