Substitutions in parameter files (#168)

Signed-off-by: Ivan Santiago Paunovic <[email protected]>

Author: ivanpauno

launch_ros/launch_ros/actions/node.py

@@ -23,7 +23,6 @@
 from typing import Optional
 from typing import Text  # noqa: F401
 from typing import Tuple  # noqa: F401
-from typing import TYPE_CHECKING
 from typing import Union
 
 import warnings
@@ -59,8 +58,8 @@
 
 import yaml
 
-if TYPE_CHECKING:
-    from ..descriptions import Parameter
+from ..descriptions import Parameter
+from ..descriptions import ParameterFile
 
 
 @expose_action('node')
@@ -257,16 +256,25 @@ def get_nested_dictionary_from_nested_key_value_pairs(params):
         normalized_params = []
         for param in params:
             from_attr = param.get_attr('from', optional=True)
+            allow_substs = param.get_attr('allow_substs', data_type=bool, optional=True)
             name = param.get_attr('name', optional=True)
             if from_attr is not None and name is not None:
                 raise RuntimeError('name and from attributes are mutually exclusive')
             elif from_attr is not None:
                 # 'from' attribute ignores 'name' attribute,
                 # it's not accepted to be nested,
                 # and it can not have children.
-                normalized_params.append(parser.parse_substitution(from_attr))
+                if isinstance(allow_substs, str):
+                    allow_substs = parser.parse_substitution(allow_substs)
+                else:
+                    allow_substs = bool(allow_substs)
+                normalized_params.append(
+                    ParameterFile(parser.parse_substitution(from_attr), allow_substs=allow_substs))
                 continue
             elif name is not None:
+                if allow_substs is not None:
+                    raise RuntimeError(
+                        "'allow_substs' can only be used together with 'from' attribute")
                 normalized_params.append(
                     get_nested_dictionary_from_nested_key_value_pairs([param]))
                 continue

launch_ros/launch_ros/descriptions/__init__.py

@@ -16,11 +16,13 @@
 
 from .composable_node import ComposableNode
 from ..parameter_descriptions import Parameter
+from ..parameter_descriptions import ParameterFile
 from ..parameter_descriptions import ParameterValue
 
 
 __all__ = [
     'ComposableNode',
     'Parameter',
+    'ParameterFile',
     'ParameterValue',
 ]

launch_ros/launch_ros/parameter_descriptions.py

@@ -14,6 +14,9 @@
 
 """Module for a description of a Parameter."""
 
+import os
+from pathlib import Path
+from tempfile import NamedTemporaryFile
 from typing import List
 from typing import Optional
 from typing import Text
@@ -24,14 +27,19 @@
 from launch import LaunchContext
 from launch import SomeSubstitutionsType
 from launch import SomeSubstitutionsType_types_tuple
+from launch.frontend.parse_substitution import parse_substitution
 from launch.substitution import Substitution
+from launch.substitutions import SubstitutionFailure
 from launch.utilities import ensure_argument_type
 from launch.utilities import normalize_to_list_of_substitutions
 from launch.utilities import perform_substitutions
 from launch.utilities.type_utils import AllowedTypesType
 from launch.utilities.type_utils import normalize_typed_substitution
 from launch.utilities.type_utils import perform_typed_substitution
 from launch.utilities.type_utils import SomeValueType
+from launch.utilities.typing_file_path import FilePath
+
+import yaml
 
 if TYPE_CHECKING:
     from .parameters_type import EvaluatedParameterValue
@@ -154,3 +162,99 @@ def evaluate(self, context: LaunchContext) -> Tuple[Text, 'EvaluatedParameterVal
         self.__evaluated_parameter_name = name
         self.__evaluated_parameter_rule = (name, value)
         return (name, value)
+
+
+class ParameterFile:
+    """Describes a ROS parameter file."""
+
+    def __init__(
+        self,
+        param_file: Union[FilePath, SomeSubstitutionsType],
+        *,
+        allow_substs: [bool, SomeSubstitutionsType] = False
+    ) -> None:
+        """
+        Construct a parameter file description.
+
+        :param param_file: Path to a parameter file.
+        :param allow_subst: Allow substitutions in the parameter file.
+        """
+        ensure_argument_type(
+            param_file,
+            SomeSubstitutionsType_types_tuple + (os.PathLike, bytes),
+            'param_file',
+            'ParameterFile()'
+        )
+        ensure_argument_type(
+            allow_substs,
+            bool,
+            'allow_subst',
+            'ParameterFile()'
+        )
+        self.__param_file: Union[List[Substitution], FilePath] = param_file
+        if isinstance(param_file, SomeSubstitutionsType_types_tuple):
+            self.__param_file = normalize_to_list_of_substitutions(param_file)
+        self.__allow_substs = normalize_typed_substitution(allow_substs, data_type=bool)
+        self.__evaluated_allow_substs: Optional[bool] = None
+        self.__evaluated_param_file: Optional[Path] = None
+        self.__created_tmp_file = False
+
+    @property
+    def param_file(self) -> Union[FilePath, List[Substitution]]:
+        """Getter for parameter file."""
+        if self.__evaluated_param_file is not None:
+            return self.__evaluated_param_file
+        return self.__param_file
+
+    @property
+    def allow_substs(self) -> Union[bool, List[Substitution]]:
+        """Getter for allow substitutions argument."""
+        if self.__evaluated_allow_substs is not None:
+            return self.__evaluated_allow_substs
+        return self.__allow_substs
+
+    def __str__(self) -> Text:
+        return (
+            'launch_ros.description.ParameterFile'
+            f'(param_file={self.param_file}, allow_substs={self.allow_substs})'
+        )
+
+    def evaluate(self, context: LaunchContext) -> Path:
+        """Evaluate and return a parameter file path."""
+        if self.__evaluated_param_file is not None:
+            return self.__evaluated_param_file
+
+        param_file = self.__param_file
+        if isinstance(param_file, list):
+            # list of substitutions
+            param_file = perform_substitutions(context, self.__param_file)
+
+        allow_substs = perform_typed_substitution(context, self.__allow_substs, data_type=bool)
+        param_file_path: Path = Path(param_file)
+        if allow_substs:
+            with open(param_file_path, 'r') as f, NamedTemporaryFile(
+                mode='w', prefix='launch_params_', delete=False
+            ) as h:
+                parsed = perform_substitutions(context, parse_substitution(f.read()))
+                try:
+                    yaml.safe_load(parsed)
+                except Exception:
+                    raise SubstitutionFailure(
+                        'The substituted parameter file is not a valid yaml file')
+                h.write(parsed)
+                param_file_path = Path(h.name)
+                self.__created_tmp_file = True
+        self.__evaluated_param_file = param_file_path
+        return param_file_path
+
+    def cleanup(self):
+        """Delete created temporary files."""
+        if self.__created_tmp_file and self.__evaluated_param_file is not None:
+            try:
+                os.unlink(self.__evaluated_param_file)
+            except FileNotFoundError:
+                pass
+            self.__evaluated_param_file = None
+
+    def __del__(self):
+        self.cleanup()

launch_ros/launch_ros/parameters_type.py

@@ -27,6 +27,7 @@
 from launch.substitution import Substitution
 
 from .parameter_descriptions import Parameter as ParameterDescription
+from .parameter_descriptions import ParameterFile
 from .parameter_descriptions import ParameterValue as ParameterValueDescription
 
 
@@ -36,7 +37,7 @@
 _MultiValueType = Union[
     Sequence[str], Sequence[int], Sequence[float], Sequence[bool], bytes]
 
-SomeParameterFile = Union[SomeSubstitutionsType, pathlib.Path]
+SomeParameterFile = Union[SomeSubstitutionsType, pathlib.Path, ParameterFile]
 SomeParameterName = Sequence[Union[Substitution, str]]
 SomeParameterValue = Union[
     ParameterValueDescription,
@@ -59,7 +60,7 @@
 # parameter names and values
 SomeParameters = Sequence[Union[SomeParameterFile, ParameterDescription, SomeParametersDict]]
 
-ParameterFile = Sequence[Substitution]
+ParameterFile = ParameterFile  # re-export
 ParameterName = Sequence[Substitution]
 ParameterValue = Union[
     Sequence[Substitution],

launch_ros/launch_ros/utilities/evaluate_parameters.py

@@ -32,6 +32,7 @@
 import yaml
 
 from ..parameter_descriptions import Parameter as ParameterDescription
+from ..parameter_descriptions import ParameterFile
 from ..parameter_descriptions import ParameterValue as ParameterValueDescription
 from ..parameters_type import EvaluatedParameters
 from ..parameters_type import EvaluatedParameterValue
@@ -140,10 +141,9 @@ def evaluate_parameters(context: LaunchContext, parameters: Parameters) -> Evalu
     """
     output_params = []  # type: List[Union[pathlib.Path, Dict[str, EvaluatedParameterValue]]]
     for param in parameters:
-        # If it's a list of substitutions then evaluate them to a string and return a pathlib.Path
-        if isinstance(param, tuple) and len(param) and isinstance(param[0], Substitution):
+        if isinstance(param, ParameterFile):
             # Evaluate a list of Substitution to a file path
-            output_params.append(pathlib.Path(perform_substitutions(context, list(param))))
+            output_params.append(param.evaluate(context))
         elif isinstance(param, ParameterDescription):
             output_params.append(param)
         elif isinstance(param, Mapping):

launch_ros/launch_ros/utilities/normalize_parameters.py

@@ -16,7 +16,6 @@
 
 from collections.abc import Mapping
 from collections.abc import Sequence
-import pathlib
 from typing import cast
 from typing import List
 from typing import Optional
@@ -34,8 +33,8 @@
 import yaml
 
 from ..parameter_descriptions import Parameter as ParameterDescription
+from ..parameter_descriptions import ParameterFile
 from ..parameter_descriptions import ParameterValue as ParameterValueDescription
-from ..parameters_type import ParameterFile  # noqa: F401
 from ..parameters_type import Parameters
 from ..parameters_type import ParametersDict
 from ..parameters_type import ParameterValue
@@ -179,16 +178,15 @@ def normalize_parameters(parameters: SomeParameters) -> Parameters:
     if isinstance(parameters, str) or not isinstance(parameters, Sequence):
         raise TypeError('Expecting list of parameters, got {}'.format(parameters))
 
-    normalized_params = []  # type: List[Union[ParameterFile, ParametersDict]]
+    normalized_params: List[Union[ParameterFile, ParametersDict, ParameterDescription]] = []
     for param in parameters:
         if isinstance(param, Mapping):
             normalized_params.append(normalize_parameter_dict(param))
         elif isinstance(param, ParameterDescription):
             normalized_params.append(param)
+        elif isinstance(param, ParameterFile):
+            normalized_params.append(param)
         else:
-            # It's a path, normalize to a list of substitutions
-            if isinstance(param, pathlib.Path):
-                param = str(param)
-            ensure_argument_type(param, SomeSubstitutionsType_types_tuple, 'parameters')
-            normalized_params.append(tuple(normalize_to_list_of_substitutions(param)))
+            # It's a path
+            normalized_params.append(ParameterFile(param))
     return tuple(normalized_params)