i/Orbbec_D2C
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修正D2C参数获取

Browse Source
This commit is contained in:
i
2026-04-21 21:31:46 +08:00
parent 17a4a819a6
commit 9eddab636f
1 changed files with 118 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+116 -21
View File
@@ -540,6 +540,100 @@ class CameraWorker(QThread):
# Batch D2C Worker Thread
# ─────────────────────────────────────────────────────────────────
class FetchParamsWorker(QThread):
log = pyqtSignal(str)
params_ready = pyqtSignal(dict, str) # params dict, temp file path
def __init__(self, color_cfg: dict, depth_cfg: dict):
super().__init__()
self._color_cfg = color_cfg
self._depth_cfg = depth_cfg
def run(self):
try:
from pyorbbecsdk import (
Pipeline, Config, OBSensorType, OBFrameAggregateOutputMode,
)
except ImportError as exc:
self.log.emit(f"导入 pyorbbecsdk 失败：{exc}")
return
def _find_profile(profile_list, cfg: dict):
for i in range(profile_list.get_count()):
p = profile_list.get_stream_profile_by_index(i)
vp = p.as_video_stream_profile()
if vp is None:
continue
if (vp.get_width() == cfg["width"]
and vp.get_height() == cfg["height"]
and vp.get_fps() == cfg["fps"]):
return vp
return profile_list.get_stream_profile_by_index(0).as_video_stream_profile()
pipeline = Pipeline()
config = Config()
try:
color_profile = _find_profile(
pipeline.get_stream_profile_list(OBSensorType.COLOR_SENSOR),
self._color_cfg)
depth_profile = _find_profile(
pipeline.get_stream_profile_list(OBSensorType.DEPTH_SENSOR),
self._depth_cfg)
except Exception as exc:
self.log.emit(f"获取流配置失败：{exc}")
return
config.enable_stream(color_profile)
config.enable_stream(depth_profile)
try:
config.set_frame_aggregate_output_mode(
OBFrameAggregateOutputMode.FULL_FRAME_REQUIRE)
except Exception:
pass
try:
pipeline.start(config)
except Exception as exc:
self.log.emit(f"Pipeline 启动失败：{exc}")
return
try:
frames = None
for _ in range(50): # 最多等待 5 秒
frames = pipeline.wait_for_frames(100)
if frames is not None:
cf = frames.get_color_frame()
df = frames.get_depth_frame()
if cf and df:
break
if frames is None or not frames.get_color_frame() or not frames.get_depth_frame():
self.log.emit("等待帧超时，无法获取参数")
return
cf = frames.get_color_frame()
df = frames.get_depth_frame()
params = CameraWorker._extract_params(cf, df)
sn = "unknown"
try:
sn = pipeline.get_device().get_device_info().get_serial_number()
except Exception:
pass
params["device"] = {"serial_number": sn}
tmp = Path(tempfile.gettempdir()) / f"orbbec_params_{sn}_{params['color']['width']}x{params['color']['height']}.json"
tmp.write_text(json.dumps(params, indent=2, ensure_ascii=False), encoding="utf-8")
self.params_ready.emit(params, str(tmp))
except Exception as exc:
self.log.emit(f"获取参数失败：{exc}")
finally:
try:
pipeline.stop()
except Exception:
pass
class BatchD2CWorker(QThread):
log = pyqtSignal(str)
done = pyqtSignal()
@@ -621,6 +715,7 @@ class MainWindow(QMainWindow):
self._worker: CameraWorker | None = None
self._batch_worker: BatchD2CWorker | None = None
self._fetch_worker: FetchParamsWorker | None = None
self._session_dir: Path | None = None
self._params_filename = "camera_params.json"
self._color_profiles_data: list[dict] = []
@@ -862,10 +957,10 @@ class MainWindow(QMainWindow):
combo_row.addWidget(self.d2c_depth_combo)
combo_row.addSpacing(12)
find_btn = QPushButton("自动查找参数文件")
find_btn.setFixedHeight(28)
find_btn.clicked.connect(self._find_params_by_combo)
combo_row.addWidget(find_btn)
fetch_btn = QPushButton("从设备获取参数")
fetch_btn.setFixedHeight(28)
fetch_btn.clicked.connect(self._fetch_params_from_device)
combo_row.addWidget(fetch_btn)
combo_row.addStretch()
pg.addLayout(combo_row)
@@ -1151,38 +1246,38 @@ class MainWindow(QMainWindow):
# Batch D2C Tab
# ─────────────────────────────────
def _find_params_by_combo(self):
"""根据当前选中的 RGB + 深度配置，在 session 目录或工作目录中查找参数文件。"""
def _fetch_params_from_device(self):
"""根据当前选中的 RGB + 深度配置，临时打开设备流并提取相机参数。"""
ci = self.d2c_color_combo.currentIndex()
di = self.d2c_depth_combo.currentIndex()
if ci < 0 or di < 0:
self.params_selected_lbl.setText("请先刷新设备并选择分辨率")
self.params_selected_lbl.setStyleSheet("color:#cc6666; font-size:12px;")
return
cc = self.d2c_color_combo.itemData(ci)
dc = self.d2c_depth_combo.itemData(di)
if not cc or not dc:
self.params_selected_lbl.setText("无配置数据，请先刷新设备")
self.params_selected_lbl.setStyleSheet("color:#cc6666; font-size:12px;")
return
pattern = (f"*_RGB{cc['width']}x{cc['height']}@{cc['fps']}"
f"_D{dc['width']}x{dc['height']}@{dc['fps']}.json")
self.params_selected_lbl.setText("正在从设备获取参数…")
self.params_selected_lbl.setStyleSheet("color:#888; font-size:12px;")
self._fetch_worker = FetchParamsWorker(cc, dc)
self._fetch_worker.log.connect(self._append_d2c_log)
self._fetch_worker.params_ready.connect(self._on_fetch_params_ready)
self._fetch_worker.start()
search_dirs = [d for d in [self._session_dir, Path(".")] if d]
found = []
for d in search_dirs:
found.extend(d.glob(pattern))
if found:
break
if found:
path = str(found[0])
def _on_fetch_params_ready(self, params: dict, path: str):
self.params_path_edit.setText(path)
self.params_selected_lbl.setText(f"✓ {path}")
self.params_selected_lbl.setText(f"✓ 已获取并缓存：{path}")
self.params_selected_lbl.setStyleSheet("color:#4aaa4a; font-size:12px;")
else:
self.params_selected_lbl.setText("未找到匹配的参数文件，请手动指定 JSON")
self.params_selected_lbl.setStyleSheet("color:#cc8844; font-size:12px;")
self._append_d2c_log(
f"[参数获取] RGB {params['color']['width']}x{params['color']['height']} "
f"/ Depth {params['depth']['width']}x{params['depth']['height']}"
)
self._fetch_worker = None
def _resolve_params_file(self) -> str | None:
explicit = self.params_path_edit.text().strip()