Python Recipes

Python 常用操作配方

适用 v0.6.0 / Python 3.12。API 字典 · 完整控制与读取示例

以下代码各自独立。真实设备示例在连接硬件的 Linux 工作站运行，先替换路径并安装相应 extra。照片脚本会打开相机，反馈脚本会访问 CAN，但不会使能机械臂。

拍一张 RGB 照片并保存

需要 video extra；先释放被网页或其他进程占用的相机。

from pathlib import Path
from openarm_video import CameraConfig, OpenCVCamera, encode_frame

config = CameraConfig("head", device="/dev/video0", width=640, height=480)
with OpenCVCamera(config) as camera:
    frame = camera.read()
    print(frame.image.shape)  # NumPy BGR 图像，实际采集尺寸
    data, mime = encode_frame(frame)
    Path("capture.jpg").write_bytes(data)

如算法需要 RGB 数组：rgb = frame.image[:, :, ::-1].copy()。文件写在当前目录，同名文件会覆盖；连续保存时请用序号生成不同文件名。

保存原始深度图

需要 video extra，以及按视觉配置说明准备的 OpenNI2 helper。

from pathlib import Path
from openarm_video import CameraConfig, OpenNIDepthCamera, encode_frame

config = CameraConfig(
    "depth", kind="depth",
    depth_helper="/path/to/openarm-depth-helper",
    width=640, height=480, fps=30,
)
with OpenNIDepthCamera(config) as camera:
    frame = camera.read()
    data, mime = encode_frame(frame, raw=True)
    Path("depth_mm.png").write_bytes(data)

输出为 uint16 毫米值，0 表示无效。回读用 cv2.imread("depth_mm.png", cv2.IMREAD_UNCHANGED)，不要默认按彩色读入；不加 raw=True 得到的是伪彩预览，不能作为原始深度训练数据。

持续读取多路画面，避免重复消费

这个例子使用软件仿真，不打开设备，安装 video extra 即可运行。对每台相机取三个不同的新鲜样本，最多等待 5 秒。

import time
from openarm_video import CameraConfig, VideoCapture

configs = [CameraConfig("head"), CameraConfig("depth", kind="depth")]
seen = {c.name: 0 for c in configs}
consumed = {c.name: 0 for c in configs}
with VideoCapture(configs, simulate=True) as capture:
    deadline = time.monotonic() + 5
    while time.monotonic() < deadline and min(consumed.values()) < 3:
        for name in seen:
            frame, error, count = capture.snapshot(name)
            if error:
                raise RuntimeError(f"{name}: {error}")
            if (frame is not None and frame.fresh(time.monotonic())
                    and count != seen[name] and consumed[name] < 3):
                seen[name] = count
                consumed[name] += 1
                print(name, frame.sequence, frame.received_at, frame.image.shape)
        time.sleep(0.01)
    if min(consumed.values()) < 3:
        raise TimeoutError("未取得足够的新鲜帧")

真实多相机使用 configs = load_cameras("/path/to/video.json") 并去掉 simulate=True。此例按每路最新样本消费，不保证无丢帧或多路同步；统一 episode 录制需要额外定义时间对齐和队列策略。

获取一份完整从臂状态并保存 JSON

仅需核心包、健康 Classic CAN 1 Mbps 链路。can0 只是占位示例；执行前确认映射，且不要与该总线的控制会话同时运行。

import json
import time
from dataclasses import asdict
from pathlib import Path
from openarm_runtime import ArmConnection

reader = ArmConnection("can0", "classic")
try:
    reader.restart()
    deadline = time.monotonic() + 10
    while time.monotonic() < deadline:
        states, error, message, replies, missed = reader.shared.snapshot()
        now = time.monotonic()
        if error:
            raise RuntimeError(error)
        if all(mid in states and states[mid].fresh(now) for mid in range(1, 9)):
            result = {"captured_at_monotonic": now,
                      "motors": {mid: asdict(states[mid]) for mid in range(1, 9)}}
            Path("arm_state.json").write_text(
                json.dumps(result, indent=2, allow_nan=False), encoding="utf-8")
            break
        time.sleep(0.02)
    else:
        raise TimeoutError("未取得八个电机的完整新鲜反馈")
finally:
    reader.close()

各电机有各自 received_at，这不是严格同时采样。JSON 对象键回读后为字符串；训练代码需明确关节顺序和单位。

控制过程中同时收集反馈

控制入口和完整可取消循环见 Python API 的 hold_arm 示例。传入 publish=on_feedback，由回调复制反馈到有界队列，另一个工作线程处理写盘。不要在同一 CAN 总线上新建 ArmConnection，也不要在回调里执行推理或保存图片。

from dataclasses import asdict
from queue import Queue, Full

feedback_queue = Queue(maxsize=100)
dropped = 0

def on_feedback(states):
    global dropped
    sample = {mid: asdict(state) for mid, state in states.items()}
    try:
        feedback_queue.put_nowait(sample)
    except Full:
        dropped += 1  # 明确记录丢样；不要阻塞电机控制循环

# 将回调传给已有的完整控制循环：
# with hold_arm(profile, "right", cancelled=stop.is_set,
#               publish=on_feedback) as arm:
#     ...

该片段只定义回调，不启动控制或消费线程，不是完整数据集录制器。训练数据还需记录 action、图像、采样时间、丢样情况和 episode 边界；不应把传入 tick 的目标偏移当成已经到达的实际位置。

常见选择

场景	选择
只取一帧	OpenCVCamera / OpenNIDepthCamera
自己的算法持续处理相机	CameraReader / VideoCapture
控制时获得关节反馈	hold_arm 的 publish 回调
无控制时独立读取关节	ArmConnection
获取主臂动作输入	LeaderReader + Mapping，参见示例教程
获取触觉	TactileReader，检查 state == "live"，参见示例教程
从 Mac/Windows 发 Python 远程控制请求	0.6.0 没有稳定的硬件 REST SDK；先在 Linux 工作站执行 Python 脚本
末端 XYZ 控制、训练数据集或模型部署	需要另加运动学、录制或策略适配层，没有可直接替代的现成公共函数