API Dictionary

Python API 字典

适用 v0.6.0 / Python 3.12。覆盖五个模块 __all__ 中的全部 27 个公开名称，以及日常调用需要的方法。签名中的 * 表示其后参数必须按名称传入；None 表示无返回值或尚无数据，具体见条目。

示例教程 · 常用操作配方 · 安装

按任务查找

我想做什么	查阅入口
加载左右臂配置	Profile
读取从臂位置、速度、力矩	ArmConnection / MotorState
保持姿态、调整关节	hold_arm / tick
执行硬件零位标定	calibrate_arm
读取主臂、换算关节角	LeaderBus / LeaderReader / Mapping
读取触觉	TactileReader / Sample
拍一张照片、保存深度	同步相机 / encode_frame；完整配方
持续读取或读取多台相机	CameraReader / VideoCapture
给自己的程序添加视频网页	create_app / serve
读取平台启动配置	PlatformConfig / load_config

公共约定

导入模块不打开硬件。LeaderBus、OpenCVCamera、OpenNIDepthCamera 构造时就打开设备；后台 reader 在启动时访问设备；hold_arm 在进入上下文时使能电机。

时间与年龄使用本机 time.monotonic() 秒数，不是 Unix 时间，也不是多设备硬件同步。快照只返回最新已知样本，可能重复或过期；读取时同时检查错误、完整性、年龄。图像作为只读数据使用，修改前 .copy()。

文件和设备路径由调用方配置。控制所需 follower extra 与原生 SDK、视频所需 video extra、串口所需 leader / tactile extra 见安装页。停止占用同一设备的网页采集或控制服务后再独立运行脚本。

下文列出主要错误，不是所有底层异常的穷举。后台采集错误通常由快照返回，构造、同步读取与清理失败通常直接抛出。不要吞掉关闭或失能失败。

openarm_runtime.Profile

Profile(left, right, mapping_verified=False,
        rate_hz=20.0, feedback_timeout_s=0.04)
Profile.load(path)  # 类方法 → Profile
profile.interface(side)  # → str

参数 / 属性	含义
left, right	不同的 SocketCAN 接口名，如 can1 / can0；不是通用接线约定
mapping_verified	是否已由操作者确认物理左右映射；控制要求为 True
rate_hz	10–50 Hz；调用方据此调度，配置本身不启动线程
feedback_timeout_s	0.005–0.1 秒
load(path)	读取 JSON，要求 format="classic"、bitrate=1000000；返回不可变配置
interface(side)	side="left" 或 "right"，返回该侧接口名

错误：字段不合法抛 ValueError，未知构造字段可能抛 TypeError，文件和 JSON 错误直接传播。加载配置不会配置 CAN 或验证物理接线。

openarm_runtime：从臂反馈

ArmConnection

ArmConnection(interface, frame_format="auto")
reader.restart()  # → None
reader.shared.snapshot()  # → (states, error, message, replies, missed)
reader.close()  # → None

推荐显式使用 frame_format="classic"，对应本项目支持的控制链路。构造不启动；restart() 停止旧线程、清空状态后启动新读取线程；close() 最多等待线程 3 秒，未停止抛 RuntimeError。本类不支持 with，使用 try/finally。

快照字段	含义
states	{motor_id: MotorState} 的字典副本；初始化或故障时可能缺少 ID
error	错误字符串；空字符串表示没有已报告错误，不代表样本必然齐全且新鲜
message	诊断文字
replies, missed	读取响应及缺失计数，用于诊断，不是同步帧序号

只发送读取请求，不使能电机。整臂采集需检查 ID 1–8 全部存在并且新鲜。控制期间不要另开同总线 ArmConnection，改用控制接口的 publish 回调。

MotorState

MotorState(motor_id, position, velocity, torque, mos_temperature,
           rotor_temperature, status, received_at, frame_format="classic")
state.fresh(now, timeout=0.4)  # → bool；0 <= 年龄 <= timeout

不可变数据对象，一般从读取器获取，不必手动构造。position 为 rad，velocity 为 rad/s，torque 为 Nm；温度字段为设备反馈温度，status 为原始状态码，received_at 为单调时钟秒数。fresh 只检查年龄，不检查故障码、关节范围或完整性。

MotorRange

MotorRange(position, velocity, torque)

