系统环境：windows 10
运行环境：uv python(3.11)
pyorbbecsdk version: 2.5.5
相机型号：336L
固件版本：1.4.6
项目描述：
scara机械臂，相机用于识别物料，输出物料的坐标
安装方式：
相机安装在机械臂末端的工具上，usb3.0接口，usb转type-c延长线长度8米，带独立供电。
线材安装于机械臂内部，和其他线材绑在一起。
相机相关的代码：
import sys
import traceback
from pyorbbecsdk import (AlignFilter, Pipeline, Config, Context, OBError,
OBFormat, OBStreamType, OBAlignMode, OBSensorType)
from src.utils import logger
class OrbbecCameraDevice:
def __init__(self, width=1280, height=800, fps=30):
self.ctx = Context()
# 设置设备插拔回调 (可选，但这里我们主要用轮询方式实现)
self.ctx.set_device_changed_callback(self._on_device_changed)
self.width = width
self.height = height
self.fps = fps
# self.pipeline = Pipeline() # 提前初始化 pipeline 以便获取 profile
# self.config = Config()
self.pipeline = None
self.config = None
self.align_filter = None
# 流配置 标志位
# self.is_stream_configured = False
# 状态标志
self.is_connected = False
# 获取设备并配置流
# self._setup_streams()
def _on_device_changed(self, removed_list, added_list):
"""设备插拔回调（底层SDK通知）"""
for dev in added_list:
logger.info(f"[Camera Hotplug] Device added: {dev.get_name()}")
for dev in removed_list:
logger.warning(f"[Camera Hotplug] Device removed: {dev.get_name()}")
self.is_connected = False # 设备拔出，标记配置失效
def _init_hardware_resources(self):
"""内部函数：创建Pipeline和Config对象，并配置流"""
try:
# 再次检查设备数量
if self.ctx.query_devices().get_count() == 0:
return False, "No device found"
# 【关键】只有确定有设备了，才创建 Pipeline
if self.pipeline is None:
self.pipeline = Pipeline()
if self.config is None:
self.config = Config()
if self.align_filter is None:
self.align_filter = AlignFilter(align_to_stream=OBStreamType.COLOR_STREAM)
# 1. 获取彩色传感器(COLOR_SENSOR)的配置列表
color_profiles = self.pipeline.get_stream_profile_list(OBSensorType.COLOR_SENSOR)
if not color_profiles:
return False, "No Color Sensor"
color_profile = None
# formats_to_try = [OBFormat.MJPG, OBFormat.RGB]
formats_to_try = [OBFormat.RGB, OBFormat.YUYV]
# 尝试队列：MJPG -> RGB -> 默认, 1280*720不支持RGB, 优先使用MJPG
for fmt in formats_to_try:
try:
color_profile = color_profiles.get_video_stream_profile(
self.width, self.height, fmt, self.fps
)
# color_profile = color_profiles.get_video_stream_profile(
# 640, 480, fmt, self.fps
# )
if color_profile:
logger.info(f"Matched Color profile: {fmt.name} {self.width}x{self.height} @{self.fps}fps")
break
except:
continue
if color_profile:
self.config.enable_stream(color_profile)
else:
logger.warning("No exact Color profile match! Using default COLOR_STREAM.")
self.config.enable_stream(OBStreamType.COLOR_STREAM)
# 2. 获取深度传感器(DEPTH_SENSOR)的配置列表
try:
depth_profiles = self.pipeline.get_stream_profile_list(OBSensorType.DEPTH_SENSOR)
# 深度图通常使用 Y16 格式
depth_profile = depth_profiles.get_video_stream_profile(
self.width, self.height, OBFormat.Y16, self.fps
)
# depth_profile = depth_profiles.get_video_stream_profile(
# 848, 480, OBFormat.Y16, self.fps
# )
self.config.enable_stream(depth_profile)
# logger.info(f"Depth stream enabled: {self.width}x{self.height} @{self.fps}fps")
except Exception as e:
logger.warning(f"Warning: Specific Depth profile not supported ({e}), using default.")
self.config.enable_stream(OBStreamType.DEPTH_STREAM)
# 3. 设置软件对齐
# gemini 336l 不支持720p下的硬件对齐(OBAlignMode.HW_MODE), 如果运行报错，可以注释掉软件对齐
# logger.info("Setting alignment mode to SW_MODE...")
# self.config.set_align_mode(OBAlignMode.SW_MODE)
# 在设置深度流之后，添加验证
logger.info("=== Stream Configuration Summary ===")
# logger.info(f"Color settings: {self.color_width}x{self.color_height} @{self.fps}fps")
# logger.info(f"Depth settings: {self.depth_width}x{self.depth_height} @{self.fps}fps")
# 尝试获取实际生效的配置
try:
# 检查config对象中实际配置的流
logger.info(
f"Config align mode: {self.config.get_align_mode() if hasattr(self.config, 'get_align_mode') else 'unknown'}")
except:
logger.warning("Could not verify align mode")
logger.info("=== Configuration Complete ===")
# 标记配置成功
# self.is_stream_configured = True
return True, "Ready"
except OBError as e:
return False, f"SDK Error: {e}"
except Exception as e:
return False, f"Setup Error: {e}"
# def _setup_streams(self):
# """修复后的配置函数：使用 OBSensorType"""
# try:
# # 先检查是否有设备，没设备直接跳过配置
# if self.ctx.query_devices().get_count() == 0:
# logger.warning("No Orbbec device found during setup_streams.")
# return False
#
# # 1. 获取彩色传感器(COLOR_SENSOR)的配置列表
# color_profiles = self.pipeline.get_stream_profile_list(OBSensorType.COLOR_SENSOR)
# color_profile = None
#
# # formats_to_try = [OBFormat.MJPG, OBFormat.RGB]
# formats_to_try = [OBFormat.RGB, OBFormat.YUYV]
# # 尝试队列：MJPG -> RGB -> 默认, 1280*720不支持RGB, 优先使用MJPG
# for fmt in formats_to_try:
# try:
# color_profile = color_profiles.get_video_stream_profile(
# self.width, self.height, fmt, self.fps
# )
# if color_profile:
# logger.info(f"Matched Color profile: {fmt.name} {self.width}x{self.height} @{self.fps}fps")
# break
# except:
# continue
#
# if color_profile:
# self.config.enable_stream(color_profile)
# else:
# logger.warning("No exact Color profile match! Using default COLOR_STREAM.")
# self.config.enable_stream(OBStreamType.COLOR_STREAM)
#
#
# # 2. 获取深度传感器(DEPTH_SENSOR)的配置列表
# try:
# depth_profiles = self.pipeline.get_stream_profile_list(OBSensorType.DEPTH_SENSOR)
# # 深度图通常使用 Y16 格式
# depth_profile = depth_profiles.get_video_stream_profile(
# self.width, self.height, OBFormat.Y16, self.fps
# )
# self.config.enable_stream(depth_profile)
# logger.info(f"Depth stream enabled: {self.width}x{self.height} @{self.fps}fps")
# except Exception as e:
# logger.warning(f"Warning: Specific Depth profile not supported ({e}), using default.")
# self.config.enable_stream(OBStreamType.DEPTH_STREAM)
#
# # 3. 设置软件对齐
# # gemini 336l 不支持720p下的硬件对齐(OBAlignMode.HW_MODE)
# logger.info("Setting alignment mode to SW_MODE...")
# # self.config.set_align_mode(OBAlignMode.SW_MODE)
#
# # 标记配置成功
# self.is_stream_configured = True
# return True
#
# except OBError as e:
# logger.error(f"SDK OBError during setup: {e} \n traceback: {traceback.format_exc()}")
# return False
# except Exception as e:
# logger.error(f"SDK Error during setup: {e} \n traceback: {traceback.format_exc()}")
# return False
def connect(self):
"""启动流水线, 支持热插拔"""
try:
# 检查是否有设备
device_list = self.ctx.query_devices()
if device_list.get_count() == 0:
# self.is_connected = False
self.disconnect()
return False, "No device connected"
# 如果之前是处于已连接但卡死的状态，必须先执行一次硬清理
# 不要尝试直接去 stop 一个旧的 pipeline，直接走毁灭流程重建最安全
if not self.is_connected and self.pipeline is not None:
self.disconnect()
# 如果还没有 Pipeline 或 Config，进行硬件资源初始化
if self.pipeline is None:
success, msg = self._init_hardware_resources()
if not success:
self.is_connected = False
return False, msg
# 注意：如果 pipeline 已经开启，先停止
# try:
# self.pipeline.stop()
# except:
# pass
# 启动 Pipeline
# self.pipeline.start(self.config)
# 3. 启动流
try:
self.pipeline.enable_frame_sync()
self.pipeline.start(self.config)
self.is_connected = True
if hasattr(self.config, 'get_align_mode'):
logger.info(f">>> Current align mode: {self.config.get_align_mode()}")
return True, "Success"
except OBError as e:
# 如果启动失败（例如被占用），清理资源以便下次重试
self.disconnect()
return False, f"Start OBError: {e}"
except Exception as e:
# self.is_stream_configured = False
self.disconnect()
logger.error(f"SDK Error during connect: {e}")
return False, str(e)
def _clean_resources(self):
"""【绝对隔离的错误吞噬区】安全的资源释放"""
self.is_connected = False
if self.pipeline is not None:
try:
# 尝试优雅停止。
# 如果此时 USB 已断开，这里必然抛出 bad magic 或超时的 OBError
self.pipeline.stop()
except OBError as e:
logger.warning(f"底层相机停止异常 (OBError已吞噬): {e}")
except Exception as e:
logger.warning(f"底层相机停止异常 (Exception已吞噬): {e}")
finally:
# 【终极保命符】
# 无论 stop() 成功与否，强制销毁 Python 侧的 pipeline 引用。
# 这样会触发底层的 C++ 析构函数，强行释放卡死的句柄。
self.pipeline = None
# 强制销毁其他相关配置对象
self.config = None
self.align_filter = None
def disconnect(self):
"""主动断开设备"""
try:
self._clean_resources()
except Exception as e:
# 万一 C++ 析构时发生了严重错误导致抛出异常，在这里被最后一道防线拦住
logger.error(f"Disconnect 发生严重崩溃并被成功拦截: {e}\n{traceback.format_exc()}")
# def disconnect(self):
# """关闭设备"""
# self.is_connected = False
# if self.pipeline:
# try:
# self.pipeline.stop()
# except OBError:
# pass
# self.pipeline = None
# self.config = None
# self.align_filter = None
def get_frames(self, timeout_ms=5000):
"""
获取一帧同步数据
返回: (status, color_frame, depth_frame)
"""
if not self.pipeline or not self.is_connected:
return False, None, None
try:
frames = self.pipeline.wait_for_frames(timeout_ms)
if not frames:
return False, None, None
if self.align_filter:
try:
aligned_frames = self.align_filter.process(frames)
if aligned_frames is None:
return False, None, None
aligned_frames = aligned_frames.as_frame_set()
except OBError as e:
logger.error(f"Alignment failed: {e}")
return False, None, None
else:
aligned_frames = frames # 如果没有滤波器，直接用原图
# 提取彩色和深度帧
color_frame = aligned_frames.get_color_frame()
depth_frame = aligned_frames.get_depth_frame()
# if color_frame and depth_frame:
# return True, color_frame, depth_frame
if color_frame is not None and depth_frame is not None:
# 安全地验证帧数据
try:
# 尝试获取数据以确保帧是有效的
color_data = color_frame.get_data()
depth_data = depth_frame.get_data()
if color_data is not None and depth_data is not None:
return True, color_frame, depth_frame
else:
logger.warning("Frame data is None")
return False, None, None
except Exception as e:
logger.error(f"Error validating frame data: {e}")
return False, None, None
except OBError as e:
logger.error(f"Wait for frames error: {e} \n {traceback.format_exc()}")
# 【核心修改】: 一旦底层超时或报错，立刻触发隔离区切断状态
self.disconnect()
except Exception as e:
logger.error(f"Unexpected error getting frames: {e}\n {traceback.format_exc()}")
# 【核心修改】
self.disconnect()
return False, None, None
def flush_frames(self, num_frames=10):
"""
排空历史缓存帧，获取最新画面，并给自动曝光留出时间
:param num_frames: 丢弃的帧数 (建议 3~5 帧)
"""
if not self.pipeline or not self.is_connected:
return
logger.info(f"清理相机底层缓存队列，丢弃前 {num_frames} 帧...")
for _ in range(num_frames):
try:
# 设定极短的超时时间，把积压在内存里的图迅速抽走抛弃
frames = self.pipeline.wait_for_frames(10)
if frames:
# 显式释放内存
color = frames.get_color_frame()
depth = frames.get_depth_frame()
if color: del color
if depth: del depth
del frames
except Exception as e:
logger.info(f"flush frames error: {e}\n {traceback.format_exc()}")
pass
def is_alive(self):
"""简单的链路健康检查"""
try:
# 尝试通过获取设备信息来判断链路是否正常
if self.is_connected and self.pipeline and self.ctx.query_devices().get_count() > 0:
return True
except:
pass
return False
def diagnose_alignment_issue(self):
"""
诊断对齐问题的工具函数 - 安全版本
"""
if not self.pipeline or not self.is_connected:
logger.error("Pipeline not started or not connected")
return
try:
logger.info("=== Starting Camera Diagnosis ===")
# 1. 获取设备信息
try:
device = self.pipeline.get_device()
device_info = device.get_device_info()
logger.info(f"Device Info: {device_info}")
except Exception as e:
logger.error(f"Failed to get device info: {e}")
# 2. 获取帧数据
logger.info("Attempting to get frames...")
frames = self.pipeline.wait_for_frames(3000)
if not frames:
logger.error("No frames received!")
return
logger.info("Frames object obtained successfully")
# 3. 安全地获取彩色帧信息
logger.info("--- Checking Color Frame ---")
color_frame = None
try:
color_frame = frames.get_color_frame()
if color_frame is None:
logger.error("Color frame is None!")
else:
# 安全地获取各种属性
try:
width = color_frame.get_width()
height = color_frame.get_height()
logger.info(f"Color frame size: {width}x{height}")
except Exception as e:
logger.error(f"Could not get color frame dimensions: {e}")
try:
fmt = color_frame.get_format()
logger.info(f"Color frame format: {fmt}")
except Exception as e:
logger.error(f"Could not get color format: {e}")
logger.info("Color frame obtained successfully")
except Exception as e:
logger.error(f"Error getting color frame: {e}")
import traceback
logger.error(f"Traceback: {traceback.format_exc()}")
# 4. 安全地获取深度帧信息
logger.info("--- Checking Depth Frame ---")
depth_frame = None
try:
depth_frame = frames.get_depth_frame()
# 不要直接打印depth_frame对象，而是检查它的存在
if depth_frame is None:
logger.error("Depth frame is None!")
else:
logger.info("Depth frame object exists")
# 安全地获取深度帧的属性
try:
width = depth_frame.get_width()
height = depth_frame.get_height()
logger.info(f"Depth frame size: {width}x{height}")
except Exception as e:
logger.error(f"Could not get depth frame dimensions: {e}")
try:
fmt = depth_frame.get_format()
logger.info(f"Depth frame format: {fmt}")
except Exception as e:
logger.error(f"Could not get depth format: {e}")
# 检查深度数据
try:
data = depth_frame.get_data()
if data is not None:
logger.info(f"Depth data size: {len(data)} bytes")
else:
logger.warning("Depth data is None")
except Exception as e:
logger.error(f"Could not get depth data: {e}")
logger.info("Depth frame obtained successfully")
except Exception as e:
logger.error(f"Error getting depth frame: {e}")
import traceback
logger.error(f"Traceback: {traceback.format_exc()}")
# 5. 对齐检查
logger.info("--- Alignment Check ---")
if color_frame is not None and depth_frame is not None:
try:
color_w = color_frame.get_width()
color_h = color_frame.get_height()
depth_w = depth_frame.get_width()
depth_h = depth_frame.get_height()
logger.info(f"Color resolution: {color_w}x{color_h}")
logger.info(f"Depth resolution: {depth_w}x{depth_h}")
if color_w == depth_w and color_h == depth_h:
logger.warning("Same resolution - check for sensor offset misalignment")
logger.info("Misalignment might be due to physical sensor offset")
else:
logger.info(f"Different resolutions - alignment mapping needed")
logger.info(f"Alignment should handle {depth_w}x{depth_h} -> {color_w}x{color_h}")
except Exception as e:
logger.error(f"Error comparing frames: {e}")
else:
logger.error("Cannot compare - one or both frames are None")
if color_frame is None:
logger.error("Color frame is missing")
if depth_frame is None:
logger.error("Depth frame is missing")
logger.info("=== Diagnosis Complete ===")
except Exception as e:
logger.error(f"Diagnosis error: {e}")
import traceback
logger.error(f"Full traceback: {traceback.format_exc()}")
if __name__ == "__main__":
camera = OrbbecCameraDevice()
status, msg = camera.connect()
if status:
logger.info(f"Connected: {msg}")
# 先清空缓存
camera.flush_frames(5)
# 运行诊断
# for i in range(5):
# camera.diagnose_alignment_issue()
# logger.info("\n\n")
else:
logger.error(f"Failed to connect: {msg}")
sys.exit(1)
相机代码的调用代码
import os
import cv2
import numpy as np
import json
import time
import queue
import threading
import traceback
from pyorbbecsdk import (OBFormat)
from src.vision.orbbec_camera import OrbbecCameraDevice
from src.depthSegmentPython.RGBDDepthSegmenterWrap import RGBDDetector
from src.utils.path_helper import get_camera_img_dir, get_logs_dir
from src.utils import logger
from src.utils import detector_logger
from src.consts import const
# 导入编译好的 C++ 模块 (cpp_algo.so)
try:
import cpp_algo
except ImportError:
logger.warning("Warning: cpp_algo module not found.")
class DetectAlgoService:
def __init__(self, product_no: str, save_dir: str = get_camera_img_dir()):
self.product_no = product_no
self.save_dir = save_dir
self.device = OrbbecCameraDevice()
self.max_retries = 10
self.depth_show = 1
# 初始化 C++ 算法
# self.algo = cpp_algo.MaterialAlgorithm()
# init_res = self.algo.initialize(self.product_no)
# if init_res.get("code") != 0:
# raise RuntimeError(f"Algorithm Init Failed: {init_res}")
self.detector = RGBDDetector()
detector_init_res = self.detector.init(product_no)
if detector_init_res.get("code") != 0:
raise RuntimeError(f"Algorithm Init Failed: {detector_init_res}")
# 异步存图队列与线程
self.save_jpg = 1
self.save_queue = queue.Queue(maxsize=100) # 限制队列长度防止内存溢出
self.stop_event = threading.Event()
self.save_thread = threading.Thread(target=self._save_worker, daemon=True)
self.save_thread.start()
self.alive_thread = threading.Thread(target=self._keep_alive_worker, daemon=True)
self.alive_thread.start()
# 初始化相机并预热
logger.info("Initializing camera hardware...")
try:
success, msg = self.device.connect()
if not success:
# 启动时没插相机，只警告，不抛异常，不退出
logger.warning(f"Camera init failed: {msg}. Hot-plug supported - waiting for device..")
else:
# 只有连接成功才预热
self._warm_up()
except Exception as e:
logger.error(f"Unexpected error during camera init: {e}")
def _warm_up(self):
"""抽取出预热逻辑：持续获取有效帧，直到帧数达标，模拟官方示例的 while True + continue 语义"""
logger.info("Warming up camera...")
warmed = 0
target = 20
max_attempts = 100
attempt = 0
while warmed < target and attempt < max_attempts:
try:
success, color_frame, depth_frame = self.device.get_frames(timeout_ms=500)
if success:
warmed += 1
# 【核心】：显式释放底层 C++ 帧缓冲！！
# 强迫 pyorbbecsdk 立即将 Buffer 归还给 SDK，防止缓存池枯竭
del color_frame
del depth_frame
else:
time.sleep(0.033)
except Exception as e:
logger.warning(f"Camera warm-up frame error: {e}")
time.sleep(0.033)
attempt += 1
logger.info(f"Camera warm-up complete (got {warmed} valid frames in {attempt} attempts).")
def _keep_alive_worker(self):
while not self.stop_event.is_set():
# logger.info(f"keep camera alive worker...")
try:
ret, color, depth = self.device.get_frames()
if ret:
del color
del depth
except Exception as e:
logger.info(f"keep_alive_worker error: {e} \n {traceback.format_exc()}")
pass
time.sleep(15)
def _save_worker(self):
"""后台存图线程函数"""
logger.info("Image save worker started.")
while not self.stop_event.is_set() or not self.save_queue.empty():
try:
# 设置超时以便能响应 stop_event
item = self.save_queue.get(timeout=1.0)
if item is None: # 约定 None 为退出信号
break
color_img, depth_img, timestamp = item
# 确定保存路径
date_str = time.strftime("%Y%m%d", time.localtime(timestamp / 1000))
path = os.path.join(self.save_dir, date_str)
if not os.path.exists(path):
os.makedirs(path, exist_ok=True)
if self.save_jpg:
# RGB -> BGR 并保存
bgr_img = cv2.cvtColor(color_img, cv2.COLOR_RGB2BGR)
cv2.imwrite(f"{path}/rgb_{timestamp}.jpg", bgr_img)
# 保存 16bit 深度图
cv2.imwrite(f"{path}/depth_{timestamp}.png", depth_img)
self.save_queue.task_done()
except queue.Empty:
continue
except Exception as e:
print(f"Error in save worker: {e}")
logger.error(f"Error in save worker: {e}")
def _save_to_local(self, color_arr, depth_arr):
"""保存图像到本地"""
timestamp = int(time.time() * 1000)
path = os.path.join(self.save_dir, time.strftime("%Y%m%d"))
os.makedirs(path, exist_ok=True)
# RGB -> BGR for OpenCV
bgr_img = cv2.cvtColor(color_arr, cv2.COLOR_RGB2BGR)
cv2.imwrite(f"{path}/rgb_{timestamp}.jpg", bgr_img)
# 深度图保存为16位PNG
cv2.imwrite(f"{path}/depth_{timestamp}.png", depth_arr)
return f"{path}/rgb_{timestamp}.jpg"
def execute_detection(self, ptype: int, detect: int=0, save_img: bool=True):
"""
对外公开的同步业务接口
:param ptype, 1-普通, 2-上料, 3-下料, 4-铝屑
:param detect, 是否执行检测，0-只拍照，不检测，1-执行检测
:return
{
"code": 0, # 正常返回0，异常返回其他值
"result": {
"ptype"：1， # 类型
"coords": [x,y,z,r], # 坐标参数
"ok": 1, # 检测结果，ok/1，ng/2
"exists": 1 # 根据ptype类型判断
1: exists == 1 表示有料，ok；
2: exists == 1 表示有料OK， 2表示空料ng；
3: exists == 2 表示空料；
4: exists == 1 有铁屑，表示ng；
}
"err_msg": "" # 异常日志，0返回空
}
"""
last_err = ""
for attempt in range(self.max_retries):
# 1. 检查并尝试重连
if not self.device.is_alive():
logger.info(f"Connection lost, retrying to connect (Attempt {attempt + 1})...")
# 确保彻底断开旧句柄
self.device.disconnect()
time.sleep(0.5)
success, msg = self.device.connect()
if not success:
last_err = msg
time.sleep(1.0)
continue
else:
# 【核心修复 2】：连接成功后，绝对不能立刻取图！
# 必须给底层光学传感器 1.5 秒的预热时间，否则立刻取图会触发假掉帧
logger.info("Camera connected, warming up optical sensor...")
time.sleep(1.5)
# 预热后，把这 1.5 秒内积攒的废图抽掉，保证曝光正常
if hasattr(self.device, 'flush_frames'):
self.device.flush_frames(num_frames=3)
# 2. 采集图像
success, color_frame, depth_frame = self.device.get_frames(timeout_ms=2000)
if not success:
last_err = "Failed to capture frames"
time.sleep(0.033) # 等待约一帧的时间 (30fps的周期)
# 只有当连续多次（例如超过3次）都拿不到图时，才真正去重启相机
if attempt >= 2:
logger.warning("Multiple consecutive frame drops, re-initializing pipeline...")
self.device.disconnect()
time.sleep(1.0) # 给 USB 驱动释放句柄的时间
# self.device.connect()
# # 采集失败通常意味着链路抖动，尝试重新初始化 pipeline
# self.device.connect()
continue
if not detect:
logger.info(f"Detect denied ...")
return {"code": 0, "result": {"ok": 1, "coords": [0, 0, 0, 0]}, "err_msg": ""}
try:
# 彩色图转换: RGB888 每个像素 3 字节 (uint8)
# color_frame.get_data() 是原始 buffer
# color_data = np.frombuffer(color_frame.get_data(), dtype=np.uint8)
# color_img = color_data.reshape((720, 1280, 3))
# color_img = np.frombuffer(color_frame.get_data(), dtype=np.uint8).reshape(
# (self.height, self.width, self.channel)).copy()
f_width = color_frame.get_width() # 动态获取当前帧的宽度
f_height = color_frame.get_height() # 动态获取当前帧的高度
color_format = color_frame.get_format()
raw_data = np.frombuffer(color_frame.get_data(), dtype=np.uint8)
if color_format == OBFormat.MJPG:
# 如果是 MJPG，使用 OpenCV 解码成 BGR
bgr_img = cv2.imdecode(raw_data, cv2.IMREAD_COLOR)
if bgr_img is None:
raise ValueError("MJPG decode failed")
# 转换为 RGB (因为你之前的逻辑是存图前转 BGR，或者算法需要 RGB)
color_img = cv2.cvtColor(bgr_img, cv2.COLOR_BGR2RGB)
elif color_format == OBFormat.RGB:
# 只有格式确实是 RGB 时才能直接 reshape
color_img = raw_data.reshape((f_height, f_width, 3)).copy()
else:
# 处理其他可能的格式（如 YUYV）
# 这里建议打印一下当前的格式，方便调试
print(f"Unsupported format for direct reshape: {color_format}")
logger.error(f"Unsupported format: {color_format}")
# 这种情况下通常需要专门的转换函数
return {"code": -1, "err_msg": f"Unsupported format {color_format}"}
# 深度图转换: Y16 每个像素 2 字节 (uint16)
# 使用 np.frombuffer 并指定 dtype=np.uint16
# depth_data = np.frombuffer(depth_frame.get_data(), dtype=np.uint16)
# depth_img = depth_data.reshape((f_height, f_width)).copy()
# depth_img = np.frombuffer(depth_frame.get_data(), dtype=np.uint16).reshape(
# (f_height, f_width)).copy()
d_height = depth_frame.get_height() # 获取深度帧实际高度
d_width = depth_frame.get_width() # 获取深度帧实际宽度
depth_img = np.frombuffer(depth_frame.get_data(), dtype=np.uint16).reshape(
(d_height, d_width)).copy()
# 本地持久化
# self._save_to_local(color_img, depth_img)
# 2. 将存图任务提交给后台队列 (非阻塞)
timestamp = int(time.time() * 1000)
try:
self.save_queue.put_nowait((color_img, depth_img, timestamp))
except queue.Full:
logger.warning("Warning: Save queue full, dropping image.")
except Exception as e:
logger.error(f"Error in image save queue: {e}")
# 5. 二进制处理 (传递给 C++ 算法)
# 直接获取原始内存 Buffer 的 bytes 形式
# rgb_binary = color_frame.get_data().tobytes()
# depth_binary = depth_frame.get_data().tobytes()
# 6. 调用 C++ 算法
# result = self.algo.detect(ptype, rgb_binary, depth_binary)
# return result
result = self.detector.detect(ptype, color_img, depth_img)
if self.depth_show:
timestamp = int(time.time() * 1000)
date_str = time.strftime("%Y%m%d", time.localtime(timestamp / 1000))
path = os.path.join(self.save_dir, date_str)
depth_color = self.detector.depth_pseudo_color(depth_img)
result_img = self.detector.draw_result_with_rotated_box(depth_color, result)
# cv2.imshow("result-line", result_img)
# if self.save_jpg:
if save_img:
cv2.imwrite(f"{path}/detect_result_horizontal_line_{timestamp}.jpg", result_img)
cv2.waitKey(0)
logger.info(f"detect result : {result}")
detector_logger.info(f"detect result : {result}")
# 显式释放底层 C++ 帧缓冲！！非常重要！！
# 让 pyorbbecsdk 立即将 Buffer 归还给 SDK，防止缓存池枯竭
del color_frame
del depth_frame
return result
# return {"code": 0, "result": {"ok": 1, "coords": [0, 0, 0, 0]}, "err_msg": ""}
except Exception as e:
last_err = str(e)
print(f"Error in image save queue: {traceback.format_exc()}")
logger.error(f"Processing error: {e} \n traceback: {traceback.format_exc()}")
# 如果发生异常，确保释放帧，防止内存泄漏
if 'color_frame' in locals(): del color_frame
if 'depth_frame' in locals(): del depth_frame
logger.info(f"Processing error: Max retries reached. {last_err}")
return {"code": -1, "err_msg": f"Max retries reached. Last error: {last_err}"}
def execute_detection_midian_depth(self, ptype:int, number:int=21, check_estop_func=None):
"""获取深度值的中位数返回值"""
# 在正式获取图像前，排空旧图
self.device.flush_frames(num_frames=5)
filter_number = 7
required_count = number - filter_number
results = []
for idx in range(number):
if idx < filter_number: # 运动到点位之后立即拍照，图像深度不稳定，前7次只触发拍照，不处理
self.execute_detection(ptype, detect=0)
time.sleep(0.01)
while len(results) < required_count:
if check_estop_func and check_estop_func():
return {"code":-99, "err_msg":f"检测异常: 系统急停"}
result = self.execute_detection(ptype, detect=1)
print(f"result is : {result}")
# 过滤有效深度的值
if ptype in (const.photo_type_loading, const.photo_type_find_head):
if result["code"] == 0 and result["result"]["coords"][2] <= const.depth_valid_filter:
results.append(result)
else:
results.append(result)
# 给底层 USB 留出喘息时间，避免阻塞 (30fps = 33ms 一张)
time.sleep(0.01)
print("#########################################")
if len(results) == 0:
return {"code":-99, "err_msg":f"检测异常: 中位数处理返回空数组"}
# 上下料，y, z取中位数，r取平均值
if ptype in (const.photo_type_loading, const.photo_type_unloading, const.photo_type_find_head):
n = len(results)
mid = n // 2
sorted_y_result = sorted(
results,
key=lambda x: x["result"]["coords"][1] # x代表每个item，取z值（索引2）
)
y_result = [item['result']['coords'][1] for item in sorted_y_result]
y_midian = y_result[mid] if n % 2 == 1 else (y_result[mid - 1] + y_result[mid]) / 2
sorted_z_result = sorted(
results,
key=lambda x: x["result"]["coords"][2] # x代表每个item，取z值（索引2）
)
z_result = [item['result']['coords'][2] for item in sorted_z_result]
z_midian = z_result[mid] if n % 2 == 1 else (z_result[mid - 1] + z_result[mid]) / 2
r_result = [item['result']['coords'][3] for item in sorted_z_result]
r_average = sum(r_result) / len(r_result)
r_midian = r_result[mid] if n % 2 == 1 else (r_result[mid - 1] + r_result[mid]) / 2
midian_result = sorted_z_result[mid]
midian_result["result"]["coords"][1] = y_midian
midian_result["result"]["coords"][2] = z_midian
midian_result["result"]["coords"][3] = r_midian
with open(f"{get_logs_dir()}/detect_algo.log", "a+") as f:
for item in sorted_z_result:
f.write(f"{item}\n")
f.write(f"midian_result: {midian_result}\n")
detector_logger.info(f"midian_result: {midian_result}")
return midian_result
return results[int(len(results)/2)]
def shutdown(self):
"""释放资源"""
logger.info("Shutting down service...")
self.stop_event.set()
self.save_queue.put(None) # 发送退出信号
self.save_thread.join(timeout=3.0)
self.alive_thread.join(timeout=3.0)
self.device.disconnect()
def update_product(self, new_product_no):
"""切换产品型号，重新加载算法配置"""
if self.product_no == new_product_no:
return
logger.info(f"Switching algorithm product to {new_product_no}...")
self.product_no = new_product_no
# 重新初始化 C++ 算法 (假设 C++ 有 reinit 接口，或者重新 new)
# self.algo.initialize(self.product_no)
self.detector.init(self.product_no)
logger.info("Algorithm updated.")
问题描述
开机启动一切正常，运行大概1小时之后，持续报错，重启程序也没有效果；
重新插拔数据线，会稳定几分钟，继续崩溃
报错日志
load extensions from D:\workspace\projects\robot-single-thread\.venv\Lib\site-packages/extensions
[05/29 13:33:55.921959][error][26920][UsbEnumeratorLibusb.cpp:403] Failed to query USB device serial number
[05/29 13:33:55.929283][error][26920][UsbEnumeratorLibusb.cpp:164] Failed to get string descriptor: error=Invalid parameter
[05/29 13:33:55.929358][error][26920][UsbEnumeratorLibusb.cpp:419] Failed to query USB device interface name
[05/29 13:33:55.936350][error][26920][UsbEnumeratorLibusb.cpp:164] Failed to get string descriptor: error=Invalid parameter
[05/29 13:33:55.936725][error][26920][UsbEnumeratorLibusb.cpp:419] Failed to query USB device interface name
[05/29 13:33:55.943743][error][26920][UsbEnumeratorLibusb.cpp:164] Failed to get string descriptor: error=Invalid parameter
[05/29 13:33:55.943775][error][26920][UsbEnumeratorLibusb.cpp:419] Failed to query USB device interface name
================================================
[05/29 13:42:57.337899][error][15668][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:42:57.507883][error][15668][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:42:57.619892][error][15668][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:42:57.787416][error][15668][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:42:58.268079][error][15668][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:42:58.510776][error][15668][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:42:58.721938][error][15668][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:42:58.896851][error][15668][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:42:59.018887][error][15668][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:42:59.199370][error][15668][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:42:59.309013][error][15668][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:43:03.822384][error][4104][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:43:03.936615][error][4104][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:43:06.075395][error][4104][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:43:06.187434][error][4104][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:43:06.301468][error][4104][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:43:06.411859][error][4104][WmfUvcDevicePort.cpp:435] setXu failed!
Error stopping pipeline: Request failed, statusCode: 5, msg: Device response with bad magic , magic=0x0, expectOpCode=0x4252
[05/29 13:43:06.523915][error][4104][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:43:06.645066][error][4104][WmfUvcDevicePort.cpp:435] setXu failed!
Error stopping pipeline: Request failed, statusCode: 5, msg: Device response with bad magic , magic=0x0, expectOpCode=0x4252
[05/29 13:43:06.758937][error][4104][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:43:06.879391][error][4104][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:43:08.532433][error][4104][WmfUvcDevicePort.cpp:435] setXu failed!
[05/29 13:43:08.640427][error][4104][WmfUvcDevicePort.cpp:435] setXu failed!
请问：这是什么原因引起的？有没有解决方案？
系统环境：windows 10
运行环境：uv python(3.11)
pyorbbecsdk version: 2.5.5
相机型号：336L
固件版本：1.4.6
项目描述：
scara机械臂，相机用于识别物料，输出物料的坐标
安装方式：
相机安装在机械臂末端的工具上，usb3.0接口，usb转type-c延长线长度8米，带独立供电。
线材安装于机械臂内部，和其他线材绑在一起。
相机相关的代码：
相机代码的调用代码
问题描述
开机启动一切正常，运行大概1小时之后，持续报错，重启程序也没有效果；
重新插拔数据线，会稳定几分钟，继续崩溃
报错日志
请问：这是什么原因引起的？有没有解决方案？