Refactor Node action into description classes

Distro A; OPSEC #4584

Signed-off-by: Roger Strain <[email protected]>

Author: Roger Strain

launch_ros/launch_ros/actions/node.py

@@ -14,57 +14,33 @@
 
 """Module for the Node action."""
 
-import os
-import pathlib
-from tempfile import NamedTemporaryFile
 from typing import Dict
 from typing import Iterable
 from typing import List
 from typing import Optional
 from typing import Text  # noqa: F401
 from typing import Tuple  # noqa: F401
-from typing import Union
 
 try:
     import importlib.metadata as importlib_metadata
 except ModuleNotFoundError:
     import importlib_metadata
 
-from launch.action import Action
+from launch.actions import ExecuteLocal
 from launch.actions import ExecuteProcess
 from launch.frontend import Entity
 from launch.frontend import expose_action
 from launch.frontend import Parser
 from launch.frontend.type_utils import get_data_type_from_identifier
 
 from launch.launch_context import LaunchContext
-import launch.logging
 from launch.some_substitutions_type import SomeSubstitutionsType
-from launch.substitutions import LocalSubstitution
-from launch.utilities import ensure_argument_type
-from launch.utilities import normalize_to_list_of_substitutions
-from launch.utilities import perform_substitutions
 
+from launch_ros.descriptions import Node as NodeDescription
+from launch_ros.descriptions import ParameterFile
+from launch_ros.descriptions import RosExecutable
 from launch_ros.parameters_type import SomeParameters
 from launch_ros.remap_rule_type import SomeRemapRules
-from launch_ros.substitutions import ExecutableInPackage
-from launch_ros.utilities import add_node_name
-from launch_ros.utilities import evaluate_parameters
-from launch_ros.utilities import get_node_name_count
-from launch_ros.utilities import make_namespace_absolute
-from launch_ros.utilities import normalize_parameters
-from launch_ros.utilities import normalize_remap_rules
-from launch_ros.utilities import plugin_support
-from launch_ros.utilities import prefix_namespace
-
-
-from rclpy.validate_namespace import validate_namespace
-from rclpy.validate_node_name import validate_node_name
-
-import yaml
-
-from ..descriptions import Parameter
-from ..descriptions import ParameterFile
 
 
 class NodeActionExtension:
@@ -110,12 +86,9 @@ def prepare_for_execute(self, context, ros_specific_arguments, node_action):
 
 
 @expose_action('node')
-class Node(ExecuteProcess):
+class Node(ExecuteLocal):
     """Action that executes a ROS node."""
 
