Source code for go2.hardware.native.native_hardware_lidar

import os
import sys
import shutil
import signal
import subprocess
from typing_extensions import override

from ..hardware_interface_lidar import HardwareInterfaceLIDAR




class NativeHardwareLIDAR(HardwareInterfaceLIDAR):
    def __init__(self) -> None:
        self._ros_proc = None
        


    @override
    def _initialize(self):
        self._launch_ros()


    def _launch_ros(self) -> None:
        kwargs = dict()
        # To detach child process: https://stackoverflow.com/questions/45911705/why-use-os-setsid-in-python
        kwargs["start_new_session"] = True

        self._ros_proc = subprocess.Popen(
            ["ros2", "launch", "bringup", "lidar_processor.launch.py"],
            **kwargs
        )



    @override
    def _shutdown(self):
        if self._ros_proc and self._ros_proc.poll() is None:
            try:
                os.killpg(os.getpgid(self._ros_proc.pid), signal.SIGINT)
                self._ros_proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                self._ros_proc.terminate()
                self._ros_proc.wait(timeout=5)