状态读取参考

状态读取参考#

八个状态 getter 都是只读操作，不申请控制权、不切换模式，也不触发运动。它们返回当前可获得的快照；除了 ServiceState.valid，返回模型不带时间戳或样本年龄，不能用于证明实时性。

gRPC 与 DDS 的读取方式#

行为	gRPC	DDS
`get_service_state()`	SDK 后台每 0.2 秒发起状态 RPC；连续失败会退避，最长 5 秒	SDK 后台每 1 秒读取 route 状态流
其余遥测 getter	每次同步 RPC，读取板端 route 的最新缓存	非阻塞地清空本地 Reader，返回 depth 1 状态流的最新样本
首帧未到	RPC 不可用后返回 `None`	返回 `None`
传输失败	默认对可重试同步 RPC 重试一次；最终失败记录日志并返回 `None`	返回已经缓存的最后样本；从未收到样本时返回 `None`
样本年龄	route 回包不包含源时间戳，无法从模型判断	SDK 缓存不记录接收时间，无法从模型判断
控制权	不需要	不需要

gRPC getter 没有公开 timeout_ms 参数，普通状态 RPC 的单次默认 deadline 为 3 秒；默认重试一次时，一次 getter 在故障中可能等待约两个 deadline 加重试间隔。DDS 遥测 getter 本身不等待新样本，但已缓存样本在连接中断后仍可能继续返回。两种后端都不能把循环调用频率当作传感器更新频率。

`get_service_state()`#

client.get_service_state() -> ServiceState | None

gRPC 和 DDS 均支持。它读取 SDK 后台维护的服务状态缓存，不在调用线程发起新 RPC。后台尚未写入首个结果时返回 None；之后返回 ServiceState。

valid 只根据缓存最后更新时间和 0.5 秒 TTL 计算，不代表 service_state=True。必须分别检查：

state = client.get_service_state()
if state is None:
    print("service-state cache has no first sample")
elif not state.valid:
    print("service-state cache is older than 0.5 s")
elif not state.service_state:
    print("service is unavailable")
else:
    print(state.fsm_state, state.controller_state)

当前 DDS 后台轮询周期为 1 秒，但缓存 TTL 为 0.5 秒，因此健康连接下 valid 也可能在相邻轮询之间短暂变为 False。连续诊断时应继续采样并同时检查 service_state；不要把单次 DDS valid=False 等同于服务已经停止。该差异不影响其他 getter，因为它们没有 valid 字段。

状态 RPC 失败时，后台缓存会写入 service_state=False 以及 ERROR 状态名。持续失败会增加轮询间隔，恢复后重新写入服务端状态。

`get_arm_joint_state()`#

client.get_arm_joint_state() -> ArmJointState | None

gRPC 和 DDS 均支持，成功返回 ArmJointState。当前 P7 返回前 7 个机械臂关节的位置、速度和力矩反馈。

以下情况返回 None：状态尚未到达、传输最终失败、三个数组长度不一致，或返回关节数少于 7。 DDS 可能在通信中断后继续返回最后缓存；gRPC 虽然每次调用 route RPC，route 回包本身仍是板端缓存，且模型没有时间戳。

`get_arm_motor_state()`#

client.get_arm_motor_state() -> ArmMotorState | None

gRPC 和 DDS 均支持，成功返回 ArmMotorState。它用于查看各机械臂电机温度和错误码，不提供关节位置。

状态未到、传输最终失败或温度/错误码数组长度不一致时返回 None。上游明确返回两个空数组时，SDK 返回包含空元组的模型，而不是 None；数组长度不是 7 时会记录警告，但仍返回收到的数据。

`get_eef_joint_state()`#

client.get_eef_joint_state() -> EEFJointState | None

gRPC 和 DDS 均支持，成功返回 EEFJointState。它从组合 JointState 中取机械臂 7 轴之后的末端执行器（End Effector, EEF）关节。

状态未到、三个数组长度不一致、运行时没有 EEF 自由度或数组不足以覆盖 EEF 时返回 None。返回数组长度由运行时 eef_dof 决定；固件信息不可用时，SDK 尝试从总关节数减 7 推断。

对于 G2、G2L、G2P 等线性 EEF，SDK API 的单位已经统一：eef_pos 使用 mm，eef_vel 使用 mm/s。gRPC 和 DDS route 之间仍使用 m、m/s 传输，SDK 在读写两端负责换算；应用不应再乘或除 1000。这项契约来自 sdk_client 18e904285779537b2fbbc3ae672e7112e1b51caf，正式 SDK/Arm App 组合和真实 EEF 行程、速度仍需按版本管理中的发布流程验证。

eef_eff 保留 EEF 驱动原生的 effort/current-like 数值，SDK 不把它换算为 A、N 或 N·m。旋转或自定义 EEF 在协议提供型号级 actuator/unit 元数据前，不应套用线性 EEF 的位置、速度单位。

`get_eef_motor_state()`#

client.get_eef_motor_state() -> EEFMotorState | None

gRPC 和 DDS 均支持，成功返回 EEFMotorState。它只包含 EEF 电机温度和错误码，与 EEF 位置/速度分开。

状态未到、传输最终失败、数组长度不一致或数组为空时返回 None。长度取决于已安装 EEF，不固定为 1。

`get_imu_state()`#

client.get_imu_state() -> ImuState | None

gRPC 和 DDS 均支持，成功返回 ImuState。两个数组都必须是固定 xyz 三元组；状态未到、传输最终失败或长度错误时返回 None。

SDK 返回值保留 arm_imu 源数据的 SI 单位，但不包含原消息的 frame_id、时间戳或协方差。把数据与其他坐标系组合前，应从整机配置确认 IMU 安装方向和坐标变换。

