Scope launch file dir/path locals to included launch file

launch/launch/actions/include_launch_description.py

@@ -24,6 +24,7 @@
 
 import launch.logging
 
+from .opaque_function import OpaqueFunction
 from .set_launch_configuration import SetLaunchConfiguration
 from ..action import Action
 from ..frontend import Entity
@@ -147,13 +148,7 @@ def _try_get_arguments_names_without_context(self):
     def execute(self, context: LaunchContext) -> List[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.
@@ -188,7 +183,45 @@ def execute(self, context: LaunchContext) -> List[LaunchDescriptionEntity]:
             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/launch/included/launch_file_dir.launch.py

@@ -0,0 +1,33 @@
+# Copyright 2025 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.
+
+from launch import LaunchContext
+from launch import LaunchDescription
+from launch.actions import OpaqueFunction
+from launch.actions import SetEnvironmentVariable
+from launch.substitutions import ThisLaunchFile
+from launch.substitutions import ThisLaunchFileDir
+
+
+def set_local(context: LaunchContext) -> None:
+    context.extend_locals({'included_local': 'context_local_value'})
+
+
+def generate_launch_description():
+    """Fixture for tests."""
+    return LaunchDescription([
+        SetEnvironmentVariable('Included_ThisLaunchFile', ThisLaunchFile()),
+        OpaqueFunction(function=set_local),
+        SetEnvironmentVariable('Included_ThisLaunchFileDir', ThisLaunchFileDir()),
+    ])

launch/test/launch/actions/launch/parent_launch_file_dir.launch.py

@@ -0,0 +1,38 @@
+# Copyright 2025 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.
+
+import os
+
+from launch import LaunchDescription
+from launch.actions import IncludeLaunchDescription
+from launch.actions import SetEnvironmentVariable
+from launch.substitutions import ThisLaunchFile
+from launch.substitutions import ThisLaunchFileDir
+
+
+def generate_launch_description():
+    """Fixture for tests."""
+    return LaunchDescription([
+        SetEnvironmentVariable('Before_ThisLaunchFile', ThisLaunchFile()),
+        SetEnvironmentVariable('Before_ThisLaunchFileDir', ThisLaunchFileDir()),
+        IncludeLaunchDescription(
+            os.path.join(
+                os.path.dirname(os.path.abspath(__file__)),
+                'included',
+                'launch_file_dir.launch.py',
+            ),
+        ),
+        SetEnvironmentVariable('After_ThisLaunchFile', ThisLaunchFile()),
+        SetEnvironmentVariable('After_ThisLaunchFileDir', ThisLaunchFileDir()),
+    ])

launch/test/launch/actions/test_include_launch_description.py

@@ -22,9 +22,13 @@
 from launch import LaunchService
 from launch.actions import DeclareLaunchArgument
 from launch.actions import IncludeLaunchDescription
+from launch.actions import OpaqueFunction
 from launch.actions import ResetLaunchConfigurations
+from launch.actions import SetEnvironmentVariable
 from launch.actions import SetLaunchConfiguration
 from launch.launch_description_sources import PythonLaunchDescriptionSource
+from launch.substitutions import ThisLaunchFile
+from launch.substitutions import ThisLaunchFileDir
 from launch.utilities import perform_substitutions
 
 import pytest
@@ -46,7 +50,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
 
@@ -56,7 +60,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
 
@@ -70,7 +74,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
 
@@ -82,11 +86,58 @@ 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 == os.path.dirname(this_file)
     assert action2.get_asyncio_future() is None
 
 
+def test_include_launch_description_launch_file_dir_location_scoped():
+    """Test that the launch file name & dir locals are scoped to the included launch file."""
+    # Rely on the test launch files to set environment variables with
+    # ThisLaunchFile()/ThisLaunchFileDir() to make testing easier
+    parent_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'launch')
+    parent_launch_file = os.path.join(parent_dir, 'parent_launch_file_dir.launch.py')
+    included_dir = os.path.join(parent_dir, 'included')
+    included_launch_file = os.path.join(included_dir, 'launch_file_dir.launch.py')
+
+    # The current launch file/dir context locals should be scoped to the included launch file
+    ld = LaunchDescription([IncludeLaunchDescription(parent_launch_file)])
+    ls = LaunchService()
+    ls.include_launch_description(ld)
+    assert 0 == ls.run()
+    lc = ls.context
+    assert lc.environment.get('Before_ThisLaunchFile') == parent_launch_file
+    assert lc.environment.get('Before_ThisLaunchFileDir') == parent_dir
+    assert lc.environment.get('Included_ThisLaunchFile') == included_launch_file
+    assert lc.environment.get('Included_ThisLaunchFileDir') == included_dir
+    assert lc.environment.get('After_ThisLaunchFile') == parent_launch_file
+    assert lc.environment.get('After_ThisLaunchFileDir') == parent_dir
+
+    # The launch file/dir context locals should be completely removed after the first included
+    # (parent) launch file, because at that point we're in a launch script and not a launch file,
+    # and therefore these substitutions should raise an error
+    ld2 = LaunchDescription([
+        IncludeLaunchDescription(parent_launch_file),
+        SetEnvironmentVariable('Outside_ThisLaunchFile', ThisLaunchFile()),
+        SetEnvironmentVariable('Outside_ThisLaunchFileDir', ThisLaunchFileDir()),
+    ])
+    ls2 = LaunchService()
+    ls2.include_launch_description(ld2)
+    assert 1 == ls2.run()
+
+    # The non-launch file/dir context locals should not be scoped to the included launch file
+    def assert_unscoped_context_local(context: LaunchContext):
+        assert context.locals.included_local == 'context_local_value'
+
+    ld3 = LaunchDescription([
+        IncludeLaunchDescription(parent_launch_file),
+        OpaqueFunction(function=assert_unscoped_context_local),
+    ])
+    ls3 = LaunchService()
+    ls3.include_launch_description(ld3)
+    assert 0 == ls3.run()
+
+
 def test_include_launch_description_launch_arguments():
     """Test the interactions between declared launch arguments and IncludeLaunchDescription."""
     # test that arguments are set when given, even if they are not declared
@@ -98,7 +149,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'
@@ -227,8 +278,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