generate_readme.py

#!/usr/bin/env python3

# Copyright Institute for Automotive Engineering (ika), RWTH Aachen University
# SPDX-License-Identifier: Apache-2.0

"""
Auto-generates ROS2 interface documentation (topics, actions, parameters)
by parsing C++ source files in a ROS2 repository and writing full package
README.md files next to package.xml.

Usage:
    python3 scripts/generate_readme.py [REPO_ROOT]

REPO_ROOT defaults to the script's parent directory.
The READMEs generated are <package_dir>/README.md for discovered ROS packages.
"""

import difflib
import ast
import re
import subprocess
import sys
from codecs import decode as codecs_decode
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Optional
from xml.etree import ElementTree

try:
    from jinja2 import Environment, FileSystemLoader, TemplateNotFound
except ModuleNotFoundError as exc:
    if exc.name == 'jinja2':
        Environment = None
        FileSystemLoader = None
        TemplateNotFound = Exception
    else:
        raise


# ---------------------------------------------------------------------------
# Data classes
# ---------------------------------------------------------------------------

@dataclass
class TopicInterface:
    name: str
    msg_type: str


TRANSPORT_MESSAGE_TYPES = {
    'image_transport': 'sensor_msgs/msg/Image',
    'point_cloud_transport': 'sensor_msgs/msg/PointCloud2',
}

TRANSPORT_SUBSCRIBE_TOPIC_ARGUMENTS = {
    'point_cloud_transport::SubscriberFilter': 1,
}


@dataclass
class ActionInterface:
    name: str
    action_type: str


@dataclass
class ServiceInterface:
    name: str
    srv_type: str


@dataclass
class Parameter:
    name: str
    ros_type: str
    default: str
    description: str


@dataclass
class NodeInterfaces:
    node_name: str
    subscribers: list = field(default_factory=list)
    publishers: list = field(default_factory=list)
    service_servers: list = field(default_factory=list)
    action_servers: list = field(default_factory=list)
    action_clients: list = field(default_factory=list)
    parameters: list = field(default_factory=list)


@dataclass
class RepoMetadata:
    host: str
    provider: str
    owner: str
    repo: str
    owner_lower: str
    pages_url: str
    repo_https_url: str
    container_image: str


@dataclass
class PackageSection:
    title: str
    kind: str
    interface_entries: list = field(default_factory=list)
    nodes: list = field(default_factory=list)
    launch_files: list = field(default_factory=list)


@dataclass
class PackageTemplateContext:
    package_name: str
    package_description: str
    sections: list[PackageSection]


@dataclass
class PackageDocEntry:
    name: str
    path: str
    description: str


@dataclass
class InterfaceTableRow:
    name: str
    interface_type: str
    description: str


@dataclass
class InterfaceDefinitionEntry:
    full_type: str
    rel_path: str
    description: str


@dataclass
class LaunchArgumentRow:
    name: str
    default: str
    description: str


@dataclass
class LaunchFileTemplateContext:
    name: str
    rel_path: str
    arguments: list[LaunchArgumentRow]


@dataclass
class NodeTemplateContext:
    node_name: str
    manual_text: str
    subscribers: list[InterfaceTableRow]
    publishers: list[InterfaceTableRow]
    service_servers: list[InterfaceTableRow]
    action_servers: list[InterfaceTableRow]
    action_clients: list[InterfaceTableRow]
    parameters: list[Parameter]


@dataclass
class TopLevelTemplateContext:
    title: str
    repo_name: str
    owner: str
    owner_lower: str
    provider: str
    pages_url: str
    repo_https_url: str
    container_image: str
    intro_block: Optional[str]
    pre_quickstart_block: str
    quickstart_body: Optional[str]
    quickstart_package: str
    quickstart_launch_file: str
    repository_packages: list[PackageDocEntry]
    licensing_extra_body: str
    acknowledgements_body: Optional[str]


# ---------------------------------------------------------------------------
# Type helpers
# ---------------------------------------------------------------------------

MANUAL_TABLE_HEADERS = {
    '| Topic | Type | Description |': 'topic',
    '| Action | Type | Description |': 'action',
    '| Service | Type | Description |': 'service',
    '| Type | Description |': 'interface',
    '| Argument | Default | Description |': 'launch_arg',
}

def cpp_ros_type(cpp_type: str, type_aliases: dict) -> str:
    """Resolve type aliases and convert C++ ROS type to ROS notation (:: -> /)."""
    t = cpp_type.strip()
    t = type_aliases.get(t, t)
    for alias, target in type_aliases.items():
        if t.startswith(f'{alias}::'):
            t = f'{target}::{t[len(alias) + 2:]}'
            break
    return t.replace('::', '/').strip()


PYTHON_PARAM_TYPE_MAP = {
    'BOOL': 'bool',
    'INTEGER': 'int',
    'DOUBLE': 'float',
    'STRING': 'string',
    'BYTE_ARRAY': 'byte[]',
    'BOOL_ARRAY': 'bool[]',
    'INTEGER_ARRAY': 'int[]',
    'DOUBLE_ARRAY': 'float[]',
    'STRING_ARRAY': 'string[]',
}


def cpp_param_type(cpp_type: str) -> str:
    """Map C++ type to ROS parameter type name."""
    t = cpp_type.strip()
    if t in ('double', 'float'):
        return 'float'
    if t in ('int', 'long', 'long int', 'int8_t', 'int16_t', 'int32_t', 'int64_t', 'uint8_t', 'uint16_t', 'uint32_t', 'uint64_t'):
        return 'int'
    if t == 'bool':
        return 'bool'
    if t == 'std::string':
        return 'string'
    vector_match = re.match(r'std::vector\s*<\s*([^>]+)\s*>', t)
    if vector_match:
        element_type = vector_match.group(1).strip()
        if element_type in ('double', 'float'):
            return 'float[]'
        if element_type in ('int', 'long', 'long int', 'int8_t', 'int16_t', 'int32_t', 'int64_t', 'uint8_t', 'uint16_t', 'uint32_t', 'uint64_t'):
            return 'int[]'
        if element_type == 'bool':
            return 'bool[]'
        if element_type == 'std::string':
            return 'string[]'
        return f'{element_type}[]'
    return t


def format_default(default_str: Optional[str], cpp_type: str, enum_value_map: dict[str, str]) -> str:
    """Format a C++ default value for Markdown display."""
    if default_str is None:
        return '[]' if 'vector' in cpp_type else ''
    formatted = default_str.strip()
    if formatted in enum_value_map:
        return enum_value_map[formatted]
    if 'vector' in cpp_type and formatted.startswith('{') and formatted.endswith('}'):
        return f'[{formatted[1:-1].strip()}]'
    return formatted


# ---------------------------------------------------------------------------
# Discovery
# ---------------------------------------------------------------------------

