Capture the environment variables in TimerAction

launch/launch/actions/__init__.py

@@ -29,6 +29,7 @@
 from .push_environment import PushEnvironment
 from .push_launch_configurations import PushLaunchConfigurations
 from .register_event_handler import RegisterEventHandler
+from .replace_environment_variables import ReplaceEnvironmentVariables
 from .reset_environment import ResetEnvironment
 from .reset_launch_configurations import ResetLaunchConfigurations
 from .set_environment_variable import SetEnvironmentVariable
@@ -57,6 +58,7 @@
     'ResetEnvironment',
     'ResetLaunchConfigurations',
     'RegisterEventHandler',
+    'ReplaceEnvironmentVariables',
     'SetEnvironmentVariable',
     'SetLaunchConfiguration',
     'Shutdown',

launch/launch/actions/replace_environment_variables.py

@@ -0,0 +1,78 @@
+# Copyright 2023 Open Source Robotics Foundation, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Module for the ReplaceEnvironmentVariables action."""
+
+from typing import List
+from typing import Mapping
+from typing import Text
+
+from ..action import Action
+from ..frontend import Entity
+from ..frontend import expose_action
+from ..frontend import Parser
+from ..launch_context import LaunchContext
+from ..utilities import normalize_to_list_of_substitutions
+from ..utilities import perform_substitutions
+
+
+@expose_action('rep_env')
+class ReplaceEnvironmentVariables(Action):
+    """
+    Action that replaces the environment variables in the current context.
+
+    The previous state can be saved by pushing the stack with the
+    :py:class:`launch.actions.PushEnvironment` action.
+    And can be restored by popping the stack with the
+    :py:class:`launch.actions.PopEnvironment` action.
+    """
+
+    def __init__(self, environment: Mapping[Text, Text] = {}, **kwargs) -> None:
+        """Create a ReplaceEnvironmentVariables action."""
+        super().__init__(**kwargs)
+        self.__environment = environment
+
+    @classmethod
+    def parse(
+        cls,
+        entity: Entity,
+        parser: Parser
+    ):
+        """Return the `ReplaceEnvironmentVariables` action and kwargs for constructing it."""
+        _, kwargs = super().parse(entity, parser)
+        env = entity.get_attr('env', data_type=List[Entity], optional=True)
+
+        if env is not None:
+            kwargs['environment'] = {
+                tuple(parser.parse_substitution(e.get_attr('name'))):
+                parser.parse_substitution(e.get_attr('value')) for e in env
+            }
+            for e in env:
+                e.assert_entity_completely_parsed()
+        return cls, kwargs
+
+    @property
+    def environment(self):
+        """Getter for environment."""
+        return self.__environment
+
+    def execute(self, context: LaunchContext):
+        """Execute the action."""
+        evaluated_environment = {}
+        for k, v in self.__environment.items():
+            evaluated_k = perform_substitutions(context, normalize_to_list_of_substitutions(k))
+            evaluated_v = perform_substitutions(context, normalize_to_list_of_substitutions(v))
+            evaluated_environment[evaluated_k] = evaluated_v
+
+        context._replace_environment(evaluated_environment)

launch/launch/actions/timer_action.py

@@ -30,6 +30,9 @@
 import launch.logging
 
 from .opaque_function import OpaqueFunction
+from .pop_environment import PopEnvironment
+from .push_environment import PushEnvironment
+from .replace_environment_variables import ReplaceEnvironmentVariables
 
 from ..action import Action
 from ..event_handler import EventHandler
