Nav2 Navigation Stack - rolling  main
ROS 2 Navigation Stack
validate_bt_xml_nodes.py
1 from __future__ import annotations
2 
3 import argparse
4 from pathlib import Path
5 import re
6 from shutil import rmtree
7 import subprocess
8 import sys
9 from typing import Any, TypedDict
10 import xml.etree.ElementTree as ET
11 
12 import yaml
13 
14 TYPE_DIRECT_MAPPINGS = {
15  'uint16_t': 'uint16',
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 }
26 
27 
28 TYPE_REGEX_TRANSFORMS = [
29  (re.compile(r'std::'), ''),
30 ]
31 
32 
33 DEFAULT_REGEX_TRANSFORMS = [
34  (re.compile(r'std::'), ''),
35 ]
36 
37 
38 TREE_NODES_MODEL_TAG = 'TreeNodesModel'
39 
40 
41 class PortData(TypedDict):
42  data_type: str
43  default: str
44  has_description: bool
45 
46 
47 type NodePorts = dict[str, PortData] # {port_name: PortData}
48 type BTNodes = dict[str, NodePorts] # {node_id: {port_name: PortData}}
49 type CPPData = dict[str, str] # {class_name: node_id}
50 type HPPData = dict[str, NodePorts] # {class_name: {port_name: PortData}}
51 
52 
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'],
57  text=True,
58  cwd=path,
59  ).strip()
60  return Path(repo_path)
61 
62 
63 def clone_sparse_github_data(
64  repo_name: str,
65  owner: str,
66  branch: str,
67  data_to_clone: list[str],
68  clone_dir: Path
69 ) -> None:
70  """Clone GitHub repository sparsely and checkout the specified directories."""
71  repo_workdir = clone_dir / repo_name
72  if repo_workdir.exists():
73  rmtree(repo_workdir)
74 
75  github_url = f'https://github.com/{owner}/{repo_name}.git'
76  print(f'Cloning data from {github_url} (branch: {branch}).')
77 
78  try:
79  subprocess.run([
80  'git', 'clone',
81  '--depth=1',
82  '--filter=blob:none',
83  '--sparse',
84  '--branch', branch,
85  github_url,
86  repo_workdir,
87  ], check=True, text=True)
88 
89  print(f'Performing sparse checkout in {repo_workdir} directory...')
90  subprocess.run([
91  'git',
92  'sparse-checkout',
93  'set',
94  '--no-cone',
95  *data_to_clone,
96  ], cwd=repo_workdir, check=True, text=True)
97  except subprocess.CalledProcessError:
98  rmtree(repo_workdir, ignore_errors=True)
99  raise
100 
101  missing_paths = []
102  for path in data_to_clone:
103  full_path = repo_workdir / path
104  if not full_path.exists():
105  missing_paths.append(path)
106 
107  if missing_paths:
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'
112  )
113 
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}')
117 
118 
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():
122 
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, []))
127 
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())
131 
132  clone_sparse_github_data(
133  repo_name=repo_name,
134  owner=repo_info['owner'],
135  branch=repo_info['branch'],
136  data_to_clone=bt_paths,
137  clone_dir=clone_dir
138  )
139 
140 
141 def update_paths_for_external_repos(config: dict, clone_dir: Path) -> None:
142  """
143  Update the directory paths for external repositories in-place.
144 
145  Points paths to the cloned location and adds repository name.
146  """
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']:
150  if key in bt:
151  updated_paths = []
152  for path in bt[key]:
153  updated_path = clone_dir / repo_name / path
154  updated_paths.append(updated_path)
155  bt[key] = updated_paths
156 
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
162 
163 
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)
171  return files
172 
173 
174 def convert_with_regex(value: str, patterns: list[tuple[re.Pattern, str]]) -> str:
175  """Apply regex patterns to convert the given value."""
176  result = value
177  for pattern, replacement in patterns:
178  result = pattern.sub(replacement, result)
179  return result
180 
181 
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)
187 
188 
189 def is_quoted_string(string: str) -> bool:
190  """Check if the given string is quoted."""
191  return (string.startswith('"') and string.endswith('"'))
192 
193 
194 def extract_template_data(content: str, template_start_pos: int) -> tuple[str, int]:
195  """
196  Extract template data starting from given position.
197 
198  Return extracted template and the position of the closing angle bracket.
199  """
200  angle_bracket_count = 0
201  template_end_pos = -1
202  pos = template_start_pos
203  while pos < len(content):
204  char = content[pos]
205  if char == '<':
206  angle_bracket_count += 1
207  if char == '>':
208  angle_bracket_count -= 1
209  if angle_bracket_count == 0:
210  template_end_pos = pos
211  break
212  pos += 1
213  else:
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)
217 
218 
219 def extract_arguments(content: str, args_start_pos: int) -> list[str]:
220  """Extract arguments from a function starting from given position."""
221  args: list[str] = []
222 
223  parentheses_count = 0
224  angle_brackets_count = 0
225  curly_brackets_count = 0
226  inside_quote = False
227  new_arg_start = args_start_pos + 1
228  pos = args_start_pos
229  while pos < len(content):
230  char = content[pos]
231 
232  if char == '"' and (pos == 0 or content[pos-1] != '\\'):
233  inside_quote = not inside_quote
234 
235  if inside_quote:
236  pos += 1
237  continue
238 
239  match char:
240  case '(':
241  parentheses_count += 1
242  case ')':
243  parentheses_count -= 1
244  if parentheses_count == 0:
245  arg = content[new_arg_start:pos].strip()
246  if arg:
247  args.append(arg)
248  break
249  case '<':
250  angle_brackets_count += 1
251  case '>':
252  angle_brackets_count -= 1
253  case '{':
254  curly_brackets_count += 1
255  case '}':
256  curly_brackets_count -= 1
257 
258  if angle_brackets_count or curly_brackets_count or parentheses_count > 1:
259  pos += 1
260  continue
261 
262  if char == ',':
263  arg = content[new_arg_start:pos].strip()
264  if arg:
265  args.append(arg)
266  pos += 1
267  new_arg_start = pos
268  continue
269  pos += 1
270  else:
271  raise ValueError('Failed to extract arguments: unmatched parentheses.')
272  return args
273 
274 
275 def extract_code_port_data(content: str) -> NodePorts:
276  """
277  Extract port information from the code.
278 
279  Returns dictionary mapping port names to their data:
280  {port_name: {'data_type': 'x', 'default': 'y', 'has_description': bool}}
281  """
282  ports: NodePorts = {}
283 
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()
287 
288  if has_leading_comments(content, start_pos, '//'):
289  continue
290 
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)
294 
295  args_start = content.find('(', port_type_end)
296  if args_start == -1:
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)
300  if not args_number:
301  raise ValueError('Failed to extract port arguments: no arguments found.')
302 
303  port_name = port_args[0]
304  is_quoted = is_quoted_string(port_name)
305  if not is_quoted:
306  raise ValueError('Port name must be a quoted string.')
307 
308  port_name = port_name.strip('"')
309  if not port_name:
310  raise ValueError('Failed to extract port name: empty string.')
311 
312  match args_number:
313  case 1:
314  # Port name only, no default value or description
315  ports[port_name] = {
316  'data_type': port_type,
317  'default': '',
318  'has_description': False
319  }
320  case 2:
321  # Port name and description exist, no default value
322  port_description_exists = bool((port_args[1]).strip('"').strip())
323  ports[port_name] = {
324  'data_type': port_type,
325  'default': '',
326  'has_description': port_description_exists
327  }
328  case 3:
329  # Port name, default value, and description exist
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())
333  ports[port_name] = {
334  'data_type': port_type,
335  'default': port_default,
336  'has_description': port_description_exists
337  }
338  return ports
339 
340 
341 def validate_bt_xml_structure(root: ET.Element) -> ET.Element:
342  if root is None or len(root) == 0:
343  raise ValueError(
344  'Invalid XML structure: '
345  f'Expected <{TREE_NODES_MODEL_TAG}> element as the first child of the root.'
346  )
347  bt_nodes_model = root[0]
348 
349  if bt_nodes_model.tag != TREE_NODES_MODEL_TAG:
350  raise ValueError(
351  'Invalid XML structure: '
352  f'Expected <{TREE_NODES_MODEL_TAG}> element as the first child of the root.'
353  )
354  return bt_nodes_model
355 
356 
357 def extract_xml_nodes_data(content: ET.ElementTree[Any]) -> BTNodes:
358  """
359  Extract Behavior Tree nodes data from the given XML data.
360 
361  Returns dictionary mapping node IDs to their port data:
362  {node_id: {port_name: {'data_type': 'x', 'default': 'y', 'has_description': bool}}}
363  """
364  root = content.getroot()
365  bt_nodes_model = validate_bt_xml_structure(root)
366 
367  bt_node_ids_xml: BTNodes = {}
368  for node in bt_nodes_model:
369  node_id = node.get('ID')
370  if not node_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}')
374 
375  ports: NodePorts = {}
376  for port in node:
377  port_name = port.get('name')
378  if not port_name:
379  raise ValueError(
380  f'Each port in {node_id} node must have a "name" attribute.'
381  )
382  if port_name in ports:
383  raise ValueError(
384  f'Duplicate port name {port_name} found in {node_id} node.'
385  )
386  port_type = port.get('type')
387  if not port_type:
388  raise ValueError(
389  f'{port_name} port in {node_id} node is missing a "type" attribute.'
390  )
391  port_default = port.get('default', '')
392  port_description_exists = bool((port.text or '').strip())
393  ports[port_name] = {
394  'data_type': port_type,
395  'default': port_default,
396  'has_description': port_description_exists
397  }
398 
399  bt_node_ids_xml[node_id] = ports
400  return bt_node_ids_xml
401 
402 
403 def extract_node_registration_data(content: str) -> dict[str, str]:
404  """
405  Extract node registration data from the given content.
406 
407  Returns dictionary mapping class names to node IDs: {class_name: node_id}
408  """
409  register_pattern = re.compile(r'register(?:NodeType|Builder)')
410 
411  register_data: dict[str, str] = {}
412  for register_match in register_pattern.finditer(content):
413  start_pos = register_match.end()
414 
415  if has_leading_comments(content, start_pos, '//'):
416  continue
417 
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.')
421 
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)
427 
428  if class_name in register_data:
429  raise ValueError(f'Duplicate {class_name} class found.')
430 
431  args_start = content.find('(', register_type_end)
432  if args_start == -1:
433  raise ValueError(
434  'Failed to extract node registration arguments: opening parenthesis not found.'
435  )
436 
437  register_args = extract_arguments(content, args_start)
438  if not register_args:
439  raise ValueError(
440  'Failed to extract node ID from node registration: no arguments found.'
441  )
442 
443  node_id = register_args[0]
444  is_quoted = is_quoted_string(node_id)
445  if not is_quoted:
446  raise ValueError('Node ID must be a quoted string.')
447 
448  node_id = node_id.strip('"')
449  if not node_id:
450  raise ValueError('Failed to extract node ID from node registration: empty string.')
451 
452  if node_id in register_data.values():
453  raise ValueError(
454  f'Duplicate node ID registration for {class_name} class found: {node_id}'
455  )
456 
457  register_data[class_name] = node_id
458 
459  if not register_data:
460  raise ValueError('No node registration found.')
461  return register_data
462 
463 
464 def extract_cpp_classes_and_ids(cpp_files: list[Path]) -> CPPData:
465  """
466  Extract class names and their corresponding node IDs from the given list of source files.
467 
468  Returns dictionary mapping class names to node IDs: {class_name: node_id}
469  """
470  node_cpp_data: CPPData = {}
471  for cpp_file in cpp_files:
472  cpp_content = cpp_file.read_text()
473  try:
474  class_names_and_ids = extract_node_registration_data(cpp_content)
475  except ValueError as exc:
476  raise ValueError(
477  f'Failed to extract node registration data from {cpp_file}: {exc}\n')
478 
479  all_classes = node_cpp_data.keys()
480  file_classes = class_names_and_ids.keys()
481  common_classes = all_classes & file_classes
482  if common_classes:
483  raise ValueError(
484  f'Duplicate class name found in {cpp_file}: '
485  f'{", ".join(common_classes)}.'
486  )
487  all_ids = set(node_cpp_data.values())
488  file_ids = set(class_names_and_ids.values())
489  common_ids = all_ids & file_ids
490  if common_ids:
491  raise ValueError(
492  f'Duplicate node ID found in {cpp_file}: '
493  f'{", ".join(common_ids)}.'
494  )
495  node_cpp_data.update(class_names_and_ids)
496  return node_cpp_data
497 
498 
499 def extract_class_definitions(
500  content: str,
501 ) -> list[tuple[str, str, str]]:
502  """
503  Extract class data from headers.
504 
505  Returns a list of tuples: (class_name, base_class_name, class_section).
506  """
507  class_pattern = re.compile(
508  r'^\s*class\s+([A-Za-z_][A-Za-z0-9_]*)',
509  re.MULTILINE
510  )
511  base_class_pattern = re.compile(
512  r'public\s+(?:[A-Za-z_][A-Za-z0-9_:]*::)?([A-Za-z_][A-Za-z0-9_]*)',
513  re.MULTILINE
514  )
515  class_definitions: list[tuple[str, str, str]] = []
516 
517  pos = 0
518  while pos < len(content):
519  class_match = class_pattern.search(content, pos)
520  if not class_match:
521  break
522  class_name = class_match.group(1)
523 
524  class_name_end = class_match.end()
525  class_brace = content.find('{', class_name_end)
526  if class_brace == -1:
527  raise ValueError(
528  'Failed to extract class definition: opening brace not found.'
529  )
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 ''
533 
534  brace_count = 1
535  pos_brace_search = class_brace + 1
536  while pos_brace_search < len(content):
537  char = content[pos_brace_search]
538  if char == '{':
539  brace_count += 1
540  if char == '}':
541  brace_count -= 1
542  if brace_count == 0:
543  pos_brace_search += 1
544  break
545  pos_brace_search += 1
546  else:
547  raise ValueError(f'Failed to parse class section for {class_name}.')
548 
549  class_section = content[class_brace:pos_brace_search]
550  class_definitions.append((class_name, base_class_name, class_section))
551 
552  pos = pos_brace_search
553 
554  return class_definitions
555 
556 
557 def extract_hpp_classes_and_ports_data(
558  hpp_files: list[Path],
559  hpp_base_classes: dict[str, Path]
560 ) -> HPPData:
561  """
562  Extract class names and their corresponding port data from the given list of header files.
563 
564  Returns dictionary mapping class names to their port data:
565  {class_name: {port_name: {'data_type': 'x', 'default': 'y', 'has_description': bool}}}
566  """
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:
572  raise ValueError(
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:
576  raise ValueError(
577  f'Duplicate class name found in {hpp_file}: {class_name}.'
578  )
579  try:
580  ports = extract_code_port_data(class_section)
581  except ValueError as exc:
582  raise ValueError(
583  f'Failed to extract port data for {class_name} class in {hpp_file}: {exc}'
584  )
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()
588  try:
589  base_ports = extract_code_port_data(base_class_content)
590  except ValueError as exc:
591  raise ValueError(
592  f'Failed to extract port data for {class_name} class '
593  f'in {base_class_path}: {exc}'
594  )
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:
599  raise ValueError(
600  f'Port name conflict between {class_name} class and '
601  f'its base class {base_class_name}: '
602  f'{", ".join(common_port_names)}.'
603  )
604  ports.update(base_ports)
605  node_hpp_data[class_name] = ports
606  return node_hpp_data
607 
608 
609 def extract_code_nodes_data(config: dict) -> BTNodes:
610  """
611  Extract BT node data from code based on the provided configuration.
612 
613  Returns dictionary mapping node IDs to their port data:
614  {node_id: {port_name: {'data_type': 'x', 'default': 'y', 'has_description': bool}}}
615  """
616  bt_node_ids_code: BTNodes = {}
617 
618  # Share base classes data between repositories
619  hpp_base_classes: dict[str, Path] = {}
620 
621  for _, repo_info in config.items():
622  bt = repo_info.get('behavior_trees', {})
623 
624  cpp_dirs = bt.get('cpp_dir_paths', [])
625  cpp_files = get_files(cpp_dirs, '*.cpp')
626 
627  hpp_dirs = bt.get('hpp_dir_paths', [])
628  hpp_files = get_files(hpp_dirs, '*.hpp')
629 
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:
634  raise ValueError(
635  f'Duplicate base class name found in configuration: {base_class_name}.'
636  )
637  hpp_base_classes[base_class_name] = base_class_path
638 
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)
641 
642  diff_classes_cpp_hpp = node_cpp_data.keys() - node_hpp_data.keys()
643  if diff_classes_cpp_hpp:
644  raise ValueError(
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.'
648  )
649  diff_classes_hpp_cpp = node_hpp_data.keys() - node_cpp_data.keys()
650  if diff_classes_hpp_cpp:
651  raise ValueError(
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.'
655  )
656 
657  # Combine data from cpp and hpp files by class names:
658  # {node_id: {port_name: {'data_type': 'x', 'default': 'y', 'has_description': bool}}}
659  for class_name, node_id in node_cpp_data.items():
660  bt_node_ids_code[node_id] = node_hpp_data[class_name]
661 
662  return bt_node_ids_code
663 
664 
665 def detect_bt_nodes_mismatches(bt_node_ids_code: BTNodes, bt_node_ids_xml: BTNodes) -> bool:
666  """
667  Compare BT node data extracted from code and XML.
668 
669  Compares node IDs, port names, data types, and default values.
670  Returns True if any mismatch is found, False otherwise.
671  """
672  is_mismatch_found = False
673 
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}')
680 
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}')
687 
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()
692 
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}')
699 
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}')
706 
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}')
716 
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
725 
726 
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
737 
738 
739 def main():
740 
741  parser = argparse.ArgumentParser()
742  parser.add_argument(
743  '--config',
744  type=Path,
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.'
749  )
750  args = parser.parse_args()
751  args_config = args.config
752 
753  try:
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}')
758  sys.exit(1)
759 
760  nav2_bt_nodes = config.get(
761  'nav2_bt_nodes_file_path',
762  'nav2_behavior_tree/nav2_tree_nodes.xml'
763  )
764  nav2_bt_nodes_file_path = Path(nav2_bt_nodes)
765 
766  try:
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}')
770  sys.exit(1)
771 
772  try:
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}')
776  sys.exit(1)
777 
778  github_repos_config = config.get('github_repositories', {})
779  if github_repos_config:
780  try:
781  # Always copy external repositories to the navigation2 root directory,
782  # regardless of the script's location
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}')
786  sys.exit(1)
787 
788  print('Cloning external repositories...')
789  try:
790  fetch_external_repos(github_repos_config, clone_dir)
791  except FileNotFoundError as exc:
792  print(
793  f'Failed to fetch external repositories: {exc}'
794  f'Review specified paths in {args_config}.'
795  )
796  sys.exit(1)
797  except (subprocess.CalledProcessError, OSError) as exc:
798  stderr = getattr(exc, 'stderr', None)
799  print(f'Failed to fetch external repositories: {stderr or exc}')
800  sys.exit(1)
801 
802  update_paths_for_external_repos(github_repos_config, clone_dir)
803 
804  local_repos_config = config.get('local_repositories', {})
805  repos_config = local_repos_config | github_repos_config
806 
807  try:
808  bt_node_ids_code = extract_code_nodes_data(repos_config)
809  except (OSError, ValueError) as exc:
810  print(
811  f'Failed to extract BT nodes data from code: {exc}\n'
812  f'Review specified files in {args_config}'
813  )
814  sys.exit(1)
815 
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)
818 
819  print('Checking for missing descriptions in XML...')
820  # Skip descriptions checking in bt_node_ids_code, as they are optional.
821  is_xml_description_missing = validate_descriptions(bt_node_ids_xml)
822 
823  if is_mismatch_found or is_xml_description_missing:
824  print(
825  'Validation failed.\n'
826  'Please review BT nodes in code and '
827  f'their corresponding XML definitions in {nav2_bt_nodes_file_path}.'
828  )
829  sys.exit(1)
830 
831  print('Validation successful. No mismatches found between code and XML BT nodes data.')
832 
833 
834 if __name__ == '__main__':
835  main()