198 lines
10 KiB
Python
198 lines
10 KiB
Python
"""Audit only the formal exports selected in the official simulator's test folder.
|
|
|
|
No encrypted payload is opened. Match the public activation-ticket digest to the
|
|
official statistics queue, and the complete package digest to its result sidecar.
|
|
Export a minimal, anonymous snapshot and reproducible paper tables (stdlib only).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import csv
|
|
import hashlib
|
|
import json
|
|
import sqlite3
|
|
import struct
|
|
from decimal import Decimal, ROUND_HALF_UP
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
FIELDS = (
|
|
"entered", "end_reason", "cleared_jammer_count", "measure_accepted_count",
|
|
"virtual_time_us", "program_run_duration_ms", "channel_switch_count",
|
|
"clear_failure_count", "state",
|
|
)
|
|
|
|
|
|
def sha(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def public_header(path: Path) -> tuple[dict, str]:
|
|
data = path.read_bytes()
|
|
if data[:8] != b"JMBFLOG1" or len(data) < 14:
|
|
raise ValueError(f"Unsupported formal envelope: {path.name}")
|
|
version, size = struct.unpack(">HI", data[8:14])
|
|
if version != 1 or not 0 < size <= len(data) - 14:
|
|
raise ValueError(f"Invalid envelope header: {path.name}")
|
|
header = json.loads(data[14:14 + size])
|
|
if header["package_type"] != "formal_behavior_log":
|
|
raise ValueError("Practice logs are not admitted")
|
|
return header, sha(data)
|
|
|
|
|
|
def fixed(value: Decimal, digits: int = 3) -> str:
|
|
return str(value.quantize(Decimal(1).scaleb(-digits), rounding=ROUND_HALF_UP))
|
|
|
|
|
|
def derive(row: dict) -> dict:
|
|
count = row["cleared_jammer_count"]
|
|
total = Decimal(row["virtual_time_us"]) / 1_000_000
|
|
movement = total - (5 * row["measure_accepted_count"]
|
|
+ row["channel_switch_count"] + 5 * count
|
|
+ 3 * row["clear_failure_count"])
|
|
if count <= 0 or movement < 0:
|
|
raise ValueError("Invalid counts or negative inferred movement time")
|
|
return {
|
|
"virtual_time_s": str(total),
|
|
"average_clear_time_s": fixed(total / count, 6),
|
|
"program_run_duration_s": fixed(Decimal(row["program_run_duration_ms"]) / 1000),
|
|
"inferred_movement_time_s": str(movement),
|
|
"inferred_distance_m": str(5 * movement),
|
|
"movement_share_percent": fixed(100 * movement / total, 6),
|
|
"source_total": None,
|
|
"clear_fraction": None,
|
|
}
|
|
|
|
|
|
def collect(sim: Path) -> list[dict]:
|
|
selected = sorted((sim / "test").rglob("*.jlog"))
|
|
if not selected:
|
|
raise ValueError("No formal exports in test/")
|
|
queue = sim / "JammersSimulatorData/formal-statistics-queue.sqlite3"
|
|
original = sim / "JammersSimulatorData/behavior-logs"
|
|
# Read the live queue including its WAL, without writing to the simulator.
|
|
with sqlite3.connect(queue.resolve().as_uri() + "?mode=ro", uri=True) as db:
|
|
db.row_factory = sqlite3.Row
|
|
query = "SELECT activation_ticket_sha256," + ",".join(FIELDS) + " FROM statistics_tasks"
|
|
statistics = {r["activation_ticket_sha256"]: dict(r) for r in db.execute(query)}
|
|
records = []
|
|
for path in selected:
|
|
header, digest = public_header(path)
|
|
encoded = header["server_ticket_b64"]
|
|
ticket_hash = sha(base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)))
|
|
sidecar = json.loads((original / (path.stem + ".result.json")).read_text("utf-8"))
|
|
for field in ("problem_no", "formal_index", "case_code"):
|
|
if header[field] != sidecar[field]:
|
|
raise ValueError(f"Mismatched case: {path.name}")
|
|
if digest != sidecar["package_sha256"] or digest != sha((original / path.name).read_bytes()):
|
|
raise ValueError(f"Mismatched formal package: {path.name}")
|
|
matched = statistics[ticket_hash]
|
|
row = {name: header[name] for name in ("problem_no", "formal_index", "case_code", "client_version")}
|
|
row.update({name: matched[name] for name in FIELDS})
|
|
row.update({
|
|
"file": path.relative_to(sim).as_posix(), "size_bytes": path.stat().st_size,
|
|
"package_sha256": digest, "activation_ticket_sha256": ticket_hash,
|
|
"created_at_utc": header["created_at_utc"],
|
|
"ended_at_utc": sidecar["ended_at_utc"],
|
|
"content_encryption": header["content_encryption"],
|
|
"identity_and_ticket_content_omitted": True,
|
|
"same_as_official_original": True,
|
|
})
|
|
row["derived"] = derive(row)
|
|
records.append(row)
|
|
keys = [(r["problem_no"], r["formal_index"]) for r in records]
|
|
if len(set(keys)) != len(keys):
|
|
raise ValueError("Duplicate formal attempt")
|
|
return records
|
|
|
|
|
|
def write_outputs(rows: list[dict], out: Path) -> None:
|
|
out.mkdir(parents=True, exist_ok=True)
|
|
snapshot = {
|
|
"schema": "paper-formal-audit-v1",
|
|
"selection": "Only Jammers-simulator-full/test/**/*.jlog",
|
|
"statistics_source": "Official formal-statistics-queue.sqlite3; matched by SHA-256 of the public activation ticket",
|
|
"limitations": [
|
|
"Encrypted action payloads are not decrypted; no trajectory or bearing observations are available.",
|
|
"Source totals and omnidirectional/directional counts are not disclosed.",
|
|
"Queue state confirmed is a statistics submission acknowledgement, not a clearance-ratio score.",
|
|
"No solver commit or configuration digest is bound to these public statistics.",
|
|
"The third formal attempt for each problem is not present in the selected data.",
|
|
],
|
|
"records": rows,
|
|
}
|
|
(out / "formal-results.json").write_text(json.dumps(snapshot, ensure_ascii=False, indent=2) + "\n", "utf-8")
|
|
columns = ["problem_no", "formal_index", "case_code", *FIELDS, "package_sha256"]
|
|
with (out / "formal-results.csv").open("w", encoding="utf-8-sig", newline="") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=columns + list(rows[0]["derived"]))
|
|
writer.writeheader()
|
|
for row in rows:
|
|
writer.writerow({**{key: row[key] for key in columns}, **row["derived"]})
|
|
for problem in (3, 4):
|
|
by_index = {r["formal_index"]: r for r in rows if r["problem_no"] == problem}
|
|
lines = ["% Generated by scripts/audit_paper_formal.py; seconds; half-up rounding."]
|
|
for index in (1, 2, 3):
|
|
row = by_index.get(index)
|
|
if row is None:
|
|
lines.append(f"测试 {index} & 未提供记录 & -- & -- & -- " + r"\\")
|
|
else:
|
|
avg = Decimal(row["virtual_time_us"]) / 1_000_000 / row["cleared_jammer_count"]
|
|
lines.append(f"测试 {index} & \\texttt{{{row['case_code']}}} & {row['cleared_jammer_count']} & {fixed(avg)} & {row['derived']['program_run_duration_s']} " + r"\\")
|
|
(out / f"formal-q{problem}-rows.tex").write_text("\n".join(lines) + "\n", "utf-8")
|
|
cost_lines, constants = [], []
|
|
for row in rows:
|
|
label = f"Q{row['problem_no']}-{row['formal_index']}"
|
|
total = Decimal(row["virtual_time_us"]) / 1_000_000
|
|
move = Decimal(row["derived"]["inferred_movement_time_s"])
|
|
cost_lines.append(f"{label} & {row['measure_accepted_count']} & {row['channel_switch_count']} & {row['clear_failure_count']} & {fixed(total)} & {fixed(move)} & {fixed(100*move/total, 2)}\\% " + r"\\")
|
|
(out / "formal-cost-rows.tex").write_text("\n".join(cost_lines) + "\n", "utf-8")
|
|
for problem, word in ((3, "Three"), (4, "Four")):
|
|
group = [r for r in rows if r["problem_no"] == problem]
|
|
count = sum(r["cleared_jammer_count"] for r in group)
|
|
total = sum(Decimal(r["virtual_time_us"]) / 1_000_000 for r in group)
|
|
mean = sum(Decimal(r["virtual_time_us"]) / 1_000_000 / r["cleared_jammer_count"] for r in group) / len(group)
|
|
constants.extend([f"\\newcommand{{\\Q{word}Count}}{{{count}}}",
|
|
f"\\newcommand{{\\Q{word}Weighted}}{{{fixed(total/count)}}}",
|
|
f"\\newcommand{{\\Q{word}Mean}}{{{fixed(mean)}}}"])
|
|
(out / "formal-macros.tex").write_text("\n".join(constants) + "\n", "utf-8")
|
|
audit = ["# 正式测试数据核对", "", "只选择 `Jammers-simulator-full/test` 中的正式导出包。",
|
|
"加密载荷不作解密。公开头部提供案例与票据；包哈希核对原件及 result 侧文件；票据哈希匹配官方正式统计数据库。",
|
|
"导出快照不包含队号、设备信息、票据内容、密钥封装或账号信息。", "",
|
|
"| 案例 | 正式序号 | 原件一致 | 包 SHA-256 |", "| --- | --- | --- | --- |"]
|
|
for row in rows:
|
|
audit.append(f"| {row['case_code']} | Q{row['problem_no']}-{row['formal_index']} | 是 | `{row['package_sha256']}` |")
|
|
audit += ["", "虚拟时间保留官方整数微秒，程序运行时间采用官方毫秒。平均定位清除时间 = 总虚拟秒数 / 清除数。",
|
|
"移动时间由总时长减去检测、换频、成功及失败清除耗时反推；不是从解密轨迹逐动作复算所得。",
|
|
"`confirmed` 仅指统计提交确认，`user_exit` 仅指主动退出。源总数未知，不能报告全清率。",
|
|
"问题三、四均缺第三次正式导出；不以演练或自建模拟器结果替代。",
|
|
"当前算法描述对应仓库代码；公开统计没有策略提交版本标识，不能用这四局作当前参数的配对消融。", ""]
|
|
(out / "audit.md").write_text("\n".join(audit), "utf-8")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--sim-root", type=Path, default=ROOT / "Jammers-simulator-full")
|
|
parser.add_argument("--out", type=Path, default=ROOT / "docs/paper/data")
|
|
parser.add_argument("--snapshot", type=Path, help="Regenerate tables from the audited anonymous snapshot, without the simulator database")
|
|
args = parser.parse_args()
|
|
if args.snapshot:
|
|
rows = json.loads(args.snapshot.read_text("utf-8"))["records"]
|
|
for row in rows:
|
|
if derive(row) != row["derived"]:
|
|
raise ValueError("Snapshot arithmetic changed")
|
|
path = args.sim_root / row["file"]
|
|
if path.exists() and public_header(path)[1] != row["package_sha256"]:
|
|
raise ValueError("Snapshot and formal export disagree")
|
|
else:
|
|
rows = collect(args.sim_root)
|
|
write_outputs(rows, args.out)
|
|
for row in rows:
|
|
print(f"Q{row['problem_no']}-{row['formal_index']} {row['case_code']}: "
|
|
f"cleared={row['cleared_jammer_count']} average={row['derived']['average_clear_time_s']} s "
|
|
f"real={row['derived']['program_run_duration_s']} s")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|