Extend simulation runtime coverage and kinematics support
This commit is contained in:
415
scripts/generate_configuration_regression.py
Normal file
415
scripts/generate_configuration_regression.py
Normal file
@@ -0,0 +1,415 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generate the deterministic K8 full-configuration regression report.
|
||||
|
||||
The LinuxCNC checkout used by this project is intentionally a source/config
|
||||
fixture rather than a complete desktop installation. This report therefore
|
||||
separates a deterministic static startup preflight from the native/WASM and
|
||||
Worker runtime probes. Missing resources and unavailable official oracles
|
||||
remain explicit blockers; they are never converted into passing goldens.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable
|
||||
|
||||
|
||||
CONTEXT_PATH = Path("context-lock.json")
|
||||
GRAPH_PATH = Path("coverage/kinematics-config-graph.report.json")
|
||||
RESOURCE_PATH = Path("coverage/config-resources.jsonl")
|
||||
RESOURCE_REPORT_PATH = Path("coverage/config-resources.report.json")
|
||||
RECOVERY_PATH = Path("coverage/workspace-recovery.report.json")
|
||||
REGISTRY_PATH = Path("coverage/kinematics-registry.report.json")
|
||||
|
||||
RUNTIME_COMMANDS: tuple[tuple[str, tuple[str, ...]], ...] = (
|
||||
("kinematics", ("build-native/sim_kinematics_test",)),
|
||||
("motion", ("build-native/sim_kinematics_motion_test",)),
|
||||
("feedback", ("build-native/sim_kinematics_feedback_test",)),
|
||||
("home_limit", ("build-native/sim_kinematics_home_limit_test",)),
|
||||
("io", ("build-native/sim_io_test",)),
|
||||
)
|
||||
WASM_COMMANDS: tuple[tuple[str, tuple[str, ...]], ...] = (
|
||||
("kinematics", ("node", "build-wasm/sim_kinematics_wasm_test.js")),
|
||||
("motion", ("node", "build-wasm/sim_kinematics_motion_wasm_test.js")),
|
||||
("feedback", ("node", "build-wasm/sim_kinematics_feedback_wasm_test.js")),
|
||||
("home_limit", ("node", "build-wasm/sim_kinematics_home_limit_wasm_test.js")),
|
||||
("io", ("node", "build-wasm/sim_io_wasm_test.js")),
|
||||
)
|
||||
WORKER_COMMANDS: tuple[tuple[str, tuple[str, ...]], ...] = (
|
||||
("replay", ("node", "web/worker/test_replay.mjs")),
|
||||
("workspace_recovery", ("node", "web/workspace/test_recovery.mjs")),
|
||||
("sim_session", ("node", "web/worker/test_sim_session.mjs")),
|
||||
)
|
||||
UNRESOLVED_STATUSES = {"unresolved", "host_or_variable_path", "outside_repo"}
|
||||
KINEMATICS_SAMPLE_KEYS = {
|
||||
"corexykins": "corexy",
|
||||
"rosekins": "rose",
|
||||
"maxkins": "maxkins",
|
||||
"rotatekins": "rotatekins",
|
||||
"5axiskins": "fiveaxiskins",
|
||||
"xyzac-trt-kins": "xyzac_trt",
|
||||
"xyzbc-trt-kins": "xyzbc_trt",
|
||||
"scarakins": "scara",
|
||||
"pumakins": "puma",
|
||||
"scorbot-kins": "scorbot",
|
||||
"genserkins": "genser",
|
||||
"lineardeltakins": "lineardelta",
|
||||
"rotarydeltakins": "rotarydelta",
|
||||
"tripodkins": "tripod",
|
||||
"genhexkins": "genhex",
|
||||
"pentakins": "pentapod",
|
||||
}
|
||||
|
||||
|
||||
def canonical(value: Any) -> str:
|
||||
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||||
|
||||
|
||||
def digest(value: Any) -> str:
|
||||
return hashlib.sha256(canonical(value).encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def sha256(path: Path) -> str:
|
||||
handle = hashlib.sha256()
|
||||
with path.open("rb") as stream:
|
||||
for block in iter(lambda: stream.read(1024 * 1024), b""):
|
||||
handle.update(block)
|
||||
return handle.hexdigest()
|
||||
|
||||
|
||||
def read_json(path: Path) -> dict[str, Any]:
|
||||
value = json.loads(path.read_text(encoding="utf-8"))
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError(f"{path} must contain an object")
|
||||
return value
|
||||
|
||||
|
||||
def read_jsonl(path: Path) -> list[dict[str, Any]]:
|
||||
rows: list[dict[str, Any]] = []
|
||||
for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
||||
if not line.strip():
|
||||
continue
|
||||
value = json.loads(line)
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError(f"{path}:{line_number} must contain an object")
|
||||
rows.append(value)
|
||||
return rows
|
||||
|
||||
|
||||
def source_record(root: Path, path: Path) -> dict[str, Any]:
|
||||
relative = path.relative_to(root).as_posix()
|
||||
return {"path": relative, "sha256": sha256(path)} if path.is_file() else {"path": relative, "sha256": None, "status": "missing"}
|
||||
|
||||
|
||||
def parse_ini_shape(path: Path) -> dict[str, Any]:
|
||||
"""Check only syntax needed by the preflight; it is not an INI oracle."""
|
||||
if not path.is_file():
|
||||
return {"file": False, "sections": [], "balanced_sections": False}
|
||||
sections: list[str] = []
|
||||
balanced = True
|
||||
for raw in path.read_text(encoding="utf-8", errors="strict").splitlines():
|
||||
line = raw.split("#", 1)[0].split(";", 1)[0].strip()
|
||||
if not line:
|
||||
continue
|
||||
if line.startswith("["):
|
||||
if not line.endswith("]"):
|
||||
balanced = False
|
||||
else:
|
||||
sections.append(line[1:-1].strip().upper())
|
||||
return {"file": True, "sections": sorted(set(sections)), "balanced_sections": balanced}
|
||||
|
||||
|
||||
def registered_kinematics(root: Path) -> set[str]:
|
||||
registry = read_json(REGISTRY_PATH if REGISTRY_PATH.is_absolute() else root / REGISTRY_PATH)
|
||||
rows = registry.get("samples", {}).get("native", {}).get("entries", [])
|
||||
return {str(row.get("name")) for row in rows if row.get("name")}
|
||||
|
||||
|
||||
def graph_entries(root: Path) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
||||
graph = read_json(root / GRAPH_PATH)
|
||||
entries = [row for row in graph.get("nodes", []) if row.get("kind") == "kinematics"]
|
||||
entries.sort(key=lambda row: str(row.get("node_id", "")))
|
||||
return graph, entries
|
||||
|
||||
|
||||
def resource_summaries(root: Path, resource_report: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
|
||||
by_entrypoint: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
||||
for row in rows:
|
||||
by_entrypoint[str(row.get("entrypoint", ""))].append(row)
|
||||
indexed = {str(path) for path in resource_report.get("entrypoint_paths", [])}
|
||||
result: dict[str, list[dict[str, Any]]] = {}
|
||||
for path in sorted(indexed):
|
||||
result[path] = by_entrypoint.get(path, [])
|
||||
return result
|
||||
|
||||
|
||||
def preflight_entry(root: Path, row: dict[str, Any], resources: dict[str, list[dict[str, Any]]], indexed_paths: set[str], registered: set[str]) -> dict[str, Any]:
|
||||
path = str(row["path"])
|
||||
module = str(row.get("module", ""))
|
||||
resource_rows = resources.get(path, [])
|
||||
missing = sorted({str(item.get("path")) for item in resource_rows if item.get("parse_status") == "missing"})
|
||||
unresolved = sorted({str(item.get("path")) for item in resource_rows if item.get("parse_status") in UNRESOLVED_STATUSES})
|
||||
ini_shape = parse_ini_shape(root / path)
|
||||
if module not in registered:
|
||||
startup_status = "blocked_unregistered_kinematics"
|
||||
elif path not in indexed_paths and path.startswith("linuxcnc/configs/sim/"):
|
||||
startup_status = "blocked_unindexed_resource_graph"
|
||||
elif missing or unresolved:
|
||||
startup_status = "blocked_resource"
|
||||
elif not ini_shape["file"] or not ini_shape["balanced_sections"]:
|
||||
startup_status = "blocked_ini_syntax"
|
||||
else:
|
||||
startup_status = "static_ready"
|
||||
if path not in indexed_paths:
|
||||
resource_status = "unindexed"
|
||||
elif missing and unresolved:
|
||||
resource_status = "blocked_missing_and_unresolved"
|
||||
elif missing:
|
||||
resource_status = "blocked_missing"
|
||||
elif unresolved:
|
||||
resource_status = "blocked_unresolved"
|
||||
else:
|
||||
resource_status = "ready"
|
||||
return {
|
||||
"config_id": row.get("node_id"),
|
||||
"path": path,
|
||||
"line": row.get("line"),
|
||||
"module": module,
|
||||
"family": row.get("family"),
|
||||
"resource_status": resource_status,
|
||||
"missing_resources": missing,
|
||||
"unresolved_resources": unresolved,
|
||||
"resource_rows": len(resource_rows),
|
||||
"ini": ini_shape,
|
||||
"kinematics_registered": module in registered,
|
||||
"startup_preflight": startup_status,
|
||||
}
|
||||
|
||||
|
||||
def run_json_command(root: Path, command: tuple[str, ...]) -> dict[str, Any]:
|
||||
executable = root / command[0] if command[0] not in {"node"} else root / command[1]
|
||||
if not executable.exists():
|
||||
return {"status": "blocked", "reason": f"missing_runtime_artifact:{executable.relative_to(root).as_posix()}"}
|
||||
completed = subprocess.run(command, cwd=root, check=False, text=True, capture_output=True)
|
||||
if completed.returncode != 0:
|
||||
return {"status": "failed", "exit_code": completed.returncode, "stderr_sha256": digest(completed.stderr)}
|
||||
json_lines = [line.strip() for line in completed.stdout.splitlines() if line.lstrip().startswith("{")]
|
||||
if not json_lines:
|
||||
return {"status": "failed", "reason": "runtime_did_not_emit_json", "stdout_sha256": digest(completed.stdout)}
|
||||
try:
|
||||
sample = json.loads(json_lines[-1])
|
||||
except json.JSONDecodeError:
|
||||
return {"status": "failed", "reason": "runtime_emitted_invalid_json", "stdout_sha256": digest(completed.stdout)}
|
||||
return {"status": "passed", "schema": sample.get("schema"), "sample": sample, "sample_sha256": digest(sample)}
|
||||
|
||||
|
||||
def run_runtime_group(root: Path, commands: Iterable[tuple[str, tuple[str, ...]]]) -> dict[str, Any]:
|
||||
records: dict[str, Any] = {}
|
||||
for name, command in commands:
|
||||
first = run_json_command(root, command)
|
||||
second = run_json_command(root, command)
|
||||
first_key = first.get("sample_sha256")
|
||||
second_key = second.get("sample_sha256")
|
||||
records[name] = {
|
||||
"command": list(command),
|
||||
"first": first,
|
||||
"second": second,
|
||||
"repeat_stable": bool(first_key and first_key == second_key),
|
||||
}
|
||||
passed = all(row["first"].get("status") == "passed" and row["repeat_stable"] for row in records.values())
|
||||
return {"status": "passed" if passed else "blocked", "records": records, "summary_sha256": digest(records)}
|
||||
|
||||
|
||||
def run_worker_group(root: Path) -> dict[str, Any]:
|
||||
records: dict[str, Any] = {}
|
||||
for name, command in WORKER_COMMANDS:
|
||||
target = root / command[-1]
|
||||
if not target.is_file():
|
||||
records[name] = {"command": list(command), "status": "blocked", "reason": "missing_worker_test"}
|
||||
continue
|
||||
first = subprocess.run(command, cwd=root, check=False, text=True, capture_output=True)
|
||||
second = subprocess.run(command, cwd=root, check=False, text=True, capture_output=True)
|
||||
first_value = {"exit_code": first.returncode, "stdout": first.stdout, "stderr": first.stderr}
|
||||
second_value = {"exit_code": second.returncode, "stdout": second.stdout, "stderr": second.stderr}
|
||||
records[name] = {
|
||||
"command": list(command),
|
||||
"status": "passed" if first.returncode == 0 else "failed",
|
||||
"first_sha256": digest(first_value),
|
||||
"second_sha256": digest(second_value),
|
||||
"repeat_stable": digest(first_value) == digest(second_value),
|
||||
"stdout": first.stdout.strip(),
|
||||
}
|
||||
passed = all(value.get("status") == "passed" and value.get("repeat_stable") for value in records.values())
|
||||
return {"status": "passed" if passed else "blocked", "records": records, "summary_sha256": digest(records)}
|
||||
|
||||
|
||||
def extract_state_hashes(value: Any, path: str = "") -> list[dict[str, Any]]:
|
||||
found: list[dict[str, Any]] = []
|
||||
if isinstance(value, dict):
|
||||
for key, child in sorted(value.items()):
|
||||
child_path = f"{path}.{key}" if path else key
|
||||
if key in {"state_hash", "hash", "repeat_hash"} and isinstance(child, (int, str)):
|
||||
found.append({"path": child_path, "value": child})
|
||||
found.extend(extract_state_hashes(child, child_path))
|
||||
elif isinstance(value, list):
|
||||
for index, child in enumerate(value):
|
||||
found.extend(extract_state_hashes(child, f"{path}[{index}]"))
|
||||
return found
|
||||
|
||||
|
||||
def build_config_state_summary(entries: list[dict[str, Any]], runtime: dict[str, Any], registry: set[str]) -> dict[str, Any]:
|
||||
samples: dict[str, Any] = {}
|
||||
for name, record in runtime.get("records", {}).items():
|
||||
sample = record.get("first", {}).get("sample")
|
||||
if isinstance(sample, dict):
|
||||
samples[name] = {
|
||||
"schema": sample.get("schema"),
|
||||
"sample_sha256": record["first"].get("sample_sha256"),
|
||||
"state_hashes": extract_state_hashes(sample),
|
||||
}
|
||||
rows = []
|
||||
for entry in entries:
|
||||
module = str(entry.get("module", ""))
|
||||
sample_key = KINEMATICS_SAMPLE_KEYS.get(module)
|
||||
rows.append({
|
||||
"config_id": entry.get("config_id"),
|
||||
"module": module,
|
||||
"sample": sample_key,
|
||||
"sample_status": "verified_sample" if sample_key in samples else "registry_only" if module in registry else "blocked_unregistered",
|
||||
"state_digest": digest({"module": module, "sample": samples.get(sample_key), "registry": module in registry}),
|
||||
})
|
||||
rows.sort(key=lambda row: str(row["config_id"]))
|
||||
return {"tick_policy": "fixed logical tick; wall clock excluded", "entries": rows, "runtime_samples": samples, "summary_sha256": digest(rows)}
|
||||
|
||||
|
||||
def generate(root: Path, output: Path) -> dict[str, Any]:
|
||||
root = root.resolve()
|
||||
context = read_json(root / CONTEXT_PATH)
|
||||
graph, entries = graph_entries(root)
|
||||
resource_report = read_json(root / RESOURCE_REPORT_PATH)
|
||||
resources_rows = read_jsonl(root / RESOURCE_PATH)
|
||||
resources = resource_summaries(root, resource_report, resources_rows)
|
||||
indexed_paths = set(resources)
|
||||
registered = registered_kinematics(root)
|
||||
preflight = [preflight_entry(root, row, resources, indexed_paths, registered) for row in entries]
|
||||
preflight.sort(key=lambda row: str(row["config_id"]))
|
||||
runtime = run_runtime_group(root, RUNTIME_COMMANDS)
|
||||
wasm = run_runtime_group(root, WASM_COMMANDS)
|
||||
worker = run_worker_group(root)
|
||||
native_wasm_equal = all(
|
||||
runtime["records"].get(name, {}).get("first", {}).get("sample_sha256")
|
||||
== wasm["records"].get(name, {}).get("first", {}).get("sample_sha256")
|
||||
and runtime["records"].get(name, {}).get("first", {}).get("status") == "passed"
|
||||
for name, _command in RUNTIME_COMMANDS
|
||||
)
|
||||
graph_orphans = list(graph.get("closure", {}).get("orphan_entry_ids", []))
|
||||
unowned = list(graph.get("closure", {}).get("unowned_template_ids", []))
|
||||
resource_orphans = sorted({str(row.get("entrypoint")) for row in resources_rows if str(row.get("entrypoint")) not in indexed_paths})
|
||||
startup_counts: dict[str, int] = defaultdict(int)
|
||||
resource_counts: dict[str, int] = defaultdict(int)
|
||||
for row in preflight:
|
||||
startup_counts[str(row["startup_preflight"])] += 1
|
||||
resource_counts[str(row["resource_status"])] += 1
|
||||
missing_resource_paths = sorted({path for row in preflight for path in row["missing_resources"]})
|
||||
unresolved_resource_paths = sorted({path for row in preflight for path in row["unresolved_resources"]})
|
||||
recovery = read_json(root / RECOVERY_PATH)
|
||||
state_summary = build_config_state_summary(entries, runtime, registered)
|
||||
report: dict[str, Any] = {
|
||||
"schema": "configuration-regression-report.v1",
|
||||
"context_id": context.get("context_id"),
|
||||
"status": "partial",
|
||||
"implementation_status": "verified",
|
||||
"scope": {"configs_root": "linuxcnc/configs/sim", "tests_root": "linuxcnc/tests", "preflight_only": True},
|
||||
"inputs": {
|
||||
# The report is itself pinned by context-lock. Recording only the
|
||||
# context ID here avoids a circular lock/report content hash.
|
||||
"context_lock": {"path": CONTEXT_PATH.as_posix(), "context_id": context.get("context_id")},
|
||||
"kinematics_graph": source_record(root, root / GRAPH_PATH),
|
||||
"config_resources": source_record(root, root / RESOURCE_PATH),
|
||||
"config_resources_report": source_record(root, root / RESOURCE_REPORT_PATH),
|
||||
"workspace_recovery": source_record(root, root / RECOVERY_PATH),
|
||||
"kinematics_registry": source_record(root, root / REGISTRY_PATH),
|
||||
},
|
||||
"counts": {
|
||||
"registered_configuration_entries": len(entries),
|
||||
"config_entrypoints": int(resource_report.get("entrypoints", 0)),
|
||||
"resource_rows": len(resources_rows),
|
||||
"resource_missing_rows": sum(1 for row in resources_rows if row.get("parse_status") == "missing"),
|
||||
"resource_unresolved_rows": sum(1 for row in resources_rows if row.get("parse_status") in UNRESOLVED_STATUSES),
|
||||
"missing_resource_paths": len(missing_resource_paths),
|
||||
"unresolved_resource_paths": len(unresolved_resource_paths),
|
||||
"resource_orphans": len(resource_orphans),
|
||||
"graph_orphans": len(graph_orphans),
|
||||
"unowned_templates": len(unowned),
|
||||
"startup_preflight": dict(sorted(startup_counts.items())),
|
||||
"resource_status": dict(sorted(resource_counts.items())),
|
||||
},
|
||||
"configuration_entries": preflight,
|
||||
"closure": {
|
||||
"graph_fingerprint_sha256": graph.get("graph_fingerprint_sha256"),
|
||||
"graph_orphan_entry_ids": graph_orphans,
|
||||
"unowned_template_ids": unowned,
|
||||
"resource_orphan_entrypoints": resource_orphans,
|
||||
"all_registered": len(entries) == int(graph.get("counts", {}).get("configuration_entries", -1)),
|
||||
},
|
||||
"runtime": {
|
||||
"native": runtime,
|
||||
"wasm": wasm,
|
||||
"native_wasm": {"status": "passed" if native_wasm_equal else "blocked", "sample_hashes_equal": native_wasm_equal},
|
||||
"worker": worker,
|
||||
},
|
||||
"state_summary": state_summary,
|
||||
"recovery_summary": {
|
||||
"status": recovery.get("status"),
|
||||
"implementation_status": recovery.get("implementation_status"),
|
||||
"checkpoint_schema": recovery.get("replay", {}).get("schema"),
|
||||
"replay_state_hash": recovery.get("replay", {}).get("state_hash"),
|
||||
"continuous_state_hash": recovery.get("replay", {}).get("continuous_state_hash"),
|
||||
"endpoints_equal": recovery.get("replay", {}).get("endpoints_equal"),
|
||||
"oracle_status": recovery.get("oracle", {}).get("status"),
|
||||
},
|
||||
"oracle": {
|
||||
"status": "blocked",
|
||||
"reason": "official LinuxCNC kinematics/recovery oracle is absent from the locked source",
|
||||
"official_root": "linuxcnc/unit_tests/kinematics",
|
||||
"derived_configuration_results_are_not_official": True,
|
||||
},
|
||||
"unclosed": {
|
||||
"resource_missing": missing_resource_paths,
|
||||
"resource_unresolved": unresolved_resource_paths,
|
||||
"unregistered_kinematics": sorted({str(row["module"]) for row in preflight if not row["kinematics_registered"]}),
|
||||
"startup_blocked_entries": [row["config_id"] for row in preflight if str(row["startup_preflight"]).startswith("blocked")],
|
||||
},
|
||||
"report_sha256": None,
|
||||
}
|
||||
report["report_sha256"] = digest({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
|
||||
|
||||
|
||||
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/configuration-regression.report.json"))
|
||||
args = parser.parse_args()
|
||||
report = generate(args.repo_root, args.repo_root / args.output)
|
||||
print(
|
||||
"configuration_regression="
|
||||
f"entries={report['counts']['registered_configuration_entries']} "
|
||||
f"startup_blocked={len(report['unclosed']['startup_blocked_entries'])} "
|
||||
f"resource_orphans={report['counts']['resource_orphans']} "
|
||||
f"native_wasm={report['runtime']['native_wasm']['status']} "
|
||||
f"worker={report['runtime']['worker']['status']}"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user