-    UNSPECIFIED_NODE_NAME = '<node_name_unspecified>'
-    UNSPECIFIED_NODE_NAMESPACE = '<node_namespace_unspecified>'
-
     def __init__(
         self, *,
         executable: SomeSubstitutionsType,
@@ -198,46 +171,20 @@ def __init__(
         :param: ros_arguments list of ROS arguments for the node
         :param: arguments list of extra arguments for the node
         """
-        if package is not None:
-            cmd = [ExecutableInPackage(package=package, executable=executable)]
-        else:
-            cmd = [executable]
-        cmd += [] if arguments is None else arguments
-        cmd += [] if ros_arguments is None else ['--ros-args'] + ros_arguments
-        # Reserve space for ros specific arguments.
-        # The substitutions will get expanded when the action is executed.
-        cmd += ['--ros-args']  # Prepend ros specific arguments with --ros-args flag
-        if name is not None:
-            cmd += ['-r', LocalSubstitution(
-                "ros_specific_arguments['name']", description='node name')]
-        if parameters is not None:
-            ensure_argument_type(parameters, (list), 'parameters', 'Node')
-            # All elements in the list are paths to files with parameters (or substitutions that
-            # evaluate to paths), or dictionaries of parameters (fields can be substitutions).
-            normalized_params = normalize_parameters(parameters)
-        # Forward 'exec_name' as to ExecuteProcess constructor
-        kwargs['name'] = exec_name
-        super().__init__(cmd=cmd, **kwargs)
-        self.__package = package
-        self.__node_executable = executable
-        self.__node_name = name
-        self.__node_namespace = namespace
-        self.__parameters = [] if parameters is None else normalized_params
-        self.__remappings = [] if remappings is None else list(normalize_remap_rules(remappings))
-        self.__ros_arguments = ros_arguments
-        self.__arguments = arguments
-
-        self.__expanded_node_name = self.UNSPECIFIED_NODE_NAME
-        self.__expanded_node_namespace = self.UNSPECIFIED_NODE_NAMESPACE
-        self.__expanded_parameter_arguments = None  # type: Optional[List[Tuple[Text, bool]]]
-        self.__final_node_name = None  # type: Optional[Text]
-        self.__expanded_remappings = None  # type: Optional[List[Tuple[Text, Text]]]
-
-        self.__substitutions_performed = False
-
-        self.__logger = launch.logging.get_logger(__name__)
-
+        self.__node_desc = NodeDescription(node_name=name, node_namespace=namespace,
+                                           parameters=parameters, remappings=remappings,
+                                           arguments=arguments)
+        self.__ros_exec = RosExecutable(package=package, executable=executable,
+                                        nodes=[self.__node_desc])
         self.__extensions = get_extensions(self.__logger)
+        super().__init__(process_description=self.__ros_exec, **kwargs)
+
+    def _perform_substitutions(self, lc: LaunchContext):
+        self.__node_desc.prepare(lc, self.__ros_exec)
+
+    def is_node_name_fully_specified(self):
+        return self.__node_desc.is_node_name_fully_specified()
+
 
     @staticmethod
     def parse_nested_parameters(params, parser):
@@ -306,7 +253,7 @@ def get_nested_dictionary_from_nested_key_value_pairs(params):
     def parse(cls, entity: Entity, parser: Parser):
         """Parse node."""
         # See parse method of `ExecuteProcess`
-        _, kwargs = super().parse(entity, parser, ignore=['cmd'])
+        _, kwargs = ExecuteProcess.parse(entity, parser, ignore=['cmd'])
         args = entity.get_attr('args', optional=True)
         if args is not None:
             kwargs['arguments'] = super()._parse_cmdline(args, parser)
@@ -348,187 +295,27 @@ def parse(cls, entity: Entity, parser: Parser):
     @property
     def node_package(self):
         """Getter for node_package."""
-        return self.__package
+        return self.__node_exec.package
 
     @property
     def node_executable(self):
         """Getter for node_executable."""
-        return self.__node_executable
+        return self.__node_exec.executable
 
     @property
     def node_name(self):
         """Getter for node_name."""
-        if self.__final_node_name is None:
-            raise RuntimeError("cannot access 'node_name' before executing action")
-        return self.__final_node_name
-
-    def is_node_name_fully_specified(self):
-        keywords = (self.UNSPECIFIED_NODE_NAME, self.UNSPECIFIED_NODE_NAMESPACE)
-        return all(x not in self.node_name for x in keywords)
-
-    def _create_params_file_from_dict(self, params):
-        with NamedTemporaryFile(mode='w', prefix='launch_params_', delete=False) as h:
-            param_file_path = h.name
-            param_dict = {
-                self.node_name if self.is_node_name_fully_specified() else '/**':
-                {'ros__parameters': params}
-            }
-            yaml.dump(param_dict, h, default_flow_style=False)
-            return param_file_path
-
-    def _get_parameter_rule(self, param: 'Parameter', context: LaunchContext):
-        name, value = param.evaluate(context)
-        return f'{name}:={yaml.dump(value)}'
-
-    def _perform_substitutions(self, context: LaunchContext) -> None:
-        # Here to avoid cyclic import
-        from ..descriptions import Parameter
-        try:
-            if self.__substitutions_performed:
-                # This function may have already been called by a subclass' `execute`, for example.
-                return
-            self.__substitutions_performed = True
-            if self.__node_name is not None:
-                self.__expanded_node_name = perform_substitutions(
-                    context, normalize_to_list_of_substitutions(self.__node_name))
-                validate_node_name(self.__expanded_node_name)
-            self.__expanded_node_name.lstrip('/')
-            expanded_node_namespace: Optional[Text] = None
-            if self.__node_namespace is not None:
-                expanded_node_namespace = perform_substitutions(
-                    context, normalize_to_list_of_substitutions(self.__node_namespace))
-            base_ns = context.launch_configurations.get('ros_namespace', None)
-            expanded_node_namespace = make_namespace_absolute(
-                prefix_namespace(base_ns, expanded_node_namespace))
-            if expanded_node_namespace is not None:
-                self.__expanded_node_namespace = expanded_node_namespace
-                cmd_extension = ['-r', LocalSubstitution("ros_specific_arguments['ns']")]
-                self.cmd.extend([normalize_to_list_of_substitutions(x) for x in cmd_extension])
-                validate_namespace(self.__expanded_node_namespace)
-        except Exception:
-            self.__logger.error(
-                "Error while expanding or validating node name or namespace for '{}':"
-                .format('package={}, executable={}, name={}, namespace={}'.format(
-                    self.__package,
-                    self.__node_executable,
-                    self.__node_name,
-                    self.__node_namespace,
-                ))
-            )
-            raise
-        self.__final_node_name = prefix_namespace(
-            self.__expanded_node_namespace, self.__expanded_node_name)
-
-        # Expand global parameters first,
-        # so they can be overridden with specific parameters of this Node
-        # The params_container list is expected to contain name-value pairs (tuples)
-        # and/or strings representing paths to parameter files.
-        params_container = context.launch_configurations.get('global_params', None)
-
-        if any(x is not None for x in (params_container, self.__parameters)):
-            self.__expanded_parameter_arguments = []
-        if params_container is not None:
-            for param in params_container:
-                if isinstance(param, tuple):
-                    name, value = param
-                    cmd_extension = ['-p', f'{name}:={value}']
-                    self.cmd.extend([normalize_to_list_of_substitutions(x) for x in cmd_extension])
-                else:
-                    param_file_path = os.path.abspath(param)
-                    self.__expanded_parameter_arguments.append((param_file_path, True))
-                    cmd_extension = ['--params-file', f'{param_file_path}']
-                    assert os.path.isfile(param_file_path)
-                    self.cmd.extend([normalize_to_list_of_substitutions(x) for x in cmd_extension])
-
-        # expand parameters too
-        if self.__parameters is not None:
-            evaluated_parameters = evaluate_parameters(context, self.__parameters)
-            for params in evaluated_parameters:
-                is_file = False
-                if isinstance(params, dict):
-                    param_argument = self._create_params_file_from_dict(params)
-                    is_file = True
-                    assert os.path.isfile(param_argument)
-                elif isinstance(params, pathlib.Path):
-                    param_argument = str(params)
-                    is_file = True
-                elif isinstance(params, Parameter):
-                    param_argument = self._get_parameter_rule(params, context)
-                else:
-                    raise RuntimeError('invalid normalized parameters {}'.format(repr(params)))
-                if is_file and not os.path.isfile(param_argument):
-                    self.__logger.warning(
-                        'Parameter file path is not a file: {}'.format(param_argument),
-                    )
-                    continue
-                self.__expanded_parameter_arguments.append((param_argument, is_file))
-                cmd_extension = ['--params-file' if is_file else '-p', f'{param_argument}']
-                self.cmd.extend([normalize_to_list_of_substitutions(x) for x in cmd_extension])
-        # expand remappings too
-        global_remaps = context.launch_configurations.get('ros_remaps', None)
-        if global_remaps or self.__remappings:
-            self.__expanded_remappings = []
-        if global_remaps:
-            self.__expanded_remappings.extend(global_remaps)
-        if self.__remappings:
-            self.__expanded_remappings.extend([
-                (perform_substitutions(context, src), perform_substitutions(context, dst))
-                for src, dst in self.__remappings
-            ])
-        if self.__expanded_remappings:
-            cmd_extension = []
-            for src, dst in self.__expanded_remappings:
-                cmd_extension.extend(['-r', f'{src}:={dst}'])
-            self.cmd.extend([normalize_to_list_of_substitutions(x) for x in cmd_extension])
-
-    def execute(self, context: LaunchContext) -> Optional[List[Action]]:
-        """
-        Execute the action.
-
-        Delegated to :meth:`launch.actions.ExecuteProcess.execute`.
-        """
-        self._perform_substitutions(context)
-        # Prepare the ros_specific_arguments list and add it to the context so that the
-        # LocalSubstitution placeholders added to the the cmd can be expanded using the contents.
-        ros_specific_arguments: Dict[str, Union[str, List[str]]] = {}
-        if self.__node_name is not None:
-            ros_specific_arguments['name'] = '__node:={}'.format(self.__expanded_node_name)
-        if self.__expanded_node_namespace != '':
-            ros_specific_arguments['ns'] = '__ns:={}'.format(self.__expanded_node_namespace)
-
-        # Give extensions a chance to prepare for execution
-        for extension in self.__extensions.values():
-            cmd_extension, ros_specific_arguments = extension.prepare_for_execute(
-                context,
-                ros_specific_arguments,
-                self
-            )
-            self.cmd.extend(cmd_extension)
-
-        context.extend_locals({'ros_specific_arguments': ros_specific_arguments})
-        ret = super().execute(context)
-
-        if self.is_node_name_fully_specified():
-            add_node_name(context, self.node_name)
-            node_name_count = get_node_name_count(context, self.node_name)
-            if node_name_count > 1:
-                execute_process_logger = launch.logging.get_logger(self.name)
-                execute_process_logger.warning(
-                    'there are now at least {} nodes with the name {} created within this '
-                    'launch context'.format(node_name_count, self.node_name)
-                )
-
-        return ret
+        return self.__node_desc.node_name
 
     @property
     def expanded_node_namespace(self):
         """Getter for expanded_node_namespace."""
-        return self.__expanded_node_namespace
+        return self.__node_desc.expanded_node_namespace
 
     @property
     def expanded_remapping_rules(self):
         """Getter for expanded_remappings."""
-        return self.__expanded_remappings
+        return self.__node_desc.expanded_remappings
 
 
 def instantiate_extension(

launch_ros/launch_ros/descriptions/__init__.py

@@ -15,14 +15,20 @@
 """descriptions Module."""
 
 from .composable_node import ComposableNode
+from .node import Node
+from .node_trait import NodeTrait
+from .ros_executable import RosExecutable
 from ..parameter_descriptions import Parameter
 from ..parameter_descriptions import ParameterFile
 from ..parameter_descriptions import ParameterValue
 
 
 __all__ = [
     'ComposableNode',
+    'Node',
+    'NodeTrait',
     'Parameter',
     'ParameterFile',
     'ParameterValue',
+    'RosExecutable',
 ]