Skip to content
AIRBOT P7 7-Axis Robotic Arm AIRBOT P7 7-Axis Robotic Arm
关节运动

关节运动#

move_joint() 接收 7 个关节的绝对目标角。它在 planning、Servo 和 MIT 模式下走不同的执行路径； 相同的 True 返回值也可能表示“规划完成”“目标到达”或“低层命令已接受”。开始运动前先选择模式， 并明确应用需要哪一种完成判据。

运动前检查#

任何关节目标都可能带动整条机械臂和末端工具。运行示例前逐项确认：

  1. 机械臂、工具、负载和线缆安装正确，工位运动范围内无人和障碍物；
  2. 现场人员能够立即操作实体急停；软件急停不能替代实体急停；
  3. ServiceState 可用，FSM 为 IDLE，7 轴关节状态和电机错误状态可读；
  4. 当前目标和到达路径适合工位，目标位于对应整机的关节限位内；
  5. 已与其他控制客户端协调，调用方显式取得控制权；
  6. 应用知道正常完成后如何切回 idle，也知道异常运动时由谁按下实体急停。

关节运动从检查状态、获取控制权到确认完成和异常处置的流程

流程中的 release_control() 只释放租约，不会停止已经接受的轨迹。只有确认命令已经完成或停止后， 才能把释放租约当作正常收尾。

set_arm_speed()#

client.set_arm_speed(arm_speed: list[float]) -> bool

该方法设置 SDK 保存的每关节速度数组，并向服务端更新 Servo 关节速度缩放。gRPC、DDS 均支持。

项目 规则
数组长度 必须等于运行时 arm_dof；当前 P7 为 7
单位 rad/s
每项范围 约 [0.5499, 7.8550] rad/s；必须是有限值，不能传 NaN 或无穷值
SDK 初始值 每轴 π/3 rad/s，约 1.0472 rad/s
越界行为 长度或任一有限元素越界时返回 False，不夹紧
公开超时 无；使用后端普通 RPC 等待设置

服务端要求有效控制租约，但此方法不会自动申请控制权。先显式调用 acquire_control()；没有租约、 服务拒绝更新或通信失败时返回 False。

当前 SDK 没有显式拒绝 NaN：比较范围时它可能通过本地校验，随后远端更新失败，而 SDK 本地数组已经 被改写。调用前应使用 math.isfinite() 检查 7 项；发生此类失败时关闭并重新创建客户端，不要继续 进入 MIT 模式。

该数组的作用取决于模式：

  • Servo：服务端根据 7 项速度的平均绝对值计算关节缩放，并继续受 Servo 运行配置限制；
  • MIT：7 项会作为低层请求的目标速度，而不只是上限；
  • planning：关节规划速度主要由 JointMoveOptions.velocity_scaling_factor 控制，不能用 set_arm_speed() 替代规划缩放参数。

SDK 在发送服务端更新前就保存新数组。如果远端更新失败，方法返回 False，但本地 MIT 目标速度仍 可能已经改变。失败后不要进入 MIT；确认控制权和服务状态，再次设置并得到 True，或关闭后重新创建 客户端。

move_joint()#

client.move_joint(
    pos: list[float],
    options: JointMoveOptions,
    timeout_ms: int = 1000,
) -> bool

gRPC、DDS 均支持。方法要求服务可用和有效控制权；缺少本地租约时 SDK 会按默认参数尝试申请一次。 运动程序应显式申请，以免隐藏多客户端协调失败。

参数 类型 单位 规则
pos list[float] rad 当前 P7 必须为 7 个有限的绝对关节角
options JointMoveOptions 见参数页 生效字段由当前模式决定
timeout_ms int ms 单次通信调用的等待上限，不是速度、规划时长或运动时长

长度错误或包含 NaN/无穷值时返回 False。目标超过 SDK 关节命令限位时不会拒绝，而是逐轴夹紧到 限位后发送，并为每个关节记录一次 warning。应用应在调用前自行拒绝越界目标；否则实际目标可能与 业务输入不同。

不同模式下的返回值#

