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',
25 'Goals':
'std::vector<geometry_msgs::msg::PoseStamped>',
29 TYPE_REGEX_TRANSFORMS = [
30 (re.compile(
r'std::'),
''),
34 DEFAULT_REGEX_TRANSFORMS = [
35 (re.compile(
r'std::'),
''),
39 TREE_NODES_MODEL_TAG =
'TreeNodesModel'
48 type NodePorts = dict[str, PortData]
49 type BTNodes = dict[str, NodePorts]
50 type CPPData = dict[str, str]
51 type HPPData = dict[str, NodePorts]
54 def git_root_path(path: Path) -> Path:
55 """Get the full path to the git root directory given any internal directory path."""
56 repo_path = subprocess.check_output(
57 [
'git',
'rev-parse',
'--show-toplevel'],
61 return Path(repo_path)
64 def clone_sparse_github_data(
68 data_to_clone: list[str],
71 """Clone GitHub repository sparsely and checkout the specified directories."""
72 repo_workdir = clone_dir / repo_name
73 if repo_workdir.exists():
76 github_url = f
'https://github.com/{owner}/{repo_name}.git'
77 print(f
'Cloning data from {github_url} (branch: {branch}).')
88 ], check=
True, text=
True)
90 print(f
'Performing sparse checkout in {repo_workdir} directory...')
97 ], cwd=repo_workdir, check=
True, text=
True)
98 except subprocess.CalledProcessError:
99 rmtree(repo_workdir, ignore_errors=
True)
103 for path
in data_to_clone:
104 full_path = repo_workdir / path
105 if not full_path.exists():
106 missing_paths.append(path)
109 rmtree(repo_workdir, ignore_errors=
True)
110 raise FileNotFoundError(
111 f
'Following paths do not exist in {github_url} (branch: {branch}): '
112 f
'{", ".join(missing_paths)}.\n'
115 print(f
'Cloned following data from {github_url} (branch: {branch}) to {repo_workdir}:')
116 for path
in data_to_clone:
117 print(f
'\t - {path}')
120 def fetch_external_repos(github_repos: dict, clone_dir: Path) ->
None:
121 """Fetch external repositories specified in the YAML configuration file."""
122 for repo_name, repo_info
in github_repos.items():
124 bt_paths: list[str] = []
125 bt = repo_info.get(
'behavior_trees', {})
126 for dir_paths
in [
'cpp_dir_paths',
'hpp_dir_paths']:
127 bt_paths.extend(bt.get(dir_paths, []))
129 base_classes: list[dict[str, str]] = bt.get(
'hpp_base_classes_paths', [])
130 for base_class
in base_classes:
131 bt_paths.extend(base_class.values())
133 clone_sparse_github_data(
135 owner=repo_info[
'owner'],
136 branch=repo_info[
'branch'],
137 data_to_clone=bt_paths,
142 def update_paths_for_external_repos(config: dict, clone_dir: Path) ->
None:
144 Update the directory paths for external repositories in-place.
146 Points paths to the cloned location and adds repository name.
148 for repo_name, repo_info
in config.items():
149 bt = repo_info.get(
'behavior_trees', {})
150 for key
in [
'cpp_dir_paths',
'hpp_dir_paths']:
154 updated_path = clone_dir / repo_name / path
155 updated_paths.append(updated_path)
156 bt[key] = updated_paths
158 base_classes: list[dict[str, str]] = bt.get(
'hpp_base_classes_paths', [])
159 for base_class
in base_classes:
160 for class_name, path
in base_class.items():
161 updated_path = clone_dir / repo_name / path
162 base_class[class_name] = updated_path
165 def get_files(directories: list[str], pattern: str) -> list[Path]:
166 """Recursively get all files matching given pattern from the specified list of directories."""
167 files: list[Path] = []
168 for directory
in directories:
169 dir_path = Path(directory)
170 dir_files = [file
for file
in dir_path.rglob(pattern)
if file.is_file()]
171 files.extend(dir_files)
175 def convert_with_regex(value: str, patterns: list[tuple[re.Pattern, str]]) -> str:
176 """Apply regex patterns to convert the given value."""
178 for pattern, replacement
in patterns:
179 result = pattern.sub(replacement, result)
183 def has_leading_comments(content: str, pos: int, comment_symbol: str) -> bool:
184 """Check if there are leading comments before the given position."""
185 line_start = content.rfind(
'\n', 0, pos) + 1
186 leading_content = content[line_start:pos].strip()
187 return leading_content.startswith(comment_symbol)
190 def is_quoted_string(string: str) -> bool:
191 """Check if the given string is quoted."""
192 return (string.startswith(
'"')
and string.endswith(
'"'))
195 def extract_template_data(content: str, template_start_pos: int) -> tuple[str, int]:
197 Extract template data starting from given position.
199 Return extracted template and the position of the closing angle bracket.
201 angle_bracket_count = 0
202 template_end_pos = -1
203 pos = template_start_pos
204 while pos < len(content):
207 angle_bracket_count += 1
209 angle_bracket_count -= 1
210 if angle_bracket_count == 0:
211 template_end_pos = pos
215 raise ValueError(
'Failed to extract template data: unmatched angle brackets.')
216 template_data = content[template_start_pos + 1:template_end_pos].strip()
217 return (template_data, template_end_pos)
220 def extract_arguments(content: str, args_start_pos: int) -> list[str]:
221 """Extract arguments from a function starting from given position."""
224 parentheses_count = 0
225 angle_brackets_count = 0
226 curly_brackets_count = 0
228 new_arg_start = args_start_pos + 1
230 while pos < len(content):
233 if char ==
'"' and (pos == 0
or content[pos-1] !=
'\\'):
234 inside_quote =
not inside_quote
242 parentheses_count += 1
244 parentheses_count -= 1
245 if parentheses_count == 0:
246 arg = content[new_arg_start:pos].strip()
251 angle_brackets_count += 1
253 angle_brackets_count -= 1
255 curly_brackets_count += 1
257 curly_brackets_count -= 1
259 if angle_brackets_count
or curly_brackets_count
or parentheses_count > 1:
264 arg = content[new_arg_start:pos].strip()
272 raise ValueError(
'Failed to extract arguments: unmatched parentheses.')
276 def extract_code_port_data(content: str) -> NodePorts:
278 Extract port information from the code.
280 Returns dictionary mapping port names to their data:
281 {port_name: {'data_type': 'x', 'default': 'y', 'has_description': bool}}
283 ports: NodePorts = {}
285 ports_code_pattern = re.compile(
r'BT::(?:Input|Output|Bidirectional)Port')
286 for port_match
in ports_code_pattern.finditer(content):
287 start_pos = port_match.end()
289 if has_leading_comments(content, start_pos,
'//'):
292 port_type, port_type_end = extract_template_data(content, start_pos)
293 port_type = TYPE_DIRECT_MAPPINGS.get(port_type, port_type)
294 port_type = convert_with_regex(port_type, TYPE_REGEX_TRANSFORMS)
296 args_start = content.find(
'(', port_type_end)
298 raise ValueError(
'Failed to extract port arguments: opening parenthesis not found.')
299 port_args = extract_arguments(content, args_start)
300 args_number = len(port_args)
302 raise ValueError(
'Failed to extract port arguments: no arguments found.')
304 port_name = port_args[0]
305 is_quoted = is_quoted_string(port_name)
307 raise ValueError(
'Port name must be a quoted string.')
309 port_name = port_name.strip(
'"')
311 raise ValueError(
'Failed to extract port name: empty string.')
317 'data_type': port_type,
319 'has_description':
False
323 port_description_exists = bool((port_args[1]).strip(
'"').strip())
325 'data_type': port_type,
327 'has_description': port_description_exists
331 port_default = port_args[1]
332 port_default = convert_with_regex(port_default, DEFAULT_REGEX_TRANSFORMS)
333 port_description_exists = bool((port_args[2]).strip(
'"').strip())
335 'data_type': port_type,
336 'default': port_default,
337 'has_description': port_description_exists
342 def validate_bt_xml_structure(root: ET.Element) -> ET.Element:
343 if root
is None or len(root) == 0:
345 'Invalid XML structure: '
346 f
'Expected <{TREE_NODES_MODEL_TAG}> element as the first child of the root.'
348 bt_nodes_model = root[0]
350 if bt_nodes_model.tag != TREE_NODES_MODEL_TAG:
352 'Invalid XML structure: '
353 f
'Expected <{TREE_NODES_MODEL_TAG}> element as the first child of the root.'
355 return bt_nodes_model
358 def extract_xml_nodes_data(content: ET.ElementTree[Any]) -> BTNodes:
360 Extract Behavior Tree nodes data from the given XML data.
362 Returns dictionary mapping node IDs to their port data:
363 {node_id: {port_name: {'data_type': 'x', 'default': 'y', 'has_description': bool}}}
365 root = content.getroot()
366 bt_nodes_model = validate_bt_xml_structure(root)
368 bt_node_ids_xml: BTNodes = {}
369 for node
in bt_nodes_model:
370 node_id = node.get(
'ID')
372 raise ValueError(
'Each BT node must have an "ID" attribute.')
373 if node_id
in bt_node_ids_xml:
374 raise ValueError(f
'Duplicate node ID found in XML: {node_id}')
376 ports: NodePorts = {}
378 port_name = port.get(
'name')
381 f
'Each port in {node_id} node must have a "name" attribute.'
383 if port_name
in ports:
385 f
'Duplicate port name {port_name} found in {node_id} node.'
387 port_type = port.get(
'type')
390 f
'{port_name} port in {node_id} node is missing a "type" attribute.'
392 port_default = port.get(
'default',
'')
393 port_description_exists = bool((port.text
or '').strip())
395 'data_type': port_type,
396 'default': port_default,
397 'has_description': port_description_exists
400 bt_node_ids_xml[node_id] = ports
401 return bt_node_ids_xml
404 def extract_node_registration_data(content: str) -> dict[str, str]:
406 Extract node registration data from the given content.
408 Returns dictionary mapping class names to node IDs: {class_name: node_id}
410 register_pattern = re.compile(
r'register(?:NodeType|Builder)')
412 register_data: dict[str, str] = {}
413 for register_match
in register_pattern.finditer(content):
414 start_pos = register_match.end()
416 if has_leading_comments(content, start_pos,
'//'):
419 register_type, register_type_end = extract_template_data(content, start_pos)
420 if not register_type:
421 raise ValueError(
'Failed to extract node registration template data.')
423 register_type = register_type.split(
'<')[0].strip()
424 class_name_match = re.match(
r'(?:.*::)?([a-zA-Z_]\w*)', register_type)
425 if not class_name_match:
426 raise ValueError(
'Failed to extract class name from node registration.')
427 class_name = class_name_match.group(1)
429 if class_name
in register_data:
430 raise ValueError(f
'Duplicate {class_name} class found.')
432 args_start = content.find(
'(', register_type_end)
435 'Failed to extract node registration arguments: opening parenthesis not found.'
438 register_args = extract_arguments(content, args_start)
439 if not register_args:
441 'Failed to extract node ID from node registration: no arguments found.'
444 node_id = register_args[0]
445 is_quoted = is_quoted_string(node_id)
447 raise ValueError(
'Node ID must be a quoted string.')
449 node_id = node_id.strip(
'"')
451 raise ValueError(
'Failed to extract node ID from node registration: empty string.')
453 if node_id
in register_data.values():
455 f
'Duplicate node ID registration for {class_name} class found: {node_id}'
458 register_data[class_name] = node_id
460 if not register_data:
461 raise ValueError(
'No node registration found.')
465 def extract_cpp_classes_and_ids(cpp_files: list[Path]) -> CPPData:
467 Extract class names and their corresponding node IDs from the given list of source files.
469 Returns dictionary mapping class names to node IDs: {class_name: node_id}
471 node_cpp_data: CPPData = {}
472 for cpp_file
in cpp_files:
473 cpp_content = cpp_file.read_text()
475 class_names_and_ids = extract_node_registration_data(cpp_content)
476 except ValueError
as exc:
478 f
'Failed to extract node registration data from {cpp_file}: {exc}\n')
480 all_classes = node_cpp_data.keys()
481 file_classes = class_names_and_ids.keys()
482 common_classes = all_classes & file_classes
485 f
'Duplicate class name found in {cpp_file}: '
486 f
'{", ".join(common_classes)}.'
488 all_ids = set(node_cpp_data.values())
489 file_ids = set(class_names_and_ids.values())
490 common_ids = all_ids & file_ids
493 f
'Duplicate node ID found in {cpp_file}: '
494 f
'{", ".join(common_ids)}.'
496 node_cpp_data.update(class_names_and_ids)
500 def extract_class_definitions(
502 ) -> list[tuple[str, str, str]]:
504 Extract class data from headers.
506 Returns a list of tuples: (class_name, base_class_name, class_section).
508 class_pattern = re.compile(
509 r'^\s*class\s+([A-Za-z_][A-Za-z0-9_]*)',
512 base_class_pattern = re.compile(
513 r'public\s+(?:[A-Za-z_][A-Za-z0-9_:]*::)?([A-Za-z_][A-Za-z0-9_]*)',
516 class_definitions: list[tuple[str, str, str]] = []
519 while pos < len(content):
520 class_match = class_pattern.search(content, pos)
523 class_name = class_match.group(1)
525 class_name_end = class_match.end()
526 class_brace = content.find(
'{', class_name_end)
527 if class_brace == -1:
529 'Failed to extract class definition: opening brace not found.'
531 base_class_area = content[class_name_end:class_brace]
532 base_class_matches = base_class_pattern.findall(base_class_area)
533 base_class_name = base_class_matches[0]
if base_class_matches
else ''
536 pos_brace_search = class_brace + 1
537 while pos_brace_search < len(content):
538 char = content[pos_brace_search]
544 pos_brace_search += 1
546 pos_brace_search += 1
548 raise ValueError(f
'Failed to parse class section for {class_name}.')
550 class_section = content[class_brace:pos_brace_search]
551 class_definitions.append((class_name, base_class_name, class_section))
553 pos = pos_brace_search
555 return class_definitions
558 def extract_hpp_classes_and_ports_data(
559 hpp_files: list[Path],
560 hpp_base_classes: dict[str, Path]
563 Extract class names and their corresponding port data from the given list of header files.
565 Returns dictionary mapping class names to their port data:
566 {class_name: {port_name: {'data_type': 'x', 'default': 'y', 'has_description': bool}}}
568 node_hpp_data: HPPData = {}
569 for hpp_file
in hpp_files:
570 content = hpp_file.read_text()
571 class_definitions = extract_class_definitions(content)
572 if not class_definitions:
574 f
'No class definitions found in {hpp_file}.')
575 for class_name, base_class_name, class_section
in class_definitions:
576 if class_name
in node_hpp_data:
578 f
'Duplicate class name found in {hpp_file}: {class_name}.'
581 ports = extract_code_port_data(class_section)
582 except ValueError
as exc:
584 f
'Failed to extract port data for {class_name} class in {hpp_file}: {exc}'
586 if base_class_name
in hpp_base_classes:
587 base_class_path = hpp_base_classes[base_class_name]
588 base_class_content = Path(base_class_path).read_text()
590 base_ports = extract_code_port_data(base_class_content)
591 except ValueError
as exc:
593 f
'Failed to extract port data for {class_name} class '
594 f
'in {base_class_path}: {exc}'
596 base_port_names = base_ports.keys()
597 node_port_names = ports.keys()
598 common_port_names = base_port_names & node_port_names
599 if common_port_names:
601 f
'Port name conflict between {class_name} class and '
602 f
'its base class {base_class_name}: '
603 f
'{", ".join(common_port_names)}.'
605 ports.update(base_ports)
606 node_hpp_data[class_name] = ports
610 def extract_code_nodes_data(config: dict) -> BTNodes:
612 Extract BT node data from code based on the provided configuration.
614 Returns dictionary mapping node IDs to their port data:
615 {node_id: {port_name: {'data_type': 'x', 'default': 'y', 'has_description': bool}}}
617 bt_node_ids_code: BTNodes = {}
620 hpp_base_classes: dict[str, Path] = {}
622 for _, repo_info
in config.items():
623 bt = repo_info.get(
'behavior_trees', {})
625 cpp_dirs = bt.get(
'cpp_dir_paths', [])
626 cpp_files = get_files(cpp_dirs,
'*.cpp')
628 hpp_dirs = bt.get(
'hpp_dir_paths', [])
629 hpp_files = get_files(hpp_dirs,
'*.hpp')
631 base_classes_config = bt.get(
'hpp_base_classes_paths', [])
632 for base_class_data
in base_classes_config:
633 for base_class_name, base_class_path
in base_class_data.items():
634 if base_class_name
in hpp_base_classes:
636 f
'Duplicate base class name found in configuration: {base_class_name}.'
638 hpp_base_classes[base_class_name] = base_class_path
640 node_cpp_data = extract_cpp_classes_and_ids(cpp_files)
641 node_hpp_data = extract_hpp_classes_and_ports_data(hpp_files, hpp_base_classes)
643 diff_classes_cpp_hpp = node_cpp_data.keys() - node_hpp_data.keys()
644 if diff_classes_cpp_hpp:
646 'Following classes are present in cpp files but missing in hpp files: '
647 f
'{", ".join(diff_classes_cpp_hpp)}.\n'
648 'Ensure that all provided cpp files have their corresponding hpp files.'
650 diff_classes_hpp_cpp = node_hpp_data.keys() - node_cpp_data.keys()
651 if diff_classes_hpp_cpp:
653 'Following classes are present in hpp files but missing in cpp files: '
654 f
'{", ".join(diff_classes_hpp_cpp)}.\n'
655 'Ensure that all provided hpp files have their corresponding cpp files.'
660 for class_name, node_id
in node_cpp_data.items():
661 bt_node_ids_code[node_id] = node_hpp_data[class_name]
663 return bt_node_ids_code
666 def detect_bt_nodes_mismatches(bt_node_ids_code: BTNodes, bt_node_ids_xml: BTNodes) -> bool:
668 Compare BT node data extracted from code and XML.
670 Compares node IDs, port names, data types, and default values.
671 Returns True if any mismatch is found, False otherwise.
673 is_mismatch_found =
False
675 diff_node_ids_xml_code = bt_node_ids_xml.keys() - bt_node_ids_code.keys()
676 if diff_node_ids_xml_code:
677 is_mismatch_found =
True
678 print(
'[ERROR] Nodes present in XML but missing in code:')
679 for node_id
in diff_node_ids_xml_code:
680 print(f
'\t - {node_id}')
682 diff_node_ids_code_xml = bt_node_ids_code.keys() - bt_node_ids_xml.keys()
683 if diff_node_ids_code_xml:
684 is_mismatch_found =
True
685 print(
'[ERROR] Nodes present in code but missing in XML:')
686 for node_id
in diff_node_ids_code_xml:
687 print(f
'\t - {node_id}')
689 common_nodes = bt_node_ids_code.keys() & bt_node_ids_xml.keys()
690 for node_name
in common_nodes:
691 ports_code = bt_node_ids_code[node_name].keys()
692 ports_xml = bt_node_ids_xml[node_name].keys()
694 diff_ports_xml_code = ports_xml - ports_code
695 if diff_ports_xml_code:
696 is_mismatch_found =
True
697 print(f
'[ERROR] {node_name} node: ports present in XML but missing in code:')
698 for port
in diff_ports_xml_code:
699 print(f
'\t - {port}')
701 diff_ports_code_xml = ports_code - ports_xml
702 if diff_ports_code_xml:
703 is_mismatch_found =
True
704 print(f
'[ERROR] {node_name} node: ports present in code but missing in XML:')
705 for port
in diff_ports_code_xml:
706 print(f
'\t - {port}')
708 common_ports = ports_code & ports_xml
709 for port
in common_ports:
710 port_type_code = bt_node_ids_code[node_name][port][
'data_type']
711 port_type_xml = bt_node_ids_xml[node_name][port][
'data_type']
712 if port_type_code != port_type_xml:
713 is_mismatch_found =
True
714 print(f
'[ERROR] {node_name} node: data type mismatch for {port} port:')
715 print(f
'\t Code: {port_type_code}')
716 print(f
'\t XML: {port_type_xml}')
718 port_default_code = bt_node_ids_code[node_name][port][
'default']
719 port_default_xml = bt_node_ids_xml[node_name][port][
'default']
720 if port_default_code != port_default_xml:
721 is_mismatch_found =
True
722 print(f
'[ERROR] {node_name} node: default value mismatch for {port} port:')
723 print(f
'\t Code: {port_default_code}')
724 print(f
'\t XML: {port_default_xml}')
725 return is_mismatch_found
728 def validate_descriptions(nodes: BTNodes) -> bool:
729 """Check for missing descriptions."""
730 is_description_missing =
False
731 for node_name, ports
in nodes.items():
732 for port, port_data
in ports.items():
733 has_description = port_data[
'has_description']
734 if not has_description:
735 print(f
'[ERROR] {node_name} node: missing description for {port} port.')
736 is_description_missing =
True
737 return is_description_missing
742 parser = argparse.ArgumentParser()
746 default=Path(
'tools/bt_nodes_validation/config.yml'),
747 help=
'YAML configuration file '
748 'containing path to the nav2_tree_nodes.xml file '
749 'and paths to repositories with code files to compare against.'
751 args = parser.parse_args()
752 args_config = args.config
755 config_yaml = args_config.read_text()
756 config = yaml.safe_load(config_yaml)
757 except (OSError, yaml.YAMLError)
as exc:
758 print(f
'Failed to load configuration file {args_config}: {exc}')
761 nav2_bt_nodes = config.get(
762 'nav2_bt_nodes_file_path',
763 'nav2_behavior_tree/nav2_tree_nodes.xml'
765 nav2_bt_nodes_file_path = Path(nav2_bt_nodes)
768 bt_nodes_content = ET.parse(nav2_bt_nodes_file_path)
769 except (OSError, ET.ParseError)
as exc:
770 print(f
'Failed to load BT nodes from {nav2_bt_nodes_file_path}: {exc}')
774 bt_node_ids_xml = extract_xml_nodes_data(bt_nodes_content)
775 except (ValueError, IndexError)
as exc:
776 print(f
'Failed to extract BT nodes data from XML: {exc}')
779 github_repos_config = config.get(
'github_repositories', {})
780 if github_repos_config:
784 clone_dir = git_root_path(Path(__file__).parent)
785 except (subprocess.CalledProcessError, FileNotFoundError)
as exc:
786 print(f
'Failed to determine git root directory: {exc}')
789 print(
'Cloning external repositories...')
791 fetch_external_repos(github_repos_config, clone_dir)
792 except FileNotFoundError
as exc:
794 f
'Failed to fetch external repositories: {exc}'
795 f
'Review specified paths in {args_config}.'
798 except (subprocess.CalledProcessError, OSError)
as exc:
799 stderr = getattr(exc,
'stderr',
None)
800 print(f
'Failed to fetch external repositories: {stderr or exc}')
803 update_paths_for_external_repos(github_repos_config, clone_dir)
805 local_repos_config = config.get(
'local_repositories', {})
806 repos_config = local_repos_config | github_repos_config
809 bt_node_ids_code = extract_code_nodes_data(repos_config)
810 except (OSError, ValueError)
as exc:
812 f
'Failed to extract BT nodes data from code: {exc}\n'
813 f
'Review specified files in {args_config}'
817 print(
'Comparing BT nodes data extracted from code and XML files...')
818 is_mismatch_found = detect_bt_nodes_mismatches(bt_node_ids_code, bt_node_ids_xml)
820 print(
'Checking for missing descriptions in XML...')
822 is_xml_description_missing = validate_descriptions(bt_node_ids_xml)
824 if is_mismatch_found
or is_xml_description_missing:
826 'Validation failed.\n'
827 'Please review BT nodes in code and '
828 f
'their corresponding XML definitions in {nav2_bt_nodes_file_path}.'
832 print(
'Validation successful. No mismatches found between code and XML BT nodes data.')
835 if __name__ ==
'__main__':