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