Implement AXIS parity UI and execution baseline
This commit is contained in:
168
scripts/generate_authoritative_kinematics_s6_04.py
Normal file
168
scripts/generate_authoritative_kinematics_s6_04.py
Normal file
@@ -0,0 +1,168 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generate deterministic S6-04 authoritative Tripod evidence."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
CONTEXT_ID = "linuxcnc-2.10.0-pre0-wasm-m1-py-c1"
|
||||
WORKER = ("node", "web/worker/test_authoritative_kinematics_s6_04.mjs")
|
||||
PAIRS = {
|
||||
"kinematics": (("build-native/sim_kinematics_test",), ("node", "build-wasm/sim_kinematics_wasm_test.js")),
|
||||
"kinematics_motion": (("build-native/sim_kinematics_motion_test",), ("node", "build-wasm/sim_kinematics_motion_wasm_test.js")),
|
||||
}
|
||||
INPUTS = (
|
||||
"coverage/authoritative-kinematics-s6.report.json",
|
||||
"coverage/kinematics-registry.report.json",
|
||||
"coverage/tripod-kinematics.report.json",
|
||||
"coverage/kinematics-motion.report.json",
|
||||
"wasm-core/include/sim_kinematics.h",
|
||||
"wasm-core/src/sim_kinematics.c",
|
||||
"wasm-core/include/sim_kinematics_motion.h",
|
||||
"wasm-core/src/sim_kinematics_motion.c",
|
||||
"web/worker/authoritative-runtime.mjs",
|
||||
"web/worker/sim-session.mjs",
|
||||
"linuxcnc/src/emc/kinematics/tripodkins.c",
|
||||
"linuxcnc/configs/sim/tklinuxcnc/tripod.ini",
|
||||
"linuxcnc/lib/hallib/tripodsim.hal",
|
||||
"wasm-core/tests/sim_kinematics_test.c",
|
||||
"web/worker/test_authoritative_kinematics_s6_04.mjs",
|
||||
)
|
||||
|
||||
|
||||
def canonical(value: Any) -> str:
|
||||
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||||
|
||||
|
||||
def digest(value: bytes) -> str:
|
||||
return hashlib.sha256(value).hexdigest()
|
||||
|
||||
|
||||
def run(root: Path, args: tuple[str, ...]) -> dict[str, Any]:
|
||||
result = subprocess.run(args, cwd=root, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
|
||||
return {
|
||||
"command": list(args),
|
||||
"returncode": result.returncode,
|
||||
"status": "PASSED" if result.returncode == 0 else "FAILED",
|
||||
"stdout_sha256": digest(result.stdout),
|
||||
"stderr_sha256": digest(result.stderr),
|
||||
"stdout": result.stdout.decode("utf-8", errors="replace"),
|
||||
"stderr": result.stderr.decode("utf-8", errors="replace"),
|
||||
}
|
||||
|
||||
|
||||
def no_output(result: dict[str, Any]) -> dict[str, Any]:
|
||||
return {key: value for key, value in result.items() if key not in {"stdout", "stderr"}}
|
||||
|
||||
|
||||
def parse(result: dict[str, Any]) -> dict[str, Any] | None:
|
||||
for line in reversed(result["stdout"].splitlines()):
|
||||
if line.lstrip().startswith("{"):
|
||||
try:
|
||||
return json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
def artifact(root: Path, path: str) -> dict[str, Any]:
|
||||
value = root / path
|
||||
if not value.is_file():
|
||||
return {"path": path, "status": "missing"}
|
||||
return {"path": path, "bytes": value.stat().st_size, "sha256": digest(value.read_bytes())}
|
||||
|
||||
|
||||
def generate(root: Path, output: Path) -> dict[str, Any]:
|
||||
worker_result = run(root, WORKER)
|
||||
worker = parse(worker_result)
|
||||
native_wasm: dict[str, Any] = {}
|
||||
pair_ok = True
|
||||
for name, (native_command, wasm_command) in PAIRS.items():
|
||||
native = run(root, native_command)
|
||||
wasm = run(root, wasm_command)
|
||||
equal = native["returncode"] == 0 and wasm["returncode"] == 0 and native["stdout"] == wasm["stdout"]
|
||||
pair_ok = pair_ok and equal
|
||||
native_wasm[name] = {
|
||||
"native": no_output(native),
|
||||
"wasm": no_output(wasm),
|
||||
"golden_diff": {
|
||||
"status": "PASSED" if equal else "FAILED",
|
||||
"byte_identical": equal,
|
||||
"native_stdout_sha256": native["stdout_sha256"],
|
||||
"wasm_stdout_sha256": wasm["stdout_sha256"],
|
||||
},
|
||||
}
|
||||
ctest = run(root, ("ctest", "--test-dir", "build-native", "-R", "^(sim_kinematics_test|sim_kinematics_motion_test)$", "--output-on-failure"))
|
||||
supported = worker.get("supported", {}) if worker else {}
|
||||
parity = worker.get("parity", {}) if worker else {}
|
||||
worker_ok = (
|
||||
worker_result["returncode"] == 0
|
||||
and worker is not None
|
||||
and worker.get("status") == "verified"
|
||||
and supported.get("type") == "tripod"
|
||||
and supported.get("source_family") == "tripodkins"
|
||||
and supported.get("pose") == [2, 3, 4]
|
||||
and len(supported.get("joints", [])) == 3
|
||||
and parity.get("mode", {}).get("equal") is True
|
||||
and parity.get("recovery", {}).get("equal") is True
|
||||
)
|
||||
report: dict[str, Any] = {
|
||||
"schema": "authoritative-kinematics-s6-04.v1",
|
||||
"task_id": "S6-04",
|
||||
"context_id": CONTEXT_ID,
|
||||
"status": "verified" if worker_ok and pair_ok and ctest["returncode"] == 0 else "blocked",
|
||||
"scope": {
|
||||
"selected_family": "parallel_tripod",
|
||||
"kinematics": "tripod",
|
||||
"source_module": "tripodkins",
|
||||
"pose": [2, 3, 4],
|
||||
"joints": ["AD", "BD", "CD"],
|
||||
"joint_count": 3,
|
||||
"geometry": {"base": [[0, 0], [10, 0], [5, 7.071]], "units": "mm"},
|
||||
"unsupported_error": "authoritative_kinematics_unsupported",
|
||||
},
|
||||
"selection": {
|
||||
"family": "parallel_tripod",
|
||||
"basis": [
|
||||
"S6-03 authoritative CoreXY report is verified and supplies the fixed-tick planner/recovery contract",
|
||||
"tripod-kinematics report has verified implementation and passed native/WASM samples with a real tripod.ini/tripodsim.hal candidate",
|
||||
"Tripod uses the existing XYZ plus three-joint authoritative shape, so no frozen ABI/protocol/schema expansion is required",
|
||||
],
|
||||
"not_promoted": ["TRT", "serial_robot", "Delta", "Hexapod", "Pentapod", "switch", "user", "official_oracle"],
|
||||
},
|
||||
"worker": {"execution": no_output(worker_result), "sample": worker},
|
||||
"native_wasm": native_wasm,
|
||||
"ctest": no_output(ctest),
|
||||
"inputs": [artifact(root, path) for path in INPUTS],
|
||||
"boundaries": [
|
||||
"TRT, serial robot, Delta, Hexapod, Pentapod and switch/user families remain queued for later slices",
|
||||
"official LinuxCNC kinematics oracle remains SOURCE_BLOCKED",
|
||||
"tripod native/WASM golden is implementation evidence, not official oracle evidence",
|
||||
],
|
||||
}
|
||||
report["report_sha256"] = digest(canonical({key: value for key, value in report.items() if key != "report_sha256"}).encode())
|
||||
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
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
|
||||
parser.add_argument("--report", type=Path, default=Path("coverage/authoritative-kinematics-s6-04.report.json"))
|
||||
args = parser.parse_args()
|
||||
root = args.repo_root.resolve()
|
||||
output = args.report if args.report.is_absolute() else root / args.report
|
||||
report = generate(root, output)
|
||||
print(f"authoritative_kinematics_s6_04={report['status']} family={report['scope']['selected_family']}")
|
||||
return 0 if report["status"] == "verified" else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user