273 lines
17 KiB
Python
273 lines
17 KiB
Python
#!/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())
|