api/ros_check.py

"""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


@dataclass
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)


@lru_cache
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)