67 lines
2.6 KiB
Python
67 lines
2.6 KiB
Python
import threading
|
||
from pyorbbecsdk import Pipeline, FrameSet
|
||
from pyorbbecsdk import Config
|
||
from pyorbbecsdk import OBSensorType, OBFormat
|
||
import cv2
|
||
from img_process.orbbec_camera.utils import frame_to_bgr_image
|
||
|
||
|
||
# 初始化摄像头并调用颜色传感器获取图像
|
||
class ColorViewer:
|
||
def __init__(self):
|
||
try:
|
||
self.pipeline = Pipeline() # 创建Pipeline对象,用于管理数据流
|
||
self.config = Config() # 创建配置对象
|
||
self.color_frame = None # 当前颜色帧
|
||
self.active = False # 控制捕获循环的标志
|
||
self.thread = None # 线程对象,用于并行执行捕获循环
|
||
self.ESC_KEY = 27 # ESC键的ASCII码,用于退出循环
|
||
self.profile_list = None # 颜色传感器的流配置列表
|
||
self.color_profile = None # 颜色传感器的流配置
|
||
self.width = 640 # 图像宽度
|
||
self.fps = 30 # 帧率
|
||
except Exception as e:
|
||
print('初始化失败:', e)
|
||
return
|
||
|
||
# 开始捕获颜色流
|
||
def start(self):
|
||
if not self.active:
|
||
self.active = True
|
||
# 传感器选择颜色传感器
|
||
self.profile_list = self.pipeline.get_stream_profile_list(OBSensorType.COLOR_SENSOR)
|
||
self.color_profile = self.profile_list.get_video_stream_profile(self.width, 0, OBFormat.RGB, self.fps)
|
||
self.config.enable_stream(self.color_profile) # 启用配置
|
||
# 创建并启动线程,目标函数是_capture_loop
|
||
self.thread = threading.Thread(target=self._capture_loop)
|
||
self.thread.start()
|
||
|
||
# 停止捕获
|
||
def stop(self):
|
||
self.active = False
|
||
if self.thread is not None:
|
||
self.thread.join() # 等待线程结束
|
||
|
||
# 捕获循环
|
||
def _capture_loop(self):
|
||
try:
|
||
self.pipeline.start(self.config)
|
||
while self.active:
|
||
try:
|
||
# 等待新帧,超时时间为100ms
|
||
frames: FrameSet = self.pipeline.wait_for_frames(100)
|
||
if frames is None:
|
||
continue
|
||
# 获取颜色帧并转换为RGB格式
|
||
self.color_frame = frame_to_bgr_image(frames.get_color_frame())
|
||
except Exception as e:
|
||
print('创建流失败{}', e)
|
||
continue
|
||
finally:
|
||
self.pipeline.stop() # 确保最终停止数据流
|
||
|
||
# 获取当前帧
|
||
def get_current_frame(self):
|
||
# 返回当前颜色的RGB格式帧
|
||
return self.color_frame
|