new ros2 api launcher for physical robots

Author: lu164

manager/manager/launcher/launcher_ros2_api.py

@@ -30,8 +30,6 @@ def run(self,callback):
         self._set_environment()
         launch_file = os.path.expandvars(self.launch_file)
 
-        print("\n\n\nLaunch file: " + str(launch_file))
-
         if (ACCELERATION_ENABLED):
             exercise_launch_cmd = f"export VGL_DISPLAY={DRI_PATH}; vglrun ros2 launch {launch_file}"
         else:
@@ -66,5 +64,4 @@ def _set_environment(self):
 
         os.environ["GAZEBO_RESOURCE_PATH"] = f"{os.environ.get('GAZEBO_RESOURCE_PATH', '')}:{':'.join(resource_folders)}"
         os.environ["GAZEBO_MODEL_PATH"] = f"{os.environ.get('GAZEBO_MODEL_PATH', '')}:{':'.join(model_folders)}"
-        os.environ["GAZEBO_PLUGIN_PATH"] = f"{os.environ.get('GAZEBO_PLUGIN_PATH', '')}:{':'.join(plugin_folders)}"
-        os.environ["PHYSICAL_ROBOT_PKG_PATH"] = f"{os.environ.get('PHYSICAL_ROBOT_PKG_PATH', '')}:{':'.join(resource_folders)}"
+        os.environ["GAZEBO_PLUGIN_PATH"] = f"{os.environ.get('GAZEBO_PLUGIN_PATH', '')}:{':'.join(plugin_folders)}"

manager/manager/launcher/launcher_ros2_api_phy.py

@@ -0,0 +1,62 @@
+import os
+from typing import List, Any
+import time
+import stat
+
+from src.manager.manager.launcher.launcher_interface import ILauncher, LauncherException
+from src.manager.manager.docker_thread.docker_thread import DockerThread
+import subprocess
+
+import logging
+
+class LauncherRos2ApiPhy(ILauncher):
+    exercise_id: str
+    type: str
+    module: str
+    resource_folders: List[str]
+    model_folders: List[str]
+    plugin_folders: List[str]
+    parameters: List[str]
+    launch_file: str
+    running = False
+    threads = []
+
+    def run(self,callback):
+        DRI_PATH = os.path.join("/dev/dri", os.environ.get("DRI_NAME", "card0"))
+        ACCELERATION_ENABLED = self.check_device(DRI_PATH)
+
+        logging.getLogger("roslaunch").setLevel(logging.CRITICAL)
+
+        # expand variables in configuration paths
+        launch_file = os.path.expandvars(self.launch_file)
+
+        print("\n\n\nLaunch file: " + str(launch_file))
+
+        if (ACCELERATION_ENABLED):
+            exercise_launch_cmd = f"export VGL_DISPLAY={DRI_PATH}; vglrun ros2 launch {launch_file}"
+        else:
+            exercise_launch_cmd = f"ros2 launch {launch_file}"
+
+        exercise_launch_thread = DockerThread(exercise_launch_cmd)
+        exercise_launch_thread.start()
+        self.threads.append(exercise_launch_thread)
+
+        self.running = True
+
+    def is_running(self):
+        return self.running
+    
+    def check_device(self, device_path):
+        try:
+            return stat.S_ISCHR(os.lstat(device_path)[stat.ST_MODE])
+        except:
+            return False
+
+    def terminate(self):
+        for thread in self.threads:
+            thread.terminate()
+            thread.join()
+        self.running = False
+
+    def died(self):
+        pass