@@ -84,6 +87,7 @@ def __init__(
         self.__period = type_utils.normalize_typed_substitution(period, float)
         self.__actions = actions
         self.__context_locals: Dict[Text, Any] = {}
+        self.__context_environment: Dict[Text, Text] = {}
         self._completed_future: Optional[asyncio.Future] = None
         self.__canceled = False
         self._canceled_future: Optional[asyncio.Future] = None
@@ -139,7 +143,12 @@ def describe_conditional_sub_entities(self) -> List[Tuple[
     def handle(self, context: LaunchContext) -> Optional[SomeEntitiesType]:
         """Handle firing of timer."""
         context.extend_locals(self.__context_locals)
-        return self.__actions
+        return [
+            PushEnvironment(),
+            ReplaceEnvironmentVariables(self.__context_environment),
+            *self.__actions,
+            PopEnvironment(),
+        ]
 
     def cancel(self) -> None:
         """
@@ -191,6 +200,8 @@ def execute(self, context: LaunchContext) -> Optional[List[LaunchDescriptionEnti
 
         # Capture the current context locals so the yielded actions can make use of them too.
         self.__context_locals = dict(context.get_locals_as_dict())  # Capture a copy
+        # Capture the current context environment so the yielded actions can make use of them too.
+        self.__context_environment = dict(context.environment)  # Capture a copy
         context.asyncio_loop.create_task(self._wait_to_fire_event(context))
 
         # By default, the 'shutdown' event will cause timers to cancel so they don't hold up the

launch/launch/launch_context.py

@@ -188,6 +188,10 @@ def _reset_environment(self):
         os.environ.clear()
         os.environ.update(self.__environment_reset)
 
+    def _replace_environment(self, environment: Mapping[Text, Text]):
+        os.environ.clear()
+        os.environ.update(environment)
+
     def _push_launch_configurations(self):
         self.__launch_configurations_stack.append(self.__launch_configurations.copy())
 

launch/test/launch/actions/test_replace_environment_variables.py

@@ -0,0 +1,93 @@
+# Copyright 2023 Open Source Robotics Foundation, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tests for the ReplaceEnvironmentVariables action classes."""
+
+import os
+
+from launch import LaunchContext
+from launch.actions import PopEnvironment
+from launch.actions import PushEnvironment
+from launch.actions import ReplaceEnvironmentVariables
+
+from temporary_environment import sandbox_environment_variables
+
+
+@sandbox_environment_variables
+def test_replace_environment_constructors():
+    """Test the constructors for ReplaceEnvironmentVariables class."""
+    ReplaceEnvironmentVariables({})
+    ReplaceEnvironmentVariables({'foo': 'bar', 'spam': 'eggs'})
+
+
+@sandbox_environment_variables
+def test_replace_environment_execute():
+    """Test the execute() of the ReplaceEnvironmentVariables class."""
+    assert isinstance(os.environ, os._Environ)
+
+    # replaces empty state
+    context = LaunchContext()
+    context.environment.clear()
+    assert len(context.environment) == 0
+    ReplaceEnvironmentVariables({'foo': 'bar', 'spam': 'eggs'}).visit(context)
+    assert len(context.environment) == 2
+    assert 'foo' in context.environment
+    assert context.environment['foo'] == 'bar'
+    assert 'spam' in context.environment
+    assert context.environment['spam'] == 'eggs'
+
+    # replaces non empty state
+    context = LaunchContext()
+    context.environment.clear()
+    assert len(context.environment) == 0
+    context.environment['quux'] = 'quuux'
+    assert len(context.environment) == 1
+    assert 'quux' in context.environment
+    assert context.environment['quux'] == 'quuux'
+    ReplaceEnvironmentVariables({'foo': 'bar', 'spam': 'eggs'}).visit(context)
+    assert len(context.environment) == 2
+    assert 'foo' in context.environment
+    assert context.environment['foo'] == 'bar'
+    assert 'spam' in context.environment
+    assert context.environment['spam'] == 'eggs'
+
+    # replaces existing environment variable
+    context = LaunchContext()
+    context.environment.clear()
+    assert len(context.environment) == 0
+    context.environment['quux'] = 'quuux'
+    assert len(context.environment) == 1
+    assert 'quux' in context.environment
+    assert context.environment['quux'] == 'quuux'
+    ReplaceEnvironmentVariables({'quux': 'eggs'}).visit(context)
+    assert len(context.environment) == 1
+    assert 'quux' in context.environment
+    assert context.environment['quux'] == 'eggs'
+
+    # Replacing the environment should not change the type of os.environ
+    assert isinstance(os.environ, os._Environ)
+
+    # does not interfere with PopEnvironment and PushEnvironment action classes
+    context = LaunchContext()
+    context.environment.clear()
+    context.environment['quux'] = 'quuux'
+    assert len(context.environment) == 1
+    assert 'quux' in context.environment
+    assert context.environment['quux'] == 'quuux'
+    PushEnvironment().visit(context)
+    ReplaceEnvironmentVariables({'foo': 'bar', 'spam': 'eggs'}).visit(context)
+    PopEnvironment().visit(context)
+    assert len(context.environment) == 1
+    assert 'quux' in context.environment
+    assert context.environment['quux'] == 'quuux'