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,41 @@
+# Copyright 2022 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 Mapping
+from typing import Text
+
+from ..action import Action
+from ..launch_context import LaunchContext
+
+
+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
+
+    def execute(self, context: LaunchContext):
+        """Execute the action."""
+        context._replace_environment(self.__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,80 @@
+# Copyright 2022 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'
+
+    # 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'