三个正有限数值分别用于位置、速度、力矩的固件编码范围，每项小于 100000。非法值抛 ValueError。这是固件解码范围，不能拿来当机械关节限位或防碰撞边界。

openarm_runtime：关节控制

hold_arm(profile, side, *, allow_unreadable_mode=False,
         cancelled=..., record=..., publish=...)
# 使用：with hold_arm(...) as arm:
arm.tick(offsets, speed, period, gain_scale=1)  # → None

hold_arm 返回上下文管理器；进入时获取总线独占、预检并使能，捕获本次基准姿态；退出时执行失能清理和关闭。返回的控制对象使用 tick，不需要自行导入内部 ArmHold 类。

参数	类型 / 语义
profile, side	Profile；"left" / "right"
cancelled	无参函数返回 bool；默认始终 False，True 时控制抛 InterruptedError
record	record(event_dict)，生命周期与采样日志，默认不处理
publish	publish(states)，{ID: MotorState}，默认不处理；可用于控制中的反馈采集
allow_unreadable_mode	默认 False；仅允许已验证的 RID10 读取超时例外，明确读到非 MIT 模式仍拒绝
offsets	{1: 度数, ..., 8: 度数}；八个 ID 必须齐全，每项 ±30°，相对本次使能姿态
speed	0.1–5 度/秒，是每周期参考目标变化的限速，不是动作完成时间
period	0.02–0.1 秒；用于目标步长计算，tick 不替调用方睡眠
gain_scale	0–1，缩放控制增益，默认 1；通常保持默认，不是速度比例

每次 tick 推进一个控制周期，不保证到达目标。调用方负责持续调度和取消；回调必须快速，磁盘写入、推理和图像编码放在循环外。非法参数抛 ValueError；预检、反馈、预测力矩或失能确认失败可能抛 RuntimeError，传输错误也会传播。

该接口不提供末端 XYZ 运动、逆运动学、轨迹规划或通用策略部署。内部 track 路径不是本页承诺兼容的公共控制 API，不能直接在普通 hold_arm 会话里调用。完整循环见示例教程。

openarm_runtime.calibrate_arm

calibrate_arm(profile, side, *, cancelled=..., record=..., publish=...,
              progress=..., allow_unreadable_mode=False)
# → dict[int, float]，ID 1–8 的硬件零位（写入前编码器坐标，rad）

同步执行真实运动、寻找止挡、到达计算零位、失能、写入并验证硬件零位。cancelled、record、publish 与控制接口含义相同；progress(message) 接收进度文字。默认取消回调返回 False，其余回调不处理。

必须先阅读硬件标定说明，提供取消机制和持久记录。失败可能已部分写入，不能自动回滚或盲目重试。此函数不是只读估计或主臂软件标定。

openarm_leader

LeaderBus

LeaderBus(path)
bus.sample()  # → openarm_leader.Sample
bus.read(mid, address, length)  # → bytes
bus.close()  # → None

path 为串口路径，推荐 /dev/serial/by-id/...。构造时打开 1 Mbps 串口并检查 ID 1–8 的 STS3215 型号。支持 with。sample() 顺序读取八个舵机，不是同时采样；read 是低层寄存器读取，参数为电机 ID、寄存器地址和字节长度，受到只读报文检查；常规业务优先 sample()。

通信失败、非电压单独报警或数据长度错误抛 OSError；型号不匹配或串口初始化失败可抛 RuntimeError。电压单独报警保留在样本中，不使有效读取失败。不写寄存器，不使能主臂。

LeaderReader

LeaderReader(path)
reader.start()  # → None；每个实例只能启动一次
reader.snapshot()  # → (sample_or_none, error, count)
reader.close()  # → None

支持 with，进入即启动。后台目标采样率 30 Hz，失败后约 1 秒重连；错误期间可能保留旧样本。使用 not error and sample is not None and sample.fresh(time.monotonic()) 检查。关闭最多等待 3 秒，未停止抛 RuntimeError；关闭后创建新实例。run() 为线程实现入口，不由应用直接调用。

Sample（主臂）

Sample(positions, torque, received_at, duration, voltages=(), errors=())
sample.fresh(now)  # → bool；0 <= 年龄 < 0.4 秒

不可变对象。positions 为 ID 1–8 原始编码器计数元组；torque 为使能状态元组，不是力矩；voltages 单位 V；errors 为故障位元组。received_at 是本轮采集起始单调时间，duration 是采集耗时秒数。