`get_end_pose()`#

client.get_end_pose() -> CartesianPose | None

gRPC 和 DDS 均支持，成功返回 CartesianPose。该位姿由当前关节状态经过正运动学（Forward Kinematics, FK）计算，表示末端相对 base_link 的位置和四元数姿态。

状态未到、传输最终失败或返回数组不是 7 个元素时返回 None。板端 FK 超时会保留上一帧位姿缓存，返回模型又没有时间戳，因此一次非 None 结果不证明它与最新关节状态同步。

`get_firmware_info()`#

client.get_firmware_info() -> ArmFirmwareInfo | None

gRPC 和 DDS 均支持，返回 ArmFirmwareInfo。SDK 只在客户端构造期间请求一次运行时信息，之后该 getter 直接返回本地缓存，不发起 RPC，也不会自动刷新。

构造时请求失败、服务报告失败或字段不足时可能返回 None。同一客户端上重复调用不会重试；先确认服务已就绪，关闭当前客户端并重新创建。硬件在客户端存活期间更换或服务重启后，也要重新创建客户端才能刷新这组信息。

连续只读诊断#

下面的完整脚本连续读取全部八类状态，不调用 acquire_control() 或任何控制接口。代码同时支持 gRPC 和 DDS；复制到本地文件即可运行，不需要打开文档仓库查找实现。

展开完整的 read_sdk_state.py

read_sdk_state.py
#!/usr/bin/env python3
"""Continuously print SDK state without acquiring control."""

from __future__ import annotations

import argparse
import time
from collections.abc import Sequence
from typing import Any

from arm_p7_sdk import AirbotClient


def collect_state(client: Any) -> dict[str, Any]:
    """Read one diagnostic snapshot; missing data remains ``None``."""
    return {
        "get_service_state": client.get_service_state(),
        "get_arm_joint_state": client.get_arm_joint_state(),
        "get_arm_motor_state": client.get_arm_motor_state(),
        "get_eef_joint_state": client.get_eef_joint_state(),
        "get_eef_motor_state": client.get_eef_motor_state(),
        "get_imu_state": client.get_imu_state(),
        "get_end_pose": client.get_end_pose(),
        "get_firmware_info": client.get_firmware_info(),
    }


def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--backend", choices=("grpc", "dds"), default="grpc")
    parser.add_argument(
        "--host", help="P7 gRPC host; required for --backend grpc"
    )
    parser.add_argument("--port", type=int, default=50071)
    parser.add_argument(
        "--domain-id", type=int, help="DDS Domain ID; required for --backend dds"
    )
    parser.add_argument("--side", choices=("none", "left", "right"), default="none")
    parser.add_argument(
        "--interval", type=float, default=0.5, help="Seconds between snapshots"
    )
    return parser.parse_args(argv)


def client_options_from_args(args: argparse.Namespace) -> dict[str, Any]:
    """Build explicit gRPC or DDS options from command-line arguments."""
    backend = args.backend
    if backend == "dds":
        if args.domain_id is None:
            raise ValueError("--domain-id is required for --backend dds")
        return {
            "backend": "dds",
            "domain_id": args.domain_id,
            "side": args.side,
        }
    if not args.host:
        raise ValueError("--host is required for --backend grpc")
    return {
        "backend": "grpc",
        "host": args.host,
        "port": args.port,
    }


def print_snapshot(snapshot: dict[str, Any]) -> None:
    """Print an observation time; it is not a sensor timestamp."""
    print(f"observed_monotonic_s={time.monotonic():.3f}")
    for name, value in snapshot.items():
        print(f"  {name}: {value}")


def main() -> None:
    try:
        args = parse_args()
        with AirbotClient(**client_options_from_args(args)) as client:
            while True:
                print_snapshot(collect_state(client))
                time.sleep(args.interval)
    except KeyboardInterrupt:
        print("stopped by user; client resources were closed")
    except ValueError as error:
        raise SystemExit(f"configuration failed: {error}") from error
    except ConnectionError as error:
        raise SystemExit(f"connection failed: {error}") from error


if __name__ == "__main__":
    main()

通过命令行显式传入 gRPC 地址后运行：

python read_sdk_state.py --backend grpc --host P7_IP_ADDRESS --port 50071

DDS 环境显式传入 Domain 和 side：

python read_sdk_state.py --backend dds --domain-id DOMAIN_FROM_P7_CONFIG --side none

按 ++ctrl+c++ 退出。with 会在正常退出和异常退出时关闭客户端，打印 None 而不是把缺失样本当作零值。采样间隔是应用观察节拍，不是实时周期：gRPC getter 可能阻塞并重试，DDS getter 可能重复返回同一缓存。

生产监控还应记录单调时钟和目标软件版本；当前模型不包含每个源样本的时间戳，不能据此构建需要确定时限的闭环控制。

读取失败后的顺序#

先检查 get_service_state() 是否为 None、陈旧或 service_state=False。
gRPC 检查地址、端口和 route 日志；DDS 检查 Domain ID、side、CORA 版本和发现状态。
单个 getter 持续为 None 时，记录 getter 名、SDK/Arm App/CORA 版本、后端和发生时间。
不要用空数组或全零模型替代 None，也不要无限高频重试。

字段解释见状态数据模型，连接层排查见连接检查和DDS QoS 与数据新鲜度。需要单独执行某个 getter 时，使用 SDK 最小可执行用例中的同名命令。

状态读取参考#

gRPC 与 DDS 的读取方式#

get_service_state()#

get_arm_joint_state()#

get_arm_motor_state()#

get_eef_joint_state()#

get_eef_motor_state()#

get_imu_state()#

get_end_pose()#

get_firmware_info()#