270 lines
12 KiB
Python
270 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate resource dependency graphs for every LinuxCNC simulation INI."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
|
|
|
|
CONFIG_ROOT = Path("linuxcnc/configs/sim")
|
|
RESOURCE_EXTENSIONS = {
|
|
".ini": "ini", ".hal": "hal", ".tbl": "tool_table", ".var": "parameter_file",
|
|
".ngc": "gcode", ".nc": "gcode", ".tap": "gcode", ".py": "python",
|
|
".xml": "ui_descriptor", ".ui": "ui_descriptor", ".glade": "ui_descriptor",
|
|
".tcl": "script", ".clp": "classicladder", ".pref": "preference",
|
|
".prefs": "preference", ".dat": "data", ".jpg": "image", ".jpeg": "image",
|
|
".png": "image", ".gif": "image", ".stl": "model", ".obj": "model",
|
|
}
|
|
KEY_KINDS = {
|
|
"HALFILE": "hal", "POSTGUI_HALFILE": "hal", "TOOL_TABLE": "tool_table",
|
|
"PARAMETER_FILE": "parameter_file", "POSITION_FILE": "data", "LOG_FILE": "log",
|
|
"MDI_HISTORY_FILE": "data", "PREFERENCE_FILE_PATH": "preference",
|
|
"INTRO_GRAPHIC": "image", "PROGRAM_PREFIX": "program_directory",
|
|
"SUBROUTINE_PATH": "subroutine_directory", "USER_COMMAND": "user_command",
|
|
"SHUTDOWN": "script", "APP": "application",
|
|
}
|
|
HAL_COMMANDS = {"source": "hal_source", "loadusr": "hal_user_component"}
|
|
TOKEN_RE = re.compile(r"(?<![A-Za-z0-9_./~-])([^\s\"'=;,]+)")
|
|
EXT_RE = re.compile(r"\.(?:ini|hal|tbl|var|ngc|nc|tap|py|xml|ui|glade|tcl|clp|pref|prefs|dat|png|jpg|jpeg|gif|stl|obj)$", re.I)
|
|
_SHA_CACHE: dict[Path, str] = {}
|
|
_LOCKED_SUBMODULE_CACHE: dict[Path, set[str] | None] = {}
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
path = path.resolve()
|
|
cached = _SHA_CACHE.get(path)
|
|
if cached:
|
|
return cached
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for block in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(block)
|
|
value = digest.hexdigest()
|
|
_SHA_CACHE[path] = value
|
|
return value
|
|
|
|
|
|
def rel(root: Path, path: Path) -> str:
|
|
return path.resolve().relative_to(root.resolve()).as_posix()
|
|
|
|
|
|
def clean_value(value: str) -> str:
|
|
value = value.split("#", 1)[0].split(";", 1)[0].strip().strip('"').strip("'")
|
|
return value
|
|
|
|
|
|
def parse_ini(path: Path) -> list[tuple[str, str, int]]:
|
|
section = ""
|
|
rows: list[tuple[str, str, int]] = []
|
|
for line_number, raw in enumerate(path.read_text(encoding="utf-8", errors="strict").splitlines(), 1):
|
|
line = raw.strip()
|
|
if not line or line.startswith(("#", ";")):
|
|
continue
|
|
if line.startswith("[") and "]" in line:
|
|
section = line[1:line.index("]")].strip().upper()
|
|
continue
|
|
if "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
rows.append((section, key.strip().upper(), clean_value(value)))
|
|
return rows
|
|
|
|
|
|
def resolve_reference(repo_root: Path, config_root: Path, base: Path, value: str) -> tuple[Path | None, str]:
|
|
value = clean_value(value)
|
|
if not value:
|
|
return None, "empty"
|
|
if value.startswith(("~", "$", "${")):
|
|
return None, "host_or_variable_path"
|
|
candidate = Path(value)
|
|
if not candidate.is_absolute():
|
|
candidate = base / candidate
|
|
candidate = candidate.resolve()
|
|
try:
|
|
candidate.relative_to(repo_root.resolve())
|
|
except ValueError:
|
|
return None, "outside_repo"
|
|
return candidate, "path"
|
|
|
|
|
|
def tokens_with_extensions(value: str) -> list[str]:
|
|
return [match.group(1) for match in TOKEN_RE.finditer(value) if EXT_RE.search(match.group(1))]
|
|
|
|
|
|
def config_files(config_root: Path) -> list[Path]:
|
|
return sorted(path for path in config_root.rglob("*") if path.is_file())
|
|
|
|
|
|
def locked_submodule_files(repo_root: Path) -> set[str] | None:
|
|
"""Return untracked LinuxCNC paths excluded from the locked inventory.
|
|
|
|
Ignored build outputs are intentionally retained because the existing
|
|
context lock was generated from that checkout. Only explicit untracked
|
|
source/config fixtures are excluded from the locked inventory.
|
|
"""
|
|
submodule = repo_root / "linuxcnc"
|
|
key = submodule.resolve()
|
|
if key in _LOCKED_SUBMODULE_CACHE:
|
|
return _LOCKED_SUBMODULE_CACHE[key]
|
|
try:
|
|
completed = subprocess.run(
|
|
["git", "-C", str(submodule), "status", "--porcelain=v1", "--untracked-files=all"],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if completed.returncode != 0:
|
|
result = None
|
|
else:
|
|
result = {line[3:] for line in completed.stdout.splitlines() if line.startswith("?? ")}
|
|
except (OSError, UnicodeError):
|
|
result = None
|
|
_LOCKED_SUBMODULE_CACHE[key] = result
|
|
return result
|
|
|
|
|
|
def is_locked_path(repo_root: Path, path: Path, locked: set[str] | None) -> bool:
|
|
if locked is None:
|
|
return True
|
|
try:
|
|
relative = path.resolve().relative_to((repo_root / "linuxcnc").resolve()).as_posix()
|
|
except ValueError:
|
|
return False
|
|
return relative not in locked
|
|
|
|
|
|
def add_row(rows: list[dict[str, object]], repo_root: Path, config_root: Path, entrypoint: Path,
|
|
target: Path | None, raw: str, kind: str, edge_kind: str, line: int, status: str | None = None,
|
|
origin: Path | None = None) -> None:
|
|
entry_id = rel(repo_root, entrypoint)
|
|
if target is None:
|
|
target_path = raw
|
|
target_sha = None
|
|
target_status = status or "unresolved"
|
|
else:
|
|
target_path = rel(repo_root, target) if target.exists() else rel(repo_root, target)
|
|
target_sha = sha256(target) if target.is_file() else None
|
|
target_status = "resolved_file" if target.is_file() else ("resolved_directory" if target.is_dir() else "missing")
|
|
rows.append({
|
|
"resource_source_id": f"resource:{entry_id}:{line}:{edge_kind}:{target_path}",
|
|
"entrypoint": entry_id,
|
|
"path": target_path,
|
|
"resource_kind": kind,
|
|
"edge_kind": edge_kind,
|
|
"line": line,
|
|
"raw_value": raw,
|
|
"source_sha": sha256(origin or entrypoint),
|
|
"target_sha256": target_sha,
|
|
"origin_path": rel(repo_root, origin or entrypoint),
|
|
"parse_status": target_status,
|
|
})
|
|
|
|
|
|
def scan_entrypoint(repo_root: Path, config_root: Path, entrypoint: Path) -> list[dict[str, object]]:
|
|
rows: list[dict[str, object]] = []
|
|
locked = locked_submodule_files(repo_root)
|
|
ini_rows = parse_ini(entrypoint)
|
|
row_numbers = {(section, key, value): index for index, (section, key, value) in enumerate(ini_rows, 1)}
|
|
for section, key, value in ini_rows:
|
|
if key in KEY_KINDS:
|
|
kind = KEY_KINDS[key]
|
|
if key in {"SUBROUTINE_PATH"}:
|
|
values = value.split(":")
|
|
elif key == "PROGRAM_PREFIX":
|
|
values = [value]
|
|
else:
|
|
values = [value]
|
|
for item in values:
|
|
target, status = resolve_reference(repo_root, config_root, entrypoint.parent, item)
|
|
add_row(rows, repo_root, config_root, entrypoint, target, item, kind, f"ini:{section}:{key}", row_numbers[(section, key, value)], status if target is None else None)
|
|
elif section == "FILTER" and key not in {"PROGRAM_EXTENSION"}:
|
|
for token in tokens_with_extensions(value):
|
|
target, status = resolve_reference(repo_root, config_root, entrypoint.parent, token)
|
|
add_row(rows, repo_root, config_root, entrypoint, target, token, "filter", "ini:filter_command", row_numbers[(section, key, value)], status if target is None else None)
|
|
elif key in {"EMBED_TAB_COMMAND", "DISPLAY", "HALCMD", "TASK", "EMCIO"}:
|
|
for token in tokens_with_extensions(value):
|
|
target, status = resolve_reference(repo_root, config_root, entrypoint.parent, token)
|
|
add_row(rows, repo_root, config_root, entrypoint, target, token, RESOURCE_EXTENSIONS.get(Path(token).suffix.lower(), "external"), f"ini:{section}:{key}", row_numbers[(section, key, value)], status if target is None else None)
|
|
|
|
seen_paths = {str(row["path"]) for row in rows}
|
|
for path in sorted(path for path in entrypoint.parent.rglob("*") if path.is_file() and path != entrypoint and is_locked_path(repo_root, path, locked)):
|
|
path_key = rel(repo_root, path)
|
|
if path.suffix.lower() in RESOURCE_EXTENSIONS and path_key not in seen_paths:
|
|
add_row(rows, repo_root, config_root, entrypoint, path, rel(repo_root, path), RESOURCE_EXTENSIONS[path.suffix.lower()], "directory_member", 0)
|
|
seen_paths.add(path_key)
|
|
|
|
seen_ids = {str(existing["resource_source_id"]) for existing in rows}
|
|
for row in parse_hal_references(repo_root, config_root, entrypoint, rows):
|
|
if row["resource_source_id"] not in seen_ids:
|
|
rows.append(row)
|
|
seen_ids.add(str(row["resource_source_id"]))
|
|
return rows
|
|
|
|
|
|
def parse_hal_references(repo_root: Path, config_root: Path, entrypoint: Path, rows: list[dict[str, object]]) -> list[dict[str, object]]:
|
|
output: list[dict[str, object]] = []
|
|
hal_paths = [Path(str(row["path"])) for row in rows if row["resource_kind"] == "hal" and row["parse_status"] == "resolved_file"]
|
|
for relative_path in hal_paths:
|
|
hal = repo_root / relative_path
|
|
for line_number, raw in enumerate(hal.read_text(encoding="utf-8", errors="strict").splitlines(), 1):
|
|
parts = raw.strip().split()
|
|
if not parts or parts[0] not in HAL_COMMANDS:
|
|
continue
|
|
command = parts[0]
|
|
candidates = tokens_with_extensions(raw)
|
|
for token in candidates:
|
|
target, status = resolve_reference(repo_root, config_root, hal.parent, token)
|
|
add_row(output, repo_root, config_root, entrypoint, target, token, RESOURCE_EXTENSIONS.get(Path(token).suffix.lower(), "external"), f"hal:{command}", line_number, status if target is None else None, hal)
|
|
return output
|
|
|
|
|
|
def generate(repo_root: Path, output: Path, report_path: Path) -> tuple[int, str]:
|
|
config_root = repo_root / CONFIG_ROOT
|
|
if not config_root.is_dir():
|
|
raise FileNotFoundError(f"required config root is missing: {CONFIG_ROOT}")
|
|
locked = locked_submodule_files(repo_root)
|
|
entrypoints = sorted(path for path in config_root.rglob("*.ini") if path.is_file() and is_locked_path(repo_root, path, locked))
|
|
rows: list[dict[str, object]] = []
|
|
for entrypoint in entrypoints:
|
|
rows.extend(scan_entrypoint(repo_root, config_root, entrypoint))
|
|
rows.sort(key=lambda row: str(row["resource_source_id"]))
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
with output.open("w", encoding="utf-8", newline="\n") as handle:
|
|
for row in rows:
|
|
handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n")
|
|
digest = sha256(output)
|
|
report = {
|
|
"schema": "config-resources-report.v1",
|
|
"entrypoints": len(entrypoints),
|
|
"resources": len(rows),
|
|
"missing": sum(1 for row in rows if row["parse_status"] == "missing"),
|
|
"unresolved": sum(1 for row in rows if row["parse_status"] in {"unresolved", "host_or_variable_path", "outside_repo"}),
|
|
"output_sha256": digest,
|
|
"entrypoint_paths": [rel(repo_root, path) for path in entrypoints],
|
|
}
|
|
report_path.parent.mkdir(parents=True, exist_ok=True)
|
|
report_path.write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8")
|
|
return len(rows), digest
|
|
|
|
|
|
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/config-resources.jsonl"))
|
|
parser.add_argument("--report", type=Path, default=Path("coverage/config-resources.report.json"))
|
|
args = parser.parse_args()
|
|
root = args.repo_root.resolve()
|
|
count, digest = generate(root, root / args.output, root / args.report)
|
|
print(f"config_resources={count} sha256={digest}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|