cmvr-es-cli/cmvr/camera_client.py

import grpc
import numpy as np
from PIL import Image
from typing import Tuple


from .enums import CMVRErrorCode
from .models import CameraState
from .models import Rs2Intrinsics
class CameraClient:
    """相机客户端"""

    def __init__(self, device_id: str, stub):
        self.device_id = device_id
        self.stub = stub
        self.generated = None  # 推迟导入

    def _import_generated(self):
        """推迟导入 generated"""
        if self.generated is None:
            try:
                # 从正确的路径导入生成的模块
                from generated.cmvr.api import common_pb2, camera_command_pb2
                self.generated = type('GeneratedModules', (), {
                    'common_pb2': common_pb2,
                    'camera_command_pb2': camera_command_pb2
                })
            except ImportError as e:
                print(f"导入生成的模块失败: {e}")
                print("请确保已生成 protobuf 代码")
                raise
        return self.generated

    def _create_command_header(self):
        generated = self._import_generated()
        """创建命令头"""
        header = generated.common_pb2.CommandHeader.Request()
        header.device_id = self.device_id
        header.timestamp.GetCurrentTime()
        return header

    def get_status(self) -> Tuple[CMVRErrorCode, CameraState]:
        """获取相机状态"""
        try:
            generated = self._import_generated()
            request = generated.camera_command_pb2.GetCameraStateCommand.Request()
            request.header.CopyFrom(self._create_command_header())

            response = self.stub.GetStatus(request)

            if not response.header.success:
                return CMVRErrorCode.CMVR_RPC_FAILED, CameraState()

            state = CameraState(
                is_initialized=response.state.is_initialized,
                is_opened=response.state.is_opened,
                is_streaming=response.state.is_streaming,
                is_recording=response.state.is_recording,
                width=response.state.width,
                height=response.state.height,
                fps=response.state.fps
            )

            return CMVRErrorCode.CMVR_SUCCESS, state

        except grpc.RpcError as e:
            print(f"获取相机状态失败: {e}")
            return CMVRErrorCode.CMVR_RPC_FAILED, CameraState()

    def start_camera(self) -> CMVRErrorCode:
        """启动相机"""
        try:
            generated = self._import_generated()
            request = generated.camera_command_pb2.StartCameraCommand.Request()
            request.header.CopyFrom(self._create_command_header())

            response = self.stub.StartCamera(request)

            return CMVRErrorCode.CMVR_SUCCESS if response.header.success else CMVRErrorCode.CMVR_RPC_FAILED

        except grpc.RpcError as e:
            print(f"启动相机失败: {e}")
            return CMVRErrorCode.CMVR_RPC_FAILED

    def stop_camera(self) -> CMVRErrorCode:
        """停止相机"""
        try:
            generated = self._import_generated()
            request = generated.camera_command_pb2.StopCameraCommand.Request()
            request.header.CopyFrom(self._create_command_header())

            response = self.stub.StopCamera(request)

            return CMVRErrorCode.CMVR_SUCCESS if response.header.success else CMVRErrorCode.CMVR_RPC_FAILED

        except grpc.RpcError as e:
            print(f"停止相机失败: {e}")
            return CMVRErrorCode.CMVR_RPC_FAILED

    def get_rgb_image(self) -> Tuple[CMVRErrorCode, np.ndarray, int, int]:
        """获取RGB图像"""
        try:
            generated = self._import_generated()
            request = generated.camera_command_pb2.GetRGBImageCommand.Request()
            request.header.CopyFrom(self._create_command_header())

            response = self.stub.GetRGBImage(request)

            if not response.header.success:
                return CMVRErrorCode.CMVR_RPC_FAILED, np.array([]), 0, 0

            width = response.color_frame.width
            height = response.color_frame.height
            size = len(response.color_frame.data)

            # 将字节数据转换为numpy数组
            img_data = np.frombuffer(response.color_frame.data, dtype=np.uint8)


            # 从 Protobuf 消息中读取 intrinsics 数据
            protobuf_intrinsics = response.intrinsics

            # 构造 Python 数据类对象（字段名称完全对应）
            python_intrinsics = Rs2Intrinsics(
                cx=protobuf_intrinsics.cx,
                cy=protobuf_intrinsics.cy,
                fx=protobuf_intrinsics.fx,
                fy=protobuf_intrinsics.fy,
                coeffs=list(protobuf_intrinsics.coeffs)  # Protobuf repeated 字段通常是列表类型，直接转换
            )

            # 示例：访问读取后的数据
            print(f"主点坐标: ({python_intrinsics.cx}, {python_intrinsics.cy})")
            print(f"焦距: ({python_intrinsics.fx}, {python_intrinsics.fy})")
            print(f"畸变系数: {python_intrinsics.coeffs}")
            # 重塑为图像格式 (height, width, channels)
            channels = 3  # 假设是RGB图像
            if size == width * height * channels:
                img_array = img_data.reshape((height, width, channels))
            else:
                # 可能是灰度图或其他格式
                img_array = img_data.reshape((height, width))

            return CMVRErrorCode.CMVR_SUCCESS, img_array, width, height

        except grpc.RpcError as e:
            print(f"获取图像失败: {e}")
            return CMVRErrorCode.CMVR_RPC_FAILED, np.array([]), 0, 0

    def start_record(self, video_path: str) -> CMVRErrorCode:
        """开始录像"""
        try:
            generated = self._import_generated()
            request = generated.camera_command_pb2.StartCameraRecordingCommand.Request()
            request.header.CopyFrom(self._create_command_header())
            request.video_path = video_path

            response = self.stub.StartRecording(request)

            return CMVRErrorCode.CMVR_SUCCESS if response.header.success else CMVRErrorCode.CMVR_RPC_FAILED

        except grpc.RpcError as e:
            print(f"开始录像失败: {e}")
            return CMVRErrorCode.CMVR_RPC_FAILED

    def stop_record(self) -> CMVRErrorCode:
        """停止录像"""
        try:
            generated = self._import_generated()
            request = generated.camera_command_pb2.StopCameraRecordingCommand.Request()
            request.header.CopyFrom(self._create_command_header())

            response = self.stub.StopRecording(request)

            return CMVRErrorCode.CMVR_SUCCESS if response.header.success else CMVRErrorCode.CMVR_RPC_FAILED

        except grpc.RpcError as e:
            print(f"停止录像失败: {e}")
            return CMVRErrorCode.CMVR_RPC_FAILED

    def create_rgb_stream_request(self):
        """创建流请求（无参数）"""
        generated = self._import_generated()
        request = generated.camera_command_pb2.GetRGBImageStreamCommand.Request()
        request.header.CopyFrom(self._create_command_header())
        # 不需要设置其他参数
        return request

    def get_rgb_stream(self, request_generator):
        """
        双向流获取传感器数据
        参数:
            request_generator: 请求生成器
        返回:
            传感器数据流迭代器
        """
        generated = self._import_generated()
        return self.stub.GetRGBImageStream(request_generator)

    def create_depth_stream_request(self):
        """创建流请求（无参数）"""
        generated = self._import_generated()
        request = generated.camera_command_pb2.GetDepthImageStreamCommand.Request()
        request.header.CopyFrom(self._create_command_header())
        # 不需要设置其他参数
        return request

    def get_depth_stream(self, request_generator):
        """
        双向流获取传感器数据
        参数:
            request_generator: 请求生成器
        返回:
            传感器数据流迭代器
        """
        generated = self._import_generated()
        return self.stub.GetDepthImageStream(request_generator)

    def create_rgbd_stream_request(self):
        """创建流请求（无参数）"""
        generated = self._import_generated()
        request = generated.camera_command_pb2.GetRGBDImageStreamCommand.Request()
        request.header.CopyFrom(self._create_command_header())
        # 不需要设置其他参数
        return request

    def get_rgbd_stream(self, request_generator):
        """
        双向流获取传感器数据
        参数:
            request_generator: 请求生成器
        返回:
            传感器数据流迭代器
        """
        generated = self._import_generated()
        return self.stub.GetRGBDImageStream(request_generator)