Mapping

Mapping(left, right, path)
mapping.device(side)  # → 内部设备标定字典；仅用于查看
mapping.capture(side, sample, kind, now)  # → None
mapping.angles(side, positions)  # → (angles, opening)
mapping.swap()  # → None
mapping.save()  # → None

left / right 为不同串口身份路径，path 为外部 JSON 标定文件。存在时加载并验证设备身份，不存在时创建未标定的内存状态；不打开硬件。

side 为 left/right。capture 的 kind 为 zero / open / closed，now=time.monotonic()；拒绝过期、已使能或夹爪端点过近的样本。只更新内存，调用 save() 才落盘。angles 接收八个原始计数，输出七个关节 rad 列表和 [0,1] 夹爪开度（未标定为 None）；关节顺序和符号由映射处理，不是直接按原始 ID 换单位。

未采集零位时使用 2048 显示基准，不能视为已标定。swap 交换左右映射后仍需保存；device 返回内部字典，不应直接修改。文件、身份、标定跨度错误会传播。本模块不负责驱动从臂跟随。

openarm_tactile

TactileReader(device=None, *, simulate=False, baudrate=921600, device_id=1)
reader.start()  # → self
reader.restart()  # → None；请求已启动实例重连
reader.snapshot()  # → dict
reader.close()  # → None
Sample(resultant, forces)
SENSOR  # "PXSR-STDIP03A"
POINT_COUNT  # 68

device 串口与 simulate=True 必须二选一；仿真也需要 tactile extra 的点位表。device_id 是 1–255 的板上模块 ID，项目验证设备用 2，不是串口编号。构造不采集，支持 with；start 不可重复，restart 只用于已经启动且未关闭的实例。失败后不自动重连，不自动切换仿真；close 最多等待 3 秒。

snapshot 字段	含义
state	waiting / live / stale / error / closed；仅 live 可作为当前样本
error	错误信息或 None
sample	Sample 或 None；失败可能保留旧值
sequence	样本序号，用于避免重复消费
age_s	最新样本年龄秒数或 None，超过 0.5 秒变 stale
rate_hz	采集速率估计；采集循环上限 60 Hz

Sample.resultant 为 XYZ 合力三元组，forces 为 68 个 XYZ 三元组；数值是原始计数，不是标定后的 N。Fx/Fy 有符号，Fz 无符号。两个部分来自独立读取事务，不保证严格同步。该 Sample 不含时间戳或 fresh 方法，使用快照的状态和年龄判断。

openarm_video：配置与帧

CameraConfig(name, device="", kind="rgb", width=640, height=480, fps=30,
             preview_width=None, preview_height=None, depth_helper=None, uri=None)
load_cameras(path)  # → tuple[CameraConfig, ...]
Frame(image, received_at, sequence=0, pixel_format="bgr8")
frame.fresh(now, max_age=3)  # → bool；0 <= 年龄 < max_age

name 以英文字母开头，只含字母、数字、下划线和连字符。kind 为 rgb/depth；RGB 的 device 为设备路径，深度的 depth_helper 为外部可执行文件，uri 为可选 OpenNI 设备标识。宽高为正整数，fps 为正有限数；深度要求宽高不超过 4096、fps 为不超过 120 的整数。

preview_width / preview_height 同时设置，只影响 HTTP 预览，不改变 Python 帧尺寸。配置本身不打开设备，也不保证硬件接受请求的分辨率，实际尺寸查看 frame.image.shape。load_cameras 读取 {"cameras": [...]} JSON。

Frame 字段	含义
image	RGB：H×W×3 uint8 BGR；深度：H×W uint16 毫米，0 无效
pixel_format	bgr8 / depth_mm
received_at	本机单调时间秒数，不是相机曝光时间
sequence	源或读取器提供的序号，不是跨相机同步序号

openarm_video：同步源与保存

OpenCVCamera(config)
OpenNIDepthCamera(config)
camera.read()  # → Frame
camera.close()  # → None
encode_frame(frame, *, raw=False, near=0.3, far=4.0)  # → (bytes, MIME字符串)
build_depth_helper(include_dir, library_dir, output, *, compiler="c++")  # → Path

