From de538a24c37cbcab258b6b89b462ed2dfd2abf9c Mon Sep 17 00:00:00 2001 From: Vinman Date: Tue, 16 Jun 2026 17:05:34 +0800 Subject: [PATCH] =?UTF-8?q?docs:=20=E7=94=9F=E6=88=90=E8=8B=B1=E6=96=87=20?= =?UTF-8?q?README=EF=BC=8C=E6=96=B0=E5=A2=9E=E6=91=84=E5=83=8F=E5=A4=B4?= =?UTF-8?q?=E6=9F=A5=E7=9C=8B=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit README 拆分为英文版和中文版并添加语言切换链接， 新增 uf_camera_view 多摄像头拼接工具， udev 规则移至项目根目录。 --- README.md | 192 +++++---- README_ZH.md | 205 ++++++++++ pyproject.toml | 1 + {src/rules => rules}/81-vive.rules | 0 .../rules => rules}/99-realsense-libusb.rules | 0 {src/rules => rules}/99-xvisio.rules | 0 {src/rules => rules}/sensor_fisheye.rules | 0 {src/rules => rules}/sensor_serial.rules | 0 .../scripts/uf_camera_view.py | 383 ++++++++++++++++++ 9 files changed, 695 insertions(+), 86 deletions(-) create mode 100644 README_ZH.md rename {src/rules => rules}/81-vive.rules (100%) rename {src/rules => rules}/99-realsense-libusb.rules (100%) rename {src/rules => rules}/99-xvisio.rules (100%) rename {src/rules => rules}/sensor_fisheye.rules (100%) rename {src/rules => rules}/sensor_serial.rules (100%) create mode 100644 src/ufactory_lerobot/scripts/uf_camera_view.py diff --git a/README.md b/README.md index a16b837..94aa10b 100644 --- a/README.md +++ b/README.md @@ -1,202 +1,222 @@ # UFACTORY LeRobot -UFACTORY 机械臂与 LeRobot 框架集成项目，支持多种遥操作方式的数据采集、策略训练和部署推理。 +> [中文版本](README_ZH.md) -## 功能特性 +UFACTORY robot arm integration with the LeRobot framework for robot learning, data collection, and policy deployment. -- 🤖 UFACTORY 机械臂控制（xArm 系列） -- 🎮 多种遥操作方式：GELLO / Pika / UMI / SpaceMouse -- 📷 多摄像头数据采集（RealSense / UMI 相机） -- 📊 数据集录制与管理（兼容 LeRobot 格式） -- 🧠 模仿学习训练（ACT / Diffusion Policy 等） -- 🚀 策略评估与实时推理 -- 🔧 Mock 机器人模拟（只用遥操作设备采集数据） +Reference project: [ufactory_teleop](https://github.com/xArm-Developer/ufactory_teleop) -## 环境要求 +## Features + +- xArm robot control (xArm series) +- Multiple teleop modes: GELLO / Pika / UMI / SpaceMouse +- Multi-camera data collection (RealSense / UMI camera) +- Dataset recording & management (LeRobot-compatible) +- Imitation learning training (ACT / Diffusion Policy / etc.) +- Policy evaluation & real-time inference +- Mock mode (no physical robot needed) + +## Requirements - Ubuntu 22.04 / 24.04 - Python >= 3.10 -- CUDA >= 12.0（GPU 训练推荐） -- UFACTORY 机械臂（xArm 系列，可选） +- CUDA >= 12.0 (recommended for GPU training) +- UFACTORY xArm (optional) -## 安装 +## Installation -### 基础项目安装 +### Base Install ```bash git clone https://github.com/xArm-Developer/ufactory_lerobot.git cd ufactory_lerobot -# 创建 conda 环境 +# Create conda environment conda create -n uf_lerobot python=3.10 -y conda activate uf_lerobot -# 安装项目 +# Install project pip install -e . ``` -包含：`lerobot==0.4.3`、`xarm-python-sdk`、`numpy`、`pyyaml`（lerobot 已自动携带 torch、opencv、wandb 等训练相关依赖）。 +Includes: `lerobot==0.4.3`, `xarm-python-sdk`, `numpy`, `pyyaml`. LeRobot already pulls in torch, opencv, wandb, etc. -### 外设模块安装 +### Peripheral Modules -外设依赖以可选模块形式提供，通过 `[模块名]` 安装。 +Peripheral dependencies are available as optional extras via `[module]` install. -#### GELLO 遥操作 +#### GELLO Teleop -适用于 GELLO 示教臂（Dynamixel 舵机方案），控制空间为关节空间。 +Dynamixel-based leader arm, joint-space control. ```bash -# 1. 安装 GELLO 模块 +# 1. Install GELLO module pip install -e ".[gello]" -# 2. 添加串口权限（重新登录后生效） +# 2. Add serial port permissions (re-login required) sudo usermod -aG dialout $USER ``` -#### Pika 遥操作 +#### Pika Teleop -适用于 Pika Sense 手持示教器 + Vive Tracker，控制空间为笛卡尔空间。 +Pika Sense handheld + Vive Tracker, task-space control. ```bash -# 1. 安装外设依赖（不需要它们的间接依赖） +# 1. Install peripheral deps (skip transitive deps) pip install pysurvive agx-pypika --no-deps -# 2. 安装 udev 规则（重新插拔设备后生效） +# 2. Install udev rules (re-plug devices afterwards) sudo cp src/rules/*.rules /etc/udev/rules.d/ sudo udevadm control --reload-rules && sudo udevadm trigger ``` -> Vive Tracker 首次使用前需校准：`uf-vive-calibrate` +> Calibrate Vive Tracker before first use: `uf-vive-calibrate` -#### UMI 遥操作 +#### UMI Teleop -适用于 UMI（Universal Manipulation Interface）方案，含 Vive Tracker 追踪，支持双机械臂。 +Universal Manipulation Interface + Vive Tracker, supports dual-arm. ```bash -# 1. 安装 XVSDK（系统级依赖，仅支持 Ubuntu Focal） +# 1. Install XVSDK (system-level, Ubuntu Focal only) sudo dpkg -i src/xvsdk/XVSDK_focal_amd64.deb sudo apt install -y --fix-broken -# 2. 安装外设依赖 +# 2. Install peripheral deps pip install pysurvive --no-deps -# 3. 安装 udev 规则（重新插拔设备后生效） +# 3. Install udev rules (re-plug devices afterwards) sudo cp src/rules/*.rules /etc/udev/rules.d/ sudo udevadm control --reload-rules && sudo udevadm trigger ``` -> Vive Tracker 首次使用前需校准：`uf-vive-calibrate` +> Calibrate Vive Tracker before first use: `uf-vive-calibrate` -**多 UMI 设备配置**（使用两台及以上时）： +**Multi-UMI device configuration** (two or more devices): ```bash -# 增加 USB 缓冲区大小 +# Increase USB buffer size sudo sed -i '/GRUB_CMDLINE_LINUX_DEFAULT/s/quiet splash/quiet splash usbcore.usbfs_memory_mb=128/' /etc/default/grub sync sudo update-grub sudo reboot ``` -#### SpaceMouse 遥操作 +#### SpaceMouse Teleop -适用于 3Dconnexion SpaceMouse / SpaceNavigator。 +3Dconnexion SpaceMouse / SpaceNavigator. ```bash -# 1. 安装 SpaceMouse 模块 +# 1. Install SpaceMouse module pip install -e ".[spacemouse]" -# 2. 安装 udev 规则（重新插拔设备后生效） +# 2. Install udev rules (re-plug device afterwards) sudo cp src/rules/*.rules /etc/udev/rules.d/ sudo udevadm control --reload-rules && sudo udevadm trigger ``` -## 使用 +## Usage -### 1. 遥操作测试 +### 1. Teleop Testing -测试遥操作设备与机械臂的联动，不录制数据。 +Test teleop-to-robot control loop without recording. ```bash uf-robot-teleop -c path/to/config.yaml -uf-robot-teleop -c path/to/config.yaml -f 60 # 指定频率 +uf-robot-teleop -c path/to/config.yaml -f 60 # specify frequency ``` -### 2. 数据采集 +### 2. Data Collection -通过遥操作录制数据集。 +Record datasets via teleop. ```bash uf-lerobot-record -c path/to/record_config.yaml -uf-lerobot-record -c path/to/config.yaml --resume # 续录 +uf-lerobot-record -c path/to/config.yaml --resume # resume recording ``` -### 3. 策略训练 +### 3. Policy Training -采集数据后，使用 LeRobot 训练管道进行模仿学习训练。 +Train imitation learning policies on collected data. ```bash lerobot-train --policy act --dataset your_dataset_name ``` -### 4. 策略评估 +### 4. Policy Evaluation + +Evaluate trained policies. ```bash uf-lerobot-eval -c path/to/eval_config.yaml ``` -### Mock 模式（无实体机械臂） +### 5. Camera Viewer -```yaml -type: "uf::mock_robot" # 单臂模拟 -type: "uf::multiple_mock_robot" # 双臂模拟 +View and stitch multiple camera feeds. + +```bash +uf-camera-view -l # list all cameras +uf-camera-view -l -T xvisio # list XVisio cameras only +uf-camera-view -T xvisio # view XVisio cameras (default 1280x1280 YU12) +uf-camera-view -T xvisio -W 640 -H 1920 -F NV12 # specify format +uf-camera-view -T other # view other camera types ``` -## 遥操作方式对比 +### Mock Mode (no physical robot) -| 特性 | GELLO | Pika | UMI | SpaceMouse | -|------|-------|------|-----|------------| -| 控制空间 | 关节空间 | 任务空间 | 任务空间 | 任务空间 | -| 跟踪方式 | Dynamixel 舵机 | Vive Tracker | UMI SLAM / Vive | 3D 鼠标 | -| 双臂支持 | ❌ | ❌ | ✅ | ❌ | -| 系统依赖 | dialout 组 | — | XVSDK deb | — | +```yaml +type: "uf::mock_robot" # single arm simulation +type: "uf::multiple_mock_robot" # dual arm simulation +``` -## 项目结构 +## Teleop Comparison + +| Feature | GELLO | Pika | UMI | SpaceMouse | +|---------|-------|------|-----|------------| +| Control space | Joint space | Task space | Task space | Task space | +| Tracking | Dynamixel servos | Vive Tracker | UMI SLAM / Vive | 3D mouse | +| Dual-arm | ❌ | ❌ | ✅ | ❌ | +| System dep | dialout group | — | XVSDK deb | — | + +## Project Structure ``` ufactory_lerobot/ ├── src/ │ ├── ufactory_lerobot/ -│ │ ├── robots/ # 机器人控制 -│ │ │ ├── uf_robot/ # xArm 实体机器人 -│ │ │ ├── uf_mock_robot/ # 仿真 Mock 机器人 -│ │ ├── teleoperators/ # 遥操作器 -│ │ │ ├── gello_teleop/ # GELLO (Dynamixel 示教臂) -│ │ │ ├── pika_teleop/ # Pika Sense (手持示教器 + Vive) -│ │ │ ├── umi_teleop/ # UMI (含双机械臂) -│ │ │ ├── space_mouse/ # SpaceMouse (3D 鼠标) -│ │ ├── cameras/ # 摄像头模块 -│ │ │ └── umi_camera/ # UMI 相机 -│ │ ├── devices/ # 外部设备驱动 -│ │ │ ├── pika/ # Pika 串口驱动 +│ │ ├── robots/ # Robot control +│ │ │ ├── uf_robot/ # xArm physical robot +│ │ │ ├── uf_mock_robot/ # Mock robot simulator +│ │ │ └── utils.py # make_robot_from_config patch +│ │ ├── teleoperators/ # Teleop drivers +│ │ │ ├── gello_teleop/ # GELLO (Dynamixel leader) +│ │ │ ├── pika_teleop/ # Pika Sense (handheld + Vive) +│ │ │ ├── umi_teleop/ # UMI (dual-arm support) +│ │ │ ├── space_mouse/ # SpaceMouse (3D mouse) +│ │ ├── cameras/ # Camera modules +│ │ │ └── umi_camera/ # UMI camera +│ │ ├── devices/ # External device drivers +│ │ │ ├── pika/ # Pika serial driver │ │ │ └── umi/ # XVLib / Vive Tracker -│ │ ├── scripts/ # 执行脚本 -│ │ │ ├── uf_robot_teleop.py # 遥操作测试 -│ │ │ ├── uf_lerobot_record.py # 数据采集 -│ │ │ ├── uf_lerobot_eval.py # 策略评估 -│ │ │ └── vive_calibrate.py # Vive Tracker 校准 -│ │ └── utils/ # 工具函数 -│ ├── rules/ # udev 设备规则 -│ └── xvsdk/ # XVSDK 系统依赖 -├── config/ # YAML 配置文件 +│ │ ├── scripts/ # Entry-point scripts +│ │ │ ├── uf_robot_teleop.py # Teleop testing +│ │ │ ├── uf_lerobot_record.py # Data recording +│ │ │ ├── uf_lerobot_eval.py # Policy evaluation +│ │ │ ├── uf_camera_view.py # camera viewer tool +│ │ │ └── vive_calibrate.py # Vive Tracker calibration +│ │ └── utils/ # Utilities +│ ├── rules/ # udev device rules +│ └── xvsdk/ # XVSDK system dependency +├── config/ # YAML config files │ ├── gello/ │ ├── pika/ │ ├── umi/ │ └── spacemouse/ ├── pyproject.toml -└── README.md +└── LICENSE ``` -## 许可证 +## License -本项目基于 Apache License 2.0 发布，详见 [LICENSE](LICENSE) 文件。 +This project is released under the Apache License 2.0. See [LICENSE](LICENSE). diff --git a/README_ZH.md b/README_ZH.md new file mode 100644 index 0000000..1f28ada --- /dev/null +++ b/README_ZH.md @@ -0,0 +1,205 @@ +# UFACTORY LeRobot + +> [English Version](README.md) + +UFACTORY 机械臂与 LeRobot 框架集成项目，支持多种遥操作方式的数据采集、策略训练和部署推理。 + +## 功能特性 + +- 🤖 UFACTORY 机械臂控制（xArm 系列） +- 🎮 多种遥操作方式：GELLO / Pika / UMI / SpaceMouse +- 📷 多摄像头数据采集（RealSense / UMI 相机） +- 📊 数据集录制与管理（兼容 LeRobot 格式） +- 🧠 模仿学习训练（ACT / Diffusion Policy 等） +- 🚀 策略评估与实时推理 +- 🔧 Mock 机器人模拟（只用遥操作设备采集数据） + +## 环境要求 + +- Ubuntu 22.04 / 24.04 +- Python >= 3.10 +- CUDA >= 12.0（GPU 训练推荐） +- UFACTORY 机械臂（xArm 系列，可选） + +## 安装 + +### 基础项目安装 + +```bash +git clone https://github.com/xArm-Developer/ufactory_lerobot.git +cd ufactory_lerobot + +# 创建 conda 环境 +conda create -n uf_lerobot python=3.10 -y +conda activate uf_lerobot + +# 安装项目 +pip install -e . +``` + +包含：`lerobot==0.4.3`、`xarm-python-sdk`、`numpy`、`pyyaml`（lerobot 已自动携带 torch、opencv、wandb 等训练相关依赖）。 + +### 外设模块安装 + +外设依赖以可选模块形式提供，通过 `[模块名]` 安装。 + +#### GELLO 遥操作 + +适用于 GELLO 示教臂（Dynamixel 舵机方案），控制空间为关节空间。 + +```bash +# 1. 安装 GELLO 模块 +pip install -e ".[gello]" + +# 2. 添加串口权限（重新登录后生效） +sudo usermod -aG dialout $USER +``` + +#### Pika 遥操作 + +适用于 Pika Sense 手持示教器 + Vive Tracker，控制空间为笛卡尔空间。 + +```bash +# 1. 安装外设依赖（不需要它们的间接依赖） +pip install pysurvive agx-pypika --no-deps + +# 2. 安装 udev 规则（重新插拔设备后生效） +sudo cp rules/*.rules /etc/udev/rules.d/ +sudo udevadm control --reload-rules && sudo udevadm trigger +``` + +> Vive Tracker 首次使用前需校准：`uf-vive-calibrate` + +#### UMI 遥操作 + +适用于 UMI（Universal Manipulation Interface）方案，含 Vive Tracker 追踪，支持双机械臂。 + +```bash +# 1. 安装 XVSDK（系统级依赖，仅支持 Ubuntu Focal） +sudo dpkg -i src/xvsdk/XVSDK_focal_amd64.deb +sudo apt install -y --fix-broken + +# 2. 安装外设依赖 +pip install pysurvive --no-deps + +# 3. 安装 udev 规则（重新插拔设备后生效） +sudo cp rules/*.rules /etc/udev/rules.d/ +sudo udevadm control --reload-rules && sudo udevadm trigger +``` + +> Vive Tracker 首次使用前需校准：`uf-vive-calibrate` + +**多 UMI 设备配置**（使用两台及以上时）： + +```bash +# 增加 USB 缓冲区大小 +sudo sed -i '/GRUB_CMDLINE_LINUX_DEFAULT/s/quiet splash/quiet splash usbcore.usbfs_memory_mb=128/' /etc/default/grub +sync +sudo update-grub +sudo reboot +``` + +#### SpaceMouse 遥操作 + +适用于 3Dconnexion SpaceMouse / SpaceNavigator。 + +```bash +# 1. 安装 SpaceMouse 模块 +pip install -e ".[spacemouse]" + +# 2. 安装 udev 规则（重新插拔设备后生效） +sudo cp rules/*.rules /etc/udev/rules.d/ +sudo udevadm control --reload-rules && sudo udevadm trigger +``` + + +## 使用 + +### 1. 遥操作测试 + +测试遥操作设备与机械臂的联动，不录制数据。 + +```bash +uf-robot-teleop -c path/to/config.yaml +uf-robot-teleop -c path/to/config.yaml -f 60 # 指定频率 +``` + +### 2. 数据采集 + +通过遥操作录制数据集。 + +```bash +uf-lerobot-record -c path/to/record_config.yaml +uf-lerobot-record -c path/to/config.yaml --resume # 续录 +``` + +### 3. 策略训练 + +采集数据后，使用 LeRobot 训练管道进行模仿学习训练。 + +```bash +lerobot-train --policy act --dataset your_dataset_name +``` + +### 4. 策略评估 + +```bash +uf-lerobot-eval -c path/to/eval_config.yaml +``` + +### Mock 模式（无实体机械臂） + +```yaml +type: "uf::mock_robot" # 单臂模拟 +type: "uf::multiple_mock_robot" # 双臂模拟 +``` + +## 遥操作方式对比 + +| 特性 | GELLO | Pika | UMI | SpaceMouse | +|------|-------|------|-----|------------| +| 控制空间 | 关节空间 | 任务空间 | 任务空间 | 任务空间 | +| 跟踪方式 | Dynamixel 舵机 | Vive Tracker | UMI SLAM / Vive | 3D 鼠标 | +| 双臂支持 | ❌ | ❌ | ✅ | ❌ | +| 系统依赖 | dialout 组 | — | XVSDK deb | — | + +## 项目结构 + +``` +ufactory_lerobot/ +├── src/ +│ ├── ufactory_lerobot/ +│ │ ├── robots/ # 机器人控制 +│ │ │ ├── uf_robot/ # xArm 实体机器人 +│ │ │ ├── uf_mock_robot/ # 仿真 Mock 机器人 +│ │ ├── teleoperators/ # 遥操作器 +│ │ │ ├── gello_teleop/ # GELLO (Dynamixel 示教臂) +│ │ │ ├── pika_teleop/ # Pika Sense (手持示教器 + Vive) +│ │ │ ├── umi_teleop/ # UMI (含双机械臂) +│ │ │ ├── space_mouse/ # SpaceMouse (3D 鼠标) +│ │ ├── cameras/ # 摄像头模块 +│ │ │ └── umi_camera/ # UMI 相机 +│ │ ├── devices/ # 外部设备驱动 +│ │ │ ├── pika/ # Pika 串口驱动 +│ │ │ └── umi/ # XVLib / Vive Tracker +│ │ ├── scripts/ # 执行脚本 +│ │ │ ├── uf_robot_teleop.py # 遥操作测试 +│ │ │ ├── uf_lerobot_record.py # 数据采集 +│ │ │ ├── uf_lerobot_eval.py # 策略评估 +│ │ │ ├── uf_camera_view.py # 摄像头查看工具 +│ │ │ └── vive_calibrate.py # Vive Tracker 校准 +│ │ └── utils/ # 工具函数 +├── config/ # YAML 配置文件 +│ ├── gello/ +│ ├── pika/ +│ ├── umi/ +│ └── spacemouse/ +├── rules/ # udev 设备规则 +├── xvsdk/ # XVSDK 系统依赖 +├── pyproject.toml +└── README.md +``` + +## 许可证 + +本项目基于 Apache License 2.0 发布，详见 [LICENSE](LICENSE) 文件。 diff --git a/pyproject.toml b/pyproject.toml index def47c4..fc49591 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,6 +36,7 @@ uf-robot-teleop = "ufactory_lerobot.scripts.uf_robot_teleop:main" uf-lerobot-record = "ufactory_lerobot.scripts.uf_lerobot_record:main" uf-lerobot-eval = "ufactory_lerobot.scripts.uf_lerobot_eval:main" uf-vive-calibrate = "ufactory_lerobot.scripts.vive_calibrate:main" +uf-camera-view = "ufactory_lerobot.scripts.uf_camera_view:main" [project.optional-dependencies] # GELLO 遥操作 diff --git a/src/rules/81-vive.rules b/rules/81-vive.rules similarity index 100% rename from src/rules/81-vive.rules rename to rules/81-vive.rules diff --git a/src/rules/99-realsense-libusb.rules b/rules/99-realsense-libusb.rules similarity index 100% rename from src/rules/99-realsense-libusb.rules rename to rules/99-realsense-libusb.rules diff --git a/src/rules/99-xvisio.rules b/rules/99-xvisio.rules similarity index 100% rename from src/rules/99-xvisio.rules rename to rules/99-xvisio.rules diff --git a/src/rules/sensor_fisheye.rules b/rules/sensor_fisheye.rules similarity index 100% rename from src/rules/sensor_fisheye.rules rename to rules/sensor_fisheye.rules diff --git a/src/rules/sensor_serial.rules b/rules/sensor_serial.rules similarity index 100% rename from src/rules/sensor_serial.rules rename to rules/sensor_serial.rules diff --git a/src/ufactory_lerobot/scripts/uf_camera_view.py b/src/ufactory_lerobot/scripts/uf_camera_view.py new file mode 100644 index 0000000..30a6758 --- /dev/null +++ b/src/ufactory_lerobot/scripts/uf_camera_view.py @@ -0,0 +1,383 @@ +#!/usr/bin/env python +"""Multi-camera viewer: XVisio / RealSense / other cameras.""" +import os +import sys +import cv2 +import time +import argparse +import threading +from pathlib import Path + +# ---------- 屏蔽 OpenCV Qt 字体警告 ---------- +os.environ["QT_LOGGING_RULES"] = "*=false" +_fake_stderr = open(os.devnull, "w") +_real_stderr = sys.stderr +sys.stderr = _fake_stderr +import cv2 as _cv2 +sys.stderr = _real_stderr +_fake_stderr.close() +# --------------------------------------------- + +BY_ID_DIR = Path("/dev/v4l/by-id") +BY_PATH_DIR = Path("/dev/v4l/by-path") + +# camera type -> (keyword, default format) +CAMERA_TYPES = { + "xvisio": (1280, 1280, "YU12"), + "realsense": (640, 480, ""), +} + + +class Camera: + """Generic camera.""" + + def __init__(self, device: str, serial: str = "", by_path: str = "", + by_id: str = "", width: int = 640, height: int = 480, + fourcc: str = ""): + self.device = device + self.by_path = by_path + self.by_id = by_id + self.serial = serial + self.width = width + self.height = height + self.fourcc = fourcc + self._cap: cv2.VideoCapture | None = None + + def open(self) -> bool: + for b in [cv2.CAP_V4L2, cv2.CAP_ANY]: + cap = cv2.VideoCapture(self.device, b) + if cap.isOpened(): + if self.fourcc: + cap.set(cv2.CAP_PROP_FOURCC, + cv2.VideoWriter_fourcc(*self.fourcc)) + cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.width) + cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.height) + rw = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + rh = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + if rw < 1 or rh < 1: + cap.release() + continue + self.width, self.height = rw, rh + self._cap = cap + return True + return False + + def read(self): + if self._cap is None: + return None + ret, frame = self._cap.read() + return frame if ret else None + + def close(self): + if self._cap: + self._cap.release() + self._cap = None + + @property + def fps(self) -> float: + return self._cap.get(cv2.CAP_PROP_FPS) if self._cap else 0.0 + + @property + def current_fourcc(self) -> str: + if not self._cap: + return "" + code = int(self._cap.get(cv2.CAP_PROP_FOURCC)) + try: + return "".join(chr((code >> (i * 8)) & 0xFF) for i in range(4)) + except (ValueError, OverflowError): + return str(code) + + @property + def is_open(self) -> bool: + return self._cap is not None and self._cap.isOpened() + + +class CameraManager: + """Camera discovery.""" + + @staticmethod + def find(cam_type: str = "other") -> list[dict]: + """Scan cameras by type. + cam_type: xvisio / realsense / other (default, excludes known types) + Uses /dev/v4l/by-path/ (unique per port), gets serial from by-id. + video-index0 = Video Capture, index1+ = depth/IR/metadata. + """ + if not BY_PATH_DIR.exists(): + return [] + + info = CAMERA_TYPES.get(cam_type) + if info: + keyword = cam_type + w, h, fc = info + else: + keyword = None + w, h, fc = (640, 480, "") + + # 建立 by-id 映射: resolved device -> by-id path + id_map = {} # {'/dev/video0': '/dev/v4l/by-id/xxxx'} + if BY_ID_DIR.exists(): + for p in BY_ID_DIR.iterdir(): + id_map[str(p.resolve())] = str(p) + + result = [] + for p in sorted(BY_PATH_DIR.iterdir()): + name = p.name + if not name.endswith("-video-index0"): + continue + + device = str(p.resolve()) + by_id = id_map.get(device, "") + + # use by-id name for type check (by-path does not contain vendor/model) + type_name = (by_id or name).lower() + if keyword and keyword not in type_name: + continue + if cam_type == 'other' and not keyword and any(k in type_name for k in CAMERA_TYPES.keys()): + continue + + if by_id: + # serial = by_id.rsplit("-video-index0", 1)[0] + # parts = serial.split("_", 2) + # serial = parts[-1] if len(parts) > 2 else serial + serial = by_id.rsplit('/', 1)[-1].split('-', 1)[-1].rsplit("-video-index0", 1)[0] + tmp = serial.split("_") + serial = '_'.join(list({val: val for val in tmp if val}.values())) + else: + serial = name.rsplit("-video-index0", 1)[0] + + result.append({ + "device": device, + "serial": serial, + "by_path": str(p), + "by_id": by_id, + "default_w": w, + "default_h": h, + "default_fourcc": fc, + }) + return result + + @staticmethod + def find_by_serial(serial: str, cam_type: str = "other") -> list[dict]: + all_devs = CameraManager.find(cam_type) + return [d for d in all_devs if serial in d["by_id"] or serial in d["by_path"] or serial in d["serial"]] + # return [d for d in all_devs if serial in d["serial"]] + + +class MultiCameraViewer: + """Multi-camera stitched viewer.""" + + FONT = cv2.FONT_HERSHEY_SIMPLEX + FONT_SCALE = 1.2 + FONT_THICK = 2 + LABEL_H = 66 + + def __init__(self, cameras: list[Camera], seconds: int = 0): + self.cameras = cameras + self.seconds = seconds + self._max_w = int(os.environ.get("DISPLAY_W", 3840)) + self._max_h = int(os.environ.get("DISPLAY_H", 2160)) + + def _stitch_frames(self, frames: list, start_monotonic: float = 0): + valid = [(cam, f) for cam, f in zip(self.cameras, frames) if f is not None] + if not valid: + return None + + resized = [] + seconds = int(time.monotonic() - start_monotonic) + for cam, frame in valid: + h, w = frame.shape[:2] + max_per_cam = self._max_h * 0.8 + if h > max_per_cam: + scale = max_per_cam / h + frame = cv2.resize(frame, (int(w * scale), int(h * scale))) + + label1 = cam.serial + label2 = f"{w}x{h} | {cam.fps:.0f}fps | {cam.current_fourcc} | #{seconds}" + padded = cv2.copyMakeBorder(frame, self.LABEL_H, 0, 0, 0, + cv2.BORDER_CONSTANT, value=(40, 40, 40)) + cv2.putText(padded, label1, (10, 30), + self.FONT, self.FONT_SCALE, (0, 255, 255), self.FONT_THICK, cv2.LINE_AA) + cv2.putText(padded, label2, (10, 80), + self.FONT, self.FONT_SCALE, (0, 200, 255), self.FONT_THICK, cv2.LINE_AA) + resized.append(padded) + + max_h = max(f.shape[0] for f in resized) + aligned = [] + for i, f in enumerate(resized): + if f.shape[0] != max_h: + f = cv2.resize(f, (int(f.shape[1] * max_h / f.shape[0]), max_h)) + # 加右边分隔线（100, 100, 100 灰色，4px 宽） + f = cv2.copyMakeBorder(f, 0, 0, 0, 4, + cv2.BORDER_CONSTANT, value=(100, 100, 100)) + aligned.append(f) + + result = cv2.hconcat(aligned) + rh, rw = result.shape[:2] + scale = min(self._max_w / rw, self._max_h / rh, 1.0) + if scale < 1.0: + result = cv2.resize(result, (int(rw * scale), int(rh * scale))) + return result + + def run(self): + has_gui = bool(os.environ.get("DISPLAY", "")) + if not has_gui: + print("No GUI, press Ctrl+C to exit.") + + if has_gui: + cv2.namedWindow("Camera Viewer", cv2.WINDOW_NORMAL) + + # 用 pynput 全局监听键盘（不依赖 OpenCV 窗口焦点） + stop_event = threading.Event() + if has_gui: + from pynput import keyboard + def _on_press(key): + try: + if key == keyboard.Key.esc or (hasattr(key, 'char') and key.char == 'q'): + stop_event.set() + except Exception: + pass + kb_listener = keyboard.Listener(on_press=_on_press) + kb_listener.daemon = True + kb_listener.start() + + print(f"\nConnected {len(self.cameras)} cameras:\n") + for i, cam in enumerate(self.cameras): + print(f"[{i+1}] {cam.serial} ({cam.width}x{cam.height} {cam.fourcc or 'default'})") + print(f" video_path: {cam.device}") + print(f" v4l_path: {cam.by_path}") + if cam.by_id: + print(f" v4l_id: {cam.by_id}") + print() + print("\nPress q or Esc to exit.\n") + + expired = 0 if self.seconds <= 0 else time.monotonic() + self.seconds + frame_count = 0 + start_monotonic = time.monotonic() + + try: + while not stop_event.is_set() and (expired == 0 or time.monotonic() < expired): + frames = [cam.read() for cam in self.cameras] + frame_count += 1 + + if has_gui: + stitched = self._stitch_frames(frames, start_monotonic) + if stitched is not None: + if frame_count == 1: + cv2.resizeWindow("Camera Viewer", + stitched.shape[1], stitched.shape[0]) + cv2.imshow("Camera Viewer", stitched) + cv2.waitKey(1) + # exit if window was closed + try: + if cv2.getWindowProperty("Camera Viewer", cv2.WND_PROP_VISIBLE) < 1: + break + except cv2.error: + break + else: + if frame_count % 30 == 0: + parts = [] + for cam, f in zip(self.cameras, frames): + status = "OK" if f is not None else "FAIL" + parts.append(f"{cam.device}:{status}") + print(f"[{time.strftime('%H:%M:%S')}] #{frame_count} | " + + " | ".join(parts), end="\r") + except KeyboardInterrupt: + pass + finally: + stop_event.set() + if has_gui: + cv2.destroyAllWindows() + for cam in self.cameras: + cam.close() + print("\nExited.") + + +def main(): + parser = argparse.ArgumentParser( + description="Multi-camera stitched viewer", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog="""Examples: + uf-camera-view -l # list all cameras + uf-camera-view -T xvisio -l # list all XVisio vSLAM + uf-camera-view -T realsense -l # list all RealSense + uf-camera-view -T xvisio # show all XVisio (default 1280x1280 YU12) + uf-camera-view -T xvisio -s 250801DR48 # filter by serial number + uf-camera-view -T xvisio -W 1280 -H 1280 -F YU12 # specify format (main camera) + uf-camera-view -T xvisio -W 640 -H 1920 -F NV12 # specify format (aux cameras) + uf-camera-view -T other # show other cameras + uf-camera-view -T other -t 30 # preview 30 seconds + DISPLAY_W=2560 DISPLAY_H=1440 uf-camera-view -T xvisio # set window size limit""") + parser.add_argument("-T", "--type", type=str, default=None, + choices=["xvisio", "realsense", "other", "all"], + help="camera type") + parser.add_argument("-s", "--serial", type=str, default=None, + help="filter by serial number") + parser.add_argument("-l", "--list", action="store_true", + help="list devices of the given type") + parser.add_argument("-t", "--time", type=int, default=0, + help="preview seconds, 0 = infinite (default: 0)") + parser.add_argument("-W", "--width", type=int, default=None, + help="capture width") + parser.add_argument("-H", "--height", type=int, default=None, + help="capture height") + parser.add_argument("-F", "--fourcc", type=str, default=None, + help="pixel format (e.g. YU12, NV12, MJPG)") + if len(sys.argv) == 1: + parser.print_help() + return + + args = parser.parse_args() + + # -l without -T defaults to all; no -l and no -T is an error + if args.list and not args.type: + args.type = "all" + elif not args.list and not args.type: + parser.error("-T/--type is required (xvisio, realsense, other, all)") + + devices = CameraManager.find(args.type) + + if args.list: + if devices: + type_label = {"all": "", "other": "Other "}.get(args.type, f'{args.type} ') + print(f"Found {len(devices)} {type_label}camera(s):\n") + for i, d in enumerate(devices): + print(f"[{i+1}] {d['serial']}") + print(f" video_path: {d['device']}") + print(f" v4l_path: {d['by_path']}") + if d.get("by_id"): + print(f" v4l_id: {d['by_id']}") + print() + else: + print(f"No {args.type} cameras found") + return + + if args.serial: + devices = CameraManager.find_by_serial(args.serial, args.type) + + if not devices: + print(f"No {args.type} cameras found. Use -l to list available devices.") + return + + cameras = [] + for d in devices: + w = args.width if args.width else d["default_w"] + h = args.height if args.height else d["default_h"] + fc = args.fourcc if args.fourcc else d["default_fourcc"] + cam = Camera(d["device"], serial=d["serial"], + by_path=d["by_path"], by_id=d.get("by_id", ""), + width=w, height=h, fourcc=fc) + if cam.open(): + cameras.append(cam) + else: + print(f"WARNING: cannot open {d['device']}") + + if not cameras: + print("No available cameras.") + return + + viewer = MultiCameraViewer(cameras, seconds=args.time) + viewer.run() + + +if __name__ == "__main__": + main() \ No newline at end of file