当前模式 命令路径 blocking=False blocking=True
planning_control PTP/OMPL 规划与轨迹执行 等到规划结果；True 表示规划成功，轨迹可能仍在执行 等到轨迹完成判据；完成、规划或执行失败分别返回 True/False
servo_control 关节 Servo 目标 发送一帧持久流目标；True 表示该帧已写入/接受 等到目标误差满足服务端判据；成功返回 True
mit_control 位置、速度、前馈力矩、kp、kd 低层命令 服务端接受命令时返回 True blocking 被忽略，含义与左列相同
idle 或重力补偿 不支持 返回 False 返回 False

SDK 根据最近一次成功的模式切换维护本地 controller 缓存。发送前仍应读取 ServiceState.fsm_state；故障、急停或其他状态变化可能让实际 FSM 与本地缓存不同。

planning 分支会再次读取当前关节角作为规划起点。当前 SDK 在读取返回 None 或数组长度不等于 7 时返回 False，不再用全零关节角继续发送规划请求。该修复仍需随正式 SDK 版本完成整机回归；应用 仍应在调用前检查状态，并把返回 False 视为本次目标没有得到可靠完成确认。

关节命令限位#

下面是 SDK 在原始机械限位内缩 0.004 rad 后采用的命令限位。单位均为 rad。

关节 P7C V3 命令范围 P7C V4 命令范围
J1 [-2.96290, 2.96290] 同 V3
J2 [-2.57910, 0.83375] 同 V3
J3 [-2.96290, 2.96290] 同 V3
J4 [-2.43950, 0.17052] [-2.52670, 0.08326]
J5 [-2.96290, 2.96290] 同 V3
J6 [-0.78139, 0.78139] 同 V3
J7 [-1.56680, 1.21770] 同 V3

SDK 从 ArmFirmwareInfo.arm_sn 选择型号：序列号中的 revision code 为 10 且 family code 为 CW 或 CS 时使用 V4；序列号缺失、过短或不匹配时回退 V3。型号识别失败不会阻止运动，因此部署 前还应根据交付资料确认实际机械臂版本。

命令限位不是无碰撞工作空间。工具、负载、底座、线缆和外部设备仍可能在限位内发生碰撞。

规划类型#

planning 模式下，JointMoveOptions.motion_type 的公开取值为：

  • "ptp"：关节空间点到点运动；
  • "lin"：当前关节接口最终按 PTP 处理，不表示末端笛卡尔直线；
  • "ompl"：OMPL 关节路径搜索，并强制启用模型碰撞检查。

Servo 和 MIT 是控制模式，不是 motion_type。完整的规划、OMPL、seed、电流阈值和 MIT 字段见 JointMoveOptions 参数。

如果任务要求末端走直线或圆弧，使用笛卡尔运动中的 LIN/CIRCLE；如果需要 经过多个关节或末端目标，使用路点运动。

阻塞、超时和完成#

三个时间概念不能互换：

参数/状态 控制什么 到期后的含义
allow_planning_time 规划器单次计算最多使用多少秒 规划可能失败；不限制轨迹执行总时长
timeout_ms SDK 等待 route 回包的 deadline 返回 False 或发生传输超时；已经接受的命令不一定被撤销
blocking 回包前是否等待当前模式的完成判据 False 时需要应用自行追踪后续执行

planning + blocking=False 仍会等待规划结果，不是“请求写入后立即返回”。planning + blocking=True 的板端最长等待配置可超过 2 分钟，所以短 timeout_ms 可能让客户端先放弃等待，而 机械臂继续执行。Servo blocking 的板端默认等待上限为 30 秒；客户端 deadline 也应留出通信余量。

任何 False 或超时都不能单独证明机械臂没有开始运动。停止发送新目标，读取 FSM、关节状态和电机 状态；无法确认停止且存在异常运动风险时，执行实体急停流程。

保守的单次 Servo 示例#

