219 lines
8.0 KiB
Python
219 lines
8.0 KiB
Python
import grpc
|
|
import numpy as np
|
|
from PIL import Image
|
|
from typing import Tuple
|
|
|
|
|
|
from .enums import CMVRErrorCode
|
|
from .models import CameraState
|
|
|
|
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)
|
|
|
|
# 重塑为图像格式 (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) |