Spaces:
Paused
Paused
| """Static validation of generated ROS 2 packages, grounded in the indexed interface definitions. | |
| No ROS installation is needed. Checks: | |
| - file paths are relative and stay inside the package tree | |
| - every package has a parseable package.xml with a name and build type, plus | |
| setup.py (ament_python) or CMakeLists.txt with ament_package() (ament_cmake) | |
| - Python files and Python launch files parse; launch files define generate_launch_description() | |
| - XML launch files parse and have a <launch> root | |
| - every message/service type used (Python imports, C++ includes and qualified names) | |
| exists in the ROS 2 distro, according to data/ros2_interfaces.json | |
| - interface packages that are used are declared as dependencies in package.xml (warning) | |
| """ | |
| from __future__ import annotations | |
| import ast | |
| import json | |
| import re | |
| import xml.etree.ElementTree as ET | |
| from dataclasses import asdict, dataclass, field | |
| from functools import lru_cache | |
| from pathlib import Path, PurePosixPath | |
| class CodeValidation: | |
| valid: bool | |
| errors: list[str] = field(default_factory=list) | |
| warnings: list[str] = field(default_factory=list) | |
| stats: dict = field(default_factory=dict) | |
| def to_dict(self) -> dict: | |
| return asdict(self) | |
| def feedback(self) -> str: | |
| lines = ["The generated package failed validation:"] + [f"- ERROR: {e}" for e in self.errors] | |
| lines += [f"- warning: {w}" for w in self.warnings] | |
| return "\n".join(lines) | |
| def load_interfaces(path: str) -> dict[str, dict[str, str]]: | |
| p = Path(path) | |
| return json.loads(p.read_text()) if p.is_file() else {} | |
| def _camel(snake: str) -> str: | |
| return "".join(part.capitalize() for part in snake.split("_")) | |
| _CPP_INCLUDE = re.compile(r'#include\s*[<"](\w+)/(msg|srv|action)/(\w+)\.hpp[>"]') | |
| _CPP_QUALIFIED = re.compile(r"\b(\w+)::(msg|srv|action)::(\w+)\b") | |
| _DEPENDS = ("depend", "exec_depend", "build_depend", "build_export_depend") | |
| def _python_interfaces(tree: ast.AST) -> set[str]: | |
| used = set() | |
| for node in ast.walk(tree): | |
| if isinstance(node, ast.ImportFrom) and node.module: | |
| parts = node.module.split(".") | |
| if len(parts) == 2 and parts[1] in ("msg", "srv", "action"): | |
| for alias in node.names: | |
| if alias.name != "*": | |
| used.add(f"{parts[0]}/{parts[1]}/{alias.name}") | |
| elif isinstance(node, ast.Attribute) and isinstance(node.value, ast.Attribute): | |
| # std_msgs.msg.String style | |
| inner = node.value | |
| if isinstance(inner.value, ast.Name) and inner.attr in ("msg", "srv", "action") and node.attr[:1].isupper(): | |
| used.add(f"{inner.value.id}/{inner.attr}/{node.attr}") | |
| return used | |
| def _cpp_interfaces(text: str) -> set[str]: | |
| used = {f"{p}/{k}/{_camel(n)}" for p, k, n in _CPP_INCLUDE.findall(text)} | |
| used |= {f"{p}/{k}/{n}" for p, k, n in _CPP_QUALIFIED.findall(text) if n[:1].isupper() and p != "std"} | |
| return used | |
| def _package_roots(files: dict[str, str]) -> dict[str, str]: | |
| """Directory -> package name for every package.xml.""" | |
| roots = {} | |
| for path, content in files.items(): | |
| if PurePosixPath(path).name == "package.xml": | |
| try: | |
| name = (ET.fromstring(content).findtext("name") or "").strip() | |
| except ET.ParseError: | |
| name = "" | |
| roots[str(PurePosixPath(path).parent)] = name | |
| return roots | |
| def _owning_root(path: str, roots: dict[str, str]) -> str | None: | |
| best = None | |
| for root in roots: | |
| prefix = "" if root == "." else root + "/" | |
| if path.startswith(prefix) and (best is None or len(root) > len(best)): | |
| best = root | |
| return best | |
| def validate_package(files: dict[str, str], distro: str, interfaces: dict[str, dict[str, str]]) -> CodeValidation: | |
| errors: list[str] = [] | |
| warnings: list[str] = [] | |
| if not files: | |
| return CodeValidation(False, ["no files were generated"]) | |
| for path in files: | |
| p = PurePosixPath(path) | |
| if p.is_absolute() or ".." in p.parts or path.startswith("~"): | |
| errors.append(f"{path}: file paths must be relative and inside the package") | |
| if errors: | |
| return CodeValidation(False, errors) | |
| known = interfaces.get(distro, {}) | |
| known_packages = {name.split("/")[0] for name in known} | |
| if not known: | |
| warnings.append(f"no interface catalogue for {distro}; message types were not checked (run rag.ingest)") | |
| # Custom interfaces defined in the generated packages themselves. | |
| defined = set() | |
| for path in files: | |
| p = PurePosixPath(path) | |
| if p.suffix in (".msg", ".srv", ".action") and p.parent.name in ("msg", "srv", "action"): | |
| roots = _package_roots(files) | |
| root = _owning_root(path, roots) | |
| pkg = roots.get(root, "") if root is not None else "" | |
| defined.add(f"{pkg}/{p.parent.name}/{p.stem}") | |
| roots = _package_roots(files) | |
| if not roots: | |
| errors.append("no package.xml found") | |
| deps_by_root: dict[str, set[str]] = {} | |
| for root, name in roots.items(): | |
| pkg_xml = files[f"{root}/package.xml" if root != "." else "package.xml"] | |
| try: | |
| tree = ET.fromstring(pkg_xml) | |
| except ET.ParseError as e: | |
| errors.append(f"{root}/package.xml: XML parse error: {e}") | |
| continue | |
| if not name: | |
| errors.append(f"{root}/package.xml: missing <name>") | |
| build_type = (tree.findtext("export/build_type") or "").strip() | |
| deps_by_root[root] = {(d.text or "").strip() for tag in _DEPENDS for d in tree.iter(tag)} | |
| prefix = "" if root == "." else root + "/" | |
| if build_type == "ament_python": | |
| if f"{prefix}setup.py" not in files: | |
| errors.append(f"{root}: ament_python package without setup.py") | |
| if name and f"{prefix}{name}/__init__.py" not in files: | |
| warnings.append(f"{root}: no {name}/__init__.py (the Python module directory)") | |
| if f"{prefix}resource/{name}" not in files: | |
| warnings.append(f"{root}: no resource/{name} marker file (ament index)") | |
| elif build_type == "ament_cmake": | |
| cmake = files.get(f"{prefix}CMakeLists.txt") | |
| if cmake is None: | |
| errors.append(f"{root}: ament_cmake package without CMakeLists.txt") | |
| elif "ament_package()" not in cmake: | |
| errors.append(f"{root}/CMakeLists.txt: missing ament_package()") | |
| else: | |
| errors.append(f"{root}/package.xml: <export><build_type> must be ament_python or ament_cmake") | |
| used_by_root: dict[str, set[str]] = {} | |
| for path, content in files.items(): | |
| p = PurePosixPath(path) | |
| used: set[str] = set() | |
| if p.suffix == ".py": | |
| try: | |
| tree = ast.parse(content, filename=path) | |
| except SyntaxError as e: | |
| errors.append(f"{path}:{e.lineno}: Python syntax error: {e.msg}") | |
| continue | |
| used = _python_interfaces(tree) | |
| if path.endswith("launch.py"): | |
| funcs = {n.name for n in tree.body if isinstance(n, ast.FunctionDef)} | |
| if "generate_launch_description" not in funcs: | |
| errors.append(f"{path}: launch file does not define generate_launch_description()") | |
| elif p.suffix in (".cpp", ".hpp", ".h", ".cc"): | |
| used = _cpp_interfaces(content) | |
| elif path.endswith(("launch.xml", ".launch")) or (p.suffix == ".xml" and "launch" in p.parts): | |
| try: | |
| root_el = ET.fromstring(content) | |
| if root_el.tag != "launch": | |
| errors.append(f"{path}: XML launch file root must be <launch>, got <{root_el.tag}>") | |
| except ET.ParseError as e: | |
| errors.append(f"{path}: XML parse error: {e}") | |
| for name in sorted(used): | |
| if name in known or name in defined: | |
| continue | |
| pkg = name.split("/")[0] | |
| if pkg in known_packages: | |
| errors.append(f"{path}: {name} does not exist in ROS 2 {distro}") | |
| else: | |
| warnings.append(f"{path}: {name} is from a package that is not indexed; not verified") | |
| owner = _owning_root(path, roots) | |
| if owner is not None: | |
| used_by_root.setdefault(owner, set()).update(used) | |
| for root, used in used_by_root.items(): | |
| own = roots.get(root) | |
| for pkg in sorted({u.split("/")[0] for u in used} - {own}): | |
| if pkg not in deps_by_root.get(root, set()): | |
| warnings.append(f"{root}/package.xml: uses {pkg} but does not declare it as a dependency") | |
| stats = { | |
| "files": len(files), | |
| "packages": sorted(n for n in roots.values() if n), | |
| "interfaces_used": sorted(set().union(*used_by_root.values())) if used_by_root else [], | |
| } | |
| return CodeValidation(not errors, errors, warnings, stats) | |