124 lines
5.9 KiB
Python
124 lines
5.9 KiB
Python
"""Offline audit of official RL practice runs (Q3/Q4); never connects to simulator.
|
|
|
|
Checks the copied official .result.json/.jlog, our session.jsonl and summary.json:
|
|
package sha256, enter/exit, accepted timestamps inside official window, unique
|
|
request ids, clear-success channel count == official jammer_count, summary
|
|
consistency, and virtual time recomputation with integer microsecond accumulation.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
|
|
def save_json(path: Path, value: dict) -> None:
|
|
path.write_text(json.dumps(value, ensure_ascii=False, indent=2,
|
|
allow_nan=False) + "\n", encoding="utf-8")
|
|
|
|
|
|
def audit(folder: Path) -> dict:
|
|
records = [json.loads(line) for line in
|
|
(folder / "session.jsonl").read_text(encoding="utf-8").splitlines()]
|
|
events = [e for e in records if e.get("event") == "accepted"]
|
|
summary = json.loads((folder / "summary.json").read_text(encoding="utf-8"))
|
|
official_paths = list(folder.glob("*.result.json"))
|
|
if len(official_paths) != 1 or not events:
|
|
raise ValueError("Expected one official result and nonempty session evidence")
|
|
official = json.loads(official_paths[0].read_text(encoding="utf-8"))
|
|
package = official_paths[0].with_name(
|
|
official_paths[0].name.replace(".result.json", ".jlog"))
|
|
checks: dict = {}
|
|
|
|
checks["official_problem_matches_summary"] = (
|
|
official["problem_no"] in (3, 4)
|
|
and summary["strategy"] == ("rl" if official["problem_no"] == 3 else "rl_q4")
|
|
and 10 <= official["jammer_count"] <= 16
|
|
and official["omnidirectional_jammer_count"] + official["directional_jammer_count"]
|
|
== official["jammer_count"])
|
|
checks["official_package_sha256_matches"] = (
|
|
hashlib.sha256(package.read_bytes()).hexdigest() == official["package_sha256"])
|
|
checks["normal_enter_and_exit"] = (
|
|
events[0]["path"] == "/enter" and events[-1]["path"] == "/exit"
|
|
and events[-1]["response"].get("exit_reason") == "user_exit")
|
|
start_ms = datetime.fromisoformat(official["window_started_at_utc"]).timestamp() * 1000
|
|
end_ms = datetime.fromisoformat(official["ended_at_utc"]).timestamp() * 1000
|
|
checks["responses_within_official_run_window"] = all(
|
|
start_ms <= e["response"]["real_timestamp_ms"] <= end_ms + 1 for e in events)
|
|
ids = [e["request"]["request_id"] for e in events]
|
|
checks["unique_accepted_request_ids"] = len(ids) == len(set(ids))
|
|
clears = [e for e in events if e["path"] == "/clear"]
|
|
successes = [e for e in clears if e["response"].get("clear_result") == "success"]
|
|
cleared = sorted({e["request"]["channel"] for e in successes})
|
|
checks["all_official_sources_cleared"] = len(cleared) == official["jammer_count"]
|
|
checks["summary_completed_and_consistent"] = (
|
|
bool(summary["completed"])
|
|
and sorted(summary["cleared_channels"]) == cleared
|
|
and summary["unresolved_channels"] == []
|
|
and summary["anomalies"] == [])
|
|
|
|
position, channel, virtual_us = (0.0, 0.0), 1, 0
|
|
mismatches = 0
|
|
for event in events:
|
|
request, response, path = event["request"], event["response"], event["path"]
|
|
distance, switches, action_s = 0.0, 0, 0
|
|
if path in ("/measure", "/clear"):
|
|
target = (request["position"]["x"], request["position"]["y"])
|
|
distance = math.dist(position, target)
|
|
position = target
|
|
if path == "/measure":
|
|
switches = int(channel != request["channel"])
|
|
channel = request["channel"]
|
|
action_s = 5
|
|
else:
|
|
action_s = 5 if response["clear_result"] == "success" else 3
|
|
virtual_us += round(distance / 5 * 1_000_000) + (switches + action_s) * 1_000_000
|
|
if round(response["virtual_time_s"] * 1_000_000) != virtual_us:
|
|
mismatches += 1
|
|
checks["all_timestamps_match_microsecond_recomputation"] = mismatches == 0
|
|
real_s = (events[-1]["response"]["real_timestamp_ms"]
|
|
- events[0]["response"]["real_timestamp_ms"]) / 1000
|
|
checks["within_real_and_virtual_limits"] = (
|
|
real_s < events[0]["response"]["remaining_real_duration_s"]
|
|
and virtual_us / 1_000_000 < events[0]["response"]["max_virtual_duration_s"])
|
|
|
|
report = {
|
|
"passed": all(checks.values()),
|
|
"source": "official_practice",
|
|
"strategy": summary["strategy"],
|
|
"problem": official["problem_no"],
|
|
"case_code": official["case_code"],
|
|
"official_jammer_count": official["jammer_count"],
|
|
"directional_jammer_count": official["directional_jammer_count"],
|
|
"cleared_channels": cleared,
|
|
"accepted_requests_including_enter_exit": len(events),
|
|
"measures": len([e for e in events if e["path"] == "/measure"]),
|
|
"clear_attempts": len(clears),
|
|
"clear_successes": len(successes),
|
|
"virtual_time_s": virtual_us / 1_000_000,
|
|
"real_elapsed_server_s": real_s,
|
|
"checks": checks,
|
|
"limits": "仅验证本官方演练案例；不触碰正式测试，不解读官方加密日志内容。",
|
|
}
|
|
save_json(folder / "verification.json", report)
|
|
manifest = {p.name: {"bytes": p.stat().st_size,
|
|
"sha256": hashlib.sha256(p.read_bytes()).hexdigest()}
|
|
for p in sorted(folder.iterdir()) if p.is_file()
|
|
and p.name != "manifest.json"}
|
|
save_json(folder / "manifest.json", manifest)
|
|
return report
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("folder", type=Path)
|
|
args = parser.parse_args()
|
|
report = audit(args.folder)
|
|
print(json.dumps({"passed": report["passed"], "case_code": report["case_code"],
|
|
"problem": report["problem"], "checks": report["checks"]},
|
|
ensure_ascii=False, indent=2))
|
|
raise SystemExit(0 if report["passed"] else 1)
|