Scope launch file dir/path locals to included launch file

Signed-off-by: Christophe Bedard <[email protected]>

Author: christophebedard

launch/launch/actions/include_launch_description.py

@@ -28,7 +28,7 @@
 
 import launch.logging
 
-
+from .opaque_function import OpaqueFunction
 from .set_launch_configuration import SetLaunchConfiguration
 from ..action import Action
 from ..frontend import Entity
@@ -212,13 +212,7 @@ def execute(self, context: LaunchContext) -> List[Union[SetLaunchConfiguration,
                                                             LaunchDescriptionEntity]]:
         """Execute the action."""
         launch_description = self.__launch_description_source.get_launch_description(context)
-        # If the location does not exist, then it's likely set to '<script>' or something.
-        context.extend_locals({
-            'current_launch_file_path': self._get_launch_file(),
-        })
-        context.extend_locals({
-            'current_launch_file_directory': self._get_launch_file_directory(),
-        })
+        self._set_launch_file_location_locals(context)
 
         # Do best effort checking to see if non-optional, non-default declared arguments
         # are being satisfied.
@@ -255,7 +249,45 @@ def execute(self, context: LaunchContext) -> List[Union[SetLaunchConfiguration,
             set_launch_configuration_actions.append(SetLaunchConfiguration(name, value))
 
         # Set launch arguments as launch configurations and then include the launch description.
-        return [*set_launch_configuration_actions, launch_description]
+        return [
+            *set_launch_configuration_actions,
+            launch_description,
+            OpaqueFunction(function=self._restore_launch_file_location_locals),
+        ]
+
+    def _set_launch_file_location_locals(self, context: LaunchContext) -> None:
+        context._push_locals()
+        # Keep the previous launch file path/dir locals so that we can restore them after
+        context_locals = context.get_locals_as_dict()
+        self.__previous_launch_file_path = context_locals.get('current_launch_file_path', None)
+        self.__previous_launch_file_dir = context_locals.get('current_launch_file_directory', None)
+        context.extend_locals({
+            'current_launch_file_path': self._get_launch_file(),
+        })
+        context.extend_locals({
+            'current_launch_file_directory': self._get_launch_file_directory(),
+        })
+
+    def _restore_launch_file_location_locals(self, context: LaunchContext) -> None:
+        # We want to keep the state of the context locals even after the include, since included
+        # launch descriptions are meant to act as if they were included literally in the parent
+        # launch description.
+        # However, we want to restore the launch file path/dir locals to their previous state, and
+        # we may have to just delete them if we're now going back to a launch script (i.e., not a
+        # launch file). However, there is no easy way to delete context locals, so save current
+        # locals, reset to the state before the include previous state and then re-apply locals,
+        # potentially minus the launch file path/dir locals.
+        context_locals = context.get_locals_as_dict()
+        if self.__previous_launch_file_path is None:
+            del context_locals['current_launch_file_path']
+        else:
+            context_locals['current_launch_file_path'] = self.__previous_launch_file_path
+        if self.__previous_launch_file_dir is None:
+            del context_locals['current_launch_file_directory']
+        else:
+            context_locals['current_launch_file_directory'] = self.__previous_launch_file_dir
+        context._pop_locals()
+        context.extend_locals(context_locals)
 
     def __repr__(self) -> Text:
         """Return a description of this IncludeLaunchDescription as a string."""

launch/test/launch/actions/test_include_launch_description.py

@@ -51,7 +51,7 @@ def test_include_launch_description_methods():
     assert isinstance(action.describe_sub_entities(), list)
     assert isinstance(action.describe_conditional_sub_entities(), list)
     # Result should only contain the launch description as there are no launch arguments.
-    assert action.visit(LaunchContext()) == [ld]
+    assert action.visit(LaunchContext())[0] == ld
     assert action.get_asyncio_future() is None
     assert len(action.launch_arguments) == 0
 
@@ -61,7 +61,7 @@ def test_include_launch_description_methods():
     assert isinstance(action2.describe_sub_entities(), list)
     assert isinstance(action2.describe_conditional_sub_entities(), list)
     # Result should only contain the launch description as there are no launch arguments.
-    assert action2.visit(LaunchContext()) == [ld2]
+    assert action2.visit(LaunchContext())[0] == ld2
     assert action2.get_asyncio_future() is None
     assert len(action2.launch_arguments) == 0
 
@@ -75,7 +75,7 @@ def test_include_launch_description_launch_file_location():
     assert isinstance(action.describe_conditional_sub_entities(), list)
     lc1 = LaunchContext()
     # Result should only contain the launch description as there are no launch arguments.
-    assert action.visit(lc1) == [ld]
+    assert action.visit(lc1)[0] == ld
     assert lc1.locals.current_launch_file_directory == '<script>'
     assert action.get_asyncio_future() is None
 
@@ -87,7 +87,7 @@ def test_include_launch_description_launch_file_location():
     assert isinstance(action2.describe_conditional_sub_entities(), list)
     lc2 = LaunchContext()
     # Result should only contain the launch description as there are no launch arguments.
-    assert action2.visit(lc2) == [ld2]
+    assert action2.visit(lc2)[0] == ld2
     assert lc2.locals.current_launch_file_directory == str(this_file.parent)
     assert action2.get_asyncio_future() is None
 
@@ -150,7 +150,7 @@ def test_include_launch_description_launch_arguments():
     assert len(action1.launch_arguments) == 1
     lc1 = LaunchContext()
     result1 = action1.visit(lc1)
-    assert len(result1) == 2
+    assert len(result1) == 3
     assert isinstance(result1[0], SetLaunchConfiguration)
     assert perform_substitutions(lc1, result1[0].name) == 'foo'
     assert perform_substitutions(lc1, result1[0].value) == 'FOO'
@@ -276,8 +276,9 @@ def test_include_python():
         assert 'IncludeLaunchDescription' in action.describe()
         assert isinstance(action.describe_sub_entities(), list)
         assert isinstance(action.describe_conditional_sub_entities(), list)
-        # Result should only contain a single launch description as there are no launch arguments.
-        assert len(action.visit(LaunchContext())) == 1
+        # Result should only contain a single launch description (+ internal action) as there are
+        # no launch arguments.
+        assert len(action.visit(LaunchContext())) == 2
         assert action.get_asyncio_future() is None
         assert len(action.launch_arguments) == 0