两种相机均构造即打开、支持 with，同步 read() 失败直接抛出。RGB 需要明确的设备路径；深度需匹配的 OpenNI2 helper，深度读取初始化最多等待 25 秒，之后帧间等待 3 秒。同步源不自动重连，需重连时使用 CameraReader。设备已被占用可能抛 RuntimeError。

encode_frame 只编码，不写文件、不启动相机。RGB 输出 JPEG / image/jpeg；深度默认输出伪彩 PNG / image/png，near / far 单位米，需满足 0 <= near < far <= 20。深度 raw=True 输出保留毫米值的 16 位 PNG；RGB 不接受 raw=True。非法格式、形状或参数抛 ValueError，编码失败抛 RuntimeError。写文件使用 Path(...).write_bytes(data)，见配方。

build_depth_helper 只在 Linux 编译，不打开相机。include_dir 必须含 OpenNI.h；library_dir 必须含同 SDK 的 libOpenNI2.so 与 OpenNI2/Drivers/liborbbec.so；output 为输出路径。缺文件抛 FileNotFoundError，编译失败或超时传播 subprocess 异常。

openarm_video：后台采集

CameraReader(config, *, source_factory=None, stale_after=3,
             retry_interval=1, simulate=False)
reader.start()  # → self
reader.snapshot()  # → (frame_or_none, error, count)
reader.status()  # → dict
reader.close()  # → None

VideoCapture(configs, *, source_factory=None, stale_after=3, simulate=False)
capture.start()  # → self
capture.snapshot(name)  # → (frame_or_none, error, count)
capture.status()  # → {相机名: 状态字典}
capture.close()  # → None

构造不采集，支持 with。configs 是非空 CameraConfig 序列，名称唯一，RGB 不能重复选择同一设备。snapshot(name) 的名称不存在抛 KeyError。

stale_after、retry_interval 为正有限秒数；前者用于状态判断，后者用于单路重连间隔。simulate 和 source_factory 不能同时设置；自定义工厂接收 config，返回提供 read() -> Frame 与有界、可重复 close() 的对象，其生命周期交给 reader 管理。

start() 对未关闭的运行实例可重复调用；关闭后不能重启。正常读取错误自动重连，清理失败会停止并报告，必要时显式再次 close 重试。多相机 close 会尝试清理全部相机并汇总异常。

状态字典包含 status（Stopped / Starting / Online / Stale / Disconnected）、error、count、kind、simulation、width、height、frame_age_s。快照可能是旧帧；消费时检查错误、非 None、frame.fresh(now, stale_after)。count 可避免重复采样。不同相机独立采集，不保证帧对齐。

openarm_video：网页呈现

create_app(capture)  # → Flask app
serve(capture, host="127.0.0.1", port=5003)  # 阻塞服务

create_app 不启动或关闭 capture，由调用方管理生命周期；适用于接入已有 Flask 应用。serve 先绑定端口、启动 capture，阻塞提供服务，退出时关闭；适用于独立单进程服务。不要让多进程 Web worker 各自重复打开同一相机。

openarm_platform

load_config(path=None)  # → PlatformConfig
PlatformConfig(port, pages, log_dir, support=(), title="OpenArm Platform")
config.services  # → support + pages
config.with_port(port)  # → 新 PlatformConfig

推荐 load_config：读取外部 JSON 并验证；None 使用默认配置，默认平台端口 8888。相对路径按配置文件所在目录解析，无文件时按当前目录。端口须为 1024–65535 的整数，平台、服务及额外端口不能重复。配置错误抛 ValueError，文件及 JSON 错误传播。

pages / support 为服务描述元组；log_dir 为 Path；title 为标题。服务描述可读取 id、label、port、description、command、cwd、exclusive_group、extra_ports；public() 返回用于显示的字典。描述对象的内部类型不作为顶层公共构造入口，优先通过 JSON 配置。

with_port 返回验证后的副本，不修改原对象；services 不启动服务。直接构造 PlatformConfig 不等于已完成完整配置验证。平台 /api/* 是内部页面接口，不是稳定的机器人远程控制协议。

源码与兼容性

以 v0.6.0 源码 为依据。0.x 接入请固定标签。未在此列出的内部传输、跟随、界面和线程函数不视为稳定 SDK；包括 openarm_viewer 实现和策略执行适配层。此字典不代表硬件已完成实机验收，也不提供尚未实现的 episode 录制或训练框架适配 API。