def git_tracked_files(repo_root: Path) -> list[Path]:
    """Return files tracked by the target repository without descending into submodules."""
    result = subprocess.run(
        ['git', 'ls-files', '-z'],
        cwd=repo_root,
        check=False,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    if result.returncode != 0:
        raise RuntimeError(result.stderr.strip() or 'git ls-files failed')
    return [repo_root / path for path in result.stdout.split('\0') if path]


def find_packages(repo_root: Path) -> list:
    """Return [(package_name, package_dir, package_description)] for ROS packages."""
    packages = []
    package_xmls = sorted(path for path in git_tracked_files(repo_root) if path.name == 'package.xml')
    for pkg_xml in package_xmls:
        if not pkg_xml.is_file():
            continue
        try:
            root = ElementTree.parse(pkg_xml).getroot()
            name_el = root.find('name')
            description_el = root.find('description')
            if name_el is not None and name_el.text:
                description = ''
                if description_el is not None and description_el.text:
                    description = ' '.join(description_el.text.split())
                packages.append((name_el.text.strip(), pkg_xml.parent, description))
        except ElementTree.ParseError:
            pass
    return packages


def find_node_sources(package_dir: Path) -> list:
    """Return .cpp files that define a ROS node (contain ': Node("..." )')."""
    node_pattern = re.compile(r':\s*(?:(?:\w+)::)*Node\s*\(')
    return [
        cpp for cpp in sorted(package_dir.rglob('*.cpp'))
        if node_pattern.search(cpp.read_text(errors='replace'))
    ]


def python_defines_node(source: str) -> bool:
    """Return whether the Python source defines a class deriving from rclpy's Node."""
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return False
    for node in ast.walk(tree):
        if isinstance(node, ast.ClassDef):
            for base in node.bases:
                base_name = base.attr if isinstance(base, ast.Attribute) else getattr(base, 'id', '')
                if base_name == 'Node':
                    return True
    return False


def find_python_node_sources(package_dir: Path) -> list:
    """Return .py files that define a ROS node (a class deriving from Node)."""
    return [
        py for py in sorted(package_dir.rglob('*.py'))
        if python_defines_node(py.read_text(errors='replace'))
    ]


def find_headers(package_dir: Path) -> list:
    """Return all header files (.hpp, .h) in the package."""
    return [
        p for ext in ('*.hpp', '*.h')
        for p in sorted(package_dir.rglob(ext))
    ]


def find_interface_files(package_dir: Path, subdir: str, ext: str) -> list:
    """Return sorted list of interface definition files in package_dir/subdir/."""
    iface_dir = package_dir / subdir
    if not iface_dir.is_dir():
        return []
    return sorted(iface_dir.glob(f'*.{ext}'))


def find_launch_files(package_dir: Path) -> list:
    """Return launch files (.py, .xml, .yaml) found in any launch/ subdirectory."""
    launch_dir = package_dir / 'launch'
    if not launch_dir.is_dir():
        return []
    return [
        p for ext in ('*.py', '*.xml', '*.yaml', '*.yml')
        for p in sorted(launch_dir.rglob(ext))
    ]


# ---------------------------------------------------------------------------
# Extraction from source/header text
# ---------------------------------------------------------------------------

def extract_node_name(source: str) -> Optional[str]:
    m = re.search(r':\s*(?:(?:\w+)::)*Node\s*\(\s*"([^"]+)"', source)
    return m.group(1) if m else None


def extract_cpp_string_symbols(source: str) -> dict[str, str]:
    """Return simple C++ string constants and local aliases found in source."""
    symbols: dict[str, str] = {}
    string_literal = r'(?:u8|u|U|L)?"((?:\\.|[^"\\])*)"'

    const_patterns = (
        rf'\b(?:static\s+)?(?:inline\s+)?(?:constexpr\s+)?(?:const\s+)?(?:std::string|char\s*(?:const)?\s*\*|const\s+char\s*\*)\s+'
        rf'((?:(?:\w+)::)?\w+)\s*=\s*({string_literal})\s*;',
        rf'\b(?:static\s+)?(?:inline\s+)?(?:constexpr\s+)?(?:const\s+)?(?:std::string|char\s*(?:const)?\s*\*|const\s+char\s*\*)\s+'
        rf'((?:(?:\w+)::)?\w+)\s*\{{\s*({string_literal})\s*\}}\s*;',
        rf'\b(?:static\s+)?(?:inline\s+)?(?:constexpr\s+)?(?:const\s+)?char\s+'
        rf'((?:(?:\w+)::)?\w+)\s*\[[^\]]*\]\s*=\s*({string_literal})\s*;',
    )
    for pattern in const_patterns:
        for match in re.finditer(pattern, source):
            name = match.group(1)
            value = decode_cpp_string_literal(match.group(3))
            symbols[name] = value
            symbols[name.split('::')[-1]] = value

    local_pattern = re.compile(r'\bstd::string\s+(\w+)\s*=\s*([^;]+);', re.DOTALL)
    for match in local_pattern.finditer(source):
        value = resolve_cpp_string_expression(match.group(2), symbols)
        if value:
            symbols[match.group(1)] = value

    return symbols


def resolve_cpp_string_expression(expr: str, symbols: dict[str, str]) -> Optional[str]:
    """Resolve string literals, known string symbols, and simple topic resolution wrappers."""
    expr = expr.strip()
    if not expr:
        return None

    resolve_match = re.search(r'resolve_topic_name\s*\((.*)\)', expr, re.DOTALL)
    if resolve_match:
        return resolve_cpp_string_expression(resolve_match.group(1), symbols)

    literal_expr_pattern = re.compile(r'(?:\s*(?:u8|u|U|L)?"((?:\\.|[^"\\])*"))+\s*\Z', re.DOTALL)
    literal_expr_match = literal_expr_pattern.fullmatch(expr)
    if literal_expr_match:
        literal_pattern = re.compile(r'(?:u8|u|U|L)?"((?:\\.|[^"\\])*)"', re.DOTALL)
        return ''.join(
            decode_cpp_string_literal(match.group(1))
            for match in literal_pattern.finditer(expr)
        )

    cleaned = re.sub(r'\bthis->', '', expr)
    cleaned = cleaned.strip('() ')
    if cleaned in symbols:
        return symbols[cleaned]
    if cleaned.split('::')[-1] in symbols:
        return symbols[cleaned.split('::')[-1]]
    return ' '.join(expr.split())


def find_cpp_templated_call_bodies(source: str, function_name: str) -> list[tuple[str, str]]:
    """Return [(template_argument, call_body)] for C++ calls with one template argument."""
    calls = []
    pattern = re.compile(rf'(?<!::)\b{re.escape(function_name)}\s*<([^>]+)>\s*\(')

    for match in pattern.finditer(source):
        start = match.end()
        depth = 1
        i = start
        while i < len(source) and depth > 0:
            if source[i] == '(':
                depth += 1
            elif source[i] == ')':
                depth -= 1
            i += 1
        if depth == 0:
            calls.append((match.group(1).strip(), source[start:i - 1]))

    return calls


def find_cpp_member_call_bodies(source: str, function_name: str) -> list[tuple[str, str]]:
    """Return [(receiver_expression, call_body)] for C++ member calls."""
    calls = []
    pattern = re.compile(
        rf'(?P<receiver>(?:this->)?\w+(?:\s*(?:->|\.)\s*\w+)*)\s*(?:->|\.)\s*{re.escape(function_name)}\s*\('
    )

    for match in pattern.finditer(source):
        start = match.end()
        depth = 1
        i = start
        while i < len(source) and depth > 0:
            if source[i] == '(':
                depth += 1
            elif source[i] == ')':
                depth -= 1
            i += 1
        if depth == 0:
            calls.append((match.group('receiver').strip(), source[start:i - 1]))

    return calls


def find_cpp_qualified_call_bodies(source: str, function_name: str) -> list[tuple[str, str]]:
    """Return [(namespace_or_class, call_body)] for C++ qualified calls."""
    calls = []
    pattern = re.compile(rf'\b((?:\w+::)+){re.escape(function_name)}\s*\(')

    for match in pattern.finditer(source):
        start = match.end()
        depth = 1
        i = start
        while i < len(source) and depth > 0:
            if source[i] == '(':
                depth += 1
            elif source[i] == ')':
                depth -= 1
            i += 1
        if depth == 0:
            calls.append((match.group(1).rstrip(':'), source[start:i - 1]))

    return calls


def unwrap_cpp_type(cpp_type: str) -> str:
    """Remove common ownership wrappers around a C++ type."""
    stripped = re.sub(r'\s+', '', cpp_type)
    wrapper_match = re.fullmatch(r'(?:std::)?(?:unique_ptr|shared_ptr|weak_ptr)<(.+)>', stripped)
    if wrapper_match:
        return wrapper_match.group(1)
    return stripped


def extract_cpp_variable_types(source: str) -> dict[str, str]:
    """Return variable/member names mapped to declared C++ types."""
    variable_types: dict[str, str] = {}
    type_token = r'(?:std::(?:unique_ptr|shared_ptr|weak_ptr)<\s*)?(?:(?:\w+)::)*\w+(?:\s*>)?'
    pattern = re.compile(rf'\b({type_token})\s+(\w+)\s*(?:[;=({{])')

    for match in pattern.finditer(source):
        cpp_type = unwrap_cpp_type(match.group(1))
        variable_types[match.group(2)] = cpp_type

    make_shared_pattern = re.compile(
        r'\bauto\s+(\w+)\s*=\s*std::make_shared\s*<\s*([^>]+)\s*>\s*\('
    )
    for match in make_shared_pattern.finditer(source):
        variable_types[match.group(1)] = match.group(2).strip()

    return variable_types


def transport_message_type_from_cpp_type(cpp_type: str, aliases: dict) -> Optional[str]:
    """Infer message type for known ROS transport helper classes."""
    resolved = aliases.get(cpp_type, cpp_type)
    for namespace, msg_type in TRANSPORT_MESSAGE_TYPES.items():
        if resolved == namespace or resolved.startswith(f'{namespace}::'):
            return msg_type
    return None


def transport_message_type_from_receiver(receiver: str, variable_types: dict[str, str], aliases: dict) -> Optional[str]:
    """Infer message type for member transport calls from the receiver object."""
    cleaned = re.sub(r'\bthis->', '', receiver).strip()
    root = re.split(r'\s*(?:->|\.)\s*', cleaned, maxsplit=1)[0]
    cpp_type = variable_types.get(root)
    if cpp_type:
        return transport_message_type_from_cpp_type(cpp_type, aliases)
    return transport_message_type_from_cpp_type(root, aliases)


def transport_subscribe_topic_argument(
    args: list[str],
    receiver: str,
    variable_types: dict[str, str],
    aliases: dict,
) -> Optional[str]:
    """Return the topic argument from a transport subscribe call."""
    cleaned = re.sub(r'\bthis->', '', receiver).strip()
    root = re.split(r'\s*(?:->|\.)\s*', cleaned, maxsplit=1)[0]
    cpp_type = variable_types.get(root, root)
    resolved_type = aliases.get(cpp_type, cpp_type)
    topic_index = TRANSPORT_SUBSCRIBE_TOPIC_ARGUMENTS.get(resolved_type, 0)
    return args[topic_index] if len(args) > topic_index else None


def is_transport_like(cpp_name: str) -> bool:
    """Return whether a C++ receiver/type/qualifier looks like a ROS transport helper."""
    return 'transport' in cpp_name.lower()


def unknown_transport_message(function_name: str, transport_name: str) -> str:
    """Return a clear error for unsupported transport helper APIs."""
    known = ', '.join(sorted(TRANSPORT_MESSAGE_TYPES))
    return (
        f'Unable to infer ROS message type for transport call {transport_name}.{function_name}(...). '
        f'Add the transport to TRANSPORT_MESSAGE_TYPES. Known transports: {known}.'
    )


def should_fail_unknown_transport(receiver: str, variable_types: dict[str, str]) -> bool:
    """Return whether an unknown member call likely belongs to a transport helper."""
    cleaned = re.sub(r'\bthis->', '', receiver).strip()
    root = re.split(r'\s*(?:->|\.)\s*', cleaned, maxsplit=1)[0]
    return is_transport_like(receiver) or is_transport_like(variable_types.get(root, ''))


def unique_topic_interfaces(interfaces: list[TopicInterface]) -> list[TopicInterface]:
    """Return interfaces with duplicate topic/type pairs removed while preserving order."""
    seen: set[tuple[str, str]] = set()
    result = []
    for interface in interfaces:
        key = (interface.name, interface.msg_type)
        if key in seen:
            continue
        seen.add(key)
        result.append(interface)
    return result


def extract_subscribers(source: str, aliases: dict, string_symbols: dict[str, str], variable_types: dict[str, str]) -> list:
    subscribers = []
    for cpp_type, call_body in find_cpp_templated_call_bodies(source, 'create_subscription'):
        args = split_cpp_arguments(call_body)
        if not args:
            continue
        topic = resolve_cpp_string_expression(args[0], string_symbols)
        if topic:
            subscribers.append(TopicInterface(name=topic, msg_type=cpp_ros_type(cpp_type, aliases)))

    for receiver, call_body in find_cpp_member_call_bodies(source, 'subscribe'):
        args = split_cpp_arguments(call_body)
        if not args:
            continue
        topic_arg = transport_subscribe_topic_argument(args, receiver, variable_types, aliases)
        topic = resolve_cpp_string_expression(topic_arg, string_symbols) if topic_arg else None
        msg_type = transport_message_type_from_receiver(receiver, variable_types, aliases)
        if topic and msg_type:
            subscribers.append(TopicInterface(name=topic, msg_type=msg_type))
        elif topic and should_fail_unknown_transport(receiver, variable_types):
            raise RuntimeError(unknown_transport_message('subscribe', receiver))

    for qualifier, call_body in find_cpp_qualified_call_bodies(source, 'create_subscription'):
        args = split_cpp_arguments(call_body)
        if len(args) < 2:
            continue
        topic = resolve_cpp_string_expression(args[1], string_symbols)
        msg_type = transport_message_type_from_cpp_type(qualifier, aliases)
        if topic and msg_type:
            subscribers.append(TopicInterface(name=topic, msg_type=msg_type))
        elif topic and is_transport_like(qualifier):
            raise RuntimeError(unknown_transport_message('create_subscription', qualifier))

    return unique_topic_interfaces(subscribers)


def extract_publishers(source: str, aliases: dict, string_symbols: dict[str, str], variable_types: dict[str, str]) -> list:
    publishers = []
    for cpp_type, call_body in find_cpp_templated_call_bodies(source, 'create_publisher'):
        args = split_cpp_arguments(call_body)
        if not args:
            continue
        topic = resolve_cpp_string_expression(args[0], string_symbols)
        if topic:
            publishers.append(TopicInterface(name=topic, msg_type=cpp_ros_type(cpp_type, aliases)))

    for receiver, call_body in find_cpp_member_call_bodies(source, 'advertise'):
        args = split_cpp_arguments(call_body)
        if not args:
            continue
        topic = resolve_cpp_string_expression(args[0], string_symbols)
        msg_type = transport_message_type_from_receiver(receiver, variable_types, aliases)
        if topic and msg_type:
            publishers.append(TopicInterface(name=topic, msg_type=msg_type))
        elif topic and should_fail_unknown_transport(receiver, variable_types):
            raise RuntimeError(unknown_transport_message('advertise', receiver))

    for qualifier, call_body in find_cpp_qualified_call_bodies(source, 'create_publisher'):
        args = split_cpp_arguments(call_body)
        if len(args) < 2:
            continue
        topic = resolve_cpp_string_expression(args[1], string_symbols)
        msg_type = transport_message_type_from_cpp_type(qualifier, aliases)
        if topic and msg_type:
            publishers.append(TopicInterface(name=topic, msg_type=msg_type))
        elif topic and is_transport_like(qualifier):
            raise RuntimeError(unknown_transport_message('create_publisher', qualifier))

    return unique_topic_interfaces(publishers)


def extract_action_servers(source: str, aliases: dict) -> list:
    return [
        ActionInterface(name=m.group(2), action_type=cpp_ros_type(m.group(1), aliases))
        for m in re.finditer(
            r'rclcpp_action::create_server\s*<([^>]+)>\s*\(\s*this\s*,\s*"([^"]+)"', source)
    ]


def extract_action_clients(source: str, aliases: dict) -> list:
    return [
        ActionInterface(name=m.group(2), action_type=cpp_ros_type(m.group(1), aliases))
        for m in re.finditer(
            r'rclcpp_action::create_client\s*<([^>]+)>\s*\(\s*this\s*,\s*"([^"]+)"', source)
    ]


def extract_service_servers(source: str, aliases: dict) -> list:
    return [
        ServiceInterface(name=m.group(2), srv_type=cpp_ros_type(m.group(1), aliases))
        for m in re.finditer(r'create_service\s*<([^>]+)>\s*\(\s*"([^"]+)"', source)
    ]


def find_cpp_call_bodies(source: str, function_name: str) -> list[str]:
    """Return the argument body of each function call, excluding declarations/definitions."""
    bodies = []
    pattern = re.compile(rf'(?<!::)\b{re.escape(function_name)}\s*\(')

    for match in pattern.finditer(source):
        start = match.end()
        depth = 1
        i = start
        in_string = False
        in_char = False
        in_line_comment = False
        in_block_comment = False
        escaped = False

        while i < len(source) and depth > 0:
            char = source[i]
            next_char = source[i + 1] if i + 1 < len(source) else ''

            if in_line_comment:
                if char == '\n':
                    in_line_comment = False
                i += 1
                continue

            if in_block_comment:
                if char == '*' and next_char == '/':
                    in_block_comment = False
                    i += 2
                else:
                    i += 1
                continue

            if in_string:
                if escaped:
                    escaped = False
                elif char == '\\':
                    escaped = True
                elif char == '"':
                    in_string = False
                i += 1
                continue

            if in_char:
                if escaped:
                    escaped = False
                elif char == '\\':
                    escaped = True
                elif char == "'":
                    in_char = False
                i += 1
                continue

            if char == '/' and next_char == '/':
                in_line_comment = True
                i += 2
                continue

            if char == '/' and next_char == '*':
                in_block_comment = True
                i += 2
                continue

            if char == '"':
                in_string = True
                i += 1
                continue

            if char == "'":
                in_char = True
                i += 1
                continue

            if char == '(':
                depth += 1
            elif char == ')':
                depth -= 1

            i += 1

        if depth == 0:
            bodies.append(source[start:i - 1])

    return bodies


def split_cpp_arguments(call_body: str) -> list[str]:
    """Split a C++ call body into top-level arguments."""
    args = []
    current = []
    paren_depth = 0
    brace_depth = 0
    bracket_depth = 0
    in_string = False
    in_char = False
    in_line_comment = False
    in_block_comment = False
    escaped = False
    i = 0

    while i < len(call_body):
        char = call_body[i]
        next_char = call_body[i + 1] if i + 1 < len(call_body) else ''

        if in_line_comment:
            current.append(char)
            if char == '\n':
                in_line_comment = False
            i += 1
            continue

        if in_block_comment:
            current.append(char)
            if char == '*' and next_char == '/':
                current.append(next_char)
                in_block_comment = False
                i += 2
            else:
                i += 1
            continue

        if in_string:
            current.append(char)
            if escaped:
                escaped = False
            elif char == '\\':
                escaped = True
            elif char == '"':
                in_string = False
            i += 1
            continue

        if in_char:
            current.append(char)
            if escaped:
                escaped = False
            elif char == '\\':
                escaped = True
            elif char == "'":
                in_char = False
            i += 1
            continue

        if char == '/' and next_char == '/':
            current.extend([char, next_char])
            in_line_comment = True
            i += 2
            continue

        if char == '/' and next_char == '*':
            current.extend([char, next_char])
            in_block_comment = True
            i += 2
            continue

        if char == '"':
            current.append(char)
            in_string = True
            i += 1
            continue

        if char == "'":
            current.append(char)
            in_char = True
            i += 1
            continue

        if char == '(':
            paren_depth += 1
        elif char == ')':
            paren_depth -= 1
        elif char == '{':
            brace_depth += 1
        elif char == '}':
            brace_depth -= 1
        elif char == '[':
            bracket_depth += 1
        elif char == ']':
            bracket_depth -= 1
        elif char == ',' and paren_depth == 0 and brace_depth == 0 and bracket_depth == 0:
            args.append(''.join(current).strip())
            current = []
            i += 1
            continue

        current.append(char)
        i += 1

    trailing = ''.join(current).strip()
    if trailing:
        args.append(trailing)

    return args


def decode_cpp_string_literal(text: str) -> str:
    """Decode the content of a regular C++ string literal."""
    return codecs_decode(text, 'unicode_escape')


def extract_cpp_string_expression(expr: str) -> str:
    """Collapse adjacent C++ string literals into a single Python string."""
    literal_pattern = re.compile(r'(?:u8|u|U|L)?"((?:\\.|[^"\\])*)"', re.DOTALL)
    literals = [decode_cpp_string_literal(match.group(1)) for match in literal_pattern.finditer(expr)]
    if not literals:
        return ' '.join(expr.split())
    return ''.join(literals)


def extract_cpp_string_literal_expression(expr: str) -> Optional[str]:
    """Collapse adjacent C++ string literals, or return None if the expression has no string literal."""
    literal_pattern = re.compile(r'(?:u8|u|U|L)?"((?:\\.|[^"\\])*)"', re.DOTALL)
    literals = [decode_cpp_string_literal(match.group(1)) for match in literal_pattern.finditer(expr)]
    if not literals:
        return None
    return ''.join(literals)


def extract_parameter_member_name(expr: str) -> Optional[str]:
    """Return the referenced member name from a parameter storage expression."""
    expr = expr.strip()
    member_match = re.fullmatch(r'(?:\w+\.)?(\w+)', expr)
    if member_match:
        return member_match.group(1)
    return None


def extract_raw_parameters(source: str) -> list:
    """Return [(param_name, member_var_name, description)] from parameter declarations."""
    params = []

    for call_body in find_cpp_call_bodies(source, 'declareAndLoadParameter'):
        args = split_cpp_arguments(call_body)
        if len(args) < 3:
            continue

        name = extract_cpp_string_literal_expression(args[0])
        member_name = extract_parameter_member_name(args[1])
        description = extract_cpp_string_expression(args[2])

        if not name or not member_name:
            continue

        params.append((name, member_name, description))

    return params


def ast_constant_string(node: "ast.AST | None") -> Optional[str]:
    """Return the value of a string-constant AST node, else None."""
    if isinstance(node, ast.Constant) and isinstance(node.value, str):
        return node.value
    return None


def python_call_argument(call: ast.Call, index: Optional[int], name: str) -> "ast.AST | None":
    """Return a call argument by keyword name, falling back to positional index."""
    for keyword in call.keywords:
        if keyword.arg == name:
            return keyword.value
    if index is not None and index < len(call.args) and not isinstance(call.args[index], ast.Starred):
        return call.args[index]
    return None


def python_call_name(call: ast.Call) -> str:
    """Return the called function/method name for a Python call node."""
    func = call.func
    if isinstance(func, ast.Attribute):
        return func.attr
    if isinstance(func, ast.Name):
        return func.id
    return ''


def build_python_import_type_map(tree: ast.AST) -> dict[str, str]:
    """Map imported interface class names to ROS 'pkg/kind/Type' strings."""
    result: dict[str, str] = {}
    for node in ast.walk(tree):
        if not isinstance(node, ast.ImportFrom) or not node.module:
            continue
        parts = node.module.split('.')
        if len(parts) < 2 or parts[-1] not in ('msg', 'srv', 'action'):
            continue
        prefix = '/'.join(parts)
        for alias in node.names:
            if alias.name == '*':
                continue
            local = alias.asname or alias.name
            result[local] = f'{prefix}/{alias.name}'
    return result


def resolve_python_interface_type(node: "ast.AST | None", type_map: dict[str, str]) -> Optional[str]:
    """Resolve a message/service/action type argument to ROS notation."""
    if isinstance(node, ast.Name):
        name = node.id
    elif isinstance(node, ast.Attribute):
        name = node.attr
    else:
        return None
    return type_map.get(name, name)


def render_python_default(node: "ast.AST | None", source: str) -> str:
    """Render a Python default-value expression for Markdown display."""
    if node is None:
        return ''
    if isinstance(node, ast.Constant):
        value = node.value
        if isinstance(value, bool):
            return 'true' if value else 'false'
        if value is None:
            return ''
        if isinstance(value, str):
            return value
        return repr(value)
    if isinstance(node, (ast.List, ast.Tuple)):
        return '[' + ', '.join(render_python_default(element, source) for element in node.elts) + ']'
    return (ast.get_source_segment(source, node) or ast.unparse(node)).strip()


def extract_python_node_name(tree: ast.AST) -> Optional[str]:
    """Return the node name from a super().__init__("name") call."""
    for node in ast.walk(tree):
        if (
            isinstance(node, ast.Call)
            and isinstance(node.func, ast.Attribute)
            and node.func.attr == '__init__'
            and isinstance(node.func.value, ast.Call)
            and isinstance(node.func.value.func, ast.Name)
            and node.func.value.func.id == 'super'
            and node.args
        ):
            name = ast_constant_string(node.args[0])
            if name:
                return name
    return None


def extract_python_parameters(tree: ast.AST, source: str) -> list:
    """Return Parameters from declare_and_load_parameter calls in Python source."""
    parameters = []
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        if python_call_name(node) not in ('declare_and_load_parameter', 'declareAndLoadParameter'):
            continue

        name = ast_constant_string(python_call_argument(node, 0, 'name'))
        if not name:
            continue

        type_node = python_call_argument(node, 1, 'param_type')
        type_key = type_node.attr if isinstance(type_node, ast.Attribute) else None
        ros_type = PYTHON_PARAM_TYPE_MAP.get(type_key, '') if type_key else ''

        description = ast_constant_string(python_call_argument(node, 2, 'description')) or ''
        default = render_python_default(python_call_argument(node, 3, 'default'), source)

        parameters.append(Parameter(name=name, ros_type=ros_type, default=default, description=description))
    return parameters


def extract_python_node_interfaces(source: str, fallback_name: str) -> Optional[NodeInterfaces]:
    """Parse a Python node source into NodeInterfaces (name, endpoints, parameters)."""
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return None

    type_map = build_python_import_type_map(tree)
    subscribers: list = []
    publishers: list = []
    service_servers: list = []
    action_servers: list = []
    action_clients: list = []

    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        call_name = python_call_name(node)

        if call_name in ('create_subscription', 'create_publisher'):
            topic = ast_constant_string(python_call_argument(node, 1, 'topic'))
            msg_type = resolve_python_interface_type(python_call_argument(node, 0, 'msg_type'), type_map)
            if topic and msg_type:
                interface = TopicInterface(name=topic, msg_type=msg_type)
                (subscribers if call_name == 'create_subscription' else publishers).append(interface)
        elif call_name == 'create_service':
            srv_name = ast_constant_string(python_call_argument(node, 1, 'srv_name'))
            srv_type = resolve_python_interface_type(python_call_argument(node, 0, 'srv_type'), type_map)
            if srv_name and srv_type:
                service_servers.append(ServiceInterface(name=srv_name, srv_type=srv_type))
        elif call_name in ('ActionServer', 'ActionClient'):
            action_name = ast_constant_string(python_call_argument(node, 2, 'action_name'))
            action_type = resolve_python_interface_type(python_call_argument(node, 1, 'action_type'), type_map)
            if action_name and action_type:
                interface = ActionInterface(name=action_name, action_type=action_type)
                (action_servers if call_name == 'ActionServer' else action_clients).append(interface)

    return NodeInterfaces(
        node_name=extract_python_node_name(tree) or fallback_name,
        subscribers=unique_topic_interfaces(subscribers),
        publishers=unique_topic_interfaces(publishers),
        service_servers=service_servers,
        action_servers=action_servers,
        action_clients=action_clients,
        parameters=extract_python_parameters(tree, source),
    )


def extract_python_launch_arguments(source: str) -> list:
    """Return [(name, default, description)] from DeclareLaunchArgument calls in a .py file."""
    results = []
    tree = ast.parse(source)

    def call_keyword(call: ast.Call, name: str) -> ast.AST | None:
        for keyword in call.keywords:
            if keyword.arg == name:
                return keyword.value
        return None

    def constant_string(node: ast.AST | None) -> Optional[str]:
        if isinstance(node, ast.Constant) and isinstance(node.value, str):
            return node.value
        return None

    def render_default(node: ast.AST | None) -> str:
        value = constant_string(node)
        if value is not None:
            return f'"{value}"'
        if node is None:
            return ''
        return (ast.get_source_segment(source, node) or ast.unparse(node)).strip()

    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        func_name = node.func.id if isinstance(node.func, ast.Name) else getattr(node.func, 'attr', '')
        if func_name != 'DeclareLaunchArgument' or not node.args:
            continue

        name = constant_string(node.args[0])
        if name is None:
            continue

        default = render_default(call_keyword(node, 'default_value'))
        description = constant_string(call_keyword(node, 'description')) or ''

        results.append((name, default, description))
    return results


def extract_xml_launch_arguments(source: str) -> list:
    """Return [(name, default, description)] from <arg> elements in a .xml launch file."""
    results = []
    for m in re.finditer(r'<arg\b([^/]*)/>', source, re.DOTALL):
        attrs = m.group(1)
        name_m = re.search(r'\bname\s*=\s*"([^"]+)"', attrs)
        if not name_m:
            continue
        default_m = re.search(r'\bdefault\s*=\s*"([^"]*)"', attrs)
        desc_m = re.search(r'\bdescription\s*=\s*"([^"]*)"', attrs)
        results.append((
            name_m.group(1),
            f'"{default_m.group(1)}"' if default_m else '',
            desc_m.group(1) if desc_m else '',
        ))
    return results


def extract_launch_arguments(path: Path) -> list:
    """Dispatch to the correct extractor based on file extension."""
    source = path.read_text(errors='replace')
    if path.suffix == '.xml':
        return extract_xml_launch_arguments(source)
    return extract_python_launch_arguments(source)


def build_member_var_map(headers: list) -> dict:
    """Return {var_name: (cpp_type, default_str)} from member variable declarations."""
    pattern = re.compile(
        r'\b(double|float|int|long|long\s+int|int8_t|int16_t|int32_t|int64_t|uint8_t|uint16_t|uint32_t|uint64_t|bool|std::string|std::vector\s*<[^>]+>)\s+(\w+)\s*'
        r'(?:=\s*([^;]+))?;'
    )
    result = {}
    for header in headers:
        for m in pattern.finditer(header.read_text(errors='replace')):
            var_name = m.group(2).strip()
            if var_name not in result:
                result[var_name] = (m.group(1).strip(), m.group(3))
    return result


def build_enum_value_map(headers: list) -> dict[str, str]:
    """Return {'ENUM::MEMBER': 'value'} for simple enum definitions in headers."""
    enum_pattern = re.compile(r'enum\s+(\w+)\s*\{(.*?)\};', re.DOTALL)
    entry_pattern = re.compile(r'(\w+)\s*=\s*(-?\d+)')
    result = {}
    for header in headers:
        source = header.read_text(errors='replace')
        for enum_match in enum_pattern.finditer(source):
            enum_name = enum_match.group(1)
            body = enum_match.group(2)
            for entry_match in entry_pattern.finditer(body):
                result[f'{enum_name}::{entry_match.group(1)}'] = entry_match.group(2)
    return result


def build_type_alias_map(headers: list) -> dict:
    """Return {alias: full_cpp_type} from 'using Alias = FullType;' declarations."""
    using_pattern = re.compile(r'\busing\s+(\w+)\s*=\s*([^;]+);')
    namespace_pattern = re.compile(r'\bnamespace\s+(\w+)\s*=\s*([^;]+);')
    result = {}
    for header in headers:
        source = header.read_text(errors='replace')
        for m in using_pattern.finditer(source):
            result[m.group(1).strip()] = m.group(2).strip()
        for m in namespace_pattern.finditer(source):
            result[m.group(1).strip()] = m.group(2).strip()
    return result


def resolve_parameters(raw_params: list, member_var_map: dict, enum_value_map: dict[str, str]) -> list:
    params = []
    for name, member_var, description in raw_params:
        cpp_type, default_raw = member_var_map.get(member_var, ('', None))
        params.append(Parameter(
            name=name,
            ros_type=cpp_param_type(cpp_type),
            default=format_default(default_raw, cpp_type, enum_value_map),
            description=description,
        ))
    return params


# ---------------------------------------------------------------------------
# Markdown rendering
# ---------------------------------------------------------------------------

def normalize_table_cell(cell: str) -> str:
    """Normalize Markdown table cell content for stable matching."""
    return ' '.join(cell.strip().split())


def sanitize_manual_content(content: str) -> str:
    """Normalize preserved description content from existing README tables."""
    return content.strip()


def render_table_cell(content: str) -> str:
    """Render Markdown table cells with an explicit placeholder for blank values."""
    normalized = content.strip()
    return normalized if normalized else 'TODO'


def render_parameter_default(content: str) -> str:
    """Render parameter defaults, using '-' when no default value exists."""
    normalized = content.strip()
    return normalized if normalized else '-'


def render_mermaid_edge_label(content: str) -> str:
    """Quote Mermaid edge labels only when their syntax requires it."""
    sanitized = content.replace('"', "'")
    if re.search(r'[\[\]{}()|]', sanitized):
        return f'"{sanitized}"'
    return sanitized


def manual_key(heading_path: tuple[str, ...], table_kind: str, row_cells: list[str]) -> tuple:
    return (
        tuple(heading_path),
        table_kind,
        tuple(normalize_table_cell(cell) for cell in row_cells),
    )


def fallback_manual_key(table_kind: str, row_cells: list[str]) -> tuple:
    return (
        table_kind,
        tuple(normalize_table_cell(cell) for cell in row_cells),
    )


def split_table_row(line: str) -> list[str]:
    """Split a simple Markdown table row into normalized cell values."""
    return [normalize_table_cell(cell) for cell in line.split('|')[1:-1]]


def extract_manual_descriptions(readme_text: str) -> dict[tuple, str]:
    """Extract manually maintained description cells from existing README tables."""
    result = {}
    heading_path: list[str] = []
    current_table_kind = None

    for line in readme_text.splitlines():
        heading_match = re.match(r'^(#{1,6})\s+(.*)$', line)
        if heading_match:
            level = len(heading_match.group(1))
            title = heading_match.group(2).strip()
            heading_path = heading_path[:level - 1]
            heading_path.append(title)
            current_table_kind = None
            continue

        stripped = line.strip()
        if stripped in MANUAL_TABLE_HEADERS:
            current_table_kind = MANUAL_TABLE_HEADERS[stripped]
            continue
        if current_table_kind and stripped.startswith('| ---'):
            continue
        if current_table_kind and stripped.startswith('|'):
            cells = split_table_row(line)
            if len(cells) >= 2:
                description = sanitize_manual_content(cells[-1])
                key_cells = cells[:-1]
                result[manual_key(tuple(heading_path), current_table_kind, key_cells)] = description
                if description:
                    result.setdefault(fallback_manual_key(current_table_kind, key_cells), description)
            continue
        if stripped:
            current_table_kind = None

    return result


def extract_manual_node_texts(readme_text: str) -> dict[str, str]:
    """Extract manually maintained free text directly below node headings."""
    result = {}
    lines = readme_text.splitlines()
    in_nodes_section = False
    i = 0

    while i < len(lines):
        line = lines[i]

        if re.match(r'^##\s+Nodes\s*$', line.strip()):
            in_nodes_section = True
            i += 1
            continue

        if in_nodes_section and re.match(r'^##\s+', line):
            in_nodes_section = False

        if not in_nodes_section:
            i += 1
            continue

        heading_match = re.match(r'^###\s+`([^`]+)`\s*$', line)
        if not heading_match:
            i += 1
            continue

        node_name = heading_match.group(1)
        i += 1
        body_lines = []

        while i < len(lines):
            current = lines[i]
            stripped = current.strip()

            if re.match(r'^##\s+', current) or re.match(r'^###\s+', current):
                break
            if stripped == '```mermaid' or re.match(r'^####\s+', current):
                break

            body_lines.append(current)
            i += 1

        trimmed = trim_blank_lines(body_lines)
        if trimmed:
            result[node_name] = '\n'.join(trimmed)

    return result


def render_manual_cell(
    manual_descriptions: dict[tuple, str],
    heading_path: tuple[str, ...],
    table_kind: str,
    row_cells: list[str],
) -> str:
    exact = manual_descriptions.get(manual_key(heading_path, table_kind, row_cells), '')
    if exact:
        return render_table_cell(exact)
    return render_table_cell(manual_descriptions.get(fallback_manual_key(table_kind, row_cells), ''))


def build_interface_rows(
    interfaces: list,
    type_attr: str,
    manual_descriptions: dict[tuple, str],
    heading_path: tuple[str, ...],
    table_kind: str,
    fallback_descriptions: Optional[dict[str, str]] = None,
) -> list[InterfaceTableRow]:
    rows = []
    fallback_descriptions = fallback_descriptions or {}
    for i in interfaces:
        description = render_manual_cell(
            manual_descriptions,
            heading_path,
            table_kind,
            [f"`{i.name}`", f"`{getattr(i, type_attr)}`"],
        )
        if description == 'TODO':
            description = render_table_cell(fallback_descriptions.get(i.name, ''))
        rows.append(
            InterfaceTableRow(
                name=i.name,
                interface_type=getattr(i, type_attr),
                description=description,
            )
        )
    return rows


def build_launch_argument_rows(
    args: list,
    manual_descriptions: dict[tuple, str],
    heading_path: tuple[str, ...],
) -> list[LaunchArgumentRow]:
    rows = []
    for name, default, description in args:
        name_cell = render_table_cell(name)
        default_cell = render_table_cell(default)
        manual_description = render_manual_cell(
            manual_descriptions,
            heading_path,
            'launch_arg',
            [f'`{name_cell}`', f'`{default_cell}`'],
        )
        rows.append(
            LaunchArgumentRow(
                name=name_cell,
                default=default_cell,
                description=render_table_cell(description) if manual_description == 'TODO' else manual_description,
            )
        )
    return rows


def build_interface_definition_entries(
    entries: list,
    manual_descriptions: dict[tuple, str],
    heading_path: tuple[str, ...],
) -> list[InterfaceDefinitionEntry]:
    return [
        InterfaceDefinitionEntry(
            full_type=full_type,
            rel_path=rel_path.as_posix(),
            description=render_manual_cell(
                manual_descriptions,
                heading_path,
                'interface',
                [f'[`{full_type}`]({rel_path.as_posix()})'],
            ),
        )
        for full_type, rel_path in entries
    ]


def build_parameter_rows(params: list[Parameter]) -> list[Parameter]:
    return [
        Parameter(
            name=render_table_cell(param.name),
            ros_type=render_table_cell(param.ros_type),
            default=render_parameter_default(param.default),
            description=render_table_cell(param.description),
        )
        for param in params
    ]


def build_launch_file_contexts(
    launch_files: list,
    doc_root: Path,
    manual_descriptions: dict[tuple, str],
) -> list[LaunchFileTemplateContext]:
    contexts = []
    for launch_file in launch_files:
        rel_path = launch_file.relative_to(doc_root).as_posix()
        contexts.append(
            LaunchFileTemplateContext(
                name=launch_file.name,
                rel_path=rel_path,
                arguments=build_launch_argument_rows(
                    extract_launch_arguments(launch_file),
                    manual_descriptions,
                    ('Launch Files', f'[`{launch_file.name}`]({rel_path})'),
                ),
            )
        )
    return contexts


def strip_rendered_string_default(value: str) -> str:
    """Remove the display quotes used for string launch defaults."""
    value = value.strip()
    if len(value) >= 2 and value[0] == '"' and value[-1] == '"':
        return value[1:-1]
    return value


def build_topic_descriptions_from_launch_files(launch_files: list) -> dict[str, str]:
    """Return topic descriptions keyed by default topic from launch remapping arguments."""
    descriptions: dict[str, str] = {}
    for launch_file in launch_files:
        for _, default, description in extract_launch_arguments(launch_file):
            topic = strip_rendered_string_default(default)
            if topic.startswith('~/') and description:
                descriptions.setdefault(topic, description)
    return descriptions


def add_dynamic_topic_parameter_descriptions(
    descriptions: dict[str, str],
    interfaces: list[TopicInterface],
    parameters: list[Parameter],
) -> dict[str, str]:
    """Describe indexed topic expressions using their corresponding parameter."""
    result = dict(descriptions)
    parameter_descriptions = {parameter.name: parameter.description for parameter in parameters}
    for interface in interfaces:
        match = re.fullmatch(r'([A-Za-z_]\w*)\s*\[[^\]]+\]', interface.name)
        if not match:
            continue
        symbol = match.group(1)
        for candidate in (symbol, symbol.rstrip('_')):
            description = parameter_descriptions.get(candidate)
            if description:
                result.setdefault(interface.name, description)
                break
    return result


def build_node_context(
    node: NodeInterfaces,
    manual_descriptions: dict[tuple, str],
    manual_node_texts: dict[str, str],
    topic_descriptions: dict[str, str],
) -> NodeTemplateContext:
    subscriber_descriptions = add_dynamic_topic_parameter_descriptions(
        topic_descriptions,
        node.subscribers,
        node.parameters,
    )
    publisher_descriptions = add_dynamic_topic_parameter_descriptions(
        topic_descriptions,
        node.publishers,
        node.parameters,
    )
    return NodeTemplateContext(
        node_name=node.node_name,
        manual_text=manual_node_texts.get(node.node_name, ''),
        subscribers=build_interface_rows(
            node.subscribers,
            'msg_type',
            manual_descriptions,
            ('Nodes', f'`{node.node_name}`', 'Subscribed Topics'),
            'topic',
            subscriber_descriptions,
        ),
        publishers=build_interface_rows(
            node.publishers,
            'msg_type',
            manual_descriptions,
            ('Nodes', f'`{node.node_name}`', 'Published Topics'),
            'topic',
            publisher_descriptions,
        ),
        service_servers=build_interface_rows(
            node.service_servers,
            'srv_type',
            manual_descriptions,
            ('Nodes', f'`{node.node_name}`', 'Service Servers'),
            'service',
        ),
        action_servers=build_interface_rows(
            node.action_servers,
            'action_type',
            manual_descriptions,
            ('Nodes', f'`{node.node_name}`', 'Action Servers'),
            'action',
        ),
        action_clients=build_interface_rows(
            node.action_clients,
            'action_type',
            manual_descriptions,
            ('Nodes', f'`{node.node_name}`', 'Action Clients'),
            'action',
        ),
        parameters=build_parameter_rows(node.parameters),
    )

def print_diff(old: str, new: str, path: Path) -> None:
    """Print a colored unified diff of old vs new to stderr."""
    RED, GREEN, CYAN, RESET = '\033[31m', '\033[32m', '\033[36m', '\033[0m'
    lines = list(difflib.unified_diff(
        old.splitlines(keepends=True),
        new.splitlines(keepends=True),
        fromfile=f'a/{path}',
        tofile=f'b/{path}',
    ))
    if not lines:
        sys.stderr.write('(no changes)\n')
        return
    for line in lines:
        if line.startswith('+') and not line.startswith('+++'):
            color = GREEN
        elif line.startswith('-') and not line.startswith('---'):
            color = RED
        elif line.startswith('@@'):
            color = CYAN
        else:
            color = ''
        sys.stderr.write(f'{color}{line}{RESET if color else ""}')


def extract_package_description(readme_text: str, fallback: str) -> str:
    """Return package.xml description plus preserved intro content before the first H2."""
    m = re.search(r'^#\s+.+?\n\n(.*?)(?=^##\s|\Z)', readme_text, re.DOTALL | re.MULTILINE)
    if not m:
        return fallback

    intro_block = m.group(1).strip()
    if not intro_block:
        return fallback

    paragraphs = re.split(r'\n\s*\n', intro_block)
    if len(paragraphs) <= 1:
        return fallback

    preserved_tail = '\n\n'.join(paragraphs[1:]).strip()
    if not preserved_tail:
        return fallback

    if fallback:
        return f'{fallback}\n\n{preserved_tail}'
    return preserved_tail


def render_package_readme(
    template_env: Environment,
    package_name: str,
    package_description: str,
    sections: list[PackageSection],
) -> str:
    context = PackageTemplateContext(
        package_name=package_name,
        package_description=package_description,
        sections=sections,
    )
    rendered = render_template(
        template_env,
        'package_readme.md.j2',
        asdict(context),
    )
    return rendered.rstrip() + '\n'


# ---------------------------------------------------------------------------
# Top-level README rendering
# ---------------------------------------------------------------------------

PRE_QUICKSTART_PLACEHOLDER = (
    '\n\n> [!IMPORTANT]\n'
    '> This repository is part of [***OpenADS***](https://openads-project.github.io/), '
    'the *Open Automated Driving Systems* project. *OpenADS* and its modules have '
    'been initiated and are currently being maintained by the '
    '[**Institute for Automotive Engineering (ika) at RWTH Aachen University**]'
    '(https://www.ika.rwth-aachen.de/de/).\n'
)


def trim_blank_lines(lines: list[str]) -> list[str]:
    """Remove leading and trailing blank lines from a list of lines."""
    trimmed = list(lines)
    while trimmed and not trimmed[0].strip():
        trimmed.pop(0)
    while trimmed and not trimmed[-1].strip():
        trimmed.pop()
    return trimmed


def build_template_environment() -> Environment:
    if Environment is None or FileSystemLoader is None:
        raise RuntimeError(
            'Missing dependency: jinja2. Install with '
            '`pip install -r .openads-dev-environment/scripts/requirements.txt`.'
        )
    templates_dir = Path(__file__).resolve().parent / 'templates'
    environment = Environment(
        loader=FileSystemLoader(str(templates_dir)),
        autoescape=False,
        keep_trailing_newline=True,
        trim_blocks=False,
        lstrip_blocks=False,
    )
    environment.filters['mermaid_edge_label'] = render_mermaid_edge_label
    return environment


def render_template(template_env: Environment, template_name: str, context: dict) -> str:
    try:
        template = template_env.get_template(template_name)
    except TemplateNotFound as exc:
        raise RuntimeError(f'Missing template: {template_name}') from exc
    return template.render(**context)


def get_origin_remote(repo_root: Path) -> str:
    """Return origin remote URL for the repository, or raise on failure."""
    return subprocess.check_output(
        ['git', 'remote', 'get-url', 'origin'],
        cwd=repo_root,
        stderr=subprocess.DEVNULL,
        text=True,
    ).strip()


def parse_repo_remote(remote: str) -> RepoMetadata:
    """Extract repository metadata from HTTPS and SSH git remotes."""
    remote = remote.strip()

    url_match = re.match(r'^[a-z][a-z0-9+.-]*://(?:(?:[^@/]+)@)?([^/]+)/(.+?)(?:\.git)?/?$', remote, re.IGNORECASE)
    ssh_match = re.match(r'^(?:[^@]+@)?([^:]+):(.+?)(?:\.git)?$', remote)

    if url_match:
        host, path = url_match.group(1), url_match.group(2)
    elif ssh_match:
        host, path = ssh_match.group(1), ssh_match.group(2)
    else:
        raise ValueError(f'Unsupported origin remote URL: {remote}')

    path_parts = [part for part in path.split('/') if part]
    if len(path_parts) < 2:
        raise ValueError(f'Unsupported origin remote URL: {remote}')

    owner = '/'.join(path_parts[:-1])
    repo = path_parts[-1]
    owner_lower = owner.lower()
    host_lower = host.lower()
    provider = 'github' if host_lower == 'github.com' else 'gitlab' if host_lower.startswith('gitlab.') else 'other'
    repo_https_url = f'https://{host}/{path}'

    pages_url = ''
    if provider == 'github':
        pages_url = f'https://{owner_lower}.github.io/{repo}'
    elif provider == 'gitlab':
        namespace_parts = owner_lower.split('/')
        if namespace_parts:
            pages_host = host_lower.removeprefix('gitlab.')
            pages_path = '/'.join([*namespace_parts[1:], repo])
            pages_url = f'https://{namespace_parts[0]}.pages.{pages_host}'
            if pages_path:
                pages_url = f'{pages_url}/{pages_path}'

    container_image = ''
    if provider == 'github':
        container_image = f'ghcr.io/{owner_lower}/{repo}:latest'
    elif provider == 'gitlab':
        container_image = f'{host}:5050/{path}:latest'

    return RepoMetadata(
        host=host,
        provider=provider,
        owner=owner,
        repo=repo,
        owner_lower=owner_lower,
        pages_url=pages_url,
        repo_https_url=repo_https_url,
        container_image=container_image,
    )




def extract_intro_block(readme_text: str) -> Optional[str]:
    """Return repo-specific intro block (headline + paragraph), if present."""
    m = re.search(
        r'^#\s+.+?\n\n(?:(?:<p align="center">\n.*?\n</p>|<table align="center">\n.*?\n</table>)\n\n)?(.*?)(?=\n<p align="center">\n  <strong>🚀|\n> \[!IMPORTANT\]|\n##[^\n]*Quick Start|\Z)',
        readme_text,
        re.DOTALL | re.MULTILINE,
    )
    if m and m.group(1).strip():
        return m.group(1).strip()
    return None


def extract_h2_section_body(readme_text: str, section_title: str) -> Optional[str]:
    """Return the body of an H2 section whose title contains section_title."""
    m = re.search(
        rf'^##[^\n]*{re.escape(section_title)}[^\n]*\n\n(.*?)(?=^##\s|\Z)',
        readme_text,
        re.DOTALL | re.MULTILINE,
    )
    if m and m.group(1).strip():
        return m.group(1).rstrip()
    return None


def extract_pre_quickstart_block(readme_text: str) -> str:
    """Return the preserved pre-Quick Start section or the initial default note."""
    m = re.search(
        r'<p align="center">\n'
        r'  <strong>🚀 <a href="#-quick-start">Quick Start</a></strong> • '
        r'<strong>💻 <a href="#-development">Development</a></strong> • '
        r'<strong>📝 <a href="#-documentation">Documentation</a></strong>\n'
        r'</p>(.*?)(?=\n##[^\n]*Quick Start|\Z)',
        readme_text,
        re.DOTALL,
    )
    if m is not None:
        return m.group(1)
    return PRE_QUICKSTART_PLACEHOLDER


def extract_quickstart_body(
    readme_text: str,
) -> Optional[str]:
    """Return repo-specific Quick Start body, if present."""
    body = extract_h2_section_body(readme_text, 'Quick Start')
    if body:
        return body
    return None


def extract_acknowledgements_body(readme_text: str) -> Optional[str]:
    """Return repo-specific acknowledgements body, if present."""
    body = extract_h2_section_body(readme_text, 'Acknowledgements')
    if body:
        return body
    return None




def extract_licensing_extra_body(readme_text: str) -> str:
    """Return repo-specific Licensing content beyond the fixed default body."""
    license_body = (
        'The source code in this repository is licensed under Apache-2.0, '
        'see [LICENSE](LICENSE). Container images provided by this repository '
        'may contain third-party software shipped with their own license terms.'
    )
    body = extract_h2_section_body(readme_text, 'Licensing')
    if not body:
        return ''

    if body.strip() == license_body:
        return ''

    if body.startswith(license_body):
        extra_lines = trim_blank_lines(body[len(license_body):].splitlines())
        if extra_lines:
            return '\n'.join(extra_lines)
        return ''

    return body


def build_package_doc_entries(
    repo_root: Path,
    packages: list,
) -> list[PackageDocEntry]:
    entries = []
    for pkg_name, pkg_dir, pkg_description in sorted(packages, key=lambda p: p[0]):
        rel_readme = (pkg_dir / 'README.md').relative_to(repo_root).as_posix()
        entries.append(
            PackageDocEntry(
                name=pkg_name,
                path=rel_readme,
                description=pkg_description or 'TODO: Describe the package purpose.',
            )
        )
    return entries


def pick_quickstart_target(packages: list) -> tuple[str, str]:
    """Choose package and launch file for quick start launch command."""
    for pkg_name, pkg_dir, _ in sorted(packages, key=lambda p: p[0]):
        launch_files = find_launch_files(pkg_dir)
        if launch_files:
            return pkg_name, launch_files[0].name
    return 'PACKAGE_NAME', 'LAUNCH_FILE'


def render_top_level_readme(
    template_env: Environment,
    repo_root: Path,
    packages: list,
    existing_readme: str,
) -> str:
    remote = get_origin_remote(repo_root)
    meta = parse_repo_remote(remote)
    intro_block = extract_intro_block(existing_readme)
    pre_quickstart_block = extract_pre_quickstart_block(existing_readme)
    ack_body = extract_acknowledgements_body(existing_readme)
    quickstart_pkg, quickstart_launch_file = pick_quickstart_target(packages)
    quickstart_body = extract_quickstart_body(existing_readme)
    package_doc_entries = build_package_doc_entries(
        repo_root,
        packages,
    )
    licensing_extra_body = extract_licensing_extra_body(existing_readme)
    context = TopLevelTemplateContext(
        title=meta.repo,
        repo_name=meta.repo,
        owner=meta.owner,
        owner_lower=meta.owner_lower,
        provider=meta.provider,
        pages_url=meta.pages_url,
        repo_https_url=meta.repo_https_url,
        container_image=meta.container_image,
        intro_block=intro_block,
        pre_quickstart_block=pre_quickstart_block,
        quickstart_body=quickstart_body,
        quickstart_package=quickstart_pkg,
        quickstart_launch_file=quickstart_launch_file,
        repository_packages=package_doc_entries,
        licensing_extra_body=licensing_extra_body,
        acknowledgements_body=ack_body,
    )
    rendered = render_template(
        template_env,
        'top_level_readme.md.j2',
        asdict(context),
    )
    return rendered.rstrip() + '\n'


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main():
    repo_root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(__file__).resolve().parents[2]
    try:
        template_env = build_template_environment()
    except RuntimeError as exc:
        print(str(exc), file=sys.stderr)
        sys.exit(2)

    packages = find_packages(repo_root)
    if not packages:
        print('Warning: no ROS packages found.', file=sys.stderr)

    for pkg_name, pkg_dir, pkg_description in packages:
        readme_path = pkg_dir / 'README.md'
        old_content = readme_path.read_text() if readme_path.exists() else ''
        pkg_description = extract_package_description(old_content, pkg_description)
        manual_descriptions = extract_manual_descriptions(old_content)
        manual_node_texts = extract_manual_node_texts(old_content)
        msgs = find_interface_files(pkg_dir, 'msg', 'msg')
        srvs = find_interface_files(pkg_dir, 'srv', 'srv')
        actions = find_interface_files(pkg_dir, 'action', 'action')
        node_sources = find_node_sources(pkg_dir)
        python_node_sources = find_python_node_sources(pkg_dir)
        launch_files = find_launch_files(pkg_dir)

        sections = []
        nodes = []

        if msgs:
            entries = [(f'{pkg_name}/msg/{f.stem}', f.relative_to(pkg_dir)) for f in msgs]
            sections.append(PackageSection(
                title='Messages',
                kind='interface_table',
                interface_entries=build_interface_definition_entries(
                    entries,
                    manual_descriptions,
                    ('Messages',),
                ),
            ))
        if srvs:
            entries = [(f'{pkg_name}/srv/{f.stem}', f.relative_to(pkg_dir)) for f in srvs]
            sections.append(PackageSection(
                title='Services',
                kind='interface_table',
                interface_entries=build_interface_definition_entries(
                    entries,
                    manual_descriptions,
                    ('Services',),
                ),
            ))
        if actions:
            entries = [(f'{pkg_name}/action/{f.stem}', f.relative_to(pkg_dir)) for f in actions]
            sections.append(PackageSection(
                title='Actions',
                kind='interface_table',
                interface_entries=build_interface_definition_entries(
                    entries,
                    manual_descriptions,
                    ('Actions',),
                ),
            ))

        if node_sources or python_node_sources or launch_files:
            headers = find_headers(pkg_dir)
            member_var_map = build_member_var_map(headers)
            type_aliases = build_type_alias_map(headers)
            topic_descriptions = build_topic_descriptions_from_launch_files(launch_files)

            for source_file in python_node_sources:
                node = extract_python_node_interfaces(
                    source_file.read_text(errors='replace'),
                    source_file.stem,
                )
                if node is not None:
                    nodes.append(node)

            for source_file in node_sources:
                source = source_file.read_text(errors='replace')
                source_context = '\n'.join([source, *(header.read_text(errors='replace') for header in headers)])
                string_symbols = extract_cpp_string_symbols(source_context)
                variable_types = extract_cpp_variable_types(source_context)
                source_type_aliases = {**type_aliases, **build_type_alias_map([source_file])}
                node = NodeInterfaces(
                    node_name=extract_node_name(source) or source_file.stem,
                    subscribers=extract_subscribers(source, source_type_aliases, string_symbols, variable_types),
                    publishers=extract_publishers(source, source_type_aliases, string_symbols, variable_types),
                    service_servers=extract_service_servers(source, source_type_aliases),
                    action_servers=extract_action_servers(source, source_type_aliases),
                    action_clients=extract_action_clients(source, source_type_aliases),
                    parameters=resolve_parameters(extract_raw_parameters(source), member_var_map, build_enum_value_map(headers)),
                )
                nodes.append(node)

            if nodes:
                sections.append(
                    PackageSection(
                        title='Nodes',
                        kind='nodes',
                        nodes=[
                            build_node_context(node, manual_descriptions, manual_node_texts, topic_descriptions)
                            for node in nodes
                        ],
                    )
                )

            if launch_files:
                sections.append(
                    PackageSection(
                        title='Launch Files',
                        kind='launch_files',
                        launch_files=build_launch_file_contexts(
                            launch_files,
                            pkg_dir,
                            manual_descriptions,
                        ),
                    )
                )

        new_content = render_package_readme(
            template_env=template_env,
            package_name=pkg_name,
            package_description=pkg_description,
            sections=sections,
        )
        readme_path.write_text(new_content)
        sys.stderr.write(f'Updated {readme_path}\n')
        print_diff(old_content, new_content, readme_path)

    root_readme_path = repo_root / 'README.md'
    old_root_readme = root_readme_path.read_text() if root_readme_path.exists() else ''
    new_root_readme = render_top_level_readme(template_env, repo_root, packages, old_root_readme)
    root_readme_path.write_text(new_root_readme)
    sys.stderr.write(f'Updated {root_readme_path}\n')
    print_diff(old_root_readme, new_root_readme, root_readme_path)


if __name__ == '__main__':
    main()