feat: add ROSLoggers and ROSLogLevel

README.md

@@ -22,6 +22,28 @@
 
 ### Templated
 
+### ROSLoggers and ROSLogLevel
+Substitutions for setting the levels of ROS node loggers.
+
+```python
+ros_arguments=["--log-level", ROSLogLevel("debug")]
+```
+
+Will set only the node's log level to `debug`. (Equivalent to `--log-level node_name:=debug`)
+
+```python
+ros_arguments=["--log-level", ROSLogLevel("debug", all_loggers=True)]
+```
+
+Will set all loggers to `debug`. (Equivalent to `--log-level debug`)
+
+```python
+ros_arguments=[*ROSLoggers("debug", {'rcl': 'debug', 'rmw_fastrtps_cpp': 'info'})]
+```
+
+Will set the node's log level to `debug`, `rcl` to `debug` and `rmw_fastrtps_cpp` to `info`.
+(Equivalent to `--log-level node_name:=debug --log-level rcl:=debug --log-level rmw_fastrtps_cpp:=info`)
+
 ### Unary
 
 ### WriteTempFile

launch_ext/actions/execute_local.py

@@ -67,7 +67,7 @@
 from launch.launch_context import LaunchContext
 from launch.launch_description import LaunchDescription
 from launch.launch_description_entity import LaunchDescriptionEntity
-from launch.some_actions_type import SomeActionsType
+from launch.some_actions_type import SomeEntitiesType
 from launch.some_substitutions_type import SomeSubstitutionsType
 from launch.substitution import Substitution  # noqa: F401
 from launch.substitutions import LaunchConfiguration
@@ -109,7 +109,7 @@
 from launch.event import Event
 from launch.events.process import ProcessIO
 from launch.launch_context import LaunchContext
-from launch.some_actions_type import SomeActionsType
+from launch.some_actions_type import SomeEntitiesType
 
 from launch.events.process import ProcessExited
 
@@ -118,15 +118,15 @@ class OnProcessIO(OnActionEventBase):
     """Convenience class for handling I/O from processes via events."""
 
     # TODO(wjwwood): make the __init__ more flexible like OnProcessExit, so
-    # that it can take SomeActionsType directly or a callable that returns it.
+    # that it can take SomeEntitiesType directly or a callable that returns it.
     def __init__(
         self,
         *,
         target_action:
             Optional[Union[Callable[['ExecuteLocalExt'], bool], 'ExecuteLocalExt']] = None,
-        on_stdin: Callable[[ProcessIO], Optional[SomeActionsType]] = None,
-        on_stdout: Callable[[ProcessIO], Optional[SomeActionsType]] = None,
-        on_stderr: Callable[[ProcessIO], Optional[SomeActionsType]] = None,
+        on_stdin: Callable[[ProcessIO], Optional[SomeEntitiesType]] = None,
+        on_stdout: Callable[[ProcessIO], Optional[SomeEntitiesType]] = None,
+        on_stderr: Callable[[ProcessIO], Optional[SomeEntitiesType]] = None,
         **kwargs
     ) -> None:
         """Create an OnProcessIO event handler."""
@@ -135,7 +135,7 @@ def __init__(
             Optional[Union[Callable[['Action'], bool], 'Action']],
             target_action)
 
-        def handle(event: Event, _: LaunchContext) -> Optional[SomeActionsType]:
+        def handle(event: Event, _: LaunchContext) -> Optional[SomeEntitiesType]:
             event = cast(ProcessIO, event)
             if event.from_stdout and on_stdout is not None:
                 return on_stdout(event)
@@ -168,8 +168,8 @@ def __init__(
             Optional[Union[Callable[['ExecuteLocalExt'], bool], 'ExecuteLocalExt']] = None,
         on_exit:
             Union[
-                SomeActionsType,
-                Callable[[ProcessExited, LaunchContext], Optional[SomeActionsType]]
+                SomeEntitiesType,
+                Callable[[ProcessExited, LaunchContext], Optional[SomeEntitiesType]]
             ],
         **kwargs
     ) -> None:
