wangdequan/linuxcncWebWASM
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement WebAssembly runtime and kinematics coverage

Browse Source
This commit is contained in:
wangdequan
2026-08-21 06:04:48 +08:00
parent 5dd614c335
commit 554b72dae8
112 changed files with 47148 additions and 807 deletions
Download Patch File Download Diff File Expand all files Collapse all files

236
scripts/generate_kinematics_config_graph.py Normal file
View File

@@ -0,0 +1,236 @@
#!/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())