get imu

Author: hatomist

openlch/__init__.py

@@ -1,3 +1,3 @@
-__version__ = "0.6.1"
+__version__ = "0.7.0"
 
 from .hal import HAL

openlch/cli.py

@@ -25,6 +25,7 @@ def cli() -> None:
     - start-video-stream: Start the video stream
     - stop-video-stream: Stop the video stream
     - get-video-stream-urls: Get the URLs for various video stream formats
+    - get-imu-data: Get current IMU sensor data (gyroscope and accelerometer readings)
 
     Use 'openlch COMMAND --help' for more information on a specific command.
     """
@@ -268,5 +269,26 @@ def set_torque_enable(settings: List[Tuple[int, str]], ip: str) -> None:
     finally:
         hal.close()
 
+@cli.command()
+@click.argument("ip", default=DEFAULT_IP)
+def get_imu_data(ip: str) -> None:
+    """Get current IMU sensor data (gyroscope and accelerometer readings)."""
+    hal = HAL(ip)
+    try:
+        imu_data = hal.imu.get_data()
+        click.echo("IMU Sensor Data:")
+        click.echo("\nGyroscope (degrees/second):")
+        click.echo(f"  X: {imu_data['gyro']['x']:.2f}")
+        click.echo(f"  Y: {imu_data['gyro']['y']:.2f}")
+        click.echo(f"  Z: {imu_data['gyro']['z']:.2f}")
+        click.echo("\nAccelerometer (g):")
+        click.echo(f"  X: {imu_data['accel']['x']:.2f}")
+        click.echo(f"  Y: {imu_data['accel']['y']:.2f}")
+        click.echo(f"  Z: {imu_data['accel']['z']:.2f}")
+    except Exception as e:
+        click.echo(f"An error occurred: {str(e)}")
+    finally:
+        hal.close()
+
 if __name__ == "__main__":
     cli()

openlch/hal.py

@@ -23,6 +23,7 @@ def __init__(self, host: str = '192.168.42.1', port: int = 50051) -> None:
         self.__stub = hal_pb_pb2_grpc.ServoControlStub(self.__channel)
         self.servo = self.Servo(self.__stub)
         self.system = self.System(self.__stub)
+        self.imu = self.IMU(self.__stub)
 
     def close(self) -> None:
         """Close the gRPC channel."""
@@ -271,3 +272,26 @@ def get_video_stream_urls(self) -> Dict[str, List[str]]:
                 'mse': list(response.mse),
                 'rtsp': list(response.rtsp)
             }
+
+    class IMU:
+        """Class for IMU-related operations."""
+
+        def __init__(self, stub):
+            self.__stub = stub
+
+        def get_data(self) -> Dict[str, Dict[str, float]]:
+            """
+            Get current IMU sensor data including gyroscope and accelerometer readings.
+
+            Returns:
+                Dict[str, Dict[str, float]]: A dictionary containing gyroscope and accelerometer data:
+                    {
+                        'gyro': {'x': float, 'y': float, 'z': float},  # Angular velocity in degrees/second
+                        'accel': {'x': float, 'y': float, 'z': float}  # Linear acceleration in g
+                    }
+            """
+            response = self.__stub.GetImuData(hal_pb_pb2.Empty())
+            return {
+                'gyro': {'x': response.gyro.x, 'y': response.gyro.y, 'z': response.gyro.z},
+                'accel': {'x': response.accel.x, 'y': response.accel.y, 'z': response.accel.z}
+            }

openlch/hal_pb_pb2.py

@@ -104,6 +104,11 @@ def __init__(self, channel):
                 request_serializer=hal__pb__pb2.TorqueEnableSettings.SerializeToString,
                 response_deserializer=hal__pb__pb2.Empty.FromString,
                 _registered_method=True)
+        self.GetImuData = channel.unary_unary(
+                '/hal_pb.ServoControl/GetImuData',
+                request_serializer=hal__pb__pb2.Empty.SerializeToString,
+                response_deserializer=hal__pb__pb2.ImuData.FromString,
+                _registered_method=True)
 
 
 class ServoControlServicer(object):
@@ -193,6 +198,12 @@ def SetTorqueEnable(self, request, context):
         context.set_details('Method not implemented!')
         raise NotImplementedError('Method not implemented!')
 
+    def GetImuData(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
 
 def add_ServoControlServicer_to_server(servicer, server):
     rpc_method_handlers = {
@@ -266,6 +277,11 @@ def add_ServoControlServicer_to_server(servicer, server):
                     request_deserializer=hal__pb__pb2.TorqueEnableSettings.FromString,
                     response_serializer=hal__pb__pb2.Empty.SerializeToString,
             ),
+            'GetImuData': grpc.unary_unary_rpc_method_handler(
+                    servicer.GetImuData,
+                    request_deserializer=hal__pb__pb2.Empty.FromString,
+                    response_serializer=hal__pb__pb2.ImuData.SerializeToString,
+            ),
     }
     generic_handler = grpc.method_handlers_generic_handler(
             'hal_pb.ServoControl', rpc_method_handlers)
@@ -654,3 +670,30 @@ def SetTorqueEnable(request,
             timeout,
             metadata,
             _registered_method=True)
+
+    @staticmethod
+    def GetImuData(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(
+            request,
+            target,
+            '/hal_pb.ServoControl/GetImuData',
+            hal__pb__pb2.Empty.SerializeToString,
+            hal__pb__pb2.ImuData.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+            _registered_method=True)