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