#!/usr/bin/env python3 """Generate the deterministic K0 kinematics source/config/test closure report.""" from __future__ import annotations import argparse import hashlib import json import os import re from collections import defaultdict from pathlib import Path from typing import Iterable KINEMATICS_ROOT = Path("linuxcnc/src/emc/kinematics") CONFIG_ROOTS = (Path("linuxcnc/configs/sim"), Path("linuxcnc/tests")) TEST_ROOTS = (Path("linuxcnc/unit_tests/kinematics"), Path("linuxcnc/tests/matrixkins")) CONFIG_EXTENSIONS = {".ini", ".inc"} KINEMATICS_EXTENSIONS = {".c", ".cc", ".h"} KIN_LINE_RE = re.compile(r"^\s*KINEMATICS\s*=\s*([^#;\r\n]+)", re.IGNORECASE) LOADRT_RE = re.compile(r"\bloadrt\s+([^\s#;]+)", re.IGNORECASE) KINEMATIC_TOKEN_RE = re.compile(r"(?:kins|kinematics|corexy_by_hal|matrixkins|millturn|xyzab_tdr_kins)", re.IGNORECASE) SWITCHKINS_RE = re.compile(r"(?:switchkins|motion\.switchkins-type)", re.IGNORECASE) HAL_TRANSFORM_RE = re.compile(r"(?:corexy_by_hal|matrixkins|xyzab_tdr_kins)", re.IGNORECASE) FUNCTION_RE = re.compile(r"^\s*(?:int|void|static\s+int|static\s+void|KINEMATICS_TYPE)\s+([A-Za-z_][A-Za-z0-9_]*)\s*\(") FAMILY_BY_MODULE = { "trivkins": "cartesian", "maxkins": "cartesian", "rotatekins": "cartesian", "corexykins": "coupled_linear", "corexy_by_hal": "coupled_linear", "rosekins": "coupled_linear", "matrixkins": "coupled_linear", "5axiskins": "five_axis", "xyzac-trt-kins": "five_axis_trt", "xyzbc-trt-kins": "five_axis_trt", "xyzab_tdr_kins": "five_axis_trt", "trtfuncs": "five_axis_trt", "lineardeltakins": "parallel_delta", "rotarydeltakins": "parallel_delta", "tripodkins": "parallel_tripod", "genhexkins": "parallel_hexapod", "pentakins": "parallel_pentapod", "genserkins": "serial_robot", "genserfuncs": "serial_robot", "scarakins": "serial_robot", "pumakins": "serial_robot", "scorbot-kins": "serial_robot", "switchkins": "switchable", "millturn": "switchable", "ugenserkins": "user_static", "userkfuncs": "user_static", "userkins": "user_static", "cubic": "user_static", } SOURCE_FAMILY_BY_MODULE = { **FAMILY_BY_MODULE, "kinematics": "common_support", "kins_util": "common_support", "lineardeltakins-common": "parallel_delta", "rotarydeltakins-common": "parallel_delta", } FAMILY_DESCRIPTIONS = { "cartesian": "axis/joint identity and mapped Cartesian kinematics", "coupled_linear": "coupled linear or HAL matrix transforms", "five_axis": "generic five-axis kinematics", "five_axis_trt": "table/rotary-tilting five-axis kinematics", "parallel_delta": "linear and rotary Delta mechanisms", "parallel_tripod": "tripod parallel mechanism", "parallel_hexapod": "general hexapod parallel mechanism", "parallel_pentapod": "pentapod parallel mechanism", "serial_robot": "serial robot, SCARA, PUMA and DH-chain kinematics", "switchable": "switchkins and switchable HAL component kinematics", "user_static": "static user/template kinematics and calibrated components", } 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 relative(root: Path, path: Path) -> str: return path.resolve().relative_to(root.resolve()).as_posix() def fingerprint(rows: Iterable[dict[str, object]]) -> str: payload = "".join(json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n" for row in rows) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def read_jsonl(path: Path) -> list[dict[str, object]]: if not path.is_file(): return [] return [value for line in path.read_text(encoding="utf-8").splitlines() if line.strip() for value in [json.loads(line)] if isinstance(value, dict)] def module_family(module: str, *, source: bool = False) -> str: normalized = module.strip().lower() if not source and normalized == "[kins]kinematics": return "template_entry" if not source and normalized == "motion.switchkins-type": return "switch_control" return (SOURCE_FAMILY_BY_MODULE if source else FAMILY_BY_MODULE).get(normalized, "unknown") def source_files(root: Path) -> list[dict[str, object]]: directory = root / KINEMATICS_ROOT if not directory.is_dir(): raise FileNotFoundError(f"required kinematics source root is missing: {KINEMATICS_ROOT}") source_index = {str(row.get("path")): row for row in read_jsonl(root / "coverage/source-index.jsonl") if row.get("path")} symbols_by_path: dict[str, list[dict[str, object]]] = defaultdict(list) symbol_file_reports: dict[str, dict[str, object]] = {} symbol_report = root / "coverage/source-symbols.report.json" if symbol_report.is_file(): payload = json.loads(symbol_report.read_text(encoding="utf-8")) symbol_file_reports = { str(row.get("path")): row for row in payload.get("file_reports", []) if isinstance(row, dict) and row.get("path") } for row in read_jsonl(root / "coverage/source-symbols.jsonl"): path = str(row.get("path", "")) if path.startswith(f"{KINEMATICS_ROOT.as_posix()}/"): symbols_by_path[path].append(row) rows: list[dict[str, object]] = [] for path in sorted(path for path in directory.iterdir() if path.is_file()): relative_path = relative(root, path) index_row = source_index.get(relative_path, {}) symbols = sorted(symbols_by_path.get(relative_path, []), key=lambda row: str(row.get("symbol_source_id", ""))) module = path.stem rows.append({ "source_id": str(index_row.get("source_id") or f"source:{relative_path}"), "path": relative_path, "bytes": path.stat().st_size, "sha256": sha256(path), "module": module, "extension": path.suffix.lower(), "source_kind": "kinematics_implementation" if path.suffix.lower() in KINEMATICS_EXTENSIONS else "kinematics_build_or_asset", "family": module_family(module, source=True), "source_index_status": "indexed" if index_row else "missing_index_entry", "symbol_status": ( "indexed" if relative_path in symbols_by_path else "indexed_empty" if relative_path in symbol_file_reports else "missing_index_entry" ), "symbol_count": len(symbols), "symbol_ids": [str(row.get("symbol_source_id")) for row in symbols], "functions": sorted({match.group(1) for line in path.read_text(encoding="utf-8", errors="strict").splitlines() if (match := FUNCTION_RE.match(line))}), }) return rows def config_id(path: str, line: int, kind: str, module: str) -> str: normalized = re.sub(r"[^A-Za-z0-9_.-]+", "_", module.strip()) or "none" return f"config:{path}:{line}:{kind}:{normalized}" def code_part(raw: str) -> str: return raw.split("#", 1)[0].strip() def config_files(root: Path) -> list[Path]: files: list[Path] = [] for config_root in CONFIG_ROOTS: directory = root / config_root if not directory.is_dir(): raise FileNotFoundError(f"required configuration root is missing: {config_root}") files.extend(path for path in directory.rglob("*") if path.is_file() and path.suffix.lower() in CONFIG_EXTENSIONS) return sorted(set(files)) def config_entries(root: Path) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] for path in config_files(root): source_sha, relative_path = sha256(path), relative(root, path) for line_number, raw in enumerate(path.read_text(encoding="utf-8", errors="strict").splitlines(), 1): match = KIN_LINE_RE.match(code_part(raw)) if not match: continue value, module = match.group(1).strip(), match.group(1).strip().split()[0] rows.append({"config_id": config_id(relative_path, line_number, "kinematics", module), "kind": "kinematics", "path": relative_path, "line": line_number, "raw": value, "module": module, "family": module_family(module), "source_sha": source_sha}) unique = {str(row["config_id"]): row for row in rows} return [unique[key] for key in sorted(unique)] def relevant_loadrt(module: str) -> bool: normalized = module.strip().lower() return normalized == "[kins]kinematics" or bool(KINEMATIC_TOKEN_RE.search(normalized)) def hal_entries(root: Path) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] scan_extensions = CONFIG_EXTENSIONS | {".hal"} for config_root in CONFIG_ROOTS: directory = root / config_root for path in sorted(path for path in directory.rglob("*") if path.suffix.lower() in scan_extensions): relative_path = relative(root, path) if not path.is_file(): rows.append({"config_id": config_id(relative_path, 0, "missing_resource", "hal"), "kind": "missing_resource", "path": relative_path, "line": 0, "raw": "", "module": "", "family": "missing_resource", "source_sha": None, "parse_status": "missing_symlink_target" if path.is_symlink() else "missing", "symlink_target": os.readlink(path) if path.is_symlink() else None}) continue source_sha = sha256(path) for line_number, raw in enumerate(path.read_text(encoding="utf-8", errors="strict").splitlines(), 1): code = code_part(raw) if not code: continue if KIN_LINE_RE.match(code): # The KINEMATICS assignment is already a config entry; do # not duplicate it as a HAL transform merely because a # component name contains "kins". transform_code = "" else: transform_code = code load_match = LOADRT_RE.search(code) if load_match and relevant_loadrt(load_match.group(1)): module = load_match.group(1) rows.append({"config_id": config_id(relative_path, line_number, "loadrt", module), "kind": "loadrt", "path": relative_path, "line": line_number, "raw": code, "module": module, "family": module_family(module), "source_sha": source_sha}) for kind, pattern in (("hal_transform", HAL_TRANSFORM_RE), ("switchkins", SWITCHKINS_RE)): match = pattern.search(transform_code) if match: module = match.group(0) rows.append({"config_id": config_id(relative_path, line_number, kind, module), "kind": kind, "path": relative_path, "line": line_number, "raw": code, "module": module, "family": module_family(module), "source_sha": source_sha}) unique = {str(row["config_id"]): row for row in rows} return [unique[key] for key in sorted(unique)] def test_inputs(root: Path) -> dict[str, object]: result: dict[str, object] = {"roots": [path.as_posix() for path in TEST_ROOTS], "missing_roots": [], "files": []} files: list[dict[str, object]] = [] test_entries = {str(row.get("path")): str(row.get("test_source_id")) for row in read_jsonl(root / "coverage/test-entries.jsonl") if row.get("path") and row.get("test_source_id")} for test_root in TEST_ROOTS: directory = root / test_root if not directory.is_dir(): result["missing_roots"].append(test_root.as_posix()) continue for path in sorted(path for path in directory.rglob("*") if path.is_file()): relative_path = relative(root, path) files.append({"test_source_id": test_entries.get(relative_path, f"test:{relative_path}"), "path": relative_path, "bytes": path.stat().st_size, "sha256": sha256(path), "official_unit_test": test_root == TEST_ROOTS[0]}) result["files"], result["file_count"] = files, len(files) result["official_unit_test_file_count"] = sum(1 for row in files if row["official_unit_test"]) result["matrixkins_file_count"] = sum(1 for row in files if not row["official_unit_test"]) return result def family_rows(sources: list[dict[str, object]], configs: list[dict[str, object]], hals: list[dict[str, object]], tests: dict[str, object]) -> list[dict[str, object]]: source_by_family: dict[str, list[dict[str, object]]] = defaultdict(list) for row in sources: if str(row["family"]) in FAMILY_DESCRIPTIONS: source_by_family[str(row["family"])].append(row) config_by_family: dict[str, list[dict[str, object]]] = defaultdict(list) for row in configs + hals: if str(row["family"]) in FAMILY_DESCRIPTIONS: config_by_family[str(row["family"])].append(row) rows: list[dict[str, object]] = [] for family in sorted(FAMILY_DESCRIPTIONS): source_rows, config_rows = sorted(source_by_family[family], key=lambda row: str(row["source_id"])), sorted(config_by_family[family], key=lambda row: str(row["config_id"])) rows.append({ "family": family, "description": FAMILY_DESCRIPTIONS[family], "modules": sorted({str(row["module"]) for row in source_rows} | {str(row["module"]) for row in config_rows if row["module"]}), "source_ids": [str(row["source_id"]) for row in source_rows], "configuration_ids": [str(row["config_id"]) for row in config_rows], "test_source_ids": [], "oracle": {"status": "blocked", "sample_plan_id": f"kinematics-oracle:{family}:fk-ik-roundtrip", "candidate_configuration_ids": [str(row["config_id"]) for row in config_rows[:8]], "reason": "linuxcnc/unit_tests/kinematics is absent from the locked source; tests/matrixkins is retained as a configuration candidate and is not promoted to an official unit-test oracle"}, }) return rows def generate(root: Path, output: Path) -> dict[str, object]: root = root.resolve() sources, configs, hals, tests = source_files(root), config_entries(root), hal_entries(root), test_inputs(root) families = family_rows(sources, configs, hals, tests) lock = json.loads((root / "context-lock.json").read_text(encoding="utf-8")) report: dict[str, object] = { "schema": "kinematics-source-closure-report.v2", "context_id": lock.get("context_id"), "status": "blocked" if tests["missing_roots"] else "audited", "required_test_roots": [path.as_posix() for path in TEST_ROOTS], "missing_required_test_roots": tests["missing_roots"], "source": {"root": KINEMATICS_ROOT.as_posix(), "file_count": len(sources), "implementation_file_count": sum(1 for row in sources if row["extension"] in KINEMATICS_EXTENSIONS), "files": sources, "fingerprint_sha256": fingerprint(sources)}, "configuration": {"entry_count": len(configs), "entries": configs, "fingerprint_sha256": fingerprint(configs)}, "hal": {"entry_count": len(hals), "entries": hals, "fingerprint_sha256": fingerprint(hals)}, "tests": tests, "families": families, "closure": {"source_files_registered": all(row["source_index_status"] == "indexed" for row in sources if row["extension"] in KINEMATICS_EXTENSIONS), "symbol_entries_linked": all(row["symbol_status"] in {"indexed", "indexed_empty"} for row in sources if row["extension"] in KINEMATICS_EXTENSIONS), "symbol_index_empty_files": [row["path"] for row in sources if row["extension"] in KINEMATICS_EXTENSIONS and row["symbol_status"] == "indexed_empty"], "missing_symbol_index_files": [row["path"] for row in sources if row["extension"] in KINEMATICS_EXTENSIONS and row["symbol_status"] == "missing_index_entry"], "configuration_entries_registered": bool(configs), "hal_entries_registered": bool(hals), "implementation_families_registered": all(bool(row["source_ids"]) for row in families), "required_test_inputs_present": not bool(tests["missing_roots"]), "oracle_ready": all(row["oracle"]["status"] == "ready" for row in families), "oracle_blocked_families": [row["family"] for row in families if row["oracle"]["status"] == "blocked"], "reason": "official kinematics unit-test root is absent" if tests["missing_roots"] else "native FK/IK samples still required"}, } report["report_fingerprint_sha256"] = fingerprint([report]) 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-source-closure.report.json")) args = parser.parse_args() report = generate(args.repo_root, args.repo_root / args.output) print(f"kinematics_closure={report['status']} sources={report['source']['implementation_file_count']} configs={report['configuration']['entry_count']} hal={report['hal']['entry_count']} families={len(report['families'])} tests={report['tests']['file_count']}") return 0 if __name__ == "__main__": raise SystemExit(main())