#!/usr/bin/env python3 """ Orbbec 数据采集 GUI - 主界面直接打开，刷新后选择分辨率 + 开关流 - 数据采集：实时预览 + 一键拍照 - Orbbec Viewer 深度图 D2C：批量软件 D2C 转换 """ import os import sys import json import time import threading import tempfile import shutil import contextlib import io import numpy as np import cv2 from pathlib import Path from datetime import datetime sys.path.insert(0, str(Path(__file__).parent)) def _load_env() -> dict: """解析脚本同级目录下的 .env 文件，返回 key=value 字典。""" env_path = Path(__file__).parent / ".env" result: dict = {} if not env_path.is_file(): return result for line in env_path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#"): continue if "=" in line: k, _, v = line.partition("=") result[k.strip()] = v.strip() return result _ENV = _load_env() _COPYRIGHT = _ENV.get("COPYRIGHT", "") from PyQt5.QtWidgets import ( QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QLineEdit, QCheckBox, QTabWidget, QComboBox, QFileDialog, QTextEdit, QFrame, QMessageBox, QSizePolicy, QGroupBox, ) from PyQt5.QtCore import Qt, QThread, pyqtSignal, QTimer from PyQt5.QtGui import QImage, QPixmap, QFont from d2c_batch import batch_convert DATA_DIR = Path("data") PARAMS_FILE = "camera_params.json" MIN_DEPTH_MM, MAX_DEPTH_MM = 200, 5000 _DISCONNECT_TIMEOUTS = 6 # 6 × 500 ms = 3 s 无帧 → 断线 # ───────────────────────────────────────────────────────────────── # Helpers # ───────────────────────────────────────────────────────────────── def imwrite_utf8(path: str, img: np.ndarray) -> bool: """cv2.imwrite Unicode 路径安全版（Windows 中文路径）。""" ext = Path(path).suffix.lower() or ".png" ok, buf = cv2.imencode(ext, img) if ok: Path(path).write_bytes(buf.tobytes()) return ok def save_ply_utf8(target_path: str, pc_frame, save_func) -> None: """先写到 ASCII 临时目录，再移动到 Unicode 目标路径。""" tmpdir = tempfile.mkdtemp() try: tmp = os.path.join(tmpdir, "pc.ply") save_func(tmp, pc_frame) shutil.move(tmp, target_path) finally: shutil.rmtree(tmpdir, ignore_errors=True) def ndarray_to_pixmap(img_bgr: np.ndarray, max_w: int, max_h: int) -> QPixmap: if img_bgr is None or img_bgr.size == 0: return QPixmap() h, w = img_bgr.shape[:2] scale = min(max_w / max(w, 1), max_h / max(h, 1), 1.0) if scale < 1.0: img_bgr = cv2.resize(img_bgr, (int(w * scale), int(h * scale)), interpolation=cv2.INTER_AREA) rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB) rh, rw = rgb.shape[:2] qimg = QImage(rgb.data.tobytes(), rw, rh, rw * 3, QImage.Format_RGB888) return QPixmap.fromImage(qimg) def depth_to_vis(depth_uint16: np.ndarray, scale: float) -> np.ndarray: depth_mm = depth_uint16.astype(np.float32) * scale norm = np.clip((depth_mm - MIN_DEPTH_MM) / (MAX_DEPTH_MM - MIN_DEPTH_MM), 0.0, 1.0) vis = cv2.applyColorMap((norm * 255).astype(np.uint8), cv2.COLORMAP_JET) vis[depth_uint16 == 0] = 0 return vis def process_ir_frame(ir_frame) -> np.ndarray | None: if ir_frame is None: return None try: from pyorbbecsdk import OBFormat vf = ir_frame.as_video_frame() data = np.asanyarray(vf.get_data()) w, h, fmt = vf.get_width(), vf.get_height(), vf.get_format() if fmt == OBFormat.Y8: arr = np.resize(data, (h, w)).astype(np.uint8) elif fmt == OBFormat.MJPG: arr = cv2.imdecode(data, cv2.IMREAD_GRAYSCALE) if arr is None: return None else: arr = np.frombuffer(data, dtype=np.uint16).reshape(h, w) arr = cv2.normalize(arr, None, 0, 255, cv2.NORM_MINMAX).astype(np.uint8) return cv2.cvtColor(arr, cv2.COLOR_GRAY2BGR) except Exception: return None def _find_profile(profile_list, cfg: dict): """在 profile_list 中找到 width/height/fps 匹配的 profile；失败时返回第一个。""" 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() def _find_profile_exact(profile_list, cfg: dict): """严格匹配 width/height/fps；找不到返回 None（不回退）。""" 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 None # ───────────────────────────────────────────────────────────────── # Camera Worker Thread # ───────────────────────────────────────────────────────────────── class CameraWorker(QThread): preview_ready = pyqtSignal(object, object) # color_bgr, depth_vis capture_done = pyqtSignal(str) # save_dir params_ready = pyqtSignal(dict) log_msg = pyqtSignal(str) disconnected = pyqtSignal() def __init__(self, color_cfg: dict, depth_cfg: dict, preset: str | None = None): super().__init__() self._color_cfg = color_cfg self._depth_cfg = depth_cfg self._preset = preset self._lock = threading.Lock() self._running = True self._d2c = False self._do_capture = False self._capture_dir = "" self._camera_params = None self._serial_number = "unknown" self._has_dual_ir = False self._ir_enabled = False self._last_emit_ts = 0.0 self._consec_timeouts = 0 self._disconnect_emitted = False self._sync_wait_count = 0 # 等待同步帧的计数器 def set_d2c(self, enabled: bool): with self._lock: self._d2c = enabled def trigger_capture(self, save_dir: str): with self._lock: self._capture_dir = save_dir self._do_capture = True def stop(self): self._running = False self.wait(6000) # ── Thread body ── def run(self): try: from pyorbbecsdk import ( Pipeline, Config, AlignFilter, PointCloudFilter, OBStreamType, OBSensorType, OBFrameType, OBFormat, OBFrameAggregateOutputMode, save_point_cloud_to_ply, ) from utils import frame_to_bgr_image except ImportError as exc: self.log_msg.emit(f"导入 pyorbbecsdk 失败：{exc}") return pipeline = Pipeline() config = Config() # 获取 SN + 探测 IR 类型 + 应用 preset try: device = pipeline.get_device() try: self._serial_number = device.get_device_info().get_serial_number() except Exception: pass sensor_list = device.get_sensor_list() for i in range(len(sensor_list)): st = sensor_list[i].get_type() if st in (OBSensorType.LEFT_IR_SENSOR, OBSensorType.RIGHT_IR_SENSOR): self._has_dual_ir = True if self._preset: try: device.load_preset(self._preset) self.log_msg.emit(f"已加载 Preset：{self._preset}") except Exception as exc: self.log_msg.emit(f"Preset 加载失败：{exc}") except Exception as exc: self.log_msg.emit(f"设备信息获取警告：{exc}") # 找到匹配的流 profile 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_msg.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 # 启用与深度流分辨率/帧率一致的 IR 流（找不到匹配则跳过，不影响主流） self._ir_enabled = False try: if self._has_dual_ir: left_ok = right_ok = False try: vp = _find_profile_exact( pipeline.get_stream_profile_list(OBSensorType.LEFT_IR_SENSOR), self._depth_cfg) if vp: config.enable_stream(vp) left_ok = True except Exception: pass try: vp = _find_profile_exact( pipeline.get_stream_profile_list(OBSensorType.RIGHT_IR_SENSOR), self._depth_cfg) if vp: config.enable_stream(vp) right_ok = True except Exception: pass self._ir_enabled = left_ok and right_ok if not self._ir_enabled: self.log_msg.emit( f"IR 流提示：当前深度配置 " f"{self._depth_cfg['width']}×{self._depth_cfg['height']}" f"@{self._depth_cfg['fps']} 无匹配 IR profile，IR 不启用") else: vp = _find_profile_exact( pipeline.get_stream_profile_list(OBSensorType.IR_SENSOR), self._depth_cfg) if vp: config.enable_stream(vp) self._ir_enabled = True else: self.log_msg.emit( f"IR 流提示：无匹配 IR profile，IR 不启用") except Exception as exc: self.log_msg.emit(f"IR 传感器警告：{exc}") try: pipeline.start(config) except Exception as exc: self.log_msg.emit(f"Pipeline 启动失败：{exc}") self.disconnected.emit() return try: pipeline.enable_frame_sync() except Exception: pass # 部分固件版本不支持，忽略 align_filter = AlignFilter(align_to_stream=OBStreamType.COLOR_STREAM) point_cloud_filter = PointCloudFilter() while self._running: try: frames = pipeline.wait_for_frames(500) if frames is None: self._consec_timeouts += 1 if (self._consec_timeouts >= _DISCONNECT_TIMEOUTS and not self._disconnect_emitted): self._disconnect_emitted = True self._running = False self.disconnected.emit() continue self._consec_timeouts = 0 with self._lock: d2c = self._d2c do_cap = self._do_capture cap_dir = self._capture_dir # 首帧提取相机参数 if self._camera_params is None: cf = frames.get_color_frame() df = frames.get_depth_frame() if cf and df: try: p = self._extract_params(cf, df) p['device'] = {'serial_number': self._serial_number} self._camera_params = p self.params_ready.emit(p) except Exception: pass # 执行拍照（等待 RGB-D 时间戳对齐后再拍） if do_cap and cap_dir: ts_diff = self._frame_ts_diff(frames) self._sync_wait_count += 1 threshold_ms = 1000 // max(self._depth_cfg.get("fps", 30), 1) force = self._sync_wait_count >= 10 if ts_diff <= threshold_ms or force: with self._lock: self._do_capture = False self._capture_dir = "" self._sync_wait_count = 0 if force and ts_diff > threshold_ms: self.log_msg.emit( f"警告：RGB-D 时间差 {ts_diff}ms 仍未对齐，强制拍摄") else: self.log_msg.emit(f"帧同步良好（RGB-D 差 {ts_diff}ms），开始拍摄") try: self._run_capture( frames, cap_dir, align_filter, point_cloud_filter, frame_to_bgr_image, save_point_cloud_to_ply, OBFrameType, OBFormat, ) except Exception as exc: self.log_msg.emit(f"拍照错误：{exc}") else: self.log_msg.emit( f"等待帧同步…RGB-D 差 {ts_diff}ms（{self._sync_wait_count}/10）") continue # 预览 view = frames if d2c: af = align_filter.process(frames) if af: view = af.as_frame_set() cf = view.get_color_frame() df = view.get_depth_frame() if not cf or not df: continue color_bgr = frame_to_bgr_image(cf) if color_bgr is None: continue depth_data = np.frombuffer(df.get_data(), dtype=np.uint16).reshape( df.get_height(), df.get_width()) depth_vis = depth_to_vis(depth_data, df.get_depth_scale()) now = time.monotonic() if now - self._last_emit_ts >= 0.033: self._last_emit_ts = now self.preview_ready.emit(color_bgr.copy(), depth_vis.copy()) except Exception as exc: if self._running: self.log_msg.emit(f"相机错误：{exc}") self._consec_timeouts += _DISCONNECT_TIMEOUTS if not self._disconnect_emitted: self._disconnect_emitted = True self._running = False self.disconnected.emit() time.sleep(0.1) try: pipeline.stop() except Exception: pass # ── Capture ── def _run_capture(self, frames, save_dir, align_filter, point_cloud_filter, frame_to_bgr_image, save_point_cloud_to_ply, OBFrameType, OBFormat): base = Path(save_dir) base.mkdir(parents=True, exist_ok=True) cf = frames.get_color_frame() df = frames.get_depth_frame() if cf: bgr = frame_to_bgr_image(cf) if bgr is not None: imwrite_utf8(str(base / "color.png"), bgr) if self._ir_enabled: if self._has_dual_ir: img = process_ir_frame(frames.get_frame(OBFrameType.LEFT_IR_FRAME)) if img is not None: imwrite_utf8(str(base / "ir_left.png"), img) img = process_ir_frame(frames.get_frame(OBFrameType.RIGHT_IR_FRAME)) if img is not None: imwrite_utf8(str(base / "ir_right.png"), img) else: img = process_ir_frame(frames.get_frame(OBFrameType.IR_FRAME)) if img is not None: imwrite_utf8(str(base / "ir.png"), img) if df: raw = np.frombuffer(df.get_data(), dtype=np.uint16).reshape( df.get_height(), df.get_width()) imwrite_utf8(str(base / "depth.png"), raw) aligned = align_filter.process(frames) if aligned: aset = aligned.as_frame_set() adf = aset.get_depth_frame() if adf: ad = np.frombuffer(adf.get_data(), dtype=np.uint16).reshape( adf.get_height(), adf.get_width()) imwrite_utf8(str(base / "depth_d2c.png"), ad) imwrite_utf8(str(base / "depth_d2c_vis.png"), depth_to_vis(ad, adf.get_depth_scale())) try: point_cloud_filter.set_create_point_format(OBFormat.POINT) pc = point_cloud_filter.process(aligned) if pc: save_ply_utf8(str(base / "point_cloud.ply"), pc, save_point_cloud_to_ply) except Exception as exc: self.log_msg.emit(f"点云警告：{exc}") try: point_cloud_filter.set_create_point_format(OBFormat.RGB_POINT) pc = point_cloud_filter.process(aligned) if pc: save_ply_utf8(str(base / "point_cloud_color.ply"), pc, save_point_cloud_to_ply) except Exception as exc: self.log_msg.emit(f"彩色点云警告：{exc}") self.capture_done.emit(save_dir) # ── Static helpers ── @staticmethod def _frame_ts_diff(frames) -> int: """返回彩色帧与深度帧时间戳差值（ms），获取失败返回 0。""" try: cf = frames.get_color_frame() df = frames.get_depth_frame() if cf and df: return abs(int(cf.get_timestamp()) - int(df.get_timestamp())) except Exception: pass return 0 @staticmethod def _extract_params(color_frame, depth_frame) -> dict: cvp = color_frame.get_stream_profile().as_video_stream_profile() dvp = depth_frame.get_stream_profile().as_video_stream_profile() def intr(i): return {"fx": float(i.fx), "fy": float(i.fy), "cx": float(i.cx), "cy": float(i.cy), "width": int(i.width), "height": int(i.height)} def dist(d): return {"k1": float(d.k1), "k2": float(d.k2), "k3": float(d.k3), "k4": float(d.k4), "k5": float(d.k5), "k6": float(d.k6), "p1": float(d.p1), "p2": float(d.p2)} ext = dvp.get_extrinsic_to(cvp) return { "color": { "width": color_frame.get_width(), "height": color_frame.get_height(), "fps": cvp.get_fps(), "format": str(cvp.get_format()).split(".")[-1], "intrinsic": intr(cvp.get_intrinsic()), "distortion": dist(cvp.get_distortion()), }, "depth": { "width": depth_frame.get_width(), "height": depth_frame.get_height(), "fps": dvp.get_fps(), "format": str(dvp.get_format()).split(".")[-1], "scale": float(depth_frame.get_depth_scale()), "intrinsic": intr(dvp.get_intrinsic()), "distortion": dist(dvp.get_distortion()), }, "extrinsic": { "rot": ext.rot.tolist(), "transform": ext.transform.tolist(), }, } # ───────────────────────────────────────────────────────────────── # Batch D2C Worker Thread # ───────────────────────────────────────────────────────────────── class BatchD2CWorker(QThread): log = pyqtSignal(str) done = pyqtSignal() def __init__(self, params: dict, depth_dir: str, splat: bool = True): super().__init__() self._params = params self._depth_dir = depth_dir self._splat = splat def run(self): class _Cap(io.StringIO): def __init__(self, sig): super().__init__() self._sig = sig def write(self, s): for line in s.splitlines(): if line.strip(): self._sig.emit(line) with contextlib.redirect_stdout(_Cap(self.log)): try: batch_convert(self._params, self._depth_dir, splat=self._splat) except Exception as exc: self.log.emit(f"Error: {exc}") self.done.emit() # ───────────────────────────────────────────────────────────────── # Toast Overlay（不阻塞鼠标，自动消失） # ───────────────────────────────────────────────────────────────── class _Toast(QLabel): def __init__(self, parent: QWidget): super().__init__(parent) self.setAlignment(Qt.AlignCenter) self.setAttribute(Qt.WA_TransparentForMouseEvents, True) self.setStyleSheet( "background: rgba(30,140,30,210);" "color: white;" "border-radius: 10px;" "font-size: 15px;" "padding: 10px 28px;" ) self._timer = QTimer(self) self._timer.setSingleShot(True) self._timer.timeout.connect(self.hide) self.hide() def show_msg(self, text: str, ms: int = 1000): self.setText(text) self.adjustSize() p = self.parent() self.move((p.width() - self.width()) // 2, p.height() - self.height() - 70) self.raise_() self.show() self._timer.start(ms) def reposition(self): if self.isVisible(): p = self.parent() self.move((p.width() - self.width()) // 2, p.height() - self.height() - 70) # ───────────────────────────────────────────────────────────────── # Main Window # ───────────────────────────────────────────────────────────────── class MainWindow(QMainWindow): def __init__(self): super().__init__() _title = "Orbbec 数据采集 & D2C" if _COPYRIGHT: _title += f" | © {_COPYRIGHT}" self.setWindowTitle(_title) self.resize(1300, 860) self._worker: CameraWorker | None = None self._batch_worker: BatchD2CWorker | None = None self._session_dir: Path | None = None self._params_filename = "camera_params.json" self._color_profiles_data: list[dict] = [] self._depth_profiles_data: list[dict] = [] self._build_ui() # toast 覆盖在 central widget 上 self._toast = _Toast(self.centralWidget()) # ───────────────────────────────── # UI Construction # ───────────────────────────────── def _build_ui(self): central = QWidget() self.setCentralWidget(central) root = QVBoxLayout(central) root.setContentsMargins(0, 0, 0, 0) root.setSpacing(0) # 版权信息栏（右上角） if _COPYRIGHT: cr_bar = QWidget() cr_bar.setFixedHeight(22) cr_bar.setStyleSheet("background: #222;") cr_layout = QHBoxLayout(cr_bar) cr_layout.setContentsMargins(8, 0, 10, 0) cr_layout.addStretch() cr_lbl = QLabel(f"© {_COPYRIGHT}") cr_lbl.setStyleSheet("color: #888; font-size: 11px;") cr_layout.addWidget(cr_lbl) root.addWidget(cr_bar) tabs = QTabWidget() tabs.addTab(self._build_capture_tab(), " 数据采集 ") tabs.addTab(self._build_d2c_tab(), " Orbbec Viewer 深度图 D2C ") root.addWidget(tabs) @staticmethod def _step_group(num: int, title: str) -> QGroupBox: grp = QGroupBox(f"第 {num} 步 · {title}") grp.setStyleSheet( "QGroupBox { font-weight:bold; font-size:12px; color:#8ec8ff;" " border:1px solid #444; border-radius:5px; margin-top:6px; padding-top:4px; }" "QGroupBox::title { subcontrol-origin:margin; left:10px; padding:0 4px; }" ) return grp def _build_capture_tab(self) -> QWidget: w = QWidget() layout = QVBoxLayout(w) layout.setContentsMargins(12, 8, 12, 8) layout.setSpacing(6) COMBO_CSS = "QComboBox { combobox-popup: 0; }" # ── 第 1 步：刷新设备 ── grp1 = self._step_group(1, "刷新设备") g1 = QHBoxLayout(grp1) g1.setContentsMargins(8, 4, 8, 6) refresh_btn = QPushButton("刷新设备") refresh_btn.setFixedHeight(28) refresh_btn.clicked.connect(self._refresh_devices) g1.addWidget(refresh_btn) self.device_status = QLabel("未连接设备，请点击「刷新设备」") self.device_status.setStyleSheet("color:#aaa; font-size:12px;") g1.addWidget(self.device_status) g1.addStretch() layout.addWidget(grp1) # ── 第 2 步：流配置 ── grp2 = self._step_group(2, "流配置") g2 = QHBoxLayout(grp2) g2.setContentsMargins(8, 4, 8, 6) g2.setSpacing(10) g2.addWidget(QLabel("RGB：")) self.color_combo = QComboBox() self.color_combo.setMinimumWidth(250) self.color_combo.setMaxVisibleItems(12) self.color_combo.setStyleSheet(COMBO_CSS) g2.addWidget(self.color_combo) g2.addSpacing(12) g2.addWidget(QLabel("深度：")) self.depth_combo = QComboBox() self.depth_combo.setMinimumWidth(250) self.depth_combo.setMaxVisibleItems(12) self.depth_combo.setStyleSheet(COMBO_CSS) g2.addWidget(self.depth_combo) g2.addSpacing(12) g2.addWidget(QLabel("Preset：")) self.preset_combo = QComboBox() self.preset_combo.setMinimumWidth(160) self.preset_combo.setMaxVisibleItems(12) self.preset_combo.setStyleSheet(COMBO_CSS) self.preset_combo.setEnabled(False) g2.addWidget(self.preset_combo) g2.addStretch() self.stream_btn = QPushButton("开启") self.stream_btn.setCheckable(True) self.stream_btn.setFixedSize(88, 30) self.stream_btn.setStyleSheet( "QPushButton{ border-radius:5px; border:none; font-weight:bold; }" "QPushButton:!checked{ background:#555; color:#ccc; }" "QPushButton:checked{ background:#2a8a2a; color:white; }" "QPushButton:disabled{ background:#444; color:#777; }" ) self.stream_btn.toggled.connect(self._on_stream_toggle) g2.addWidget(self.stream_btn) layout.addWidget(grp2) # ── 第 3 步：保存目录 ── grp3 = self._step_group(3, "保存目录") g3 = QHBoxLayout(grp3) g3.setContentsMargins(8, 4, 8, 6) g3.setSpacing(8) today = datetime.now() self.session_edit = QLineEdit(f"{today.year}年{today.month}月{today.day}日") self.session_edit.setMinimumWidth(160) g3.addWidget(self.session_edit) create_btn = QPushButton("创建 / 载入") create_btn.setFixedHeight(28) create_btn.clicked.connect(self._create_session) g3.addWidget(create_btn) self.session_status = QLabel("（未创建）") self.session_status.setStyleSheet("color:gray; font-size:12px;") g3.addWidget(self.session_status) g3.addStretch() self.d2c_check = QCheckBox("启用 D2C 对齐预览") self.d2c_check.setToolTip("勾选后预览使用 D2C 对齐；拍照时始终保存全部数据") self.d2c_check.stateChanged.connect( lambda s: self._worker and self._worker.set_d2c(s == Qt.Checked)) g3.addWidget(self.d2c_check) layout.addWidget(grp3) # ── 预览区 ── preview = QFrame() preview.setFrameShape(QFrame.StyledPanel) preview.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) pv = QHBoxLayout(preview) pv.setSpacing(6) pv.setContentsMargins(6, 6, 6, 6) for attr, title in (("color_lbl", "彩色图"), ("depth_lbl", "深度伪彩图")): box = QVBoxLayout() hdr = QLabel(title) hdr.setAlignment(Qt.AlignCenter) hdr.setStyleSheet("font-weight:bold; font-size:13px;") box.addWidget(hdr) lbl = QLabel("等待画面…") lbl.setAlignment(Qt.AlignCenter) lbl.setMinimumSize(540, 360) lbl.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) lbl.setStyleSheet("background:#111; border:1px solid #555; color:#666;") setattr(self, attr, lbl) box.addWidget(lbl) pv.addLayout(box) layout.addWidget(preview, stretch=1) # ── 第 4 步：拍照 ── step4_row = QHBoxLayout() step4_lbl = QLabel("第 4 步 · 拍照") step4_lbl.setStyleSheet( "color:#8ec8ff; font-weight:bold; font-size:12px;") step4_row.addWidget(step4_lbl) step4_row.addStretch() hint = QLabel("快捷键 S") hint.setStyleSheet("color:#666; font-size:11px;") step4_row.addWidget(hint) self.capture_btn = QPushButton("拍照") self.capture_btn.setFixedSize(160, 44) self.capture_btn.setEnabled(False) f = QFont(); f.setPointSize(13); f.setBold(True) self.capture_btn.setFont(f) self.capture_btn.setStyleSheet( "QPushButton{ background:#2a6ebb; color:white; border-radius:8px; border:none; }" "QPushButton:hover{ background:#3a80dd; }" "QPushButton:pressed{ background:#1a559e; }" "QPushButton:disabled{ background:#444; color:#777; }" ) self.capture_btn.clicked.connect(self._on_capture) step4_row.addWidget(self.capture_btn) layout.addLayout(step4_row) return w def _build_d2c_tab(self) -> QWidget: w = QWidget() layout = QVBoxLayout(w) layout.setContentsMargins(12, 8, 12, 8) layout.setSpacing(6) COMBO_CSS = "QComboBox { combobox-popup: 0; }" # ── 第 1 步：刷新设备 ── grp1 = self._step_group(1, "刷新设备") g1 = QHBoxLayout(grp1) g1.setContentsMargins(8, 4, 8, 6) d2c_refresh_btn = QPushButton("刷新设备") d2c_refresh_btn.setFixedHeight(28) d2c_refresh_btn.clicked.connect(self._refresh_devices) g1.addWidget(d2c_refresh_btn) self.d2c_device_status = QLabel("未连接设备，请点击「刷新设备」") self.d2c_device_status.setStyleSheet("color:#aaa; font-size:12px;") g1.addWidget(self.d2c_device_status) g1.addStretch() layout.addWidget(grp1) # ── 第 2 步：相机参数 ── grp2 = self._step_group(2, "相机参数（用于软件 D2C 计算）") pg = QVBoxLayout(grp2) pg.setContentsMargins(8, 4, 8, 6) pg.setSpacing(6) combo_row = QHBoxLayout() combo_row.addWidget(QLabel("RGB：")) self.d2c_color_combo = QComboBox() self.d2c_color_combo.setMinimumWidth(240) self.d2c_color_combo.setMaxVisibleItems(12) self.d2c_color_combo.setStyleSheet(COMBO_CSS) combo_row.addWidget(self.d2c_color_combo) combo_row.addSpacing(12) combo_row.addWidget(QLabel("深度：")) self.d2c_depth_combo = QComboBox() self.d2c_depth_combo.setMinimumWidth(240) self.d2c_depth_combo.setMaxVisibleItems(12) self.d2c_depth_combo.setStyleSheet(COMBO_CSS) 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) combo_row.addStretch() pg.addLayout(combo_row) sep = QLabel("── 或直接指定 JSON 文件 ──") sep.setAlignment(Qt.AlignCenter) sep.setStyleSheet("color:#666; font-size:11px;") pg.addWidget(sep) json_row = QHBoxLayout() self.params_path_edit = QLineEdit() self.params_path_edit.setPlaceholderText("留空则使用上方自动查找结果…") json_row.addWidget(self.params_path_edit) br2 = QPushButton("浏览…"); br2.setFixedWidth(70) br2.clicked.connect(lambda: self._browse_json(self.params_path_edit)) json_row.addWidget(br2) pg.addLayout(json_row) self.params_selected_lbl = QLabel("（尚未选定参数文件）") self.params_selected_lbl.setStyleSheet("color:#888; font-size:12px;") pg.addWidget(self.params_selected_lbl) layout.addWidget(grp2) # ── 第 3 步：深度图目录 ── grp3 = self._step_group(3, "深度图目录") g3 = QHBoxLayout(grp3) g3.setContentsMargins(8, 4, 8, 6) g3.setSpacing(8) self.d2c_dir_edit = QLineEdit() self.d2c_dir_edit.setPlaceholderText("选择包含 PNG 深度图的文件夹…") g3.addWidget(self.d2c_dir_edit) br1 = QPushButton("浏览…"); br1.setFixedWidth(70) br1.clicked.connect(lambda: self._browse_dir(self.d2c_dir_edit)) g3.addWidget(br1) layout.addWidget(grp3) # ── 第 4 步：开始转换 ── step4_row = QHBoxLayout() step4_lbl = QLabel("第 4 步 · 开始转换") step4_lbl.setStyleSheet("color:#8ec8ff; font-weight:bold; font-size:12px;") step4_row.addWidget(step4_lbl) self.splat_check = QCheckBox("Splat 3×3 填充（减少 D2C 空洞，默认开启）") self.splat_check.setChecked(True) step4_row.addWidget(self.splat_check) step4_row.addStretch() self.run_d2c_btn = QPushButton("▶ 开始批量 D2C 转换") self.run_d2c_btn.setFixedHeight(36) self.run_d2c_btn.setStyleSheet( "QPushButton{ background:#2a7a2a; color:white; border-radius:6px; border:none; }" "QPushButton:hover{ background:#3a9a3a; }" "QPushButton:disabled{ background:#444; color:#777; }" ) self.run_d2c_btn.clicked.connect(self._run_batch_d2c) step4_row.addWidget(self.run_d2c_btn) layout.addLayout(step4_row) # ── 日志 ── layout.addWidget(QLabel("输出日志：")) self.d2c_log = QTextEdit() self.d2c_log.setReadOnly(True) self.d2c_log.setStyleSheet( "font-family: Consolas,'Courier New',monospace; font-size:12px;") layout.addWidget(self.d2c_log, stretch=1) return w # ───────────────────────────────── # Device Refresh # ───────────────────────────────── def _refresh_devices(self): """连接相机，枚举并填充所有分辨率下拉框（Tab 1 + Tab 2）。""" self.device_status.setText("正在扫描设备…") QApplication.processEvents() self.color_combo.clear() self.depth_combo.clear() self.preset_combo.clear() self.d2c_color_combo.clear() self.d2c_depth_combo.clear() self._color_profiles_data.clear() self._depth_profiles_data.clear() try: from pyorbbecsdk import Pipeline, OBSensorType p = Pipeline() sn = "unknown" try: sn = p.get_device().get_device_info().get_serial_number() except Exception: pass color_list = p.get_stream_profile_list(OBSensorType.COLOR_SENSOR) for i in range(color_list.get_count()): vp = color_list.get_stream_profile_by_index(i).as_video_stream_profile() if vp is None: continue fmt = str(vp.get_format()).split(".")[-1] cfg = {"width": vp.get_width(), "height": vp.get_height(), "fps": vp.get_fps(), "format": fmt} label = f"{vp.get_width()}×{vp.get_height()} @ {vp.get_fps()} fps [{fmt}]" self._color_profiles_data.append(cfg) self.color_combo.addItem(label, cfg) self.d2c_color_combo.addItem(label, cfg) depth_list = p.get_stream_profile_list(OBSensorType.DEPTH_SENSOR) for i in range(depth_list.get_count()): vp = depth_list.get_stream_profile_by_index(i).as_video_stream_profile() if vp is None: continue fmt = str(vp.get_format()).split(".")[-1] cfg = {"width": vp.get_width(), "height": vp.get_height(), "fps": vp.get_fps(), "format": fmt} label = f"{vp.get_width()}×{vp.get_height()} @ {vp.get_fps()} fps [{fmt}]" self._depth_profiles_data.append(cfg) self.depth_combo.addItem(label, cfg) self.d2c_depth_combo.addItem(label, cfg) # 枚举 preset try: preset_list = p.get_device().get_available_preset_list() current_preset = p.get_device().get_current_preset_name() if preset_list and len(preset_list) > 0: self.preset_combo.addItem("── 不切换 ──", None) for name in preset_list: self.preset_combo.addItem(name, name) # 选中当前 preset idx = self.preset_combo.findData(current_preset) if idx >= 0: self.preset_combo.setCurrentIndex(idx) self.preset_combo.setEnabled(True) else: self.preset_combo.addItem("（设备不支持）", None) self.preset_combo.setEnabled(False) except Exception: self.preset_combo.addItem("（不支持）", None) self.preset_combo.setEnabled(False) nc, nd = len(self._color_profiles_data), len(self._depth_profiles_data) msg = f"已连接 SN: {sn} 彩色 {nc} 项 / 深度 {nd} 项" self.device_status.setText(msg) self.device_status.setStyleSheet("color:#4aaa4a; font-size:12px;") self.d2c_device_status.setText(msg) self.d2c_device_status.setStyleSheet("color:#4aaa4a; font-size:12px;") except Exception as exc: msg = f"未找到设备：{exc}" self.device_status.setText(msg) self.device_status.setStyleSheet("color:#cc6666; font-size:12px;") self.d2c_device_status.setText(msg) self.d2c_device_status.setStyleSheet("color:#cc6666; font-size:12px;") # ───────────────────────────────── # Stream Toggle # ───────────────────────────────── def _on_stream_toggle(self, checked: bool): if checked: if not self._color_profiles_data or not self._depth_profiles_data: self._set_stream_btn(False) QMessageBox.warning(self, "提示", "请先点击「刷新设备」以获取可用配置。") return ci = self.color_combo.currentIndex() di = self.depth_combo.currentIndex() if ci < 0 or di < 0: self._set_stream_btn(False) return self._start_stream(self._color_profiles_data[ci], self._depth_profiles_data[di]) else: self._stop_stream() def _start_stream(self, color_cfg: dict, depth_cfg: dict): self._stop_stream() # 确保旧 worker 已停止 preset = self.preset_combo.currentData() # None 表示不切换 self._worker = CameraWorker(color_cfg, depth_cfg, preset=preset) self._worker.preview_ready.connect(self._update_preview) self._worker.capture_done.connect(self._on_capture_done) self._worker.params_ready.connect(self._on_params_ready) self._worker.log_msg.connect(self._append_d2c_log) self._worker.disconnected.connect(self._on_camera_disconnected) self._worker.start() self.stream_btn.setText("关闭") self.color_combo.setEnabled(False) self.depth_combo.setEnabled(False) self.capture_btn.setEnabled(True) def _stop_stream(self): if self._worker: self._worker.stop() self._worker = None self.capture_btn.setEnabled(False) self.color_combo.setEnabled(True) self.depth_combo.setEnabled(True) self.stream_btn.setText("开启") # 清空预览 for lbl in (self.color_lbl, self.depth_lbl): lbl.setText("等待画面…") lbl.setPixmap(QPixmap()) def _set_stream_btn(self, state: bool): """不触发 toggled 信号地设置按钮状态。""" self.stream_btn.blockSignals(True) self.stream_btn.setChecked(state) self.stream_btn.setText("关闭" if state else "开启") self.stream_btn.blockSignals(False) def _on_camera_disconnected(self): """相机断开：静默关闭流，不弹窗。""" if self._worker: self._worker.deleteLater() self._worker = None self._set_stream_btn(False) self.capture_btn.setEnabled(False) self.color_combo.setEnabled(True) self.depth_combo.setEnabled(True) for lbl in (self.color_lbl, self.depth_lbl): lbl.setText("设备已断开，请刷新后重新开启") lbl.setPixmap(QPixmap()) self.device_status.setText("设备已断开连接，请重新插入后点击「刷新设备」") self.device_status.setStyleSheet("color:#cc8844; font-size:12px;") # ───────────────────────────────── # Preview # ───────────────────────────────── def _update_preview(self, color_bgr: np.ndarray, depth_vis: np.ndarray): w, h = self.color_lbl.width(), self.color_lbl.height() self.color_lbl.setPixmap(ndarray_to_pixmap(color_bgr, w, h)) w2, h2 = self.depth_lbl.width(), self.depth_lbl.height() self.depth_lbl.setPixmap(ndarray_to_pixmap(depth_vis, w2, h2)) # ───────────────────────────────── # Session / Capture # ───────────────────────────────── def _create_session(self): name = self.session_edit.text().strip() if not name: QMessageBox.warning(self, "提示", "请输入目录名称。") return session_dir = DATA_DIR / name session_dir.mkdir(parents=True, exist_ok=True) self._session_dir = session_dir if self._worker and self._worker._camera_params: self._save_params_to_session(self._worker._camera_params) self.session_status.setText(f"✓ {session_dir}") self.session_status.setStyleSheet("color:#2a9a2a; font-size:12px;") def _on_params_ready(self, params: dict): sn = params.get("device", {}).get("serial_number", "unknown") cw = params["color"]["width"]; ch = params["color"]["height"] cfps = params["color"]["fps"] dw = params["depth"]["width"]; dh = params["depth"]["height"] dfps = params["depth"]["fps"] self._params_filename = f"SN{sn}_RGB{cw}x{ch}@{cfps}_D{dw}x{dh}@{dfps}.json" if self._session_dir and self._session_dir.exists(): self._save_params_to_session(params) def _save_params_to_session(self, params: dict): try: (self._session_dir / self._params_filename).write_text( json.dumps(params, indent=2, ensure_ascii=False), encoding="utf-8") except Exception: pass def _on_capture(self): if self._session_dir is None: self._create_session() if self._session_dir is None: return ts = datetime.now().strftime("%Y%m%d%H%M%S") self.capture_btn.setEnabled(False) self.capture_btn.setText("拍摄中…") self._worker.trigger_capture(str(self._session_dir / ts)) def _on_capture_done(self, save_dir: str): self.capture_btn.setEnabled(True) self.capture_btn.setText("拍照") self._toast.show_msg("拍摄完成！", 1000) # ───────────────────────────────── # Batch D2C Tab # ───────────────────────────────── def _find_params_by_combo(self): """根据当前选中的 RGB + 深度配置，在 session 目录或工作目录中查找参数文件。""" ci = self.d2c_color_combo.currentIndex() di = self.d2c_depth_combo.currentIndex() if ci < 0 or di < 0: self.params_selected_lbl.setText("请先刷新设备并选择分辨率") 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("无配置数据，请先刷新设备") return pattern = (f"*_RGB{cc['width']}x{cc['height']}@{cc['fps']}" f"_D{dc['width']}x{dc['height']}@{dc['fps']}.json") 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]) self.params_path_edit.setText(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;") def _resolve_params_file(self) -> str | None: explicit = self.params_path_edit.text().strip() if explicit and Path(explicit).is_file(): return explicit # session 目录下 SN*.json if self._session_dir and self._session_dir.exists(): hits = sorted(self._session_dir.glob("SN*.json")) if hits: return str(hits[0]) fb = self._session_dir / PARAMS_FILE if fb.is_file(): return str(fb) hits = sorted(Path(".").glob("SN*.json")) if hits: return str(hits[0]) if Path(PARAMS_FILE).is_file(): return PARAMS_FILE return None def _browse_dir(self, edit: QLineEdit): d = QFileDialog.getExistingDirectory(self, "选择文件夹") if d: edit.setText(d) def _browse_json(self, edit: QLineEdit): f, _ = QFileDialog.getOpenFileName(self, "选择相机参数文件", "", "JSON (*.json)") if f: edit.setText(f) self.params_selected_lbl.setText(f"✓ {f}") self.params_selected_lbl.setStyleSheet("color:#4aaa4a; font-size:12px;") def _run_batch_d2c(self): depth_dir = self.d2c_dir_edit.text().strip() if not depth_dir: QMessageBox.warning(self, "错误", "请先选择深度图目录。") return params_file = self._resolve_params_file() if params_file is None: QMessageBox.warning( self, "错误", "找不到相机参数文件。\n" "请先开启摄像头流（自动生成）或手动选择 JSON 文件。") return with open(params_file, encoding="utf-8") as fh: params = json.load(fh) self.d2c_log.clear() self.d2c_log.append(f"[参数文件] {params_file}") self.d2c_log.append(f"[深度图目录] {depth_dir}\n") self.run_d2c_btn.setEnabled(False) self._batch_worker = BatchD2CWorker(params, depth_dir, splat=self.splat_check.isChecked()) self._batch_worker.log.connect(self.d2c_log.append) self._batch_worker.done.connect(self._on_batch_done) self._batch_worker.start() def _on_batch_done(self): self.run_d2c_btn.setEnabled(True) self.d2c_log.append("\n[ 转换完成 ]") def _append_d2c_log(self, msg: str): self.d2c_log.append(msg) # ───────────────────────────────── # Keyboard & Resize # ───────────────────────────────── def keyPressEvent(self, event): if event.key() == Qt.Key_S: focused = QApplication.focusWidget() if not isinstance(focused, (QLineEdit, QTextEdit)) and self.capture_btn.isEnabled(): self._on_capture() return super().keyPressEvent(event) def resizeEvent(self, event): super().resizeEvent(event) self._toast.reposition() # ───────────────────────────────── # Close # ───────────────────────────────── def closeEvent(self, event): self._stop_stream() super().closeEvent(event) # ───────────────────────────────────────────────────────────────── # Entry Point # ───────────────────────────────────────────────────────────────── def main(): app = QApplication(sys.argv) app.setStyle("Fusion") from PyQt5.QtGui import QPalette, QColor pal = QPalette() pal.setColor(QPalette.Window, QColor(45, 45, 45)) pal.setColor(QPalette.WindowText, QColor(220, 220, 220)) pal.setColor(QPalette.Base, QColor(30, 30, 30)) pal.setColor(QPalette.AlternateBase, QColor(50, 50, 50)) pal.setColor(QPalette.ToolTipBase, QColor(255, 255, 220)) pal.setColor(QPalette.ToolTipText, QColor(0, 0, 0)) pal.setColor(QPalette.Text, QColor(220, 220, 220)) pal.setColor(QPalette.Button, QColor(60, 60, 60)) pal.setColor(QPalette.ButtonText, QColor(220, 220, 220)) pal.setColor(QPalette.BrightText, QColor(255, 0, 0)) pal.setColor(QPalette.Link, QColor(42, 130, 218)) pal.setColor(QPalette.Highlight, QColor(42, 130, 218)) pal.setColor(QPalette.HighlightedText, QColor(0, 0, 0)) app.setPalette(pal) win = MainWindow() win.show() sys.exit(app.exec_()) if __name__ == "__main__": main()