@@ -179,8 +179,8 @@ def __init__(
             target_action)
         on_exit = cast(
             Union[
-                SomeActionsType,
-                Callable[[Event, LaunchContext], Optional[SomeActionsType]]],
+                SomeEntitiesType,
+                Callable[[Event, LaunchContext], Optional[SomeEntitiesType]]],
             on_exit)
         super().__init__(
             action_matcher=target_action,
@@ -240,8 +240,8 @@ def __init__(
         cached_output: bool = False,
         log_cmd: bool = False,
         on_exit: Optional[Union[
-            SomeActionsType,
-            Callable[[ProcessExited, LaunchContext], Optional[SomeActionsType]]
+            SomeEntitiesType,
+            Callable[[ProcessExited, LaunchContext], Optional[SomeEntitiesType]]
         ]] = None,
         respawn: Union[bool, SomeSubstitutionsType] = False,
         respawn_delay: Optional[float] = None,
@@ -476,7 +476,7 @@ def __on_signal_process_event(
     def __on_process_stdin(
         self,
         event: ProcessIO
-    ) -> Optional[SomeActionsType]:
+    ) -> Optional[SomeEntitiesType]:
         self.__logger.warning(
             "in ExecuteProcess('{}').__on_process_stdin_event()".format(id(self)),
         )
@@ -485,7 +485,7 @@ def __on_process_stdin(
 
     def __on_process_output(
         self, event: ProcessIO, buffer: io.TextIOBase, logger: logging.Logger
-    ) -> Optional[SomeActionsType]:
+    ) -> Optional[SomeEntitiesType]:
         to_write = event.text.decode(errors='replace')
         if buffer.closed:
             # buffer was probably closed by __flush_buffers on shutdown.  Output without
@@ -536,7 +536,7 @@ def __flush_buffers(self, event, context):
 
     def __on_process_output_cached(
         self, event: ProcessIO, buffer, logger
-    ) -> Optional[SomeActionsType]:
+    ) -> Optional[SomeEntitiesType]:
         to_write = event.text.decode(errors='replace')
         last_cursor = buffer.tell()
         buffer.seek(0, os.SEEK_END)  # go to end of buffer
@@ -563,7 +563,7 @@ def __flush_cached_buffers(self, event, context):
                 self.__output_format.format(line=line, this=self)
             )
 
-    def __on_shutdown(self, event: Event, context: LaunchContext) -> Optional[SomeActionsType]:
+    def __on_shutdown(self, event: Event, context: LaunchContext) -> Optional[SomeEntitiesType]:
         due_to_sigint = cast(Shutdown, event).due_to_sigint
         return self._shutdown_process(
             context,

launch_ext/substitutions/__init__.py

@@ -1,13 +1,16 @@
 """substitutions Module."""
 
 from .file_contents import FileContents
+from .ros_log_level import ROSLogLevel, ROSLoggers
 from .templated import Templated
 from .unary import Unary
 from .write_temp_file import WriteTempFile
 from .yaml_to_file import YAMLToFile
 
 __all__ = [
     'FileContents',
+    'ROSLogLevel',
+    'ROSLoggers',
     'Templated',
     'Unary',
     'WriteTempFile',

launch_ext/substitutions/ros_log_level.py

@@ -0,0 +1,65 @@
+"""Module for the ROSLogLevel substitution."""
+
+from typing import Text, List
+
+from launch.launch_context import LaunchContext
+from launch.substitution import Substitution
+from launch.some_substitutions_type import SomeSubstitutionsType
+from launch.utilities import normalize_to_list_of_substitutions
+from launch.utilities import perform_substitutions
+from launch.substitutions.substitution_failure import SubstitutionFailure
+
+def ROSLoggers(node_level: SomeSubstitutionsType = None, *args):
+    out = []
+    for a in args:
+        if type(a) == dict:
+            for k,v in a.items():
+                out.append("--log-level")
+                out.append(ROSLogLevel(v, name=k))
+
+    if node_level is not None:
+        out.append("--log-level")
+        out.append(ROSLogLevel(v))
+    return out
+
+class ROSLogLevel(Substitution):
+    """Substitution that contains the file contents of a file path."""
+
+    def __init__(self, level: SomeSubstitutionsType, name: SomeSubstitutionsType = None, all_loggers:bool = False) -> None:
+        """Create an ROSLogLevel substitution."""
+        super().__init__()
+
+        self.__node_name = None
+        self.__node_level = normalize_to_list_of_substitutions(level)
+        self.__all_loggers = all_loggers
+
+        if not all_loggers and name is not None:
+            self.__node_name = normalize_to_list_of_substitutions(name)
+            # raise SubstitutionFailure(f"Cannot specify all loggers and specific log levels at the same time.")
+
+    @property
+    def level(self) -> Substitution:
+        """Getter for path."""
+        return self.__node_level
+
+    def describe(self) -> Text:
+        """Return a description of this substitution as a string."""
+        return 'ROSLogLevel(level={})'.format(self.__node_level.describe())
+
+    def perform(self, context: LaunchContext) -> Text:
+        """Perform the substitution by obtaining the Node name from the context"""
+        level = perform_substitutions(context, self.__node_level)
+
+        try:
+            if self.__node_name is not None:
+                node_name = perform_substitutions(context, self.__node_name)
+            else:
+                node_name = context.get_locals_as_dict()['ros_specific_arguments']['name']
+                _, node_name = node_name.split(":=")
+        except KeyError:
+            raise SubstitutionFailure(f"Invalid context for ROSLogLevel substitution.")
+
+        if self.__all_loggers:
+            return level
+        else:
+            return f"{node_name}:={level}"