#!/usr/bin/env python3 """Generate a deterministic HAL component manifest and config graph report.""" from __future__ import annotations import argparse import hashlib import json import re from collections import Counter, defaultdict from pathlib import Path from typing import Any from generate_config_resources import locked_submodule_files CONTEXT = Path("context-lock.json") HAL_REPORT = Path("coverage/hal-runtime.report.json") CONFIG_CLOSURE = Path("coverage/configuration-startup-closure.report.json") RESOURCES = Path("coverage/config-resources.jsonl") GATE = Path("coverage/release-candidate-gate.report.json") COMPONENT_ROOT = Path("linuxcnc/src/hal/components") OUTPUT = Path("coverage/hal-component-manifest.report.json") SCHEMA = "hal-component-manifest.v1" COMPONENT_RE = re.compile(r"^\s*component\s+([A-Za-z_][A-Za-z0-9_]*)\b") PIN_RE = re.compile(r"^\s*pin\s+(in|out|io)\s+([A-Za-z_][A-Za-z0-9_]*)\s+([A-Za-z_][A-Za-z0-9_.#-]*)") PARAM_RE = re.compile(r"^\s*param\s+(r|rw)\s+([A-Za-z_][A-Za-z0-9_]*)\s+([A-Za-z_][A-Za-z0-9_.-]*)(?:\s*=\s*([^\s]+))?") FUNCTION_RE = re.compile(r"^\s*function\s+([^;]+);") VARIABLE_RE = re.compile(r"^\s*variable\s+([A-Za-z_][A-Za-z0-9_]*)\s+([A-Za-z_][A-Za-z0-9_.-]*)") OPTION_RE = re.compile(r"^\s*option\s+([A-Za-z_][A-Za-z0-9_]*)\s*(?:=\s*([^\s]+))?") LOADRT_RE = re.compile(r"^\s*loadrt\s+([^\s]+)(?:\s+(.*))?$") LOADUSR_RE = re.compile(r"^\s*loadusr\s+(.+)$") ADDF_RE = re.compile(r"^\s*addf\s+([^\s]+)\s+([^\s]+)") LINE_COMMANDS = {"loadrt", "loadusr", "addf", "net", "setp", "source", "newsig", "linksp", "linkps", "linkpp"} 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 file_digest(path: Path) -> str | None: if not path.is_file(): return None 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 artifact(root: Path, path: Path, role: str) -> dict[str, Any]: full = root / path sha = file_digest(full) if sha is None: return {"path": path.as_posix(), "role": role, "status": "missing"} return {"path": path.as_posix(), "role": role, "status": "present", "bytes": full.stat().st_size, "sha256": sha} def relative(root: Path, path: Path) -> str: return path.resolve().relative_to(root.resolve()).as_posix() def parse_comp(root: Path, path: Path) -> dict[str, Any]: component = None pins: list[dict[str, Any]] = [] params: list[dict[str, Any]] = [] functions: list[dict[str, Any]] = [] variables: list[dict[str, Any]] = [] options: list[dict[str, Any]] = [] errors: list[str] = [] try: lines = path.read_text(encoding="utf-8", errors="strict").splitlines() except (OSError, UnicodeError) as exc: return {"path": relative(root, path), "source_sha256": file_digest(path), "status": "blocked", "errors": [str(exc)], "component": None, "pins": [], "params": [], "functions": [], "variables": [], "options": []} for line_number, raw in enumerate(lines, 1): line = raw.split("//", 1)[0].strip() match = COMPONENT_RE.match(line) if match: if component is not None and component != match.group(1): errors.append(f"duplicate_component:{line_number}") component = match.group(1) continue match = PIN_RE.match(line) if match: pins.append({"direction": match.group(1), "type": match.group(2), "name": match.group(3), "line": line_number}) continue match = PARAM_RE.match(line) if match: params.append({"mode": match.group(1), "type": match.group(2), "name": match.group(3), "default": match.group(4).rstrip(";") if match.group(4) else None, "line": line_number}) continue match = FUNCTION_RE.match(line) if match: for name in match.group(1).split(): if name: functions.append({"name": name, "line": line_number}) continue match = VARIABLE_RE.match(line) if match: variables.append({"type": match.group(1), "name": match.group(2), "line": line_number}) continue match = OPTION_RE.match(line) if match: options.append({"name": match.group(1), "value": match.group(2), "line": line_number}) if component is None: errors.append("component_declaration_missing") names = [row["name"] for row in pins] + [row["name"] for row in params] + [row["name"] for row in variables] if len(names) != len(set(names)): errors.append("duplicate_declared_name") return {"path": relative(root, path), "source_sha256": file_digest(path), "status": "verified" if not errors else "blocked", "errors": errors, "component": component, "pins": pins, "params": params, "functions": functions, "variables": variables, "options": options} def parse_hal(root: Path, path: Path, component_names: set[str], resource_source_id: str | None) -> list[dict[str, Any]]: records: list[dict[str, Any]] = [] try: lines = path.read_text(encoding="utf-8", errors="strict").splitlines() except (OSError, UnicodeError) as exc: return [{"path": relative(root, path), "line": 0, "command": "file", "status": "blocked", "code": "decode_error", "detail": str(exc), "resource_source_id": resource_source_id}] for line_number, raw in enumerate(lines, 1): line = raw.split("#", 1)[0].strip() if not line: continue command = line.split(None, 1)[0] if command not in LINE_COMMANDS: continue base = {"path": relative(root, path), "line": line_number, "command": command, "raw": line, "resource_source_id": resource_source_id} if command == "loadrt": match = LOADRT_RE.match(line) module = match.group(1) if match else "" names = re.search(r"(?:^|\s)names=([^\s]+)", match.group(2) or "" if match else "") instances = names.group(1).split(",") if names else [] if module.startswith("[") or module not in component_names: records.append({**base, "status": "blocked", "code": "undeclared_or_variable_component", "module": module, "instances": instances}) else: records.append({**base, "status": "declared", "code": None, "module": module, "instances": instances}) elif command == "loadusr": match = LOADUSR_RE.match(line) records.append({**base, "status": "blocked", "code": "host_dynamic_loader", "program": match.group(1).split()[0] if match else ""}) elif command == "addf": match = ADDF_RE.match(line) records.append({**base, "status": "checked", "code": None if match else "invalid_addf", "function": match.group(1) if match else None, "thread": match.group(2) if match else None}) else: tokens = line.split() records.append({**base, "status": "checked", "code": None, "tokens": tokens[1:]}) return records def generate(root: Path, output: Path = OUTPUT) -> dict[str, Any]: root = root.resolve() lock = json.loads((root / CONTEXT).read_text(encoding="utf-8")) hal_runtime = json.loads((root / HAL_REPORT).read_text(encoding="utf-8")) config_closure = json.loads((root / CONFIG_CLOSURE).read_text(encoding="utf-8")) gate = json.loads((root / GATE).read_text(encoding="utf-8")) resource_rows = [json.loads(line) for line in (root / RESOURCES).read_text(encoding="utf-8").splitlines() if line.strip()] untracked = locked_submodule_files(root) component_root = root / COMPONENT_ROOT component_files = sorted(path for path in component_root.rglob("*.comp") if path.is_file() and (untracked is None or relative(root, path).removeprefix("linuxcnc/") not in untracked)) manifest = [parse_comp(root, path) for path in component_files] manifest.sort(key=lambda row: str(row["component"] or row["path"])) component_names = {str(row["component"]) for row in manifest if row.get("component")} hal_resource_ids: dict[str, str] = {} for row in resource_rows: if row.get("resource_kind") == "hal" and row.get("parse_status") in {"resolved_file", "resolved_directory"}: hal_resource_ids.setdefault(str(row.get("path")), str(row.get("resource_source_id"))) hal_paths = sorted(path for path in hal_resource_ids if (root / path).is_file()) graph: list[dict[str, Any]] = [] for path in hal_paths: graph.extend(parse_hal(root, root / path, component_names, hal_resource_ids[path])) graph.sort(key=lambda row: (str(row.get("path")), int(row.get("line", 0)), str(row.get("command")), str(row.get("raw", "")))) statuses = Counter(str(row.get("status")) for row in graph) codes = Counter(str(row.get("code")) for row in graph if row.get("code")) component_names_list = [str(row.get("component")) for row in manifest if row.get("component")] checks = { "manifest_sources": {"status": "verified" if len(manifest) == 145 and len(component_names_list) == len(set(component_names_list)) and all(row.get("source_sha256") for row in manifest) else "blocked", "components": len(manifest)}, "manifest_declarations": {"status": "verified" if all(row.get("status") == "verified" and row.get("component") for row in manifest) else "blocked", "blocked_components": sum(row.get("status") != "verified" for row in manifest)}, "hal_resource_edges": {"status": "verified" if all(row.get("resource_source_id") for row in graph) else "blocked", "hal_files": len(hal_paths), "graph_records": len(graph)}, "capacity_contract": {"status": "verified" if hal_runtime.get("pin_capacity") == 256 and hal_runtime.get("max_period_ticks") == 1000000 else "blocked", "pin_capacity": hal_runtime.get("pin_capacity"), "max_period_ticks": hal_runtime.get("max_period_ticks")}, "no_dynamic_loader_acceptance": {"status": "verified" if all(row.get("code") != "dynamic_library_accepted" for row in graph) else "blocked", "host_dynamic_loaders": codes.get("host_dynamic_loader", 0)}, "configuration_dependency": {"status": "verified" if config_closure.get("implementation_status") == "verified" else "blocked"}, "release_stop_preserved": {"status": "verified" if gate.get("release_action") == "STOP_RELEASE" and gate.get("candidate_eligible") is False else "blocked", "release_action": gate.get("release_action")}, } checks_ok = all(value["status"] == "verified" for value in checks.values()) report: dict[str, Any] = { "schema": SCHEMA, "context_id": lock.get("context_id"), "task_id": "M3-RT-02", "status": "partial" if checks_ok and statuses.get("blocked", 0) else "blocked" if not checks_ok else "verified", "implementation_status": "verified" if checks_ok else "blocked", "candidate_eligible": False, "release_action": "STOP_RELEASE", "inputs": [artifact(root, CONTEXT, "context_lock"), artifact(root, RESOURCES, "config_resources"), artifact(root, CONFIG_CLOSURE, "configuration_startup_closure"), artifact(root, HAL_REPORT, "hal_runtime"), artifact(root, GATE, "release_candidate_gate")], "counts": {"component_files": len(manifest), "components": len(component_names_list), "pins": sum(len(row["pins"]) for row in manifest), "params": sum(len(row["params"]) for row in manifest), "functions": sum(len(row["functions"]) for row in manifest), "hal_files": len(hal_paths), "graph_records": len(graph), "graph_statuses": dict(sorted(statuses.items())), "graph_codes": dict(sorted(codes.items()))}, "checks": checks, "component_manifest": manifest, "hal_graph": graph, "errors": {}, "next_transition": "execute M3-RT-03 virtual device fault injection and replay", "report_sha256": None, } report["report_sha256"] = digest({key: value for key, value in report.items() if key != "report_sha256"}) destination = output if output.is_absolute() else root / output destination.parent.mkdir(parents=True, exist_ok=True) destination.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=OUTPUT) args = parser.parse_args() report = generate(args.repo_root.resolve(), args.output) print(f"hal_component_manifest={report['status']} components={report['counts']['components']} graph_records={report['counts']['graph_records']}") return 0 if __name__ == "__main__": raise SystemExit(main())