下面的 safe_joint_motion.py 只支持 gRPC，并执行以下动作：

  • 要求初始 FSM 为 IDLE，关节和电机错误状态完整；
  • 打印目标并要求操作员输入 MOVE；
  • 显式获取控制权，切入 SERVO_CONTROL 并再次确认 FSM；
  • 把 Servo 速度设为每轴 0.55 rad/s；
  • 仅让 J7 向 0 rad 移动，最大变化 0.03 rad，其他 6 轴保持当前反馈；
  • 以 blocking=True 发送目标，读取最终反馈，再切回 idle 并释放租约。

小幅角度不等于无碰撞。脚本不包含工位几何模型，必须先完成本页的现场检查。完整代码默认折叠， 展开后可以直接复制保存：

完整代码：safe_joint_motion.py
safe_joint_motion.py
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
#!/usr/bin/env python3
"""Run one small, blocking J7 Servo move after explicit safety checks."""

from __future__ import annotations

import math
import argparse
import sys
import time
from typing import Any, Callable


class MotionCheckError(RuntimeError):
    """A required state, command, or cleanup check failed."""


def make_small_j7_target(angles: tuple[float, ...], step_rad: float = 0.03) -> list[float]:
    """Move J7 toward zero by no more than step_rad; keep all other joints fixed."""
    if len(angles) != 7 or not all(math.isfinite(value) for value in angles):
        raise MotionCheckError("expected seven finite arm joint angles")
    if not math.isfinite(step_rad) or step_rad <= 0.0 or step_rad > 0.03:
        raise ValueError("step_rad must be in (0, 0.03]")

    target = list(angles)
    current = target[6]
    if current > 0.0:
        target[6] = max(0.0, current - step_rad)
    elif current < 0.0:
        target[6] = min(0.0, current + step_rad)
    return target


def require_ready_snapshot(client: Any) -> tuple[float, ...]:
    """Require a fresh IDLE service snapshot and error-free arm feedback."""
    service = client.get_service_state()
    if service is None or not service.valid or not service.service_state:
        raise MotionCheckError("service state is unavailable or stale")
    if service.fsm_state != "IDLE":
        raise MotionCheckError(f"expected IDLE before motion, got {service.fsm_state}")

    joints = client.get_arm_joint_state()
    if joints is None:
        raise MotionCheckError("arm joint state is unavailable")
    angles = tuple(joints.angles)
    if len(angles) != 7 or not all(math.isfinite(value) for value in angles):
        raise MotionCheckError("arm joint state must contain seven finite angles")

    motors = client.get_arm_motor_state()
    if motors is None or len(motors.error_ids) != 7:
        raise MotionCheckError("arm motor error state is unavailable or incomplete")
    if any(code != 0 for code in motors.error_ids):
        raise MotionCheckError(f"arm motor errors are present: {motors.error_ids}")
    return angles


def wait_for_fsm(
    client: Any,
    expected: str,
    *,
    timeout_s: float = 2.0,
    poll_interval_s: float = 0.1,
    monotonic: Callable[[], float] = time.monotonic,
    sleep: Callable[[float], None] = time.sleep,
) -> None:
    """Wait for the asynchronous service-state cache to report one FSM state."""
    deadline = monotonic() + timeout_s
    while monotonic() < deadline:
        state = client.get_service_state()
        if state is not None and state.service_state and state.fsm_state == expected:
            return
        sleep(poll_interval_s)
    raise MotionCheckError(f"FSM did not report {expected} before the timeout")


