93 lines
3.8 KiB
Python
93 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate deterministic S2-03 encoder/coolant/interlock/ack evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
OUTPUT = ROOT / "coverage/device-io-s2.report.json"
|
|
|
|
INPUTS = (
|
|
"context-lock.json",
|
|
"wasm-core/include/sim_device_io.h",
|
|
"wasm-core/src/sim_device_io.c",
|
|
"wasm-core/tests/sim_device_io_test.c",
|
|
"web/worker/device-io.mjs",
|
|
"web/worker/test_device_io.mjs",
|
|
"web/worker/io-ack.mjs",
|
|
"web/worker/sim-session.mjs",
|
|
"linuxcnc/src/hal/components/sim_encoder.c",
|
|
"linuxcnc/src/hal/components/encoder.c",
|
|
"linuxcnc/configs/sim/axis/cooling.hal",
|
|
)
|
|
|
|
|
|
def sha256_bytes(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def canonical(value: object) -> bytes:
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
|
|
|
|
def run(command: list[str]) -> dict[str, object]:
|
|
completed = subprocess.run(command, cwd=ROOT, text=True, capture_output=True, check=False)
|
|
sample = None
|
|
for line in reversed(completed.stdout.splitlines()):
|
|
if line.lstrip().startswith("{"):
|
|
try:
|
|
sample = json.loads(line)
|
|
break
|
|
except json.JSONDecodeError:
|
|
continue
|
|
return {
|
|
"command": command,
|
|
"returncode": completed.returncode,
|
|
"status": "verified" if completed.returncode == 0 else "blocked",
|
|
"stdout_sha256": sha256_bytes(completed.stdout.encode()),
|
|
"stderr_sha256": sha256_bytes(completed.stderr.encode()),
|
|
"sample": sample,
|
|
}
|
|
|
|
|
|
def generate(output: Path = OUTPUT) -> dict[str, object]:
|
|
context = json.loads((ROOT / "context-lock.json").read_text(encoding="utf-8"))
|
|
native = run([str(ROOT / "build-native/sim_device_io_test")])
|
|
wasm = run(["node", str(ROOT / "build-wasm/sim_device_io_wasm_test.js")])
|
|
worker = run(["node", str(ROOT / "web/worker/test_device_io.mjs")])
|
|
native_sample = native["sample"]
|
|
wasm_sample = wasm["sample"]
|
|
report: dict[str, object] = {
|
|
"schema": "device-io-s2-report.v1",
|
|
"task_id": "S2-03",
|
|
"context_id": context["context_id"],
|
|
"status": "verified" if all(item["status"] == "verified" for item in (native, wasm, worker)) and native_sample == wasm_sample else "blocked",
|
|
"features": [
|
|
"fixed-tick quadrature encoder and index pulse",
|
|
"encoder position/velocity/count scaling",
|
|
"coolant mist/flood/lube interlock gating",
|
|
"door and latched interlock fault/recovery",
|
|
"deterministic IO ack issue/duplicate/timeout/late/abort",
|
|
"native/WASM byte-identical state sample",
|
|
"Worker checkpoint/replay device state",
|
|
],
|
|
"capacities": {"encoders": 8, "ack_entries": 64, "max_tick_step": 1000000},
|
|
"boundaries": ["real encoder transport and physical safety circuits remain REAL-EXCLUDED", "nonfinite/range/unsafe commands fail closed"],
|
|
"inputs": [{"path": path, "sha256": sha256_bytes((ROOT / path).read_bytes())} for path in INPUTS],
|
|
"tests": {"native": native, "wasm": wasm, "worker": worker, "native_wasm_byte_identical": native_sample == wasm_sample},
|
|
"report_sha256": None,
|
|
}
|
|
report["report_sha256"] = sha256_bytes(canonical({key: value for key, value in report.items() if key != "report_sha256"}))
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8")
|
|
return report
|
|
|
|
|
|
if __name__ == "__main__":
|
|
result = generate()
|
|
print(f"device_io_s2={result['status']} native_wasm={result['tests']['native_wasm_byte_identical']}")
|