237 lines
14 KiB
Python
237 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate the K0 kinematics configuration graph and oracle sample definitions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
|
|
|
|
HAL_KEYS = {"HALFILE", "POSTGUI_HALFILE", "POSTTASK_HALFILE"}
|
|
COMPONENT_PATHS = {
|
|
"corexy_by_hal": "linuxcnc/src/hal/components/corexy_by_hal.comp",
|
|
"matrixkins": "linuxcnc/src/hal/components/matrixkins.comp",
|
|
"millturn": "linuxcnc/src/hal/components/millturn.comp",
|
|
"userkins": "linuxcnc/src/hal/components/userkins.comp",
|
|
"xyzab_tdr_kins": "linuxcnc/src/hal/components/xyzab_tdr_kins.comp",
|
|
}
|
|
SAMPLE_INPUTS = {
|
|
"cartesian": {"joint": [10.0, -5.0, 2.5]},
|
|
"coupled_linear": {"joint": [12.0, 4.0, 3.0]},
|
|
"five_axis": {"joint": [10.0, 20.0, 30.0, 15.0, -20.0]},
|
|
"five_axis_trt": {"joint": [100.0, 50.0, 200.0, 30.0, 45.0]},
|
|
"parallel_delta": {"pose": {"tran": [5.0, -3.0, -250.0], "abcuvw": [0.0] * 6}},
|
|
"parallel_tripod": {"pose": {"tran": [1.0, 2.0, 100.0], "abcuvw": [2.0, -1.0, 0.0, 0.0, 0.0, 0.0]}},
|
|
"parallel_hexapod": {"pose": {"tran": [0.0, 0.0, 100.0], "abcuvw": [1.0, -1.0, 2.0, 0.0, 0.0, 0.0]}},
|
|
"parallel_pentapod": {"pose": {"tran": [0.0, 0.0, 75.0], "abcuvw": [1.0, 2.0, 0.0, 0.0, 0.0, 0.0]}},
|
|
"serial_robot": {"joint": [0.0, -30.0, 45.0, 10.0, 20.0, -15.0]},
|
|
"switchable": {"joint": [5.0, 6.0, 7.0, 30.0], "switch_sequence": [0, 1, 0]},
|
|
"user_static": {"pose": {"tran": [10.0, 20.0, 30.0], "abcuvw": [0.0] * 6}},
|
|
}
|
|
|
|
|
|
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 stable_hash(rows: Iterable[dict[str, object]]) -> str:
|
|
data = "".join(json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n" for row in rows)
|
|
return hashlib.sha256(data.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def read_jsonl(path: Path) -> list[dict[str, object]]:
|
|
if not path.is_file():
|
|
return []
|
|
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
|
|
|
|
|
|
def relative(root: Path, path: Path) -> str:
|
|
return path.resolve().relative_to(root.resolve()).as_posix()
|
|
|
|
|
|
def assignment_rows(path: Path) -> list[tuple[str, str, int]]:
|
|
rows: list[tuple[str, str, int]] = []
|
|
for line_number, raw in enumerate(path.read_text(encoding="utf-8", errors="strict").splitlines(), 1):
|
|
line = raw.split("#", 1)[0].split(";", 1)[0].strip()
|
|
if not line or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
rows.append((key.strip().upper(), value.strip().strip('"').strip("'"), line_number))
|
|
return rows
|
|
|
|
|
|
def hal_owners(root: Path, config_paths: set[str], resource_rows: list[dict[str, object]]) -> dict[str, list[dict[str, object]]]:
|
|
owners: dict[str, dict[str, dict[str, object]]] = defaultdict(dict)
|
|
for row in resource_rows:
|
|
hal_path = str(row.get("path", ""))
|
|
entrypoint = str(row.get("entrypoint", ""))
|
|
if row.get("resource_kind") == "hal" and row.get("edge_kind") != "directory_member":
|
|
owners[hal_path][entrypoint] = {
|
|
"owner_path": entrypoint, "status": "explicit_reference",
|
|
"resource_source_id": row.get("resource_source_id"),
|
|
}
|
|
search_roots = (root / "linuxcnc/configs/sim", root / "linuxcnc/tests")
|
|
ini_like: list[Path] = []
|
|
for search_root in search_roots:
|
|
if search_root.is_dir():
|
|
ini_like.extend(path for path in search_root.rglob("*") if path.is_file() and (path.suffix.lower() in {".ini", ".inc"} or ".ini." in path.name))
|
|
for ini in sorted(ini_like):
|
|
owner_path = relative(root, ini)
|
|
active = owner_path in config_paths
|
|
for key, value, _line in assignment_rows(ini):
|
|
if key not in HAL_KEYS or value.startswith("LIB:"):
|
|
continue
|
|
target = ini.parent / value.split()[0]
|
|
if not target.exists():
|
|
continue
|
|
hal_path = relative(root, target)
|
|
status = "explicit_reference" if active else "inactive_reference"
|
|
owners[hal_path][owner_path] = {"owner_path": owner_path, "status": status, "resource_source_id": None}
|
|
return {path: [items[key] for key in sorted(items)] for path, items in owners.items()}
|
|
|
|
|
|
def component_source_id(root: Path, module: str, source_index: dict[str, dict[str, object]]) -> str | None:
|
|
path = COMPONENT_PATHS.get(module.lower())
|
|
if path and (root / path).is_file():
|
|
row = source_index.get(path)
|
|
return str(row.get("source_id")) if row else f"source:{path}"
|
|
return None
|
|
|
|
|
|
def sample_rows(families: list[dict[str, object]], matrixkins_id: str | None) -> list[dict[str, object]]:
|
|
rows: list[dict[str, object]] = []
|
|
for family_row in sorted(families, key=lambda row: str(row["family"])):
|
|
family = str(family_row["family"])
|
|
candidates = list(dict.fromkeys(str(value) for value in family_row.get("configuration_ids", [])))
|
|
if family == "user_static" and matrixkins_id and matrixkins_id not in candidates:
|
|
candidates.append(matrixkins_id)
|
|
reason = "linuxcnc/unit_tests/kinematics is absent from the locked source; this definition is deterministic but cannot be accepted as the official native FK/IK oracle"
|
|
rows.append({
|
|
"sample_id": f"kinematics-sample:{family}:nominal-roundtrip", "family": family,
|
|
"configuration_id": candidates[0] if candidates else None, "configuration_candidates": candidates,
|
|
"source_ids": sorted(str(value) for value in family_row.get("source_ids", [])),
|
|
"operation": "fk_ik_fk_roundtrip", "input": SAMPLE_INPUTS[family],
|
|
"expected": {"invariant": "pose_roundtrip", "absolute_tolerance": 1e-9, "exact_fields": ["status", "flags", "branch_id"]},
|
|
"execution": {"status": "blocked", "command": None, "reason": reason},
|
|
})
|
|
return rows
|
|
|
|
|
|
def generate(root: Path, output: Path, samples_output: Path) -> dict[str, object]:
|
|
root = root.resolve()
|
|
closure_path = root / "coverage/kinematics-source-closure.report.json"
|
|
closure = json.loads(closure_path.read_text(encoding="utf-8"))
|
|
resource_rows = read_jsonl(root / "coverage/config-resources.jsonl")
|
|
source_index = {str(row.get("path")): row for row in read_jsonl(root / "coverage/source-index.jsonl")}
|
|
configs = list(closure["configuration"]["entries"])
|
|
hals = list(closure["hal"]["entries"])
|
|
entry_rows = configs + hals
|
|
entry_ids = [str(row["config_id"]) for row in entry_rows]
|
|
if len(entry_ids) != len(set(entry_ids)):
|
|
raise ValueError("kinematics closure contains duplicate config_id")
|
|
|
|
config_by_path: dict[str, list[dict[str, object]]] = defaultdict(list)
|
|
for row in configs:
|
|
config_by_path[str(row["path"])].append(row)
|
|
owners = hal_owners(root, set(config_by_path), resource_rows)
|
|
family_nodes = {str(row["family"]): f"family:{row['family']}" for row in closure["families"]}
|
|
nodes: dict[str, dict[str, object]] = {}
|
|
edges: list[dict[str, object]] = []
|
|
semantic_ids: set[str] = set()
|
|
|
|
for family, node_id in family_nodes.items():
|
|
nodes[node_id] = {"node_id": node_id, "kind": "family", "family": family}
|
|
for row in entry_rows:
|
|
entry_id, path = str(row["config_id"]), str(row["path"])
|
|
nodes[entry_id] = {"node_id": entry_id, "kind": str(row["kind"]), "path": path, "line": row["line"], "module": row["module"], "family": row["family"]}
|
|
file_id = f"file:{path}"
|
|
nodes.setdefault(file_id, {"node_id": file_id, "kind": "file", "path": path})
|
|
edges.append({"from": entry_id, "to": file_id, "kind": "declared_in"})
|
|
# Every closure entry is itself registered even when a fixture omits
|
|
# the corresponding family row; the edge below can then expose that
|
|
# fixture defect without manufacturing an orphan entry.
|
|
semantic_ids.add(entry_id)
|
|
family = str(row["family"])
|
|
if family in family_nodes:
|
|
edges.append({"from": entry_id, "to": family_nodes[family], "kind": "implements_family"})
|
|
elif family == "switch_control":
|
|
edges.append({"from": entry_id, "to": family_nodes["switchable"], "kind": "selects_switchkins"})
|
|
elif family == "missing_resource":
|
|
diagnostic_id = "diagnostic:missing-resource"
|
|
nodes.setdefault(diagnostic_id, {"node_id": diagnostic_id, "kind": "diagnostic"})
|
|
edges.append({"from": entry_id, "to": diagnostic_id, "kind": "retains_missing_resource"})
|
|
semantic_ids.add(entry_id)
|
|
elif family == "template_entry":
|
|
owner_rows = owners.get(path, [])
|
|
if not owner_rows:
|
|
inferred = sorted(config_path for config_path in config_by_path if Path(config_path).parent == Path(path).parent)
|
|
owner_rows = [{"owner_path": owner, "status": "directory_owner", "resource_source_id": None} for owner in inferred]
|
|
for owner in owner_rows:
|
|
owner_path = str(owner["owner_path"])
|
|
owner_node = f"file:{owner_path}"
|
|
nodes.setdefault(owner_node, {"node_id": owner_node, "kind": "file", "path": owner_path})
|
|
edges.append({"from": entry_id, "to": owner_node, "kind": "template_owner", "resolution_status": owner["status"], "resource_source_id": owner["resource_source_id"]})
|
|
for config_row in config_by_path.get(owner_path, []):
|
|
edges.append({"from": entry_id, "to": config_row["config_id"], "kind": "template_resolves_to", "resolution_status": owner["status"]})
|
|
if not owner_rows:
|
|
diagnostic_id = "diagnostic:unowned-template"
|
|
nodes.setdefault(diagnostic_id, {"node_id": diagnostic_id, "kind": "diagnostic"})
|
|
edges.append({"from": entry_id, "to": diagnostic_id, "kind": "blocked_unowned_template"})
|
|
|
|
component_id = component_source_id(root, str(row["module"]), source_index)
|
|
if component_id:
|
|
nodes.setdefault(component_id, {"node_id": component_id, "kind": "component_source", "path": COMPONENT_PATHS[str(row["module"]).lower()]})
|
|
edges.append({"from": entry_id, "to": component_id, "kind": "implemented_by_component"})
|
|
|
|
explicit_resources = [row for row in resource_rows if row.get("edge_kind") != "directory_member" and str(row.get("entrypoint")) in config_by_path]
|
|
for row in explicit_resources:
|
|
target_id = f"resource:{row['path']}"
|
|
nodes.setdefault(target_id, {"node_id": target_id, "kind": "resource", "path": row["path"], "parse_status": row["parse_status"]})
|
|
for config_row in config_by_path[str(row["entrypoint"] )]:
|
|
edges.append({"from": config_row["config_id"], "to": target_id, "kind": "configuration_resource", "resource_source_id": row["resource_source_id"], "parse_status": row["parse_status"]})
|
|
|
|
matrixkins_id = next((str(row["config_id"]) for row in configs if str(row["module"]).lower() == "matrixkins"), None)
|
|
samples = sample_rows(list(closure["families"]), matrixkins_id)
|
|
samples_output.parent.mkdir(parents=True, exist_ok=True)
|
|
samples_output.write_text("".join(json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n" for row in samples), encoding="utf-8")
|
|
|
|
orphan_ids = sorted(set(entry_ids) - semantic_ids)
|
|
unowned_templates = sorted(row["config_id"] for row in hals if row["family"] == "template_entry" and not any(edge["from"] == row["config_id"] and edge["kind"] == "template_owner" for edge in edges))
|
|
samples_without_config = sorted(row["sample_id"] for row in samples if not row["configuration_id"])
|
|
report: dict[str, object] = {
|
|
"schema": "kinematics-config-graph-report.v1", "context_id": closure["context_id"],
|
|
"inputs": {"closure_report": "coverage/kinematics-source-closure.report.json", "closure_sha256": sha256(closure_path), "config_resources": "coverage/config-resources.jsonl"},
|
|
"counts": {"entry_nodes": len(entry_ids), "configuration_entries": len(configs), "hal_entries": len(hals), "family_nodes": len(family_nodes), "nodes": len(nodes), "edges": len(edges), "samples": len(samples)},
|
|
"nodes": [nodes[key] for key in sorted(nodes)], "edges": sorted(edges, key=lambda row: (str(row["from"]), str(row["kind"]), str(row["to"]))),
|
|
"samples": {"path": relative(root, samples_output), "sha256": sha256(samples_output), "sample_ids": [row["sample_id"] for row in samples]},
|
|
"closure": {"expected_entry_nodes": int(closure["configuration"]["entry_count"]) + int(closure["hal"]["entry_count"]), "all_entry_ids_registered": len(entry_ids) == len(set(entry_ids)), "orphan_entry_ids": orphan_ids, "unowned_template_ids": unowned_templates, "samples_without_configuration": samples_without_config, "official_oracle_status": "blocked", "official_oracle_reason": "linuxcnc/unit_tests/kinematics is absent from the locked source"},
|
|
}
|
|
report["graph_fingerprint_sha256"] = stable_hash(report["nodes"] + report["edges"])
|
|
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-config-graph.report.json"))
|
|
parser.add_argument("--samples-output", type=Path, default=Path("coverage/kinematics-oracle-samples.jsonl"))
|
|
args = parser.parse_args()
|
|
report = generate(args.repo_root, args.repo_root / args.output, args.repo_root / args.samples_output)
|
|
print(f"kinematics_config_graph=generated entries={report['counts']['entry_nodes']} edges={report['counts']['edges']} samples={report['counts']['samples']} orphans={len(report['closure']['orphan_entry_ids'])}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|