From 4c21a9af7f0ce226865113820a2b203bf29f79c9 Mon Sep 17 00:00:00 2001
From: Gwilym-Rutherford <gmarutherford1@sheffield.ac.uk>
Date: Mon, 18 Aug 2025 13:59:19 +0100
Subject: [PATCH] Add ROS2_launch.py

---
 srunner/tools/ROS2_launch.py | 54 ++++++++++++++++++++++++++++++++++++
 1 file changed, 54 insertions(+)
 create mode 100644 srunner/tools/ROS2_launch.py

diff --git a/srunner/tools/ROS2_launch.py b/srunner/tools/ROS2_launch.py
new file mode 100644
index 0000000..695a437
--- /dev/null
+++ b/srunner/tools/ROS2_launch.py
@@ -0,0 +1,54 @@
+import subprocess
+import threading
+
+class ROS2Launch(object):
+    threads = dict()
+    stop_event = threading.Event()
+        
+    @classmethod
+    def launch_file(self, package_name: str, launch_file: str, args: dict) -> None:
+        """will launch the file specified from the package specified
+
+        Args:
+            package_name (str): ROS2 package name
+            launch_file (str): launch file name
+            args (dict): dictionary of arguments to pass to launch file
+        """
+        command = self._construct_command(package_name, launch_file, args)
+        thread = threading.Thread(target=self._run_command, args=(command, self.stop_event))
+        self.threads[self.package_name] = {
+            "thread": thread,
+            "stop": False
+        }
+        thread.start()
+    
+    @classmethod
+    def cleanup(self, package_name: str) -> None:
+        """provide package name to stop running
+
+        Args:
+            package_name (str): name of the ROS2 package to kill
+        """
+        if package_name in self.threads:
+            thread = self.threads[package_name]
+            thread['stop'] = True
+        
+    def _run_command(self, command: list[str], package_name) -> None:
+        process = subprocess.Popen(command, check=True)
+        
+        if package_name in self.threads:
+            while(not self.threads[package_name]['stop']):
+                if process.poll() is not None:
+                    break
+                
+            if process.poll() is None:
+                process.kill()
+        
+    def _construct_command(self, package_name, launch_file, args) -> list[str]:
+        command = []
+        command.append("ros2 launch").append(package_name).append(launch_file)
+        for key, value in args.dict():
+            command.append(f"{key}:={value}")
+        return command
+            
+        
\ No newline at end of file