1 from __future__
import annotations
4 from pathlib
import Path
6 from shutil
import rmtree
9 from typing
import Any, TypedDict
10 import xml.etree.ElementTree
as ET
14 TYPE_DIRECT_MAPPINGS = {
16 'unsigned short':
'uint16',
17 'ActionResult::_error_code_type':
'uint16',
18 'Action::Result::_error_code_type':
'uint16',
19 'ActionResult::_num_retries_type':
'uint16',
20 'ActionResult::_planning_time_type':
'builtin_interfaces::msg::Duration',
21 'ActionResult::_coverage_path_type':
'opennav_coverage_msgs::msg::PathComponents',
22 'ActionResult::_success_type':
'bool',
23 'ActionResult::_route_type':
'nav2_msgs::msg::Route',
24 'ActionResult::_total_elapsed_time_type':
'builtin_interfaces::msg::Duration',
28 TYPE_REGEX_TRANSFORMS = [
29 (re.compile(
r'std::'),
''),
33 DEFAULT_REGEX_TRANSFORMS = [
34 (re.compile(
r'std::'),
''),
38 TREE_NODES_MODEL_TAG =
'TreeNodesModel'
47 type NodePorts = dict[str, PortData]
48 type BTNodes = dict[str, NodePorts]
49 type CPPData = dict[str, str]
50 type HPPData = dict[str, NodePorts]
53 def git_root_path(path: Path) -> Path:
54 """Get the full path to the git root directory given any internal directory path."""
55 repo_path = subprocess.check_output(
56 [
'git',
'rev-parse',
'--show-toplevel'],
60 return Path(repo_path)
63 def clone_sparse_github_data(
67 data_to_clone: list[str],
70 """Clone GitHub repository sparsely and checkout the specified directories."""
71 repo_workdir = clone_dir / repo_name
72 if repo_workdir.exists():
75 github_url = f
'https://github.com/{owner}/{repo_name}.git'
76 print(f
'Cloning data from {github_url} (branch: {branch}).')
87 ], check=
True, text=
True)
89 print(f
'Performing sparse checkout in {repo_workdir} directory...')
96 ], cwd=repo_workdir, check=
True, text=
True)
97 except subprocess.CalledProcessError:
98 rmtree(repo_workdir, ignore_errors=
True)
102 for path
in data_to_clone:
103 full_path = repo_workdir / path
104 if not full_path.exists():
105 missing_paths.append(path)
108 rmtree(repo_workdir, ignore_errors=
True)
109 raise FileNotFoundError(
110 f
'Following paths do not exist in {github_url} (branch: {branch}): '
111 f
'{", ".join(missing_paths)}.\n'
114 print(f
'Cloned following data from {github_url} (branch: {branch}) to {repo_workdir}:')
115 for path
in data_to_clone:
116 print(f
'\t - {path}')
119 def fetch_external_repos(github_repos: dict, clone_dir: Path) ->
None:
120 """Fetch external repositories specified in the YAML configuration file."""
121 for repo_name, repo_info
in github_repos.items():
123 bt_paths: list[str] = []
124 bt = repo_info.get(
'behavior_trees', {})
125 for dir_paths
in [
'cpp_dir_paths',
'hpp_dir_paths']:
126 bt_paths.extend(bt.get(dir_paths, []))
128 base_classes: list[dict[str, str]] = bt.get(
'hpp_base_classes_paths', [])
129 for base_class
in base_classes:
130 bt_paths.extend(base_class.values())
132 clone_sparse_github_data(
134 owner=repo_info[
'owner'],
135 branch=repo_info[
'branch'],
136 data_to_clone=bt_paths,
141 def update_paths_for_external_repos(config: dict, clone_dir: Path) ->
None:
143 Update the directory paths for external repositories in-place.
145 Points paths to the cloned location and adds repository name.
147 for repo_name, repo_info
in config.items():
148 bt = repo_info.get(
'behavior_trees', {})
149 for key
in [
'cpp_dir_paths',
'hpp_dir_paths']:
153 updated_path = clone_dir / repo_name / path
154 updated_paths.append(updated_path)
155 bt[key] = updated_paths
157 base_classes: list[dict[str, str]] = bt.get(
'hpp_base_classes_paths', [])
158 for base_class
in base_classes:
159 for class_name, path
in base_class.items():
160 updated_path = clone_dir / repo_name / path
161 base_class[class_name] = updated_path
164 def get_files(directories: list[str], pattern: str) -> list[Path]:
165 """Recursively get all files matching given pattern from the specified list of directories."""
166 files: list[Path] = []
167 for directory
in directories:
168 dir_path = Path(directory)
169 dir_files = [file
for file
in dir_path.rglob(pattern)
if file.is_file()]
170 files.extend(dir_files)
174 def convert_with_regex(value: str, patterns: list[tuple[re.Pattern, str]]) -> str:
175 """Apply regex patterns to convert the given value."""
177 for pattern, replacement
in patterns:
178 result = pattern.sub(replacement, result)
182 def has_leading_comments(content: str, pos: int, comment_symbol: str) -> bool:
183 """Check if there are leading comments before the given position."""
184 line_start = content.rfind(
'\n', 0, pos) + 1
185 leading_content = content[line_start:pos].strip()
186 return leading_content.startswith(comment_symbol)
189 def is_quoted_string(string: str) -> bool:
190 """Check if the given string is quoted."""
191 return (string.startswith(
'"')
and string.endswith(
'"'))
194 def extract_template_data(content: str, template_start_pos: int) -> tuple[str, int]:
196 Extract template data starting from given position.
198 Return extracted template and the position of the closing angle bracket.
200 angle_bracket_count = 0
201 template_end_pos = -1
202 pos = template_start_pos
203 while pos < len(content):
206 angle_bracket_count += 1
208 angle_bracket_count -= 1
209 if angle_bracket_count == 0:
210 template_end_pos = pos
214 raise ValueError(
'Failed to extract template data: unmatched angle brackets.')
215 template_data = content[template_start_pos + 1:template_end_pos].strip()
216 return (template_data, template_end_pos)
219 def extract_arguments(content: str, args_start_pos: int) -> list[str]:
220 """Extract arguments from a function starting from given position."""
223 parentheses_count = 0
224 angle_brackets_count = 0
225 curly_brackets_count = 0
227 new_arg_start = args_start_pos + 1
229 while pos < len(content):
232 if char ==
'"' and (pos == 0
or content[pos-1] !=
'\\'):
233 inside_quote =
not inside_quote
241 parentheses_count += 1
243 parentheses_count -= 1
244 if parentheses_count == 0:
245 arg = content[new_arg_start:pos].strip()
250 angle_brackets_count += 1
252 angle_brackets_count -= 1
254 curly_brackets_count += 1
256 curly_brackets_count -= 1
258 if angle_brackets_count
or curly_brackets_count
or parentheses_count > 1:
263 arg = content[new_arg_start:pos].strip()
271 raise ValueError(
'Failed to extract arguments: unmatched parentheses.')
275 def extract_code_port_data(content: str) -> NodePorts:
277 Extract port information from the code.
279 Returns dictionary mapping port names to their data:
280 {port_name: {'data_type': 'x', 'default': 'y', 'has_description': bool}}
282 ports: NodePorts = {}
284 ports_code_pattern = re.compile(
r'BT::(?:Input|Output|Bidirectional)Port')
285 for port_match
in ports_code_pattern.finditer(content):
286 start_pos = port_match.end()
288 if has_leading_comments(content, start_pos,
'//'):
291 port_type, port_type_end = extract_template_data(content, start_pos)
292 port_type = TYPE_DIRECT_MAPPINGS.get(port_type, port_type)
293 port_type = convert_with_regex(port_type, TYPE_REGEX_TRANSFORMS)
295 args_start = content.find(
'(', port_type_end)
297 raise ValueError(
'Failed to extract port arguments: opening parenthesis not found.')
298 port_args = extract_arguments(content, args_start)
299 args_number = len(port_args)
301 raise ValueError(
'Failed to extract port arguments: no arguments found.')
303 port_name = port_args[0]
304 is_quoted = is_quoted_string(port_name)
306 raise ValueError(
'Port name must be a quoted string.')
308 port_name = port_name.strip(
'"')
310 raise ValueError(
'Failed to extract port name: empty string.')
316 'data_type': port_type,
318 'has_description':
False
322 port_description_exists = bool((port_args[1]).strip(
'"').strip())
324 'data_type': port_type,
326 'has_description': port_description_exists
330 port_default = port_args[1]
331 port_default = convert_with_regex(port_default, DEFAULT_REGEX_TRANSFORMS)
332 port_description_exists = bool((port_args[2]).strip(
'"').strip())
334 'data_type': port_type,
335 'default': port_default,
336 'has_description': port_description_exists
341 def validate_bt_xml_structure(root: ET.Element) -> ET.Element:
342 if root
is None or len(root) == 0:
344 'Invalid XML structure: '
345 f
'Expected <{TREE_NODES_MODEL_TAG}> element as the first child of the root.'
347 bt_nodes_model = root[0]
349 if bt_nodes_model.tag != TREE_NODES_MODEL_TAG:
351 'Invalid XML structure: '
352 f
'Expected <{TREE_NODES_MODEL_TAG}> element as the first child of the root.'
354 return bt_nodes_model
357 def extract_xml_nodes_data(content: ET.ElementTree[Any]) -> BTNodes:
359 Extract Behavior Tree nodes data from the given XML data.
361 Returns dictionary mapping node IDs to their port data:
362 {node_id: {port_name: {'data_type': 'x', 'default': 'y', 'has_description': bool}}}
364 root = content.getroot()
365 bt_nodes_model = validate_bt_xml_structure(root)
367 bt_node_ids_xml: BTNodes = {}
368 for node
in bt_nodes_model:
369 node_id = node.get(
'ID')
371 raise ValueError(
'Each BT node must have an "ID" attribute.')
372 if node_id
in bt_node_ids_xml:
373 raise ValueError(f
'Duplicate node ID found in XML: {node_id}')
375 ports: NodePorts = {}
377 port_name = port.get(
'name')
380 f
'Each port in {node_id} node must have a "name" attribute.'
382 if port_name
in ports:
384 f
'Duplicate port name {port_name} found in {node_id} node.'
386 port_type = port.get(
'type')
389 f
'{port_name} port in {node_id} node is missing a "type" attribute.'
391 port_default = port.get(
'default',
'')
392 port_description_exists = bool((port.text
or '').strip())
394 'data_type': port_type,
395 'default': port_default,
396 'has_description': port_description_exists
399 bt_node_ids_xml[node_id] = ports
400 return bt_node_ids_xml
403 def extract_node_registration_data(content: str) -> dict[str, str]:
405 Extract node registration data from the given content.
407 Returns dictionary mapping class names to node IDs: {class_name: node_id}
409 register_pattern = re.compile(
r'register(?:NodeType|Builder)')
411 register_data: dict[str, str] = {}
412 for register_match
in register_pattern.finditer(content):
413 start_pos = register_match.end()
415 if has_leading_comments(content, start_pos,
'//'):
418 register_type, register_type_end = extract_template_data(content, start_pos)
419 if not register_type:
420 raise ValueError(
'Failed to extract node registration template data.')
422 register_type = register_type.split(
'<')[0].strip()
423 class_name_match = re.match(
r'(?:.*::)?([a-zA-Z_]\w*)', register_type)
424 if not class_name_match:
425 raise ValueError(
'Failed to extract class name from node registration.')
426 class_name = class_name_match.group(1)
428 if class_name
in register_data:
429 raise ValueError(f
'Duplicate {class_name} class found.')
431 args_start = content.find(
'(', register_type_end)
434 'Failed to extract node registration arguments: opening parenthesis not found.'
437 register_args = extract_arguments(content, args_start)
438 if not register_args:
440 'Failed to extract node ID from node registration: no arguments found.'
443 node_id = register_args[0]
444 is_quoted = is_quoted_string(node_id)
446 raise ValueError(
'Node ID must be a quoted string.')
448 node_id = node_id.strip(
'"')
450 raise ValueError(
'Failed to extract node ID from node registration: empty string.')
452 if node_id
in register_data.values():
454 f
'Duplicate node ID registration for {class_name} class found: {node_id}'
457 register_data[class_name] = node_id
459 if not register_data:
460 raise ValueError(
'No node registration found.')
464 def extract_cpp_classes_and_ids(cpp_files: list[Path]) -> CPPData:
466 Extract class names and their corresponding node IDs from the given list of source files.
468 Returns dictionary mapping class names to node IDs: {class_name: node_id}
470 node_cpp_data: CPPData = {}
471 for cpp_file
in cpp_files:
472 cpp_content = cpp_file.read_text()
474 class_names_and_ids = extract_node_registration_data(cpp_content)
475 except ValueError
as exc:
477 f
'Failed to extract node registration data from {cpp_file}: {exc}\n')
479 all_classes = node_cpp_data.keys()
480 file_classes = class_names_and_ids.keys()
481 common_classes = all_classes & file_classes
484 f
'Duplicate class name found in {cpp_file}: '
485 f
'{", ".join(common_classes)}.'
487 all_ids = set(node_cpp_data.values())
488 file_ids = set(class_names_and_ids.values())
489 common_ids = all_ids & file_ids
492 f
'Duplicate node ID found in {cpp_file}: '
493 f
'{", ".join(common_ids)}.'
495 node_cpp_data.update(class_names_and_ids)
499 def extract_class_definitions(
501 ) -> list[tuple[str, str, str]]:
503 Extract class data from headers.
505 Returns a list of tuples: (class_name, base_class_name, class_section).
507 class_pattern = re.compile(
508 r'^\s*class\s+([A-Za-z_][A-Za-z0-9_]*)',
511 base_class_pattern = re.compile(
512 r'public\s+(?:[A-Za-z_][A-Za-z0-9_:]*::)?([A-Za-z_][A-Za-z0-9_]*)',
515 class_definitions: list[tuple[str, str, str]] = []
518 while pos < len(content):
519 class_match = class_pattern.search(content, pos)
522 class_name = class_match.group(1)
524 class_name_end = class_match.end()
525 class_brace = content.find(
'{', class_name_end)
526 if class_brace == -1:
528 'Failed to extract class definition: opening brace not found.'
530 base_class_area = content[class_name_end:class_brace]
531 base_class_matches = base_class_pattern.findall(base_class_area)
532 base_class_name = base_class_matches[0]
if base_class_matches
else ''
535 pos_brace_search = class_brace + 1
536 while pos_brace_search < len(content):
537 char = content[pos_brace_search]
543 pos_brace_search += 1
545 pos_brace_search += 1
547 raise ValueError(f
'Failed to parse class section for {class_name}.')
549 class_section = content[class_brace:pos_brace_search]
550 class_definitions.append((class_name, base_class_name, class_section))
552 pos = pos_brace_search
554 return class_definitions
557 def extract_hpp_classes_and_ports_data(
558 hpp_files: list[Path],
559 hpp_base_classes: dict[str, Path]
562 Extract class names and their corresponding port data from the given list of header files.
564 Returns dictionary mapping class names to their port data:
565 {class_name: {port_name: {'data_type': 'x', 'default': 'y', 'has_description': bool}}}
567 node_hpp_data: HPPData = {}
568 for hpp_file
in hpp_files:
569 content = hpp_file.read_text()
570 class_definitions = extract_class_definitions(content)
571 if not class_definitions:
573 f
'No class definitions found in {hpp_file}.')
574 for class_name, base_class_name, class_section
in class_definitions:
575 if class_name
in node_hpp_data:
577 f
'Duplicate class name found in {hpp_file}: {class_name}.'
580 ports = extract_code_port_data(class_section)
581 except ValueError
as exc:
583 f
'Failed to extract port data for {class_name} class in {hpp_file}: {exc}'
585 if base_class_name
in hpp_base_classes:
586 base_class_path = hpp_base_classes[base_class_name]
587 base_class_content = Path(base_class_path).read_text()
589 base_ports = extract_code_port_data(base_class_content)
590 except ValueError
as exc:
592 f
'Failed to extract port data for {class_name} class '
593 f
'in {base_class_path}: {exc}'
595 base_port_names = base_ports.keys()
596 node_port_names = ports.keys()
597 common_port_names = base_port_names & node_port_names
598 if common_port_names:
600 f
'Port name conflict between {class_name} class and '
601 f
'its base class {base_class_name}: '
602 f
'{", ".join(common_port_names)}.'
604 ports.update(base_ports)
605 node_hpp_data[class_name] = ports
609 def extract_code_nodes_data(config: dict) -> BTNodes:
611 Extract BT node data from code based on the provided configuration.
613 Returns dictionary mapping node IDs to their port data:
614 {node_id: {port_name: {'data_type': 'x', 'default': 'y', 'has_description': bool}}}
616 bt_node_ids_code: BTNodes = {}
619 hpp_base_classes: dict[str, Path] = {}
621 for _, repo_info
in config.items():
622 bt = repo_info.get(
'behavior_trees', {})
624 cpp_dirs = bt.get(
'cpp_dir_paths', [])
625 cpp_files = get_files(cpp_dirs,
'*.cpp')
627 hpp_dirs = bt.get(
'hpp_dir_paths', [])
628 hpp_files = get_files(hpp_dirs,
'*.hpp')
630 base_classes_config = bt.get(
'hpp_base_classes_paths', [])
631 for base_class_data
in base_classes_config:
632 for base_class_name, base_class_path
in base_class_data.items():
633 if base_class_name
in hpp_base_classes:
635 f
'Duplicate base class name found in configuration: {base_class_name}.'
637 hpp_base_classes[base_class_name] = base_class_path
639 node_cpp_data = extract_cpp_classes_and_ids(cpp_files)
640 node_hpp_data = extract_hpp_classes_and_ports_data(hpp_files, hpp_base_classes)
642 diff_classes_cpp_hpp = node_cpp_data.keys() - node_hpp_data.keys()
643 if diff_classes_cpp_hpp:
645 'Following classes are present in cpp files but missing in hpp files: '
646 f
'{", ".join(diff_classes_cpp_hpp)}.\n'
647 'Ensure that all provided cpp files have their corresponding hpp files.'
649 diff_classes_hpp_cpp = node_hpp_data.keys() - node_cpp_data.keys()
650 if diff_classes_hpp_cpp:
652 'Following classes are present in hpp files but missing in cpp files: '
653 f
'{", ".join(diff_classes_hpp_cpp)}.\n'
654 'Ensure that all provided hpp files have their corresponding cpp files.'
659 for class_name, node_id
in node_cpp_data.items():
660 bt_node_ids_code[node_id] = node_hpp_data[class_name]
662 return bt_node_ids_code
665 def detect_bt_nodes_mismatches(bt_node_ids_code: BTNodes, bt_node_ids_xml: BTNodes) -> bool:
667 Compare BT node data extracted from code and XML.
669 Compares node IDs, port names, data types, and default values.
670 Returns True if any mismatch is found, False otherwise.
672 is_mismatch_found =
False
674 diff_node_ids_xml_code = bt_node_ids_xml.keys() - bt_node_ids_code.keys()
675 if diff_node_ids_xml_code:
676 is_mismatch_found =
True
677 print(
'[ERROR] Nodes present in XML but missing in code:')
678 for node_id
in diff_node_ids_xml_code:
679 print(f
'\t - {node_id}')
681 diff_node_ids_code_xml = bt_node_ids_code.keys() - bt_node_ids_xml.keys()
682 if diff_node_ids_code_xml:
683 is_mismatch_found =
True
684 print(
'[ERROR] Nodes present in code but missing in XML:')
685 for node_id
in diff_node_ids_code_xml:
686 print(f
'\t - {node_id}')
688 common_nodes = bt_node_ids_code.keys() & bt_node_ids_xml.keys()
689 for node_name
in common_nodes:
690 ports_code = bt_node_ids_code[node_name].keys()
691 ports_xml = bt_node_ids_xml[node_name].keys()
693 diff_ports_xml_code = ports_xml - ports_code
694 if diff_ports_xml_code:
695 is_mismatch_found =
True
696 print(f
'[ERROR] {node_name} node: ports present in XML but missing in code:')
697 for port
in diff_ports_xml_code:
698 print(f
'\t - {port}')
700 diff_ports_code_xml = ports_code - ports_xml
701 if diff_ports_code_xml:
702 is_mismatch_found =
True
703 print(f
'[ERROR] {node_name} node: ports present in code but missing in XML:')
704 for port
in diff_ports_code_xml:
705 print(f
'\t - {port}')
707 common_ports = ports_code & ports_xml
708 for port
in common_ports:
709 port_type_code = bt_node_ids_code[node_name][port][
'data_type']
710 port_type_xml = bt_node_ids_xml[node_name][port][
'data_type']
711 if port_type_code != port_type_xml:
712 is_mismatch_found =
True
713 print(f
'[ERROR] {node_name} node: data type mismatch for {port} port:')
714 print(f
'\t Code: {port_type_code}')
715 print(f
'\t XML: {port_type_xml}')
717 port_default_code = bt_node_ids_code[node_name][port][
'default']
718 port_default_xml = bt_node_ids_xml[node_name][port][
'default']
719 if port_default_code != port_default_xml:
720 is_mismatch_found =
True
721 print(f
'[ERROR] {node_name} node: default value mismatch for {port} port:')
722 print(f
'\t Code: {port_default_code}')
723 print(f
'\t XML: {port_default_xml}')
724 return is_mismatch_found
727 def validate_descriptions(nodes: BTNodes) -> bool:
728 """Check for missing descriptions."""
729 is_description_missing =
False
730 for node_name, ports
in nodes.items():
731 for port, port_data
in ports.items():
732 has_description = port_data[
'has_description']
733 if not has_description:
734 print(f
'[ERROR] {node_name} node: missing description for {port} port.')
735 is_description_missing =
True
736 return is_description_missing
741 parser = argparse.ArgumentParser()
745 default=Path(
'tools/bt_nodes_validation/config.yml'),
746 help=
'YAML configuration file '
747 'containing path to the nav2_tree_nodes.xml file '
748 'and paths to repositories with code files to compare against.'
750 args = parser.parse_args()
751 args_config = args.config
754 config_yaml = args_config.read_text()
755 config = yaml.safe_load(config_yaml)
756 except (OSError, yaml.YAMLError)
as exc:
757 print(f
'Failed to load configuration file {args_config}: {exc}')
760 nav2_bt_nodes = config.get(
761 'nav2_bt_nodes_file_path',
762 'nav2_behavior_tree/nav2_tree_nodes.xml'
764 nav2_bt_nodes_file_path = Path(nav2_bt_nodes)
767 bt_nodes_content = ET.parse(nav2_bt_nodes_file_path)
768 except (OSError, ET.ParseError)
as exc:
769 print(f
'Failed to load BT nodes from {nav2_bt_nodes_file_path}: {exc}')
773 bt_node_ids_xml = extract_xml_nodes_data(bt_nodes_content)
774 except (ValueError, IndexError)
as exc:
775 print(f
'Failed to extract BT nodes data from XML: {exc}')
778 github_repos_config = config.get(
'github_repositories', {})
779 if github_repos_config:
783 clone_dir = git_root_path(Path(__file__).parent)
784 except (subprocess.CalledProcessError, FileNotFoundError)
as exc:
785 print(f
'Failed to determine git root directory: {exc}')
788 print(
'Cloning external repositories...')
790 fetch_external_repos(github_repos_config, clone_dir)
791 except FileNotFoundError
as exc:
793 f
'Failed to fetch external repositories: {exc}'
794 f
'Review specified paths in {args_config}.'
797 except (subprocess.CalledProcessError, OSError)
as exc:
798 stderr = getattr(exc,
'stderr',
None)
799 print(f
'Failed to fetch external repositories: {stderr or exc}')
802 update_paths_for_external_repos(github_repos_config, clone_dir)
804 local_repos_config = config.get(
'local_repositories', {})
805 repos_config = local_repos_config | github_repos_config
808 bt_node_ids_code = extract_code_nodes_data(repos_config)
809 except (OSError, ValueError)
as exc:
811 f
'Failed to extract BT nodes data from code: {exc}\n'
812 f
'Review specified files in {args_config}'
816 print(
'Comparing BT nodes data extracted from code and XML files...')
817 is_mismatch_found = detect_bt_nodes_mismatches(bt_node_ids_code, bt_node_ids_xml)
819 print(
'Checking for missing descriptions in XML...')
821 is_xml_description_missing = validate_descriptions(bt_node_ids_xml)
823 if is_mismatch_found
or is_xml_description_missing:
825 'Validation failed.\n'
826 'Please review BT nodes in code and '
827 f
'their corresponding XML definitions in {nav2_bt_nodes_file_path}.'
831 print(
'Validation successful. No mismatches found between code and XML BT nodes data.')
834 if __name__ ==
'__main__':