diff --git a/robotnik_common/robotnik_common/launch/__init__.py b/robotnik_common/robotnik_common/launch/__init__.py index 5c06587..bcb8500 100644 --- a/robotnik_common/robotnik_common/launch/__init__.py +++ b/robotnik_common/robotnik_common/launch/__init__.py @@ -14,3 +14,5 @@ from .rewritten_yaml import RewrittenYaml from .add_launch_args import add_launch_args +from .add_launch_args import ExtendedArgument +from .add_launch_args import AddArgumentParser diff --git a/robotnik_common/robotnik_common/launch/add_launch_args.py b/robotnik_common/robotnik_common/launch/add_launch_args.py index eb32bb8..81e269f 100644 --- a/robotnik_common/robotnik_common/launch/add_launch_args.py +++ b/robotnik_common/robotnik_common/launch/add_launch_args.py @@ -1,4 +1,4 @@ -# Copyright (c) 2023, Robotnik Automation S.L.L. +# Copyright (c) 2023, Robotnik Automation S.L. # All rights reserved. # # Redistribution and use in source and binary forms, with or without @@ -23,20 +23,108 @@ # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. -import launch +from launch.substitutions import EnvironmentVariable +from launch.substitutions import LaunchConfiguration +from launch import LaunchDescription +from launch.actions import DeclareLaunchArgument +from dataclasses import dataclass, field -def add_launch_args(ld : launch.LaunchDescription, params : list[tuple[str, str, str]]): # name, description, default_value - ret={} +@dataclass +class ExtendedArgument: + """ Extended Arguments item dataclass """ + name: str + default_value: str + description: str + # Use environment variable + use_env: bool = field(default=True) + # Environment variable name to use + environment: str = field(default=None) - for param in params: - # Declare the launch options - ld.add_action(launch.actions.DeclareLaunchArgument( - name=param[0], description=param[1], - default_value=launch.substitutions.EnvironmentVariable(param[0].upper(), default_value=param[2]), - )) + def __post_init__(self): + if not self.environment: + self.environment = self.name.upper() - # Get the launch configuration variables - ret[param[0]] = launch.substitutions.LaunchConfiguration(param[0]) - return ret +class AddArgumentParser(): + """ Add the argument with environment variable """ + + def __init__( + self, + ld: LaunchDescription, + ): + self._arg_list = [] + self._ld = ld + + def add_arg( + self, + arg: ExtendedArgument, + ): + self._arg_list.append(arg) + + def __add_launch_arg( + self, + arg: ExtendedArgument, + ): + if not arg.environment: + arg.environment = arg.name.upper() + if arg.use_env: + default_value = EnvironmentVariable( + name=arg.environment, + default_value=arg.default_value, + ) + arg = DeclareLaunchArgument( + name=arg.name, + description=arg.description, + default_value=default_value, + ) + else: + arg = DeclareLaunchArgument( + name=arg.name, + description=arg.description, + default_value=arg.default_value, + ) + self._ld.add_action(arg) + pass + + def process_arg( + self, + ): + params = {} + for arg in self._arg_list: + self.__add_launch_arg(arg) + params[arg.name] = LaunchConfiguration(arg.name) + return params + + +def add_launch_args( + ld: LaunchDescription, + params: list[ + tuple[ + str, + str, + str + ] + ] +): + # name, description, default_value + + ret = {} + + for param in params: + # Declare the launch options + ld.add_action( + DeclareLaunchArgument( + name=param[0], + description=param[1], + default_value=EnvironmentVariable( + param[0].upper(), + default_value=param[2], + ), + ) + ) + + # Get the launch configuration variables + ret[param[0]] = LaunchConfiguration(param[0]) + + return ret diff --git a/robotnik_common/robotnik_common/launch/rewritten_yaml.py b/robotnik_common/robotnik_common/launch/rewritten_yaml.py index 8c77151..2ed05eb 100644 --- a/robotnik_common/robotnik_common/launch/rewritten_yaml.py +++ b/robotnik_common/robotnik_common/launch/rewritten_yaml.py @@ -20,6 +20,7 @@ import yaml import tempfile import launch +from launch.utilities import perform_substitutions class DictItemReference: @@ -41,12 +42,14 @@ class RewrittenYaml(launch.Substitution): Used in launch system """ - def __init__(self, + def __init__( + self, source_file: launch.SomeSubstitutionsType, param_rewrites: Dict, root_key: Optional[launch.SomeSubstitutionsType] = None, key_rewrites: Optional[Dict] = None, - convert_types = False) -> None: + convert_types=False + ) -> None: super().__init__() """ Construct the substitution @@ -55,20 +58,26 @@ def __init__(self, :param: param_rewrites mappings to replace :param: root_key if provided, the contents are placed under this key :param: key_rewrites keys of mappings to replace - :param: convert_types whether to attempt converting the string to a number or boolean + :param: convert_types whether to attempt converting the + string to a number or boolean """ - from launch.utilities import normalize_to_list_of_substitutions # import here to avoid loop + # import here to avoid loop + from launch.utilities \ + import normalize_to_list_of_substitutions + self.__source_file = normalize_to_list_of_substitutions(source_file) self.__param_rewrites = {} self.__key_rewrites = {} self.__convert_types = convert_types self.__root_key = None for key in param_rewrites: - self.__param_rewrites[key] = normalize_to_list_of_substitutions(param_rewrites[key]) + value = normalize_to_list_of_substitutions(param_rewrites[key]) + self.__param_rewrites[key] = value if key_rewrites is not None: for key in key_rewrites: - self.__key_rewrites[key] = normalize_to_list_of_substitutions(key_rewrites[key]) + value = normalize_to_list_of_substitutions(key_rewrites[key]) + self.__key_rewrites[key] = value if root_key is not None: self.__root_key = normalize_to_list_of_substitutions(root_key) @@ -82,14 +91,17 @@ def describe(self) -> Text: return '' def perform(self, context: launch.LaunchContext) -> Text: - yaml_filename = launch.utilities.perform_substitutions(context, self.name) + yaml_filename = perform_substitutions(context, self.name) rewritten_yaml = tempfile.NamedTemporaryFile(mode='w', delete=False) param_rewrites, keys_rewrites = self.resolve_rewrites(context) data = yaml.safe_load(open(yaml_filename, 'r')) self.substitute_params(data, param_rewrites) self.substitute_keys(data, keys_rewrites) if self.__root_key is not None: - root_key = launch.utilities.perform_substitutions(context, self.__root_key) + root_key = perform_substitutions( + context, + self.__root_key + ) name = '' while len(root_key) != 0: @@ -98,9 +110,9 @@ def perform(self, context: launch.LaunchContext) -> Text: name = '' else: name = root_key[-1] + name - + root_key = root_key[:-1] - + data = {name: data} yaml.dump(data, rewritten_yaml) @@ -110,10 +122,16 @@ def perform(self, context: launch.LaunchContext) -> Text: def resolve_rewrites(self, context): resolved_params = {} for key in self.__param_rewrites: - resolved_params[key] = launch.utilities.perform_substitutions(context, self.__param_rewrites[key]) + resolved_params[key] = perform_substitutions( + context, + self.__param_rewrites[key] + ) resolved_keys = {} for key in self.__key_rewrites: - resolved_keys[key] = launch.utilities.perform_substitutions(context, self.__key_rewrites[key]) + resolved_keys[key] = perform_substitutions( + context, + self.__key_rewrites[key] + ) return resolved_params, resolved_keys def substitute_params(self, yaml, param_rewrites): @@ -132,14 +150,17 @@ def substitute_params(self, yaml, param_rewrites): yaml_keys = path.split('.') yaml = self.updateYamlPathVals(yaml, yaml_keys, rewrite_val) - def updateYamlPathVals(self, yaml, yaml_key_list, rewrite_val): for key in yaml_key_list: if key == yaml_key_list[-1]: yaml[key] = rewrite_val break key = yaml_key_list.pop(0) - yaml[key] = self.updateYamlPathVals(yaml.get(key, {}), yaml_key_list, rewrite_val) + yaml[key] = self.updateYamlPathVals( + yaml.get(key, {}), + yaml_key_list, + rewrite_val + ) return yaml @@ -192,7 +213,7 @@ def convert(self, text_value): return True if text_value.lower() == "false": return False - + # try converting array if text_value[0] == '[': return eval(text_value)