#!/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, QDialog, QAction, ) from PyQt5.QtCore import Qt, QThread, pyqtSignal, QTimer from PyQt5.QtGui import QImage, QPixmap, QFont from d2c_batch import batch_convert, software_d2c, software_d2c_inv PARAMS_FILE = "camera_params.json" MIN_DEPTH_MM, MAX_DEPTH_MM = 200, 5000 _DISCONNECT_TIMEOUTS = 6 # 6 × 500 ms = 3 s 无帧 → 断线 _UI_SCALE: float = 1.0 # initialized in main() before window creation def _fs(n: float) -> int: """Scale a font-size (px or pt) value.""" return max(8, int(n * _UI_SCALE)) def _sz(n: float) -> int: """Scale a UI dimension (px).""" return max(1, int(n * _UI_SCALE)) # ───────────────────────────────────────────────────────────────── # App Settings # ───────────────────────────────────────────────────────────────── class AppSettings: _APP_DIR = Path(os.environ.get("APPDATA", Path.home())) / "OrbbecD2C" _CONFIG_PATH = _APP_DIR / "app_settings.json" FONT_SCALE_AUTO = "auto" def __init__(self): self.font_scale: str = self.FONT_SCALE_AUTO self.splat_radius: int = 1 # 0=关闭, 1=3×3, 2=5×5, 3=7×7 self.save_base_dir: str = "" # 空 = 当前目录 self.capture_rgb: bool = True self.capture_depth: bool = True self.capture_d2c: bool = True self.capture_irl: bool = False self.capture_irr: bool = False self.capture_pc: bool = False self.capture_pc_color:bool = False self._load() def _load(self): if self._CONFIG_PATH.is_file(): try: d = json.loads(self._CONFIG_PATH.read_text(encoding="utf-8")) self.font_scale = d.get("font_scale", self.FONT_SCALE_AUTO) self.splat_radius = int(d.get("splat_radius", 1)) self.save_base_dir = d.get("save_base_dir", "") self.capture_rgb = bool(d.get("capture_rgb", True)) self.capture_depth = bool(d.get("capture_depth", True)) self.capture_d2c = bool(d.get("capture_d2c", True)) self.capture_irl = bool(d.get("capture_irl", False)) self.capture_irr = bool(d.get("capture_irr", False)) self.capture_pc = bool(d.get("capture_pc", False)) self.capture_pc_color= bool(d.get("capture_pc_color", False)) except Exception: pass def save(self): self._APP_DIR.mkdir(parents=True, exist_ok=True) self._CONFIG_PATH.write_text( json.dumps( { "font_scale": self.font_scale, "splat_radius": self.splat_radius, "save_base_dir": self.save_base_dir, "capture_rgb": self.capture_rgb, "capture_depth": self.capture_depth, "capture_d2c": self.capture_d2c, "capture_irl": self.capture_irl, "capture_irr": self.capture_irr, "capture_pc": self.capture_pc, "capture_pc_color": self.capture_pc_color, }, ensure_ascii=False, indent=2, ), encoding="utf-8", ) @staticmethod def compute_auto_scale() -> float: screen = QApplication.primaryScreen() if screen: dpr = screen.devicePixelRatio() return round(max(0.8, min(2.0, dpr)), 2) return 1.0 def get_scale_factor(self) -> float: if self.font_scale == self.FONT_SCALE_AUTO: return self.compute_auto_scale() try: return float(self.font_scale) except ValueError: return 1.0 # ───────────────────────────────────────────────────────────────── # Settings Dialog # ───────────────────────────────────────────────────────────────── class SettingsDialog(QDialog): def __init__(self, settings: "AppSettings", parent=None): super().__init__(parent) self.setWindowTitle("程序设置") self.setFixedSize(_sz(520), _sz(360)) self.setWindowFlags(self.windowFlags() & ~Qt.WindowContextHelpButtonHint) self._settings = settings self._build_ui() def _build_ui(self): layout = QVBoxLayout(self) layout.setContentsMargins(_sz(16), _sz(14), _sz(16), _sz(14)) layout.setSpacing(_sz(10)) row = QHBoxLayout() lbl = QLabel("字体缩放比例：") lbl.setMinimumWidth(_sz(110)) row.addWidget(lbl) self._scale_combo = QComboBox() self._scale_combo.setSizeAdjustPolicy(QComboBox.AdjustToContents) for key, label in [ ("auto", "自动（根据屏幕分辨率和缩放比例）"), ("0.8", "0.8× 小"), ("1.0", "1.0× 标准"), ("1.2", "1.2× 稍大"), ("1.5", "1.5× 大"), ("2.0", "2.0× 特大"), ]: self._scale_combo.addItem(label, key) idx = self._scale_combo.findData(self._settings.font_scale) if idx >= 0: self._scale_combo.setCurrentIndex(idx) row.addWidget(self._scale_combo) row.addStretch() layout.addLayout(row) auto_val = AppSettings.compute_auto_scale() info = QLabel(f"修改后需重启程序生效。当前屏幕自动比例：×{auto_val:.2f}") info.setStyleSheet(f"color:#888; font-size:{_fs(11)}px;") layout.addWidget(info) splat_row = QHBoxLayout() splat_lbl = QLabel("Splat 填充大小：") splat_lbl.setMinimumWidth(_sz(110)) splat_row.addWidget(splat_lbl) self._splat_combo = QComboBox() self._splat_combo.setSizeAdjustPolicy(QComboBox.AdjustToContents) for val, label in [(0, "关闭"), (1, "3×3"), (2, "5×5"), (3, "7×7")]: self._splat_combo.addItem(label, val) idx = self._splat_combo.findData(self._settings.splat_radius) if idx >= 0: self._splat_combo.setCurrentIndex(idx) splat_row.addWidget(self._splat_combo) splat_row.addStretch() layout.addLayout(splat_row) splat_info = QLabel("Splat 填充可减少深度图 D2C 后的空洞（修改立即对当前会话生效）") splat_info.setStyleSheet(f"color:#888; font-size:{_fs(11)}px;") layout.addWidget(splat_info) # ── 分隔 ── sep = QLabel("── 采集设置 ──") sep.setAlignment(Qt.AlignCenter) sep.setStyleSheet(f"color:#666; font-size:{_fs(11)}px; margin-top:{_sz(4)}px;") layout.addWidget(sep) # 保存根目录 dir_row = QHBoxLayout() dir_lbl = QLabel("根目录：") dir_lbl.setMinimumWidth(_sz(80)) dir_row.addWidget(dir_lbl) self._base_dir_edit = QLineEdit(self._settings.save_base_dir) self._base_dir_edit.setPlaceholderText("留空 = 当前工作目录") dir_row.addWidget(self._base_dir_edit) browse_btn = QPushButton("浏览…") browse_btn.setFixedWidth(_sz(60)) browse_btn.clicked.connect(self._browse_save_dir) dir_row.addWidget(browse_btn) layout.addLayout(dir_row) # 默认采集内容 cap_lbl = QLabel("默认采集内容：") layout.addWidget(cap_lbl) self._cap_checks: dict = {} cap_items = [ ("capture_rgb", "RGB"), ("capture_depth", "Depth"), ("capture_d2c", "Depth D2C"), ("capture_irl", "IR-L"), ("capture_irr", "IR-R"), ("capture_pc", "点云"), ("capture_pc_color", "颜色点云"), ] cap_row1 = QHBoxLayout() cap_row1.setSpacing(_sz(8)) for attr, label in cap_items[:5]: cb = QCheckBox(label) cb.setChecked(getattr(self._settings, attr, True)) cap_row1.addWidget(cb) self._cap_checks[attr] = cb cap_row1.addStretch() layout.addLayout(cap_row1) cap_row2 = QHBoxLayout() cap_row2.setSpacing(_sz(8)) for attr, label in cap_items[5:]: cb = QCheckBox(label) cb.setChecked(getattr(self._settings, attr, False)) cap_row2.addWidget(cb) self._cap_checks[attr] = cb cap_row2.addStretch() layout.addLayout(cap_row2) layout.addStretch() btn_row = QHBoxLayout() btn_row.addStretch() ok_btn = QPushButton("确定") ok_btn.setFixedWidth(_sz(80)) ok_btn.clicked.connect(self._save_and_close) cancel_btn = QPushButton("取消") cancel_btn.setFixedWidth(_sz(80)) cancel_btn.clicked.connect(self.reject) btn_row.addWidget(ok_btn) btn_row.addWidget(cancel_btn) layout.addLayout(btn_row) def _browse_save_dir(self): d = QFileDialog.getExistingDirectory( self, "选择数据保存根目录", self._base_dir_edit.text() or str(Path.cwd())) if d: self._base_dir_edit.setText(d) def _save_and_close(self): self._settings.font_scale = self._scale_combo.currentData() self._settings.splat_radius = self._splat_combo.currentData() self._settings.save_base_dir = self._base_dir_edit.text().strip() for attr, cb in self._cap_checks.items(): setattr(self._settings, attr, cb.isChecked()) self._settings.save() self.accept() # ───────────────────────────────────────────────────────────────── # 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 _splat_depth(depth: np.ndarray, radius: int) -> np.ndarray: """Dilate non-zero depth pixels outward to fill projection gaps, then restore originals.""" if radius <= 0: return depth kernel = np.ones((2 * radius + 1, 2 * radius + 1), dtype=np.uint8) dilated = cv2.dilate(depth, kernel) valid = depth > 0 dilated[valid] = depth[valid] return dilated 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 # 等待同步帧的计数器 self._splat_radius = 0 self._d2c_method = "sdk" self._capture_opts: dict = {} def set_d2c(self, enabled: bool): with self._lock: self._d2c = enabled def set_d2c_method(self, method: str): with self._lock: self._d2c_method = method def set_splat(self, radius: int): with self._lock: self._splat_radius = radius def trigger_capture(self, save_dir: str, capture_opts: dict | None = None): with self._lock: self._capture_dir = save_dir self._capture_opts = capture_opts or {} 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 d2c_method = self._d2c_method do_cap = self._do_capture cap_dir = self._capture_dir splat_r = self._splat_radius cap_opts = self._capture_opts # 首帧提取相机参数 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, cap_opts, 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 # 预览 cf = frames.get_color_frame() df = frames.get_depth_frame() if not cf or not df: continue color_bgr = frame_to_bgr_image(cf) if color_bgr is None: continue if d2c: if d2c_method == "forward" and self._camera_params is not None: raw = np.frombuffer(df.get_data(), dtype=np.uint16).reshape( df.get_height(), df.get_width()) depth_data = software_d2c( raw, self._camera_params, splat_radius=splat_r) depth_scale = df.get_depth_scale() else: af = align_filter.process(frames) if af: view = af.as_frame_set() df2 = view.get_depth_frame() if df2: df = df2 depth_data = np.frombuffer(df.get_data(), dtype=np.uint16).reshape( df.get_height(), df.get_width()) depth_scale = df.get_depth_scale() else: depth_data = np.frombuffer(df.get_data(), dtype=np.uint16).reshape( df.get_height(), df.get_width()) if splat_r > 0: depth_data = _splat_depth(depth_data, splat_r) depth_scale = df.get_depth_scale() depth_vis = depth_to_vis(depth_data, 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, capture_opts, 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 capture_opts.get("rgb", True) and 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: if capture_opts.get("irl", True): img = process_ir_frame(frames.get_frame(OBFrameType.LEFT_IR_FRAME)) if img is not None: imwrite_utf8(str(base / "ir_left.png"), img) if capture_opts.get("irr", True): 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: if capture_opts.get("irl", True): img = process_ir_frame(frames.get_frame(OBFrameType.IR_FRAME)) if img is not None: imwrite_utf8(str(base / "ir.png"), img) if capture_opts.get("depth", True) and df: raw = np.frombuffer(df.get_data(), dtype=np.uint16).reshape( df.get_height(), df.get_width()) imwrite_utf8(str(base / "depth.png"), raw) imwrite_utf8(str(base / "depth_vis.png"), depth_to_vis(raw, df.get_depth_scale())) d2c_method = capture_opts.get("d2c_method", "sdk") need_aligned = ((d2c_method == "sdk" and capture_opts.get("d2c", True)) or capture_opts.get("pc", False) or capture_opts.get("pc_color", False)) aset = None if need_aligned: aligned = align_filter.process(frames) if aligned: aset = aligned.as_frame_set() # D2C 深度图保存 if capture_opts.get("d2c", True): if d2c_method == "sdk" and aset: 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())) elif self._camera_params is not None and d2c_method == "forward": raw = np.frombuffer(df.get_data(), dtype=np.uint16).reshape( df.get_height(), df.get_width()) splat_r = capture_opts.get("splat_radius", 1) aligned_img = software_d2c(raw, self._camera_params, splat_radius=splat_r) imwrite_utf8(str(base / "depth_d2c.png"), aligned_img) imwrite_utf8(str(base / "depth_d2c_vis.png"), depth_to_vis(aligned_img, df.get_depth_scale())) # 点云保存（始终使用 SDK AlignFilter 结果） if aset: if capture_opts.get("pc", False): 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}") if capture_opts.get("pc_color", False): 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 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"SN{sn}_RGB{params['color']['width']}x{params['color']['height']}@{params['color']['fps']}" f"_D{params['depth']['width']}x{params['depth']['height']}@{params['depth']['fps']}.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() def __init__(self, params: dict, depth_dir: str, method: str = "inverse", splat_radius: int = 1): super().__init__() self._params = params self._depth_dir = depth_dir self._method = method self._splat_radius = splat_radius 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, method=self._method, splat_radius=self._splat_radius) 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( f"background: rgba(30,140,30,210);" f"color: white;" f"border-radius: {_sz(10)}px;" f"font-size: {_fs(15)}px;" f"padding: {_sz(10)}px {_sz(28)}px;" ) 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() - _sz(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() - _sz(70)) # ───────────────────────────────────────────────────────────────── # Main Window # ───────────────────────────────────────────────────────────────── class MainWindow(QMainWindow): def __init__(self): super().__init__() _title = "Orbbec 数据采集 & D2C" if _COPYRIGHT: _title += f" | © {_COPYRIGHT}" self.setWindowTitle(_title) self.resize(_sz(1300), _sz(860)) 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] = [] self._depth_profiles_data: list[dict] = [] self._app_settings = AppSettings() self._build_ui() # toast 覆盖在 central widget 上 self._toast = _Toast(self.centralWidget()) # ───────────────────────────────── # UI Construction # ───────────────────────────────── def _build_ui(self): # 菜单栏 settings_menu = self.menuBar().addMenu("设置") action_pref = QAction("程序设置…", self) action_pref.triggered.connect(self._show_settings) settings_menu.addAction(action_pref) if _COPYRIGHT: cr_widget = QWidget() cr_layout = QHBoxLayout(cr_widget) cr_layout.setContentsMargins(0, 0, 10, 0) cr_lbl = QLabel(f"© {_COPYRIGHT}") cr_lbl.setStyleSheet(f"color: #888; font-size:{_fs(11)}px;") cr_layout.addWidget(cr_lbl) cr_action = QWidgetAction(self) cr_action.setDefaultWidget(cr_widget) self.menuBar().addAction(cr_action) central = QWidget() self.setCentralWidget(central) root = QVBoxLayout(central) root.setContentsMargins(0, 0, 0, 0) root.setSpacing(0) tabs = QTabWidget() tabs.addTab(self._build_capture_tab(), " 数据采集 ") tabs.addTab(self._build_d2c_tab(), " Orbbec Viewer 深度图 D2C ") root.addWidget(tabs) def _browse_base_dir(self): d = QFileDialog.getExistingDirectory( self, "选择数据保存根目录", self.base_dir_edit.text() or str(Path.cwd())) if d: self.base_dir_edit.setText(d) self._app_settings.save_base_dir = d self._app_settings.save() def _show_settings(self): dlg = SettingsDialog(self._app_settings, self) if dlg.exec_() == QDialog.Accepted: # 同步 Splat 下拉框 idx = self.capture_splat_combo.findData(self._app_settings.splat_radius) if idx >= 0: self.capture_splat_combo.setCurrentIndex(idx) # 立即更新当前流的 splat 半径 if self._worker: self._worker.set_splat(self._app_settings.splat_radius) # 同步根目录显示 self.base_dir_edit.setText( self._app_settings.save_base_dir or str(Path.cwd())) # 同步采集内容复选框为新默认值 for key, cb in self._capture_checks.items(): cb.setChecked(getattr(self._app_settings, f"capture_{key}", cb.isChecked())) def _on_capture_d2c_method_changed(self): method = self.capture_d2c_method_combo.currentData() if self._worker: self._worker.set_d2c_method(method) self.capture_splat_combo.setEnabled(method == "forward") def _on_capture_splat_combo_changed(self): radius = self.capture_splat_combo.currentData() if self._worker: self._worker.set_splat(radius) def _on_d2c_method_changed(self): method = self.d2c_method_combo.currentData() self.d2c_splat_combo.setEnabled(method == "forward") @staticmethod def _step_group(num: int, title: str) -> QGroupBox: grp = QGroupBox(f"第 {num} 步 · {title}") grp.setStyleSheet( f"QGroupBox {{ font-weight:bold; font-size:{_fs(12)}px; color:#8ec8ff;" f" border:1px solid #444; border-radius:5px; margin-top:6px; padding-top:4px; }}" f"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(_sz(28)) refresh_btn.clicked.connect(self._refresh_devices) g1.addWidget(refresh_btn) self.device_status = QLabel("未连接设备，请点击「刷新设备」") self.device_status.setStyleSheet(f"color:#aaa; font-size:{_fs(12)}px;") 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(_sz(340)) self.color_combo.setSizeAdjustPolicy(QComboBox.AdjustToContents) self.color_combo.setMaxVisibleItems(15) 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(_sz(340)) self.depth_combo.setSizeAdjustPolicy(QComboBox.AdjustToContents) self.depth_combo.setMaxVisibleItems(15) 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(_sz(220)) self.preset_combo.setSizeAdjustPolicy(QComboBox.AdjustToContents) 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(_sz(88), _sz(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, "保存目录 && 采集内容") vg3 = QVBoxLayout(grp3) vg3.setContentsMargins(8, 4, 8, 8) vg3.setSpacing(5) # 行 1：根目录选择 row_base = QHBoxLayout() row_base.setSpacing(6) lbl_base = QLabel("根目录：") lbl_base.setFixedWidth(_sz(80)) row_base.addWidget(lbl_base) self.base_dir_edit = QLineEdit( self._app_settings.save_base_dir or str(Path.cwd())) row_base.addWidget(self.base_dir_edit) browse_base_btn = QPushButton("浏览…") browse_base_btn.setFixedWidth(_sz(58)) browse_base_btn.setFixedHeight(_sz(26)) browse_base_btn.clicked.connect(self._browse_base_dir) row_base.addWidget(browse_base_btn) vg3.addLayout(row_base) # 行 2：子目录 + 状态 + D2C + Splat row_ses = QHBoxLayout() row_ses.setSpacing(8) lbl_ses = QLabel("子目录：") lbl_ses.setFixedWidth(_sz(80)) row_ses.addWidget(lbl_ses) today = datetime.now() self.session_edit = QLineEdit(f"{today.year}年{today.month}月{today.day}日") self.session_edit.setMinimumWidth(_sz(140)) row_ses.addWidget(self.session_edit) create_btn = QPushButton("创建 / 载入") create_btn.setFixedHeight(_sz(26)) create_btn.clicked.connect(self._create_session) row_ses.addWidget(create_btn) self.session_status = QLabel("（未创建）") self.session_status.setStyleSheet(f"color:gray; font-size:{_fs(12)}px;") row_ses.addWidget(self.session_status) row_ses.addStretch() self.d2c_check = QCheckBox("D2C 预览") self.d2c_check.setToolTip("勾选后预览使用 D2C 对齐；拍照时始终保存全部选中的数据") self.d2c_check.setChecked(True) self.d2c_check.stateChanged.connect( lambda s: self._worker and self._worker.set_d2c(s == Qt.Checked)) row_ses.addWidget(self.d2c_check) vg3.addLayout(row_ses) # 行 3：采集内容复选框 row_cap = QHBoxLayout() row_cap.setSpacing(_sz(8)) lbl_cap = QLabel("采集内容：") lbl_cap.setFixedWidth(_sz(80)) row_cap.addWidget(lbl_cap) self._capture_checks: dict = {} for key, label, default in [ ("rgb", "RGB", self._app_settings.capture_rgb), ("depth", "Depth", self._app_settings.capture_depth), ("d2c", "Depth D2C", self._app_settings.capture_d2c), ("irl", "IR-L", self._app_settings.capture_irl), ("irr", "IR-R", self._app_settings.capture_irr), ("pc", "点云", self._app_settings.capture_pc), ("pc_color", "颜色点云", self._app_settings.capture_pc_color), ]: cb = QCheckBox(label) cb.setChecked(default) row_cap.addWidget(cb) self._capture_checks[key] = cb row_cap.addStretch() vg3.addLayout(row_cap) # 行 4：D2C 转换方式 row_d2c_method = QHBoxLayout() row_d2c_method.setSpacing(8) lbl_d2c_method = QLabel("D2C 方式：") lbl_d2c_method.setFixedWidth(_sz(80)) row_d2c_method.addWidget(lbl_d2c_method) self.capture_d2c_method_combo = QComboBox() self.capture_d2c_method_combo.setMinimumWidth(_sz(220)) self.capture_d2c_method_combo.setSizeAdjustPolicy(QComboBox.AdjustToContents) self.capture_d2c_method_combo.addItem("逆向投影（SDK 同等）", "sdk") self.capture_d2c_method_combo.addItem("正向投影 + Splat 填充", "forward") self.capture_d2c_method_combo.currentIndexChanged.connect( self._on_capture_d2c_method_changed) row_d2c_method.addWidget(self.capture_d2c_method_combo) lbl_splat = QLabel("Splat：") self.capture_splat_combo = QComboBox() self.capture_splat_combo.setMinimumWidth(_sz(80)) self.capture_splat_combo.setSizeAdjustPolicy(QComboBox.AdjustToContents) for val, label in [(0, "关闭"), (1, "3×3"), (2, "5×5"), (3, "7×7")]: self.capture_splat_combo.addItem(label, val) idx = self.capture_splat_combo.findData(self._app_settings.splat_radius) if idx >= 0: self.capture_splat_combo.setCurrentIndex(idx) self.capture_splat_combo.currentIndexChanged.connect( self._on_capture_splat_combo_changed) row_d2c_method.addWidget(lbl_splat) row_d2c_method.addWidget(self.capture_splat_combo) self.capture_splat_combo.setEnabled(False) row_d2c_method.addStretch() vg3.addLayout(row_d2c_method) 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(f"font-weight:bold; font-size:{_fs(13)}px;") box.addWidget(hdr) lbl = QLabel("等待画面…") lbl.setAlignment(Qt.AlignCenter) lbl.setMinimumSize(_sz(540), _sz(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( f"color:#8ec8ff; font-weight:bold; font-size:{_fs(12)}px;") step4_row.addWidget(step4_lbl) step4_row.addStretch() hint = QLabel("快捷键 S") hint.setStyleSheet(f"color:#666; font-size:{_fs(11)}px;") step4_row.addWidget(hint) self.capture_btn = QPushButton("拍照") self.capture_btn.setFixedSize(_sz(160), _sz(44)) self.capture_btn.setEnabled(False) f = QFont(); f.setPointSize(_fs(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(_sz(28)) d2c_refresh_btn.clicked.connect(self._refresh_devices) g1.addWidget(d2c_refresh_btn) self.d2c_device_status = QLabel("未连接设备，请点击「刷新设备」") self.d2c_device_status.setStyleSheet(f"color:#aaa; font-size:{_fs(12)}px;") 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(_sz(340)) self.d2c_color_combo.setSizeAdjustPolicy(QComboBox.AdjustToContents) self.d2c_color_combo.setMaxVisibleItems(15) 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(_sz(340)) self.d2c_depth_combo.setSizeAdjustPolicy(QComboBox.AdjustToContents) self.d2c_depth_combo.setMaxVisibleItems(15) self.d2c_depth_combo.setStyleSheet(COMBO_CSS) combo_row.addWidget(self.d2c_depth_combo) combo_row.addSpacing(12) fetch_btn = QPushButton("从设备获取参数") fetch_btn.setFixedHeight(_sz(28)) fetch_btn.clicked.connect(self._fetch_params_from_device) combo_row.addWidget(fetch_btn) combo_row.addStretch() pg.addLayout(combo_row) sep = QLabel("── 或直接指定 JSON 文件 ──") sep.setAlignment(Qt.AlignCenter) sep.setStyleSheet(f"color:#666; font-size:{_fs(11)}px;") 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(_sz(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(f"color:#888; font-size:{_fs(12)}px;") 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(_sz(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(f"color:#8ec8ff; font-weight:bold; font-size:{_fs(12)}px;") step4_row.addWidget(step4_lbl) self.d2c_method_combo = QComboBox() self.d2c_method_combo.setMinimumWidth(_sz(220)) self.d2c_method_combo.setSizeAdjustPolicy(QComboBox.AdjustToContents) self.d2c_method_combo.addItem("正向投影 + Splat 填充", "forward") self.d2c_method_combo.addItem("逆向投影（SDK 同等）", "sdk") self.d2c_method_combo.currentIndexChanged.connect( self._on_d2c_method_changed) step4_row.addWidget(self.d2c_method_combo) self.d2c_splat_combo = QComboBox() self.d2c_splat_combo.setMinimumWidth(_sz(80)) self.d2c_splat_combo.setSizeAdjustPolicy(QComboBox.AdjustToContents) for val, label in [(0, "关闭"), (1, "3×3"), (2, "5×5"), (3, "7×7")]: self.d2c_splat_combo.addItem(label, val) idx = self.d2c_splat_combo.findData(self._app_settings.splat_radius) if idx >= 0: self.d2c_splat_combo.setCurrentIndex(idx) step4_row.addWidget(QLabel("Splat：")) step4_row.addWidget(self.d2c_splat_combo) step4_row.addStretch() self.run_d2c_btn = QPushButton("▶ 开始批量 D2C 转换") self.run_d2c_btn.setFixedHeight(_sz(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( f"font-family: Consolas,'Courier New',monospace; font-size:{_fs(12)}px;") 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 i in range(len(preset_list)): name = preset_list[i] 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 as exc: self.preset_combo.addItem(f"（不支持: {exc}）", 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(f"color:#4aaa4a; font-size:{_fs(12)}px;") self.d2c_device_status.setText(msg) self.d2c_device_status.setStyleSheet(f"color:#4aaa4a; font-size:{_fs(12)}px;") except Exception as exc: msg = f"未找到设备：{exc}" self.device_status.setText(msg) self.device_status.setStyleSheet(f"color:#cc6666; font-size:{_fs(12)}px;") self.d2c_device_status.setText(msg) self.d2c_device_status.setStyleSheet(f"color:#cc6666; font-size:{_fs(12)}px;") # ───────────────────────────────── # 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.set_d2c(self.d2c_check.isChecked()) self._worker.set_d2c_method(self.capture_d2c_method_combo.currentData()) self._worker.set_splat(self.capture_splat_combo.currentData()) 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(f"color:#cc8844; font-size:{_fs(12)}px;") # ───────────────────────────────── # 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 base = Path(self.base_dir_edit.text().strip() or ".") session_dir = base / 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(f"color:#2a9a2a; font-size:{_fs(12)}px;") 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("拍摄中…") capture_opts = {k: cb.isChecked() for k, cb in self._capture_checks.items()} capture_opts["d2c_method"] = self.capture_d2c_method_combo.currentData() capture_opts["splat_radius"] = self.capture_splat_combo.currentData() self._worker.trigger_capture(str(self._session_dir / ts), capture_opts) 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 _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(f"color:#cc6666; font-size:{_fs(12)}px;") 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(f"color:#cc6666; font-size:{_fs(12)}px;") return self.params_selected_lbl.setText("正在从设备获取参数…") self.params_selected_lbl.setStyleSheet(f"color:#888; font-size:{_fs(12)}px;") 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() 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.setStyleSheet(f"color:#4aaa4a; font-size:{_fs(12)}px;") self._append_d2c_log( f"[参数获取] RGB {params['color']['width']}x{params['color']['height']} " f"/ Depth {params['depth']['width']}x{params['depth']['height']}" ) # 在日志区打印完整参数 c = params["color"] d = params["depth"] ci = c["intrinsic"] di = d["intrinsic"] cd = c["distortion"] dd = d["distortion"] ext = params["extrinsic"] R = np.array(ext["rot"]).reshape(3, 3) t = np.array(ext["transform"]) sn = params.get("device", {}).get("serial_number", "unknown") self._append_d2c_log("--- Camera Parameters ---") self._append_d2c_log(f" Device: SN {sn}") self._append_d2c_log(f" RGB : {c['width']}x{c['height']} @ {c['fps']}fps {c['format']}") self._append_d2c_log(f" intrinsic fx={ci['fx']:.4f} fy={ci['fy']:.4f} cx={ci['cx']:.4f} cy={ci['cy']:.4f}") self._append_d2c_log(f" distortion k1={cd['k1']:.6f} k2={cd['k2']:.6f} k3={cd['k3']:.6f} k4={cd['k4']:.6f}") self._append_d2c_log(f" k5={cd['k5']:.6f} k6={cd['k6']:.6f} p1={cd['p1']:.6f} p2={cd['p2']:.6f}") self._append_d2c_log(f" Depth : {d['width']}x{d['height']} @ {d['fps']}fps {d['format']}") self._append_d2c_log(f" intrinsic fx={di['fx']:.4f} fy={di['fy']:.4f} cx={di['cx']:.4f} cy={di['cy']:.4f}") self._append_d2c_log(f" distortion k1={dd['k1']:.6f} k2={dd['k2']:.6f} k3={dd['k3']:.6f} k4={dd['k4']:.6f}") self._append_d2c_log(f" k5={dd['k5']:.6f} k6={dd['k6']:.6f} p1={dd['p1']:.6f} p2={dd['p2']:.6f}") self._append_d2c_log(f" scale={d['scale']:.6f} mm/unit") self._append_d2c_log(f" Extrinsic R : {R[0].tolist()}") self._append_d2c_log(f" {R[1].tolist()}") self._append_d2c_log(f" {R[2].tolist()}") self._append_d2c_log(f" Extrinsic t : {t.tolist()} (mm)") self._fetch_worker = None 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(f"color:#4aaa4a; font-size:{_fs(12)}px;") 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) method = self.d2c_method_combo.currentData() splat_r = self.d2c_splat_combo.currentData() if method == "forward" else 1 self._batch_worker = BatchD2CWorker(params, depth_dir, method=method, splat_radius=splat_r) 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") global _UI_SCALE _cfg = AppSettings() _UI_SCALE = _cfg.get_scale_factor() if _UI_SCALE != 1.0: base = app.font() base.setPointSizeF(base.pointSizeF() * _UI_SCALE) app.setFont(base) 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()