Feature clear launch configs

launch/launch/actions/__init__.py

@@ -25,6 +25,7 @@
 from .pop_launch_configurations import PopLaunchConfigurations
 from .push_launch_configurations import PushLaunchConfigurations
 from .register_event_handler import RegisterEventHandler
+from .reset_launch_configurations import ResetLaunchConfigurations
 from .set_environment_variable import SetEnvironmentVariable
 from .set_launch_configuration import SetLaunchConfiguration
 from .shutdown_action import Shutdown
@@ -44,6 +45,7 @@
     'OpaqueFunction',
     'PopLaunchConfigurations',
     'PushLaunchConfigurations',
+    'ResetLaunchConfigurations',
     'RegisterEventHandler',
     'SetEnvironmentVariable',
     'SetLaunchConfiguration',

launch/launch/actions/group_action.py

@@ -21,6 +21,7 @@
 
 from .pop_launch_configurations import PopLaunchConfigurations
 from .push_launch_configurations import PushLaunchConfigurations
+from .reset_launch_configurations import ResetLaunchConfigurations
 from .set_launch_configuration import SetLaunchConfiguration
 from ..action import Action
 from ..frontend import Entity
@@ -34,26 +35,47 @@
 @expose_action('group')
 class GroupAction(Action):
     """
-    Action that yields other actions, optionally scoping launch configurations.
+    Action that yields other actions, optionally scoping and forwarding launch configurations.
 
     This action is used to nest other actions without including a separate
     launch description, while also optionally having a condition (like all
-    other actions), scoping launch configurations, and/or declaring launch
-    configurations for just the group and its yielded actions.
+    other actions), scoping launch configurations, forwarding launch
+    configurations, and/or declaring launch configurations for just the
+    group and its yielded actions.
+
+    When scoped=True, changes to launch configurations are limited to the
+    scope of actions in the group action.
+
+    When scoped=True and forwarding=True, all existing launch configurations
+    are available in the scoped context.
+
+    When scoped=True and forwarding=False, all existing launch configurations
+    are removed from the scoped context.
+
+    Any launch configuration defined in the launch_configurations dictionary
+    will be set in the current context.
+    When scoped=False these configurations will persist even after the
+    GroupAction has completed.
+    When scoped=True these configurations will only be available to actions in
+    the GroupAction.
+    When scoped=True and forwarding=False, the launch_configurations dictionary
+    is evaluated before clearing, and then re-set in the cleared scoped context.
     """
 
     def __init__(
         self,
         actions: Iterable[Action],
         *,
         scoped: bool = True,
+        forwarding: bool = True,
         launch_configurations: Optional[Dict[SomeSubstitutionsType, SomeSubstitutionsType]] = None,
         **left_over_kwargs
     ) -> None:
         """Create a GroupAction."""
         super().__init__(**left_over_kwargs)
         self.__actions = actions
         self.__scoped = scoped
+        self.__forwarding = forwarding
         if launch_configurations is not None:
             self.__launch_configurations = launch_configurations
         else:
@@ -65,24 +87,49 @@ def parse(cls, entity: Entity, parser: Parser):
         """Return `GroupAction` action and kwargs for constructing it."""
         _, kwargs = super().parse(entity, parser)
         scoped = entity.get_attr('scoped', data_type=bool, optional=True)
+        forwarding = entity.get_attr('forwarding', data_type=bool, optional=True)
+        keeps = entity.get_attr('keep', data_type=List[Entity], optional=True)
         if scoped is not None:
             kwargs['scoped'] = scoped
-        kwargs['actions'] = [parser.parse_action(e) for e in entity.children]
+        if forwarding is not None:
+            kwargs['forwarding'] = forwarding
+        if keeps is not None:
+            kwargs['launch_configurations'] = {
+                    tuple(parser.parse_substitution(e.get_attr('name'))):
+                    parser.parse_substitution(e.get_attr('value')) for e in keeps
+            }
+            for e in keeps:
+                e.assert_entity_completely_parsed()
+        kwargs['actions'] = [parser.parse_action(e) for e in entity.children
+                             if e.type_name != 'keep']
         return cls, kwargs
 
     def get_sub_entities(self) -> List[LaunchDescriptionEntity]:
         """Return subentities."""
         if self.__actions_to_return is None:
-            self.__actions_to_return = []  # type: List[Action]
-            self.__actions_to_return += [
+            self.__actions_to_return = list(self.__actions)
+            configuration_sets = [
                 SetLaunchConfiguration(k, v) for k, v in self.__launch_configurations.items()
             ]
-            self.__actions_to_return += list(self.__actions)
             if self.__scoped:
+                if self.__forwarding:
+                    self.__actions_to_return = [
+                        PushLaunchConfigurations(),
+                        *configuration_sets,
+                        *self.__actions_to_return,
+                        PopLaunchConfigurations()
+                    ]
+                else:
+                    self.__actions_to_return = [
+                        PushLaunchConfigurations(),
+                        ResetLaunchConfigurations(self.__launch_configurations),
+                        *self.__actions_to_return,
+                        PopLaunchConfigurations()
+                    ]
+            else:
                 self.__actions_to_return = [
-                    PushLaunchConfigurations(),
-                    *self.__actions_to_return,
-                    PopLaunchConfigurations()
+                    *configuration_sets,
+                    *self.__actions_to_return
                 ]
         return self.__actions_to_return
 

launch/launch/actions/reset_launch_configurations.py

@@ -0,0 +1,87 @@
+# Copyright 2021 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 ResetLaunchConfigurations action."""
+
+from typing import Dict
+from typing import List
+from typing import Optional
+
+from ..action import Action
+from ..frontend import Entity
+from ..frontend import expose_action
+from ..frontend import Parser
+from ..launch_context import LaunchContext
+from ..some_substitutions_type import SomeSubstitutionsType
+from ..utilities import normalize_to_list_of_substitutions
+from ..utilities import perform_substitutions
+
+
+@expose_action('reset')
+class ResetLaunchConfigurations(Action):
+    """
+    Action that resets launch configurations in the current context.
+
+    This action can be used to clear the launch configurations from the
+    context it was called in.
+    It optionally can be given a dictionary with launch configurations
+    to be set after clearing.
+    Launch configurations given in the dictionary are evaluated before
+    the context launch configurations are cleared.
+    This allows launch configurations to be passed through the clearing
+    of the context.
+
+    If launch_configurations is None or an empty dict then all launch configurations
+    will be cleared.
+
+    If launch_configurations has entries (i.e. {'foo': 'FOO'}) then these will be
+    set after the clearing operation.
+    """
+
+    def __init__(
+        self,
+        launch_configurations: Optional[Dict[SomeSubstitutionsType, SomeSubstitutionsType]] = None,
+        **kwargs
+    ) -> None:
+        """Create an ResetLaunchConfigurations action."""
+        super().__init__(**kwargs)
+        self.__launch_configurations = launch_configurations
+
+    @classmethod
+    def parse(cls, entity: Entity, parser: Parser):
+        """Return `ResetLaunchConfigurations` action and kwargs for constructing it."""
+        _, kwargs = super().parse(entity, parser)
+        keeps = entity.get_attr('keep', data_type=List[Entity], optional=True)
+        if keeps is not None:
+            kwargs['launch_configurations'] = {
+                    tuple(parser.parse_substitution(e.get_attr('name'))):
+                    parser.parse_substitution(e.get_attr('value')) for e in keeps
+            }
+            for e in keeps:
+                e.assert_entity_completely_parsed()
+        return cls, kwargs
+
+    def execute(self, context: LaunchContext):
+        """Execute the action."""
+        if self.__launch_configurations is None:
+            context.launch_configurations.clear()
+        else:
+            evaluated_configurations = {}
+            for k, v in self.__launch_configurations.items():
+                evaluated_k = perform_substitutions(context, normalize_to_list_of_substitutions(k))
+                evaluated_v = perform_substitutions(context, normalize_to_list_of_substitutions(v))
+                evaluated_configurations[evaluated_k] = evaluated_v
+
+            context.launch_configurations.clear()
+            context.launch_configurations.update(evaluated_configurations)

launch/test/launch/actions/test_group_action.py

@@ -19,15 +19,18 @@
 from launch.actions import GroupAction
 from launch.actions import PopLaunchConfigurations
 from launch.actions import PushLaunchConfigurations
+from launch.actions import ResetLaunchConfigurations
 from launch.actions import SetLaunchConfiguration
+from launch.substitutions import LaunchConfiguration
 
 
 def test_group_action_constructors():
     """Test the constructors for the GroupAction class."""
     GroupAction([])
     GroupAction([Action()])
     GroupAction([Action()], scoped=False)
-    GroupAction([Action()], scoped=False, launch_configurations={'foo': 'FOO'})
+    GroupAction([Action()], scoped=False, forwarding=False)
+    GroupAction([Action()], scoped=False, forwarding=False, launch_configurations={'foo': 'FOO'})
 
 
 def test_group_action_execute():
@@ -39,16 +42,26 @@ def test_group_action_execute():
     assert len(lc1.launch_configurations) == 0
 
     assert len(lc1.launch_configurations) == 0
-    result = GroupAction([]).visit(lc1)
-    assert len(result) == 2  # push and pop actions, due to scope=True
+    result = GroupAction([], forwarding=True).visit(lc1)
+    assert len(result) == 2  # push and pop actions, due to scope=True, forwarded=True
     assert isinstance(result[0], PushLaunchConfigurations)
     assert isinstance(result[1], PopLaunchConfigurations)
     for a in result:
         a.visit(lc1)
     assert len(lc1.launch_configurations) == 0
 
     assert len(lc1.launch_configurations) == 0
-    result = GroupAction([], launch_configurations={'foo': 'FOO'}).visit(lc1)
+    result = GroupAction([], forwarding=False).visit(lc1)
+    assert len(result) == 3  # push, reset, pop actions, due to scope=True, forwarded=False
+    assert isinstance(result[0], PushLaunchConfigurations)
+    assert isinstance(result[1], ResetLaunchConfigurations)
+    assert isinstance(result[2], PopLaunchConfigurations)
+    for a in result:
+        a.visit(lc1)
+    assert len(lc1.launch_configurations) == 0
+
+    assert len(lc1.launch_configurations) == 0
+    result = GroupAction([], forwarding=True, launch_configurations={'foo': 'FOO'}).visit(lc1)
     assert len(result) == 3  # push, set 1 launch_configurations, and pop actions
     assert isinstance(result[0], PushLaunchConfigurations)
     assert isinstance(result[1], SetLaunchConfiguration)
@@ -58,7 +71,8 @@ def test_group_action_execute():
     assert len(lc1.launch_configurations) == 0
 
     assert len(lc1.launch_configurations) == 0
-    result = GroupAction([], launch_configurations={'foo': 'FOO', 'bar': 'BAR'}).visit(lc1)
+    result = GroupAction([], forwarding=True,
+                         launch_configurations={'foo': 'FOO', 'bar': 'BAR'}).visit(lc1)
     assert len(result) == 4  # push, set 2 launch_configurations, and pop actions
     assert isinstance(result[0], PushLaunchConfigurations)
     assert isinstance(result[1], SetLaunchConfiguration)
@@ -68,6 +82,17 @@ def test_group_action_execute():
         a.visit(lc1)
     assert len(lc1.launch_configurations) == 0
 
+    assert len(lc1.launch_configurations) == 0
+    result = GroupAction([], forwarding=False,
+                         launch_configurations={'foo': 'FOO', 'bar': 'BAR'}).visit(lc1)
+    assert len(result) == 3  # push, reset (which will set launch_configurations), and pop actions
+    assert isinstance(result[0], PushLaunchConfigurations)
+    assert isinstance(result[1], ResetLaunchConfigurations)
+    assert isinstance(result[2], PopLaunchConfigurations)
+    for a in result:
+        a.visit(lc1)
+    assert len(lc1.launch_configurations) == 0
+
     assert len(lc1.launch_configurations) == 0
     PushLaunchConfigurations().visit(lc1)
     result = GroupAction([], scoped=False, launch_configurations={'foo': 'FOO'}).visit(lc1)
@@ -92,7 +117,8 @@ def test_group_action_execute():
     assert len(lc1.launch_configurations) == 0
 
     assert len(lc1.launch_configurations) == 0
-    result = GroupAction([Action()], launch_configurations={'foo': 'FOO'}).visit(lc1)
+    result = GroupAction([Action()], forwarding=True,
+                         launch_configurations={'foo': 'FOO'}).visit(lc1)
     assert len(result) == 4  # push, set 1 launch_configurations, the 1 action, and pop actions
     assert isinstance(result[0], PushLaunchConfigurations)
     assert isinstance(result[1], SetLaunchConfiguration)
@@ -101,3 +127,65 @@ def test_group_action_execute():
     for a in result:
         a.visit(lc1)
     assert len(lc1.launch_configurations) == 0
+
+    assert len(lc1.launch_configurations) == 0
+    lc1.launch_configurations['foo'] = 'FOO'
+    lc1.launch_configurations['bar'] = 'BAR'
+    result = GroupAction([Action()], forwarding=False,
+                         launch_configurations={'bar': LaunchConfiguration('bar'),
+                                                'baz': 'BAZ'}).visit(lc1)
+    # push, reset (which will set launch_configurations), 1 action, and pop actions
+    assert len(result) == 4
+    assert isinstance(result[0], PushLaunchConfigurations)
+    assert isinstance(result[1], ResetLaunchConfigurations)
+    assert isinstance(result[2], Action)
+    assert isinstance(result[3], PopLaunchConfigurations)
+    result[0].visit(lc1)  # Push
+    assert 'foo' in lc1.launch_configurations.keys()  # Copied to new scope, before Reset
+    assert lc1.launch_configurations['foo'] == 'FOO'
+    assert 'bar' in lc1.launch_configurations.keys()  # Copied to new scope, before Reset
+    assert lc1.launch_configurations['bar'] == 'BAR'
+    result[1].visit(lc1)  # Reset
+    assert 'foo' not in lc1.launch_configurations.keys()  # Cleared from scope in Reset
+    assert 'bar' in lc1.launch_configurations.keys()  # Evaluated and forwarded in Reset
+    assert lc1.launch_configurations['bar'] == 'BAR'
+    assert 'baz' in lc1.launch_configurations.keys()  # Evaluated and added in Reset
+    assert lc1.launch_configurations['baz'] == 'BAZ'
+    result[2].visit(lc1)  # Action
+    result[3].visit(lc1)  # Pop
+    assert 'foo' in lc1.launch_configurations.keys()  # Still in original scope
+    assert lc1.launch_configurations['foo'] == 'FOO'
+    assert 'bar' in lc1.launch_configurations.keys()  # Still in original scope
+    assert lc1.launch_configurations['bar'] == 'BAR'
+    assert 'baz' not in lc1.launch_configurations.keys()  # Out of scope from pop, no longer exists
+    assert len(lc1.launch_configurations) == 2
+    lc1.launch_configurations.clear()
+
+    assert len(lc1.launch_configurations) == 0
+    lc1.launch_configurations['foo'] = 'FOO'
+    lc1.launch_configurations['bar'] = 'BAR'
+    result = GroupAction([Action()], forwarding=True,
+                         launch_configurations={'foo': 'OOF'}).visit(lc1)
+    # push, 1 set (overwrite), 1 action, and pop actions
+    assert len(result) == 4
+    assert isinstance(result[0], PushLaunchConfigurations)
+    assert isinstance(result[1], SetLaunchConfiguration)
+    assert isinstance(result[2], Action)
+    assert isinstance(result[3], PopLaunchConfigurations)
+    result[0].visit(lc1)  # Push
+    assert 'foo' in lc1.launch_configurations.keys()  # Copied to new scope, before Set
+    assert lc1.launch_configurations['foo'] == 'FOO'
+    assert 'bar' in lc1.launch_configurations.keys()  # Copied to new scope
+    assert lc1.launch_configurations['bar'] == 'BAR'
+    result[1].visit(lc1)  # Set
+    assert 'foo' in lc1.launch_configurations.keys()  # Overwritten in Set
+    assert lc1.launch_configurations['foo'] == 'OOF'
+    assert 'bar' in lc1.launch_configurations.keys()  # Untouched
+    assert lc1.launch_configurations['bar'] == 'BAR'
+    result[2].visit(lc1)  # Action
+    result[3].visit(lc1)  # Pop
+    assert 'foo' in lc1.launch_configurations.keys()  # Still in original scope with original value
+    assert lc1.launch_configurations['foo'] == 'FOO'
+    assert 'bar' in lc1.launch_configurations.keys()  # Still in original scope with original value
+    assert lc1.launch_configurations['bar'] == 'BAR'
+    lc1.launch_configurations.clear()