114 lines
4.7 KiB
Python
114 lines
4.7 KiB
Python
import threading
|
||
|
||
from pyorbbecsdk import Pipeline, FrameSet
|
||
from pyorbbecsdk import Config
|
||
from pyorbbecsdk import OBSensorType, OBFormat
|
||
from pyorbbecsdk import OBError
|
||
import cv2
|
||
import numpy as np
|
||
import time
|
||
|
||
|
||
# 定义一个用于实现时间滤波的类
|
||
class DepthViewer:
|
||
def __init__(self, ):
|
||
# 创建配置对象和管道对象
|
||
self.config = Config()
|
||
self.pipeline = Pipeline()
|
||
self.alpha = 0.5 # 厂家给出的参数
|
||
self.previous_frame = None
|
||
self.ESC_KEY = 27
|
||
self.MIN_DEPTH = 20 # 最小深度值为20mm
|
||
self.MAX_DEPTH = 10000 # 最大深度值为10000mm
|
||
self.active = False # 控制捕获循环的标志
|
||
self.thread = None # 线程对象,用于并行执行捕获循环
|
||
self.color_profile = None # 颜色传感器的流配置
|
||
self.depth_frame = None # 当前深度帧
|
||
self.width = 640 # 图像宽度
|
||
self.fps = 30 # 帧率
|
||
self.depth_image = None # 深度图像
|
||
self.center_distance = 0 # 中心点的深度值
|
||
|
||
def start(self):
|
||
if not self.active:
|
||
self.active = True
|
||
# 选择深度传感器
|
||
profile_list = self.pipeline.get_stream_profile_list(OBSensorType.DEPTH_SENSOR)
|
||
# 获取指定分辨率和格式的深度流配置
|
||
try:
|
||
depth_profile = profile_list.get_video_stream_profile(self.width, 0, OBFormat.Y16, self.fps)
|
||
print("深度流配置获取成功:", depth_profile)
|
||
except OBError as e:
|
||
print("获取深度流配置失败: ", e)
|
||
# 如果出错,则获取默认的深度流配置
|
||
depth_profile = profile_list.get_default_video_stream_profile()
|
||
assert depth_profile is not None
|
||
# 将深度流配置加入到配置对象中
|
||
print('将深度流配置加入到配置对象中')
|
||
self.config.enable_stream(depth_profile)
|
||
# 创建并启动线程,目标函数是_capture_loop
|
||
self.thread = threading.Thread(target=self._capture_loop)
|
||
print('创建并启动线程,目标函数是_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:
|
||
print("管道启动")
|
||
self.pipeline.start(self.config)
|
||
print("管道启动成功")
|
||
while True:
|
||
try:
|
||
frames: FrameSet = self.pipeline.wait_for_frames(100)
|
||
if frames is None:
|
||
continue
|
||
self.depth_frame = frames.get_depth_frame()
|
||
if self.depth_frame is None:
|
||
continue
|
||
|
||
# 对当前帧进行数据处理
|
||
width = self.depth_frame.get_width()
|
||
height = self.depth_frame.get_height()
|
||
scale = self.depth_frame.get_depth_scale()
|
||
|
||
# 获取深度数据并进行处理
|
||
depth_data = np.frombuffer(self.depth_frame.get_data(), dtype=np.uint16)
|
||
depth_data = depth_data.reshape((height, width))
|
||
|
||
depth_data = depth_data.astype(np.float32) * scale
|
||
depth_data = np.where((depth_data > self.MIN_DEPTH) & (depth_data < self.MAX_DEPTH), depth_data, 0)
|
||
depth_data = depth_data.astype(np.uint16)
|
||
|
||
# 使用时间滤波
|
||
if self.previous_frame is None:
|
||
pass
|
||
else:
|
||
depth_data = cv2.addWeighted(depth_data, self.alpha, self.previous_frame, 1 - self.alpha, 0)
|
||
self.previous_frame = depth_data
|
||
|
||
# 获取图像中心点的深度值
|
||
center_y = int(height / 2)
|
||
center_x = int(width / 2)
|
||
self.center_distance = depth_data[center_y, center_x]
|
||
# 将深度数据转换为灰度图像,并使用伪彩色表示深度信息
|
||
depth_image = cv2.normalize(depth_data, None, 0, 255, cv2.NORM_MINMAX, dtype=cv2.CV_8U)
|
||
self.depth_image = cv2.applyColorMap(depth_image, cv2.COLORMAP_JET)
|
||
except Exception as e:
|
||
print(print('创建流失败{}', e))
|
||
|
||
finally:
|
||
self.pipeline.stop()
|
||
|
||
# 获取当前帧
|
||
def get_current_image(self):
|
||
# 返回当前深度图
|
||
return self.depth_image
|
||
|
||
def get_depth(self):
|
||
# 返回图像中心点的深度值
|
||
return self.center_distance
|