112 lines
6.3 KiB
Python
112 lines
6.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Run the native kinematics smoke and emit strict FK/IK oracle readiness evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
|
|
SCHEMA = "kinematics-oracle-report.v1"
|
|
GOLDEN_SCHEMA = "kinematics-golden.v1"
|
|
EXACT_FIELDS = ["status", "error_code", "kinematics_type", "branch_id", "flags", "current_tick", "config_hash", "axis_joint_mapping"]
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for block in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(block)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def sha256_bytes(value: bytes) -> str:
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
|
|
def fixed_env(root: Path) -> dict[str, str]:
|
|
return {
|
|
"PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
|
|
"HOME": ".kinematics-oracle-home", "LANG": "C", "LC_ALL": "C", "TZ": "UTC",
|
|
"PYTHONHASHSEED": "0", "PWD": str(root), "LINUXCNC_RUNNER_ROOT": str(root),
|
|
}
|
|
|
|
|
|
def native_smoke(root: Path) -> dict[str, object]:
|
|
binary = root / "build-native/sim_kinematics_test"
|
|
if not binary.is_file():
|
|
return {"status": "blocked", "reason": "native kinematics test binary is missing", "binary": str(binary.relative_to(root))}
|
|
try:
|
|
completed = subprocess.run([str(binary)], cwd=root, env=fixed_env(root), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30, check=False)
|
|
except subprocess.TimeoutExpired:
|
|
return {"status": "timeout", "binary": str(binary.relative_to(root))}
|
|
return {
|
|
"status": "passed" if completed.returncode == 0 else "failed", "binary": str(binary.relative_to(root)),
|
|
"exit_code": completed.returncode, "stdout_sha256": sha256_bytes(completed.stdout),
|
|
"stderr_sha256": sha256_bytes(completed.stderr), "result_sha256": sha256_bytes(completed.stdout + b"\0" + completed.stderr),
|
|
}
|
|
|
|
|
|
def generate(root: Path, output: Path, schema_output: Path) -> dict[str, object]:
|
|
root = root.resolve()
|
|
samples_path = root / "coverage/kinematics-oracle-samples.jsonl"
|
|
closure_path = root / "coverage/kinematics-source-closure.report.json"
|
|
samples = [json.loads(line) for line in samples_path.read_text(encoding="utf-8").splitlines() if line.strip()]
|
|
closure = json.loads(closure_path.read_text(encoding="utf-8"))
|
|
missing_roots = list(closure.get("missing_required_test_roots", []))
|
|
smoke = native_smoke(root)
|
|
schema = {
|
|
"$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "linuxcnc-web-wasm/kinematics-golden.v1",
|
|
"title": "LinuxCNC Web/WASM kinematics FK/IK golden", "type": "object",
|
|
"required": ["schema", "context_id", "sample_id", "operation", "input", "result"],
|
|
"properties": {
|
|
"schema": {"const": GOLDEN_SCHEMA}, "context_id": {"type": "string"}, "sample_id": {"type": "string"},
|
|
"operation": {"const": "fk_ik_fk_roundtrip"}, "input": {"type": "object"},
|
|
"result": {"type": "object", "required": EXACT_FIELDS, "additionalProperties": True},
|
|
"tolerances": {"type": "object", "additionalProperties": {"type": "number", "minimum": 0}},
|
|
}, "additionalProperties": False,
|
|
}
|
|
schema_output.parent.mkdir(parents=True, exist_ok=True)
|
|
schema_output.write_text(json.dumps(schema, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8")
|
|
records: list[dict[str, object]] = []
|
|
for sample in sorted(samples, key=lambda row: str(row["sample_id"])):
|
|
reason = str(sample.get("execution", {}).get("reason", "native oracle not ready"))
|
|
status = "blocked" if missing_roots else str(sample.get("execution", {}).get("status", "blocked"))
|
|
records.append({
|
|
"schema": GOLDEN_SCHEMA, "context_id": closure.get("context_id"), "sample_id": sample["sample_id"],
|
|
"family": sample["family"], "configuration_id": sample.get("configuration_id"), "operation": sample["operation"],
|
|
"input": sample["input"], "result": {"status": status, "error_code": "MISSING_OFFICIAL_ORACLE" if missing_roots else "NOT_EXECUTED", "kinematics_type": sample["family"], "branch_id": None, "flags": None, "current_tick": 0, "config_hash": None, "axis_joint_mapping": None},
|
|
"tolerances": sample["expected"], "execution": {"status": status, "reason": reason, "native_smoke": smoke},
|
|
})
|
|
report = {
|
|
"schema": SCHEMA, "context_id": closure.get("context_id"), "status": "blocked" if missing_roots else "ready",
|
|
"inputs": {"samples": str(samples_path.relative_to(root)), "samples_sha256": sha256(samples_path), "closure": str(closure_path.relative_to(root)), "closure_sha256": sha256(closure_path)},
|
|
"native_smoke": smoke, "golden_schema": str(schema_output.relative_to(root)), "golden_schema_sha256": sha256(schema_output),
|
|
"sample_count": len(records), "ready_count": sum(1 for row in records if row["execution"]["status"] == "ready"),
|
|
"blocked_count": sum(1 for row in records if row["execution"]["status"] == "blocked"), "records": records,
|
|
"blocking_reasons": sorted(set(missing_roots)) if missing_roots else [],
|
|
}
|
|
report["report_fingerprint_sha256"] = hashlib.sha256(json.dumps(report, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()
|
|
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("--output", type=Path, default=Path("coverage/kinematics-oracle.report.json"))
|
|
parser.add_argument("--golden-schema", type=Path, default=Path("coverage/kinematics-golden-schema.json"))
|
|
args = parser.parse_args()
|
|
report = generate(args.repo_root, args.repo_root / args.output, args.repo_root / args.golden_schema)
|
|
print(f"kinematics_oracle={report['status']} samples={report['sample_count']} ready={report['ready_count']} blocked={report['blocked_count']} native_smoke={report['native_smoke']['status']}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|