def run_small_servo_move(
    client: Any,
    *,
    controller: Any,
    options_factory: Callable[..., Any],
    confirm: Callable[[list[float]], bool],
    wait_for_state: Callable[[Any, str], None] = wait_for_fsm,
) -> list[float]:
    """Execute one confirmed J7 target and always release the control lease."""
    current = require_ready_snapshot(client)
    target = make_small_j7_target(current)
    if not confirm(target):
        raise MotionCheckError("motion was not confirmed")

    acquired = False
    servo_selected = False
    try:
        acquired = client.acquire_control(lease_ms=15_000, renew_period_s=5.0)
        if not acquired:
            raise MotionCheckError("control lease was not acquired")

        if not client.switch_controller(controller.servo_control, timeout_ms=5_000):
            raise MotionCheckError("SERVO_CONTROL switch failed")
        servo_selected = True
        wait_for_state(client, "SERVO_CONTROL")

        if not client.set_arm_speed([0.55] * 7):
            raise MotionCheckError("Servo speed update failed")

        options = options_factory(blocking=True)
        if not client.move_joint(target, options, timeout_ms=35_000):
            raise MotionCheckError("blocking Servo move failed or timed out")

        final_state = client.get_arm_joint_state()
        if final_state is None or len(final_state.angles) != 7:
            raise MotionCheckError("final arm joint state is unavailable")
        if abs(final_state.angles[6] - target[6]) > 0.02:
            raise MotionCheckError("J7 feedback is not within the example check tolerance")
        return target
    finally:
        idle_cleanup_ok = not servo_selected
        if servo_selected:
            try:
                idle_cleanup_ok = bool(
                    client.switch_controller(controller.idle, timeout_ms=5_000)
                )
            except Exception:
                idle_cleanup_ok = False
        if acquired:
            client.release_control()
        if not idle_cleanup_ok:
            message = (
                "cleanup could not confirm IDLE; keep the workcell guarded and "
                "inspect the current FSM state"
            )
            active_error = sys.exc_info()[1]
            if active_error is not None and hasattr(active_error, "add_note"):
                active_error.add_note(message)
            else:
                raise MotionCheckError(message)


def main() -> None:
    from arm_p7_sdk import AirbotClient, Controller, JointMoveOptions

    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--host", required=True)
    parser.add_argument("--port", type=int, default=50071)
    args = parser.parse_args()

    def confirm(target: list[float]) -> bool:
        print("目标关节角（rad）:", target)
        answer = input("确认现场已清空、实体急停可用；输入 MOVE 执行: ")
        return answer == "MOVE"

    with AirbotClient(
        host=args.host,
        port=args.port,
        backend="grpc",
        client_name="docs-safe-joint-motion",
    ) as client:
        target = run_small_servo_move(
            client,
            controller=Controller,
            options_factory=JointMoveOptions,
            confirm=confirm,
        )
        print("命令完成，目标（rad）:", target)


if __name__ == "__main__":
    main()

保存代码后，把目标地址和端口作为命令行参数传入：

python safe_joint_motion.py --host P7_IP_ADDRESS --port 50071

预期先打印 7 个目标角和确认提示。只有输入完全匹配的 MOVE 才会申请控制权并发出目标；成功后打印 目标数组。示例中的 0.02 rad 反馈差值仅是脚本的保守结果复核阈值，不是产品精度承诺。

示例已经通过 fake client 测试，验证了未确认时不申请控制权、成功和失败路径都尝试切回 idle 并 释放租约。它没有在真实机械臂、具体工具或具体工位上执行；首次真机运行必须由负责该工位安全的人员 复核目标和停止流程。

停止和异常处理#

正常结束#

  1. 使用阻塞调用，或由应用确认非阻塞任务已经完成；
  2. 读取关节状态和 ServiceState；
  3. 调用 switch_controller(Controller.idle) 并确认 fsm_state == "IDLE"；
  4. 调用 release_control()，最后关闭客户端。

返回 False、超时或状态异常#

  1. 立即停止产生和发送新目标；
  2. 不要把 release_control() 当作停止命令；它无法撤销已接受的规划轨迹；
  3. 如果运动正常且没有即时危险，保持现场警戒，读取 FSM、关节和电机状态，确认任务是否仍在执行；
  4. 出现意外运动、人员侵入或失控风险时，由现场人员按下实体急停；
  5. 运动已经停止且通信可用时，尝试切回 idle，再释放控制权；
  6. 保存目标、返回值、状态、错误码、SDK/Arm App 版本和日志，再按 安全与恢复命令处理。

软件急停可以作为仍可通信时的附加停止请求，但不能代替上述实体急停动作。急停复位后，旧 Servo session 和旧控制模式不会恢复；必须从最新状态重新开始。

模式选择、规划期碰撞检查和 planning/Servo 的完整边界见 运动规划。move_joint() 和关节路点的同名运行入口见 SDK 最小可执行用例。