串口搜索和摄像头搜索
This commit is contained in:
@@ -32,8 +32,8 @@ class Demo02Config:
|
||||
video2: str = "1"
|
||||
rotation1: dict | str = field(default_factory=lambda: copy.deepcopy(DEFAULT_ROTATION1))
|
||||
rotation2: dict | str = field(default_factory=lambda: copy.deepcopy(DEFAULT_ROTATION2))
|
||||
sensor_config1: str = "./config/ddjx01.json"
|
||||
sensor_config2: str = "./config/ddjx01.json"
|
||||
sensor_config1: str = "./config/ddjx01_q6a.json"
|
||||
sensor_config2: str = "./config/ddjx01_q6a.json"
|
||||
sensor_fps: float = 30.0
|
||||
motion_threshold: float = 0.0
|
||||
sensor_cpu: bool = False
|
||||
|
||||
@@ -0,0 +1,162 @@
|
||||
"""Discover USB cameras without coupling discovery to the sensor runtime."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import os
|
||||
from pathlib import Path
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
|
||||
import cv2
|
||||
|
||||
|
||||
# Most USB camera stacks expose devices from 0 upward. Keeping this bounded
|
||||
# prevents a search from spending seconds probing empty DirectShow devices.
|
||||
DEFAULT_MAX_CAMERA_INDEX = 3
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CameraInfo:
|
||||
index: int
|
||||
backend_name: str
|
||||
backend_api: int
|
||||
width: int
|
||||
height: int
|
||||
|
||||
@property
|
||||
def source(self):
|
||||
if sys.platform.startswith("linux"):
|
||||
return f"/dev/video{self.index}"
|
||||
return str(self.index)
|
||||
|
||||
def label(self):
|
||||
device = self.source if sys.platform.startswith("linux") else f"相机 {self.index}"
|
||||
return f"{device} {self.width}x{self.height} [{self.backend_name}]"
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _suppress_opencv_noise():
|
||||
previous_level = None
|
||||
set_level = None
|
||||
try:
|
||||
logging = getattr(getattr(cv2, "utils", None), "logging", None)
|
||||
if logging is not None:
|
||||
set_level = getattr(logging, "setLogLevel", None)
|
||||
get_level = getattr(logging, "getLogLevel", None)
|
||||
silent = getattr(logging, "LOG_LEVEL_SILENT", 0)
|
||||
if set_level is not None:
|
||||
if get_level is not None:
|
||||
previous_level = get_level()
|
||||
set_level(silent)
|
||||
except Exception:
|
||||
previous_level = None
|
||||
|
||||
devnull = open(os.devnull, "w")
|
||||
try:
|
||||
with contextlib.redirect_stderr(devnull):
|
||||
try:
|
||||
stderr_fd = sys.stderr.fileno()
|
||||
saved_fd = os.dup(stderr_fd)
|
||||
except (AttributeError, OSError):
|
||||
yield
|
||||
else:
|
||||
try:
|
||||
os.dup2(devnull.fileno(), stderr_fd)
|
||||
yield
|
||||
finally:
|
||||
os.dup2(saved_fd, stderr_fd)
|
||||
os.close(saved_fd)
|
||||
finally:
|
||||
devnull.close()
|
||||
if set_level is not None and previous_level is not None:
|
||||
try:
|
||||
set_level(previous_level)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _backend_candidates():
|
||||
candidates = []
|
||||
if sys.platform == "win32":
|
||||
if hasattr(cv2, "CAP_DSHOW"):
|
||||
candidates.append(("DSHOW", cv2.CAP_DSHOW))
|
||||
if hasattr(cv2, "CAP_MSMF"):
|
||||
candidates.append(("MSMF", cv2.CAP_MSMF))
|
||||
elif sys.platform.startswith("linux") and hasattr(cv2, "CAP_V4L2"):
|
||||
candidates.append(("V4L2", cv2.CAP_V4L2))
|
||||
return candidates or [("ANY", cv2.CAP_ANY)]
|
||||
|
||||
|
||||
def camera_source_index(source):
|
||||
"""Return the numeric index for ``0`` or ``/dev/video0`` sources."""
|
||||
text = str(source or "").strip()
|
||||
if text.isdigit():
|
||||
return int(text)
|
||||
if text.startswith("/dev/video"):
|
||||
suffix = text[len("/dev/video"):]
|
||||
if suffix.isdigit():
|
||||
return int(suffix)
|
||||
return None
|
||||
|
||||
|
||||
def _candidate_indices(max_index):
|
||||
if sys.platform.startswith("linux"):
|
||||
indices = []
|
||||
for path in Path("/dev").glob("video*"):
|
||||
index = camera_source_index(str(path))
|
||||
if index is not None:
|
||||
indices.append(index)
|
||||
if indices:
|
||||
return sorted(set(indices))
|
||||
return list(range(max_index + 1))
|
||||
|
||||
|
||||
def _try_open(index, backend_name, backend_api):
|
||||
capture = cv2.VideoCapture(index, backend_api)
|
||||
try:
|
||||
if not capture.isOpened():
|
||||
return None
|
||||
ok, frame = capture.read()
|
||||
width = int(capture.get(cv2.CAP_PROP_FRAME_WIDTH) or 0)
|
||||
height = int(capture.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0)
|
||||
if ok and frame is not None:
|
||||
frame_height, frame_width = frame.shape[:2]
|
||||
width = width or int(frame_width)
|
||||
height = height or int(frame_height)
|
||||
elif width <= 0 or height <= 0:
|
||||
return None
|
||||
return CameraInfo(index, backend_name, backend_api, width, height)
|
||||
finally:
|
||||
capture.release()
|
||||
|
||||
|
||||
def list_cameras(
|
||||
max_index=DEFAULT_MAX_CAMERA_INDEX,
|
||||
skip_indices=None,
|
||||
fallback_backends=False,
|
||||
):
|
||||
"""Probe camera indices and return one working backend for each index."""
|
||||
found = []
|
||||
seen_indices = set()
|
||||
for source in skip_indices or ():
|
||||
index = camera_source_index(source)
|
||||
if index is not None:
|
||||
seen_indices.add(index)
|
||||
backends = _backend_candidates()
|
||||
if not fallback_backends:
|
||||
backends = backends[:1]
|
||||
with _suppress_opencv_noise():
|
||||
# Probe one backend across all indices before falling back to another
|
||||
# backend. On Windows this avoids opening every empty index twice when
|
||||
# DirectShow is available.
|
||||
for backend_name, backend_api in backends:
|
||||
for index in _candidate_indices(max_index):
|
||||
if index in seen_indices:
|
||||
continue
|
||||
info = _try_open(index, backend_name, backend_api)
|
||||
if info is not None:
|
||||
found.append(info)
|
||||
seen_indices.add(index)
|
||||
found.sort(key=lambda camera: camera.index)
|
||||
return found
|
||||
@@ -272,8 +272,8 @@ class TwoSensorForceReader:
|
||||
self,
|
||||
video1=0,
|
||||
video2=1,
|
||||
config1="./config/ddjx01.json",
|
||||
config2="./config/ddjx01.json",
|
||||
config1="./config/ddjx01_q6a.json",
|
||||
config2="./config/ddjx01_q6a.json",
|
||||
rotation1=None,
|
||||
rotation2=None,
|
||||
target_fps=30,
|
||||
@@ -433,8 +433,8 @@ def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--video1", "-v1", type=str, default="0")
|
||||
parser.add_argument("--video2", "-v2", type=str, default="1")
|
||||
parser.add_argument("--config1", type=str, default="./config/ddjx01.json")
|
||||
parser.add_argument("--config2", type=str, default="./config/ddjx01.json")
|
||||
parser.add_argument("--config1", type=str, default="./config/ddjx01_q6a.json")
|
||||
parser.add_argument("--config2", type=str, default="./config/ddjx01_q6a.json")
|
||||
parser.add_argument("--rotation1", "-r1", type=str, default=None)
|
||||
parser.add_argument("--rotation2", "-r2", type=str, default=None)
|
||||
parser.add_argument("--target-fps", type=float, default=30)
|
||||
|
||||
+319
-5
@@ -18,6 +18,11 @@ if sys.platform.startswith("linux") and os.environ.get("WSL_DISTRO_NAME"):
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
try:
|
||||
from serial.tools import list_ports as serial_list_ports
|
||||
except Exception:
|
||||
serial_list_ports = None
|
||||
|
||||
|
||||
def _fix_qt_plugin_path():
|
||||
try:
|
||||
@@ -42,6 +47,7 @@ try:
|
||||
QApplication,
|
||||
QButtonGroup,
|
||||
QCheckBox,
|
||||
QComboBox,
|
||||
QDialog,
|
||||
QFrame,
|
||||
QGraphicsOpacityEffect,
|
||||
@@ -71,7 +77,7 @@ except Exception as exc:
|
||||
QColor = QLinearGradient = QPainter = QPen = QPolygonF = QRadialGradient = QVector3D = None
|
||||
QGridLayout = QHBoxLayout = QLabel = QPushButton = QSizeGrip = QSizePolicy = QSplitter = None
|
||||
QVBoxLayout = QWidget = None
|
||||
QButtonGroup = QCheckBox = QDialog = QFrame = QGraphicsOpacityEffect = QLineEdit = QListWidget = QMessageBox = QMainWindow = object
|
||||
QButtonGroup = QCheckBox = QComboBox = QDialog = QFrame = QGraphicsOpacityEffect = QLineEdit = QListWidget = QMessageBox = QMainWindow = object
|
||||
QProgressBar = QScrollArea = QSlider = QSpinBox = QStackedWidget = QTextEdit = None
|
||||
|
||||
try:
|
||||
@@ -92,6 +98,7 @@ else:
|
||||
|
||||
from ..app import build_controller
|
||||
from ..common.paths import PACKAGE_DIR, REPO_ROOT
|
||||
from ..sensors.camera_discovery import camera_source_index, list_cameras
|
||||
|
||||
|
||||
PLOT_POINTS = 180
|
||||
@@ -599,7 +606,7 @@ QSpinBox {
|
||||
border-radius: 4px;
|
||||
padding: 3px 5px;
|
||||
}
|
||||
QLineEdit, QTextEdit {
|
||||
QLineEdit, QTextEdit, QComboBox {
|
||||
background-color: #ffffff;
|
||||
color: #1f2933;
|
||||
border: 1px solid #d7dee8;
|
||||
@@ -608,7 +615,7 @@ QLineEdit, QTextEdit {
|
||||
font-family: "Cascadia Code", "Consolas";
|
||||
font-size: 12px;
|
||||
}
|
||||
QLineEdit:focus, QTextEdit:focus {
|
||||
QLineEdit:focus, QTextEdit:focus, QComboBox:focus {
|
||||
border: 1px solid #7aa7e8;
|
||||
}
|
||||
QListWidget {
|
||||
@@ -733,6 +740,12 @@ def _draw_top_icon(kind, color="#344054", size=22):
|
||||
painter.setBrush(QColor("#ffffff"))
|
||||
painter.drawEllipse(QPointF(knob_x, y), size * 0.075, size * 0.075)
|
||||
painter.setBrush(Qt.NoBrush)
|
||||
elif kind == "search":
|
||||
painter.drawEllipse(QRectF(size * 0.25, size * 0.23, size * 0.40, size * 0.40))
|
||||
painter.drawLine(
|
||||
QPointF(size * 0.60, size * 0.60),
|
||||
QPointF(size * 0.76, size * 0.76),
|
||||
)
|
||||
elif kind == "minimize":
|
||||
painter.drawLine(
|
||||
QPointF(size * 0.32, size * 0.62),
|
||||
@@ -932,6 +945,7 @@ class ImageBox(QFrame):
|
||||
layout.setContentsMargins(6, 6, 6, 6)
|
||||
layout.setSpacing(5)
|
||||
header = QHBoxLayout()
|
||||
self.header_layout = header
|
||||
header.setSpacing(6)
|
||||
self.title_label = QLabel(title)
|
||||
self.title_label.setObjectName("SectionTitle")
|
||||
@@ -994,6 +1008,10 @@ class FloatingImageBox(ImageBox):
|
||||
self.image_label.setScaledContents(False)
|
||||
self.mode_toggle_button.setObjectName("FlowModeButton")
|
||||
self.mode_toggle_button.setText("光流")
|
||||
self.camera_search_button = TopIconButton("search", "搜索并设置左右摄像头")
|
||||
self.camera_search_button.setFixedSize(24, 24)
|
||||
self.camera_search_button.setIconSize(QSize(17, 17))
|
||||
self.header_layout.insertWidget(self.header_layout.count() - 1, self.camera_search_button)
|
||||
self._drag_mode = None
|
||||
self._drag_start_global = None
|
||||
self._start_geometry = None
|
||||
@@ -2203,6 +2221,141 @@ class SliderControl(QFrame):
|
||||
return int(self.slider.value())
|
||||
|
||||
|
||||
class CameraDiscoveryDialog(QDialog):
|
||||
def __init__(self, viewer):
|
||||
super().__init__(viewer)
|
||||
self.viewer = viewer
|
||||
self._scan_id = 0
|
||||
self._scan_result = None
|
||||
self.setWindowTitle("双摄像头设置")
|
||||
self.setWindowIcon(_load_company_icon(64))
|
||||
# Keep the main window responsive while OpenCV probes devices in the
|
||||
# worker thread. A modeless child also avoids a full-screen compositor
|
||||
# flash caused by switching into a nested modal event loop.
|
||||
self.setModal(False)
|
||||
self.setMinimumWidth(520)
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(14, 14, 14, 14)
|
||||
layout.setSpacing(10)
|
||||
self.status_label = QLabel("正在搜索摄像头...")
|
||||
self.status_label.setObjectName("MetricLabel")
|
||||
layout.addWidget(self.status_label)
|
||||
|
||||
camera_grid = QGridLayout()
|
||||
camera_grid.setHorizontalSpacing(10)
|
||||
camera_grid.setVerticalSpacing(8)
|
||||
self.camera_combos = {}
|
||||
for row, (side, title) in enumerate((("left", "左侧摄像头"), ("right", "右侧摄像头"))):
|
||||
label = QLabel(title)
|
||||
label.setObjectName("MetricLabel")
|
||||
combo = QComboBox()
|
||||
combo.setMinimumHeight(32)
|
||||
self.camera_combos[side] = combo
|
||||
camera_grid.addWidget(label, row, 0)
|
||||
camera_grid.addWidget(combo, row, 1)
|
||||
camera_grid.setColumnStretch(1, 1)
|
||||
layout.addLayout(camera_grid)
|
||||
|
||||
actions = QHBoxLayout()
|
||||
self.refresh_button = AnimatedButton("重新搜索")
|
||||
self.refresh_button.setObjectName("ManualButton")
|
||||
self.refresh_button.clicked.connect(self._start_scan)
|
||||
self.apply_button = AnimatedButton("应用左右摄像头")
|
||||
self.apply_button.setObjectName("OpenActionButton")
|
||||
self.apply_button.clicked.connect(self._apply)
|
||||
cancel_button = AnimatedButton("取消")
|
||||
cancel_button.setObjectName("ManualButton")
|
||||
cancel_button.clicked.connect(self.reject)
|
||||
actions.addWidget(self.refresh_button)
|
||||
actions.addStretch(1)
|
||||
actions.addWidget(self.apply_button)
|
||||
actions.addWidget(cancel_button)
|
||||
layout.addLayout(actions)
|
||||
|
||||
self.poll_timer = QTimer(self)
|
||||
self.poll_timer.setInterval(100)
|
||||
self.poll_timer.timeout.connect(self._poll_scan)
|
||||
self._start_scan()
|
||||
|
||||
def _start_scan(self):
|
||||
self._scan_id += 1
|
||||
scan_id = self._scan_id
|
||||
self._scan_result = None
|
||||
self.status_label.setText("正在快速搜索摄像头(编号 0-3)...")
|
||||
current_sources = self.viewer.camera_sources()
|
||||
skip_indices = set()
|
||||
for side, combo in self.camera_combos.items():
|
||||
combo.clear()
|
||||
current = current_sources.get(side, "")
|
||||
if current:
|
||||
combo.addItem(f"当前配置 {current}(可能正在使用)", current)
|
||||
combo.setEnabled(True)
|
||||
current_index = camera_source_index(current)
|
||||
if current_index is not None:
|
||||
skip_indices.add(current_index)
|
||||
else:
|
||||
combo.setEnabled(False)
|
||||
self.refresh_button.setEnabled(False)
|
||||
self.apply_button.setEnabled(
|
||||
all(combo.count() > 0 for combo in self.camera_combos.values())
|
||||
)
|
||||
|
||||
def scan():
|
||||
try:
|
||||
self._scan_result = (
|
||||
scan_id,
|
||||
list_cameras(skip_indices=skip_indices),
|
||||
None,
|
||||
)
|
||||
except Exception as exc:
|
||||
self._scan_result = (scan_id, [], str(exc))
|
||||
|
||||
threading.Thread(target=scan, daemon=True).start()
|
||||
self.poll_timer.start()
|
||||
|
||||
def _poll_scan(self):
|
||||
result = self._scan_result
|
||||
if result is None:
|
||||
return
|
||||
scan_id, cameras, error = result
|
||||
if scan_id != self._scan_id:
|
||||
return
|
||||
self.poll_timer.stop()
|
||||
self.refresh_button.setEnabled(True)
|
||||
current_sources = self.viewer.camera_sources()
|
||||
found_sources = set()
|
||||
for side, combo in self.camera_combos.items():
|
||||
combo.clear()
|
||||
for camera in cameras:
|
||||
combo.addItem(camera.label(), camera.source)
|
||||
found_sources.add(camera.source)
|
||||
current = current_sources.get(side, "")
|
||||
if current and current not in found_sources:
|
||||
combo.addItem(f"当前配置 {current}(可能正在使用)", current)
|
||||
current_index = combo.findData(current)
|
||||
if current_index >= 0:
|
||||
combo.setCurrentIndex(current_index)
|
||||
if error:
|
||||
self.status_label.setText(f"搜索失败:{error}")
|
||||
elif cameras:
|
||||
self.status_label.setText(f"找到 {len(cameras)} 个可用摄像头")
|
||||
else:
|
||||
self.status_label.setText("未发现空闲摄像头;正在使用的当前配置仍可保留")
|
||||
has_both = all(combo.count() > 0 for combo in self.camera_combos.values())
|
||||
for combo in self.camera_combos.values():
|
||||
combo.setEnabled(combo.count() > 0)
|
||||
self.apply_button.setEnabled(has_both)
|
||||
|
||||
def _apply(self):
|
||||
left = self.camera_combos["left"].currentData()
|
||||
right = self.camera_combos["right"].currentData()
|
||||
if left is None or right is None:
|
||||
return
|
||||
if self.viewer.apply_camera_sources(str(left), str(right)):
|
||||
self.accept()
|
||||
|
||||
|
||||
class VisualSettingsDialog(QDialog):
|
||||
CONFIG_GROUPS = [
|
||||
(
|
||||
@@ -2595,6 +2748,21 @@ class VisualSettingsDialog(QDialog):
|
||||
return scroll
|
||||
|
||||
def _make_control(self, key, value):
|
||||
if key == "port":
|
||||
container = QWidget()
|
||||
layout = QHBoxLayout(container)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
layout.setSpacing(6)
|
||||
control = QComboBox()
|
||||
control.setEditable(True)
|
||||
control.addItem(str(value), str(value))
|
||||
control.setToolTip("port")
|
||||
search_button = TopIconButton("search", "搜索可用串口")
|
||||
search_button.clicked.connect(self._search_serial_ports)
|
||||
layout.addWidget(control, 1)
|
||||
layout.addWidget(search_button)
|
||||
self.controls[key] = control
|
||||
return container
|
||||
if isinstance(value, bool):
|
||||
control = QCheckBox()
|
||||
control.setChecked(value)
|
||||
@@ -2612,6 +2780,11 @@ class VisualSettingsDialog(QDialog):
|
||||
|
||||
def _control_value(self, key, original):
|
||||
control = self.controls[key]
|
||||
if isinstance(control, QComboBox):
|
||||
selected = control.currentData()
|
||||
if control.currentIndex() >= 0 and selected is not None:
|
||||
return str(selected)
|
||||
return control.currentText().strip()
|
||||
if isinstance(original, bool):
|
||||
return bool(control.isChecked())
|
||||
if isinstance(original, (dict, list)):
|
||||
@@ -2636,11 +2809,50 @@ class VisualSettingsDialog(QDialog):
|
||||
return
|
||||
if isinstance(control, QCheckBox):
|
||||
control.setChecked(bool(value))
|
||||
elif isinstance(control, QComboBox):
|
||||
text = "" if value is None else str(value)
|
||||
index = control.findData(text)
|
||||
if index < 0:
|
||||
control.addItem(text, text)
|
||||
index = control.count() - 1
|
||||
control.setCurrentIndex(index)
|
||||
elif isinstance(control, QTextEdit):
|
||||
control.setPlainText(json.dumps(value, ensure_ascii=False, indent=2))
|
||||
elif isinstance(control, QLineEdit):
|
||||
control.setText("null" if value is None else str(value))
|
||||
|
||||
def _search_serial_ports(self):
|
||||
control = self.controls.get("port")
|
||||
if not isinstance(control, QComboBox):
|
||||
return
|
||||
current = self._control_value("port", self.config_data.get("port", ""))
|
||||
if serial_list_ports is None:
|
||||
QMessageBox.warning(self, "搜索失败", "PySerial 串口枚举模块未安装。")
|
||||
return
|
||||
try:
|
||||
ports = list(serial_list_ports.comports())
|
||||
except Exception as exc:
|
||||
QMessageBox.warning(self, "搜索失败", f"无法枚举串口:\n{exc}")
|
||||
return
|
||||
|
||||
control.clear()
|
||||
selected_index = -1
|
||||
for port in ports:
|
||||
device = str(port.device)
|
||||
description = str(port.description or "未知设备")
|
||||
control.addItem(f"{device} [{description}]", device)
|
||||
if device == current:
|
||||
selected_index = control.count() - 1
|
||||
if current and selected_index < 0:
|
||||
control.addItem(f"{current} [当前配置]", current)
|
||||
selected_index = control.count() - 1
|
||||
if selected_index >= 0:
|
||||
control.setCurrentIndex(selected_index)
|
||||
if not ports:
|
||||
QMessageBox.information(self, "搜索串口", "未发现可用串口,已保留当前配置。")
|
||||
return
|
||||
control.showPopup()
|
||||
|
||||
def _reload(self):
|
||||
self._load_config()
|
||||
for key, value in self.config_data.items():
|
||||
@@ -2665,12 +2877,26 @@ class VisualSettingsDialog(QDialog):
|
||||
QMessageBox.warning(self, "保存失败", f"无法写入配置文件:\n{exc}")
|
||||
return
|
||||
|
||||
try:
|
||||
layout_path = self.viewer.save_current_layout()
|
||||
except Exception as exc:
|
||||
QMessageBox.warning(self, "画面保存失败", f"参数已写入,但画面旋转无法保存:\n{exc}")
|
||||
return
|
||||
|
||||
self.config_data = updated
|
||||
QMessageBox.information(self, "已保存", "配置已写入 gripper.json。重启控制程序后完整生效。")
|
||||
QMessageBox.information(
|
||||
self,
|
||||
"已保存",
|
||||
f"配置已写入 gripper.json,画面旋转已写入:\n{layout_path}\n重启控制程序后完整生效。",
|
||||
)
|
||||
|
||||
def _adjust(self, side, action):
|
||||
self.viewer.adjust_visual_transform(side, action)
|
||||
self._refresh_labels()
|
||||
try:
|
||||
self.viewer.save_current_layout()
|
||||
except Exception as exc:
|
||||
QMessageBox.warning(self, "保存失败", f"画面已调整,但无法保存旋转配置:\n{exc}")
|
||||
|
||||
def _refresh_labels(self):
|
||||
for side, title in (("left", "左侧光流"), ("right", "右侧光流")):
|
||||
@@ -2726,7 +2952,7 @@ class GripperDemo02Viewer(QMainWindow):
|
||||
if pg is None:
|
||||
raise RuntimeError(f"pyqtgraph import failed: {_PYQTGRAPH_IMPORT_ERROR}")
|
||||
|
||||
self.setWindowTitle("Gripper Demo 02")
|
||||
self.setWindowTitle("Orisys 夹爪")
|
||||
self.setWindowIcon(_load_company_icon(64))
|
||||
self.setStyleSheet(STYLE)
|
||||
self.controller = None
|
||||
@@ -2745,6 +2971,7 @@ class GripperDemo02Viewer(QMainWindow):
|
||||
self.argv = argv
|
||||
self.layout_path = GUI_LAYOUT_PATH
|
||||
self.visual_settings_dialog = None
|
||||
self.camera_discovery_dialog = None
|
||||
self._pending_runtime_speed = None
|
||||
self._pending_position_target = None
|
||||
self._pending_position_since = 0.0
|
||||
@@ -2898,6 +3125,91 @@ class GripperDemo02Viewer(QMainWindow):
|
||||
config_path = REPO_ROOT / config_path
|
||||
return config_path
|
||||
|
||||
def camera_sources(self):
|
||||
sources = {"left": "", "right": ""}
|
||||
try:
|
||||
with open(self._control_config_path(), "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
sources["left"] = str(data.get("video1", ""))
|
||||
sources["right"] = str(data.get("video2", ""))
|
||||
except Exception:
|
||||
pass
|
||||
return sources
|
||||
|
||||
def open_camera_search(self):
|
||||
existing = self.camera_discovery_dialog
|
||||
if existing is not None and existing.isVisible():
|
||||
existing.raise_()
|
||||
existing.activateWindow()
|
||||
return
|
||||
dialog = CameraDiscoveryDialog(self)
|
||||
self.camera_discovery_dialog = dialog
|
||||
dialog.finished.connect(lambda _result: setattr(self, "camera_discovery_dialog", None))
|
||||
dialog.show()
|
||||
dialog.raise_()
|
||||
|
||||
def _update_camera_argv(self, config_key, source):
|
||||
args = list(self.argv or [])
|
||||
option_names = {
|
||||
"video1": ("--video1", "-v1"),
|
||||
"video2": ("--video2", "-v2"),
|
||||
}[config_key]
|
||||
for index, arg in enumerate(args):
|
||||
if arg in option_names and index + 1 < len(args):
|
||||
args[index + 1] = source
|
||||
self.argv = args
|
||||
return
|
||||
for option in option_names:
|
||||
if str(arg).startswith(option + "="):
|
||||
args[index] = option + "=" + source
|
||||
self.argv = args
|
||||
return
|
||||
|
||||
def apply_camera_sources(self, left_source, right_source):
|
||||
if left_source == right_source:
|
||||
QMessageBox.warning(self, "摄像头冲突", "左右两侧需要选择不同的摄像头。")
|
||||
return False
|
||||
sources = self.camera_sources()
|
||||
if left_source == sources.get("left") and right_source == sources.get("right"):
|
||||
return True
|
||||
|
||||
config_path = self._control_config_path()
|
||||
temp_path = config_path.with_suffix(config_path.suffix + ".tmp")
|
||||
try:
|
||||
with open(config_path, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError("配置文件根节点不是 JSON 对象")
|
||||
data["video1"] = left_source
|
||||
data["video2"] = right_source
|
||||
config_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(temp_path, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||||
f.write("\n")
|
||||
os.replace(temp_path, config_path)
|
||||
except Exception as exc:
|
||||
try:
|
||||
if temp_path.exists():
|
||||
temp_path.unlink()
|
||||
except OSError:
|
||||
pass
|
||||
QMessageBox.warning(self, "保存失败", f"无法保存摄像头配置:\n{exc}")
|
||||
return False
|
||||
|
||||
self._update_camera_argv("video1", left_source)
|
||||
self._update_camera_argv("video2", right_source)
|
||||
if self.visual_settings_dialog is not None:
|
||||
for config_key, source in (("video1", left_source), ("video2", right_source)):
|
||||
self.visual_settings_dialog.config_data[config_key] = source
|
||||
self.visual_settings_dialog._set_control_value(config_key, source)
|
||||
QMessageBox.information(
|
||||
self,
|
||||
"双摄像头已切换",
|
||||
f"左侧摄像头: {left_source}\n右侧摄像头: {right_source}\n控制程序将自动重启。",
|
||||
)
|
||||
self.restart_control()
|
||||
return True
|
||||
|
||||
def _load_position_config(self):
|
||||
open_pos = 0
|
||||
close_pos = 9000
|
||||
@@ -3007,6 +3319,8 @@ class GripperDemo02Viewer(QMainWindow):
|
||||
self.right_flow = self.gripper_stage.right_flow
|
||||
self.left_flow.set_transform_label(self.visual_transforms["left"])
|
||||
self.right_flow.set_transform_label(self.visual_transforms["right"])
|
||||
self.left_flow.camera_search_button.clicked.connect(self.open_camera_search)
|
||||
self.right_flow.camera_search_button.clicked.connect(self.open_camera_search)
|
||||
visual_layout.addWidget(self.gripper_stage, 0, 0)
|
||||
visual_layout.setColumnStretch(0, 1)
|
||||
center_splitter.addWidget(visual_panel)
|
||||
|
||||
Reference in New Issue
Block a user