orisys/ros_contorl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

注释与提示信息全面中文化：源码 66 个 .py + 脚本/配置 45 个文件

Browse Source 
背景：现场读代码的人只读中文。此前代码里的 docstring 是英文、行内注释是中文，
一半英文一半中文反而更难看懂 —— 新手打开一个 .py 不知道这个模块为什么存在。

范围（本仓库自己的代码，厂商上游不动）
  src/arm_application/**   37 个文件：servo_node / perception_node / handeye_* / dh_pgea /
                           yolo_detector / kalman_filter / goto_ee_plane / dual_gripper …
  src/arm_interfaces/**    11 个文件：paths / manifest / loader / config_guard / errors …
  src/arm_control/**        5 个文件
  src/arm_bringup/**        3 个 launch
  scripts/**                6 个 .py + 29 个 .sh（scripts/ 是"唯一操作入口"，用户屏幕上
                           看到的就是这些提示，所以提示信息也一并中文化）
  config/ handeye/ devices/ src/arm_*/config/   17 个 .yaml/.xml/.xacro/.cfg

口径（写进 docs/FRAMEWORK_RULES.md 新增的 R6.5，作为长期约束）
  注释、docstring、print/日志/异常里给人看的句子 = 中文；
  标识符、API 名、话题名、参数名、字段名、命令名 = 保留英文原文（用 `` `` 包起来）；
  单位、十六进制、协议报文、CRC 值、命令示例 = 原样保留；
  src/flexiv_*/ = 厂商上游代码，不许改也不许翻译（R4.4）。
  翻译保原意与原有详细程度：那些"为什么不能用另一种做法""踩过什么坑"的说明
  全部完整保留 —— 它们才是这份代码里最贵的部分。

新增 scripts/check_text_only.py：只改文字时的自证工具
  人工 review 靠不住（115 个文件、3000 多行 diff）。这个脚本把改动前后的源码解析成
  语法树，删掉 docstring、抹平字符串常量、把 f-string 只留"插值了几次 + 插值是什么"，
  再比对：完全相同 = 改动范围内没有逻辑变化。
  留了一个口子：f-string 的插值按内容排序后再比 —— 因为中文语序常常要和英文反着来
  （f"on {port} read {idx} failed" → f"在 {port} 上读 {idx} 失败"），只换显示顺序不该判失败；
  但多插一个、少插一个、换掉被插入的表达式，立刻会报出来。
  本次 66 个 .py 全部通过该闸门。

验收
  改动的 66 个 .py：scripts/check_text_only.py 全部通过（零逻辑变化）
  45 个非 .py 文件：去注释后与 HEAD 逐字节相同（零配置变化）
  docstring 总数：没有一个文件减少（不存在"翻译时顺手删了说明"）
  单测：arm_application 233 / arm_driver_flexiv 69 / arm_interfaces 46 /
        arm_kinematics 24 / arm_sim_gazebo 236 = 608 passed，与改动前一致
  全部 .sh：bash -n 通过；并额外实际执行了 --help 验证（见下一个提交：
        bash -n 对 CRLF 脚本会漏报，只有真跑才暴露）

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
arm_dev
2026-09-23 17:40:55 +08:00
parent d78e66ce98
commit 72353f3773
115 changed files with 1758 additions and 1573 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/arm_interfaces/arm_interfaces/__init__.py
+10 -10
View File
@@ -1,14 +1,14 @@
"""Device contract layer for the robotic arm stack.
"""机械臂栈的设备契约层。
Public entry points::
公开入口::
from arm_interfaces import load_device_stack, check_requirement
stack = load_device_stack() # raises ContractError with a fix hint
check_requirement(stack, "handeye") # raises CapabilityError if unmet
stack = load_device_stack() # 失败时抛出带修复提示的 ContractError
check_requirement(stack, "handeye") # 能力不满足时抛出 CapabilityError
Everything device-selection related funnels through here so that the rest of the
framework never reads ``devices/`` or guesses at ``site.yaml`` structure itself.
所有与设备选型相关的事都汇集到这里，框架其余部分因此永远不必自己读
``devices/``，也不必猜测 ``site.yaml`` 的结构。
"""
from .capabilities import (
@@ -42,7 +42,7 @@ from .registry import Registry, list_available_brands, load_registry
__version__ = "0.1.0"
__all__ = [
# loading
# 加载
"load_device_stack",
"load_registry",
"load_manifest",
@@ -54,18 +54,18 @@ __all__ = [
"effective_capabilities",
"list_available_brands",
"manifest_path",
# types
# 类型
"DeviceStack",
"Manifest",
"Registry",
# capabilities
# 能力
"ALL_CAPABILITIES",
"CAPABILITIES_BY_KIND",
"REQUIREMENTS",
"ROS_INTERFACE",
"capabilities_for_kind",
"describe",
# errors
# 错误
"ContractError",
"RegistryError",
"ManifestError",
src/arm_interfaces/arm_interfaces/capabilities.py
+31 -35
View File
@@ -1,23 +1,19 @@
"""Capability vocabulary and the requirement sets upper layers depend on.
"""能力词表，以及上层依赖的能力需求集。
Two distinct ideas live here, and keeping them separate matters:
这里有彼此不同的两个概念，把它们分开很重要：
``Capability``
What a *driver* can do. Declared by each brand's ``manifest.yaml``. Adding a
new brand never changes this file — only edit it when there is genuinely a
new kind of motion or sensing to express.
*驱动*能做什么。由每个品牌的 ``manifest.yaml`` 声明。新增品牌从不改动
本文件——只有当确实要表达一种新的运动或感知方式时才编辑它。
``Requirement``
What an *upper layer* needs. A flow (tracking, hand-eye calibration, the
through-gauge assembly state machine) declares the set of capabilities it
will call. The contract layer checks the selected drivers satisfy it at
startup, so a missing capability fails immediately with a clear message
instead of failing mid-motion.
*上层*需要什么。一个流程（视觉跟踪、手眼标定、通止规装配状态机）声明
它将要调用的能力集合。契约层在启动时检查所选驱动是否满足它，因此缺失
能力会立刻报出清晰的消息，而不是在运动中途才失败。
The ``ROS_INTERFACE`` table maps each capability to the concrete ROS 2 endpoint
that implements it. It exists so error messages can name the interface an
operator should check, and so future drivers know exactly what to publish
or serve.
``ROS_INTERFACE`` 表把每项能力映射到实现它的具体 ROS 2 端点。它的存在，
一是让错误消息能点名操作员该检查哪个接口，二是让未来的驱动清楚自己该
发布或提供什么。
"""
from __future__ import annotations
@@ -26,17 +22,17 @@ from typing import Dict, FrozenSet, NamedTuple, Tuple
class CapabilityInfo(NamedTuple):
"""Documentation for one capability."""
"""对某一项能力的说明。"""
owner: str
"""Which device kind provides it: robot | gripper | sensor | camera."""
"""由哪类设备提供：robot | gripper | sensor | camera。"""
summary: str
"""One-line description of what the capability does."""
"""一句话说明该能力做什么。"""
# ---------------------------------------------------------------------------
# Robot capabilities
# 机器人能力
# ---------------------------------------------------------------------------
ROBOT_CAPABILITIES: Dict[str, CapabilityInfo] = {
"get_arm_state": CapabilityInfo(
@@ -72,7 +68,7 @@ ROBOT_CAPABILITIES: Dict[str, CapabilityInfo] = {
}
# ---------------------------------------------------------------------------
# Gripper capabilities
# 夹爪能力
# ---------------------------------------------------------------------------
GRIPPER_CAPABILITIES: Dict[str, CapabilityInfo] = {
"open": CapabilityInfo("gripper", "张开。"),
@@ -85,7 +81,7 @@ GRIPPER_CAPABILITIES: Dict[str, CapabilityInfo] = {
}
# ---------------------------------------------------------------------------
# Sensor capabilities
# 传感器能力
# ---------------------------------------------------------------------------
SENSOR_CAPABILITIES: Dict[str, CapabilityInfo] = {
"read_tangential": CapabilityInfo(
@@ -100,7 +96,7 @@ SENSOR_CAPABILITIES: Dict[str, CapabilityInfo] = {
}
# ---------------------------------------------------------------------------
# Camera capabilities
# 相机能力
# ---------------------------------------------------------------------------
CAMERA_CAPABILITIES: Dict[str, CapabilityInfo] = {
"color": CapabilityInfo("camera", "彩色图。推理与跟踪依赖它。"),
@@ -116,7 +112,7 @@ CAMERA_CAPABILITIES: Dict[str, CapabilityInfo] = {
"set_exposure": CapabilityInfo("camera", "设置曝光。跟踪需要限制曝光避免运动拖影。"),
}
#: Every capability name known to the framework, tagged with its device kind.
#: 框架已知的全部能力名，并标注其设备类别。
ALL_CAPABILITIES: Dict[str, CapabilityInfo] = {
**ROBOT_CAPABILITIES,
**GRIPPER_CAPABILITIES,
@@ -124,7 +120,7 @@ ALL_CAPABILITIES: Dict[str, CapabilityInfo] = {
**CAMERA_CAPABILITIES,
}
#: Device kind -> the capability names that kind is allowed to declare.
#: 设备类别 -> 该类设备允许声明的能力名。
CAPABILITIES_BY_KIND: Dict[str, FrozenSet[str]] = {
"robot": frozenset(ROBOT_CAPABILITIES),
"gripper": frozenset(GRIPPER_CAPABILITIES),
@@ -134,10 +130,10 @@ CAPABILITIES_BY_KIND: Dict[str, FrozenSet[str]] = {
# ---------------------------------------------------------------------------
# ROS 2 interface mapping
# ROS 2 接口映射
# ---------------------------------------------------------------------------
class RosInterface(NamedTuple):
"""The concrete ROS 2 endpoint implementing a capability."""
"""实现某项能力的具体 ROS 2 端点。"""
namespace: str
name: str
@@ -145,10 +141,10 @@ class RosInterface(NamedTuple):
type: str
#: capability -> ROS 2 endpoint. Names are relative to the device namespace and
#: are resolved per-instance (two arms never share a namespace).
#: capability -> ROS 2 端点。名称相对于设备命名空间，且按实例解析
#:（两条机械臂绝不共用命名空间）。
ROS_INTERFACE: Dict[str, RosInterface] = {
# --- robot ---
# --- 机器人 ---
"get_arm_state": RosInterface("", "get_arm_state", "service", "arm_msgs/srv/GetArmState"),
"follow_joint_trajectory": RosInterface(
"", "follow_joint_trajectory", "action", "arm_msgs/action/FollowJointTrajectory"
@@ -164,19 +160,19 @@ ROS_INTERFACE: Dict[str, RosInterface] = {
"enable": RosInterface("", "enable", "service", "std_srvs/srv/SetBool"),
"clear_fault": RosInterface("", "clear_fault", "service", "std_srvs/srv/Trigger"),
"soft_drag": RosInterface("", "soft_drag", "service", "std_srvs/srv/SetBool"),
# --- gripper ---
# --- 夹爪 ---
"open": RosInterface("", "open", "service", "arm_msgs/srv/GripperExecute"),
"close": RosInterface("", "close", "service", "arm_msgs/srv/GripperExecute"),
"move": RosInterface("", "move", "service", "arm_msgs/srv/GripperExecute"),
"rotate": RosInterface("", "rotate", "service", "arm_msgs/srv/GripperExecute"),
"get_status": RosInterface("", "status", "topic", "arm_msgs/msg/GripperStatus"),
# --- sensor ---
# --- 传感器 ---
"read_tangential": RosInterface("", "tangential", "topic", "sensor_msgs/msg/Float32MultiArray"),
"read_normal": RosInterface("", "normal", "topic", "sensor_msgs/msg/Float32"),
"read_depth": RosInterface("", "depth", "topic", "sensor_msgs/msg/Image"),
"read_force_vector": RosInterface("", "force_vector", "topic", "sensor_msgs/msg/Image"),
"measure_baseline": RosInterface("", "measure_baseline", "service", "std_srvs/srv/Trigger"),
# --- camera ---
# --- 相机 ---
"color": RosInterface("", "color/image_raw", "topic", "sensor_msgs/msg/Image"),
"depth": RosInterface("", "depth/image_raw", "topic", "sensor_msgs/msg/Image"),
"aligned_depth_to_color": RosInterface(
@@ -192,10 +188,10 @@ ROS_INTERFACE: Dict[str, RosInterface] = {
# ---------------------------------------------------------------------------
# Requirement sets
# 需求集
# ---------------------------------------------------------------------------
class Requirement(NamedTuple):
"""A named flow and the capabilities it needs, grouped by device kind."""
"""一个具名流程及其所需能力，按设备类别分组。"""
summary: str
robot: Tuple[str, ...]
@@ -261,12 +257,12 @@ REQUIREMENTS: Dict[str, Requirement] = {
def capabilities_for_kind(kind: str) -> FrozenSet[str]:
"""Return the set of capability names valid for a device kind."""
"""返回某类设备合法的能力名集合。"""
return CAPABILITIES_BY_KIND.get(kind, frozenset())
def describe(capability: str) -> str:
"""Human-readable description used in error messages."""
"""用于错误消息的、给人看的说明。"""
info = ALL_CAPABILITIES.get(capability)
if info is None:
return f"{capability}（框架未登记的能力名）"
src/arm_interfaces/arm_interfaces/cli.py
+9 -9
View File
@@ -1,19 +1,19 @@
"""``device_check`` — validate the device stack from the command line.
"""``device_check`` —— 从命令行校验设备栈。
Usage::
用法::
ros2 run arm_interfaces device_check # load everything, print summary
ros2 run arm_interfaces device_check # 全量加载并打印摘要
ros2 run arm_interfaces device_check --requirement handeye
ros2 run arm_interfaces device_check --json
ros2 run arm_interfaces device_check --list-brands
ros2 run arm_interfaces device_check --no-assets # skip vendor asset checks
ros2 run arm_interfaces device_check --no-assets # 跳过厂商物料检查
Exit codes are meaningful so scripts can branch on them:
退出码有明确含义，脚本可据此分支：
=== ==========================================================
0 contract satisfied
2 contract violation (missing brand, config, asset, capability)
3 contract layer itself broke (unexpected exception)
0 契约满足
2 契约违规（缺少品牌、配置、物料或能力）
3 契约层自身出错（意料之外的异常）
=== ==========================================================
"""
@@ -119,7 +119,7 @@ def main(argv: Optional[List[str]] = None) -> int:
print(format_error(err), file=sys.stderr)
return EXIT_CONTRACT
except Exception as exc: # noqa: BLE001 - the CLI is the last line of defence
except Exception as exc: # noqa: BLE001 - CLI 是最后一道防线
print(
f"契约层内部错误（这本身是框架 bug，请上报）: {type(exc).__name__}: {exc}",
file=sys.stderr,
src/arm_interfaces/arm_interfaces/config_guard.py
+44 -52
View File
@@ -1,20 +1,19 @@
"""Validate ``config/site.yaml`` against a manifest's declared ``config_schema``.
"""按清单声明的 ``config_schema`` 校验 ``config/site.yaml``。
Two classes of bug motivated this module:
有两类 bug 促成了本模块：
1. **Silent no-ops.** ``run_handeye.sh`` passed five parameters
(``settle_qd_max``, ``pin_holds``, ...) to a node that never declared them.
YAML accepted the values, nothing read them, tuning had no effect.
2. **Unit confusion.** A value typed in millimetres but interpreted as metres
produces a plausible-looking but wrong motion.
1. **静默失效。** ``run_handeye.sh`` 把五个参数
（``settle_qd_max``、``pin_holds``……）传给了一个从未声明它们的节点。
YAML 接受了这些值，却没人读它们，调参毫无效果。
2. **单位混淆。** 按毫米输入却被当成米解释的值，会产生看似合理但错误的
运动。
The guard addresses both: every declared field is checked for presence and type,
and ``unit`` is echoed in error messages so the operator sees the expected unit
next to the value they wrote.
配置守卫同时解决这两点：每个已声明的字段都会检查是否存在、类型是否正确，
并且 ``unit`` 会回显在错误消息里，让操作员在写下的值旁边就能看到期望的单位。
``site_path`` supports a list-expansion token so per-instance config can be
declared once: ``robot.robots[].ip`` resolves to ``robot.robots[0].ip`` and
``robot.robots[1].ip`` and requires *every* instance to supply the value.
``site_path`` 支持列表展开记号，使按实例的配置只需声明一次：
``robot.robots[].ip`` 会解析为 ``robot.robots[0].ip`` 和
``robot.robots[1].ip``，并要求*每一个*实例都提供该值。
"""
from __future__ import annotations
@@ -27,9 +26,8 @@ import yaml
from .errors import BadConfigValueError, MissingConfigError
from .manifest import ConfigField, Manifest
#: Python types accepted for each declared type name. ``bool`` is deliberately
#: excluded from the numeric types: in YAML ``true`` is a bool, and silently
#: using it as 1.0 has caused real bugs.
#: 每个已声明类型名所接受的 Python 类型。``bool`` 被刻意排除在数值类型之外：
#: 在 YAML 里 ``true`` 是布尔值，把它悄悄当作 1.0 用曾引发过真实 bug。
_TYPE_CHECKS = {
"str": (str,),
"int": (int,),
@@ -41,20 +39,19 @@ _TYPE_CHECKS = {
class ResolvedValue(NamedTuple):
"""One concrete value found in ``site.yaml`` for a declared field."""
"""在 ``site.yaml`` 中为某个已声明字段找到的一个具体值。"""
path: str
"""Concrete path, e.g. ``robot.robots[1].ip``."""
"""具体路径，例如 ``robot.robots[1].ip``。"""
value: Any
def load_site(path: Path | None = None) -> Dict[str, Any]:
"""Read ``config/site.yaml``.
"""读取 ``config/site.yaml``。
Returns an empty dict when the file is absent, matching the lenient
behaviour the old ``site_io.load_site_dict`` had — absence is reported by
the guard as missing required config rather than as a parse crash.
文件不存在时返回空字典，与旧 ``site_io.load_site_dict`` 的宽松行为保持
一致——缺失会由配置守卫报告为“缺少必需配置”，而不是解析崩溃。
"""
from . import paths
@@ -66,30 +63,27 @@ def load_site(path: Path | None = None) -> Dict[str, Any]:
def resolve_path(site: Dict[str, Any], site_path: str) -> List[ResolvedValue]:
"""Resolve a (possibly list-expanding) dotted path against parsed YAML.
"""在解析后的 YAML 上解析一个（可能带列表展开的）点分路径。
The ``[]`` token expands a collection into one entry per member, so
per-instance config can be declared once and still require *every* instance
to supply it:
``[]`` 记号会把一个集合展开成每个成员一条记录，因此按实例的配置只需
声明一次，却仍要求*每一个*实例都提供它：
* sequence — ``robot.robots[].sn`` → ``robot.robots[0].sn``, ``...[1].sn``
* mapping — ``cameras[].serial`` → ``cameras.d405_left.serial``, ...
* 序列 — ``robot.robots[].sn`` → ``robot.robots[0].sn``、``...[1].sn``
* 映射 — ``cameras[].serial`` → ``cameras.d405_left.serial``……
Mappings are supported because ``site.yaml`` registers cameras (and other
named devices) as mappings keyed by device name rather than as sequences.
之所以支持映射，是因为 ``site.yaml`` 把相机（以及其他具名设备）登记为
以设备名为键的映射，而不是序列。
Missing branches produce no results; the caller decides whether that is an
error. A *scalar* where a collection is expected (``cameras: 5``) is
reported as a single value resolved at the collection level, so the
type check can flag it instead of the operator being told, misleadingly,
that the value is simply absent.
缺失的分支不产生结果；是否算错误由调用方决定。在本该是集合的位置出现
*标量*（``cameras: 5``）时，会作为在集合层级解析出的单个值上报，让类型
检查能指出问题，而不是误导性地告诉操作员“该值不存在”。
"""
tokens = [t.strip() for t in site_path.split(".") if t.strip()]
return _resolve(site, tokens, "")
def _resolve(node: Any, tokens: List[str], prefix: str) -> List[ResolvedValue]:
"""Recursive worker for :func:`resolve_path`."""
""":func:`resolve_path` 的递归工作函数。"""
if not tokens:
return [ResolvedValue(path=prefix, value=node)]
@@ -122,8 +116,8 @@ def _resolve(node: Any, tokens: List[str], prefix: str) -> List[ResolvedValue]:
if child is None:
return []
# Scalar where a collection was declared. Stop here and let the type check
# produce a precise complaint rather than a misleading "missing".
# 声明为集合的位置却是标量。在这里停下，让类型检查给出精确的报错，
# 而不是误导性的“缺失”。
return [ResolvedValue(path=child_path, value=child)]
@@ -132,17 +126,17 @@ def check_config(
site: Dict[str, Any],
site_file: Optional[Path] = None,
) -> Dict[str, List[ResolvedValue]]:
"""Validate all declared fields of a manifest against parsed site config.
"""按解析后的现场配置校验清单中所有已声明的字段。
Returns a mapping of ``field.key`` to the resolved values, so callers that
need the values anyway do not have to resolve them a second time.
返回 ``field.key`` 到已解析值的映射，这样本来就需要这些值的调用方不必
再解析第二遍。
Raises
------
MissingConfigError
A required field has no value in ``site.yaml``.
某个必需字段在 ``site.yaml`` 中没有值。
BadConfigValueError
A field is present but has the wrong YAML type.
某个字段存在，但字段值的 YAML 类型不对。
"""
resolved: Dict[str, List[ResolvedValue]] = {}
problems: List[str] = []
@@ -156,9 +150,8 @@ def check_config(
problems.append(_missing_message(field))
continue
# A required field with per-instance expansion needs *every* instance
# populated. Reporting only when all were blank would let one
# unconfigured arm slip through.
# 带按实例展开的必需字段需要*每一个*实例都填好。若只在全部为空时
# 才报告，就会放过某一台未配置的机械臂。
if field.required:
blank_paths = [v.path for v in values if _is_blank(v.value)]
if blank_paths:
@@ -204,7 +197,7 @@ def _default_site_file() -> Path:
def _is_blank(value: Any) -> bool:
"""Whether a value should be treated as 'not supplied'."""
"""判断某个值是否应被视为“未提供”。"""
if value is None:
return True
if isinstance(value, str) and not value.strip():
@@ -217,7 +210,7 @@ def _is_blank(value: Any) -> bool:
def _matches_type(value: Any, declared: str) -> bool:
expected = _TYPE_CHECKS.get(declared)
if expected is None:
# Unknown declared type is rejected at manifest load time.
# 未知的已声明类型会在清单加载阶段被拒。
return True
if declared in ("int", "float") and isinstance(value, bool):
return False
@@ -225,11 +218,10 @@ def _matches_type(value: Any, declared: str) -> bool:
def _missing_message(field: ConfigField, blank_paths: Optional[List[str]] = None) -> str:
"""Render one missing-config problem.
"""渲染一条缺失配置问题。
``blank_paths`` lists concrete per-instance paths that were present but
empty (e.g. ``robot.robots[1].sn``), which is far more actionable than
repeating the abstract declaration path.
``blank_paths`` 列出存在但为空的具体按实例路径（例如
``robot.robots[1].sn``），这远比重复抽象的声明路径更具可操作性。
"""
lines = [f" 缺: {field.key}"]
if blank_paths:
src/arm_interfaces/arm_interfaces/errors.py
+20 -22
View File
@@ -1,13 +1,12 @@
"""Structured contract errors.
"""结构化的契约错误。
Every error carries a ``fix_hint`` that names the exact file and key a human
should edit. This is deliberate: the whole point of the contract layer is to
replace "something silently did nothing" (the old behaviour, where
``run_handeye.sh`` passed five parameters the node never declared) with an
actionable message.
每个错误都带有 ``fix_hint``，明确指出人应该去改哪个文件、哪个键。这是刻意
设计的：契约层的全部意义，就是把“某个东西静默地什么都没做”（旧行为，例如
``run_handeye.sh`` 传给节点的五个参数节点根本不曾声明）替换成一条可执行的
提示。
Callers should format errors with :func:`format_error`, which renders the
hint on its own indented line so it is impossible to miss in a terminal.
调用方应当用 :func:`format_error` 格式化错误，它会把修复提示单独渲染成
一行缩进文本，在终端里绝不会被漏看。
"""
from __future__ import annotations
@@ -16,18 +15,17 @@ from typing import Optional
class ContractError(Exception):
"""Base class for every device-contract failure.
"""所有设备契约失败的基类。
Attributes
----------
what:
One-line description of what is wrong.
一句话说明哪里不对。
fix_hint:
Concrete instruction naming the file and key to edit. ``None`` means
the caller could not determine a remedy, which callers should treat as
a bug in the contract layer rather than user error.
指明要改哪个文件、哪个键的具体指示。``None`` 表示调用方无法确定
补救办法，此时调用方应把它当作契约层的 bug，而不是用户错误。
detail:
Optional extra context (paths tried, values found, ...).
可选的补充上下文（尝试过的路径、读到的值……）。
"""
def __init__(
@@ -41,36 +39,36 @@ class ContractError(Exception):
self.fix_hint = fix_hint
self.detail = detail
def __str__(self) -> str: # pragma: no cover - exercised via format_error
def __str__(self) -> str: # pragma: no cover - 通过 format_error 覆盖
return format_error(self)
class RegistryError(ContractError):
"""``devices/registry.yaml`` is missing, unreadable, or incomplete."""
"""``devices/registry.yaml`` 缺失、不可读或不完整。"""
class ManifestError(ContractError):
"""A per-brand ``manifest.yaml`` is missing, unreadable, or malformed."""
"""某个品牌的 ``manifest.yaml`` 缺失、不可读或格式错误。"""
class MissingAssetError(ContractError):
"""A vendor asset declared as required in a manifest is not on disk."""
"""清单中声明为必需的厂商物料不在磁盘上。"""
class MissingConfigError(ContractError):
"""``config/site.yaml`` does not supply a required declared config key."""
"""``config/site.yaml`` 未提供某个必需的已声明配置键。"""
class CapabilityError(ContractError):
"""An upper layer asked for a capability the selected driver does not declare."""
"""上层请求了所选驱动未声明的能力。"""
class BadConfigValueError(ContractError):
"""A config key exists but its value has the wrong type or shape."""
"""某个配置键存在，但值的类型或结构不对。"""
def format_error(err: ContractError) -> str:
"""Render a contract error as a multi-line, human-actionable message."""
"""把契约错误渲染成多行、可执行的中文提示。"""
lines = [f"设备契约校验失败: {err.what}"]
if err.detail:
for line in str(err.detail).splitlines():
src/arm_interfaces/arm_interfaces/loader.py
+35 -42
View File
@@ -1,18 +1,16 @@
"""Load the whole device stack and check flows against it.
"""加载整个设备栈，并据此校验各流程。
This is the single entry point the rest of the framework uses. It chains:
这是框架其余部分使用的唯一入口。它依次串联：
1. :mod:`arm_interfaces.registry` — which brand implements each kind
2. :mod:`arm_interfaces.manifest` — that brand's capabilities, assets, config
3. :mod:`arm_interfaces.config_guard` — does ``site.yaml`` supply that config
1. :mod:`arm_interfaces.registry` —— 每一类设备由哪个品牌实现
2. :mod:`arm_interfaces.manifest` —— 该品牌的能力、物料、配置
3. :mod:`arm_interfaces.config_guard` —— ``site.yaml`` 是否提供了那份配置
then exposes :func:`check_requirement`, which verifies that a named flow
(tracking, hand-eye, gauge assembly, ...) only asks for capabilities the
selected drivers actually declare.
然后暴露 :func:`check_requirement`，用来验证某个具名流程
（跟踪、手眼、通止规装配……）只索要所选驱动确实声明了的能力。
The payoff is at startup: instead of discovering on the first motion that the
selected gripper cannot rotate, the run refuses to start and names the missing
capability, the interface it implies, and which manifest to change.
收益体现在启动时：不会等到第一次运动才发现所选夹爪不能旋转，而是直接
拒绝启动，并点出缺失的能力、它对应的接口，以及该改哪个清单。
"""
from __future__ import annotations
@@ -29,31 +27,31 @@ from .registry import REQUIRED_KINDS, Registry, load_registry
class DeviceStack(NamedTuple):
"""Everything about the currently selected hardware."""
"""当前所选硬件的全部信息。"""
registry: Registry
manifests: Dict[str, Manifest]
"""Registry kind -> loaded manifest, e.g. ``{"robots": <Manifest>}``."""
"""选型类别 -> 已加载的清单，例如 ``{"robots": <Manifest>}``。"""
config: Dict[str, List[ResolvedValue]]
"""``f"{kind}.{field.key}"`` -> resolved site.yaml values."""
"""``f"{kind}.{field.key}"`` -> 解析出的 site.yaml 值。"""
site_path: Path
"""The ``config/site.yaml`` that was validated."""
"""被校验过的 ``config/site.yaml``。"""
def manifest(self, kind: str) -> Manifest:
"""Return the manifest for a registry kind."""
"""返回某个选型类别的清单。"""
return self.manifests[kind]
def capabilities(self, kind: str) -> Set[str]:
"""Effective capabilities of the selected driver for a kind."""
"""某类设备所选驱动的有效能力。"""
manifest = self.manifests.get(kind)
if manifest is None:
return set()
return set(effective_capabilities(manifest, self._selected_models(kind)))
def _selected_models(self, kind: str) -> List[str]:
"""Model ids selected in site.yaml for a kind (may be empty)."""
"""某类设备在 site.yaml 中选中的型号 id（可能为空）。"""
manifest = self.manifests.get(kind)
if manifest is None:
return []
@@ -70,12 +68,12 @@ class DeviceStack(NamedTuple):
def effective_capabilities(
manifest: Manifest, selected_models: Sequence[str] = ()
) -> Set[str]:
"""Capabilities a driver provides, given which of its models are selected.
"""在给定选中哪些型号的前提下，驱动提供的能力。
Top-level ``capabilities`` are what *every* model of the brand supports.
A model may add more (RGI adds ``rotate`` on top of the shared open/close).
Declared capabilities are never removed, because narrowing silently would
turn a clear startup error into a confusing runtime failure.
顶层 ``capabilities`` 是该品牌*每一个*型号都支持的能力。某个型号可以
追加更多（RGI 在共有的 open/close 之上增加了 ``rotate``）。已声明的
能力绝不会被移除，因为悄悄收窄会把清晰的启动报错变成令人困惑的运行时
失败。
"""
caps: Set[str] = set(manifest.capabilities)
for model_id in selected_models:
@@ -91,19 +89,16 @@ def load_device_stack(
*,
verify_assets: bool = True,
) -> DeviceStack:
"""Load and validate the entire device stack.
"""加载并校验整个设备栈。
Validation failures across *all* device kinds are collected and reported
together. Reporting only the first one would force the operator to re-run
the check once per missing key, which is exactly the kind of round-trip this
layer exists to eliminate.
*所有*设备类别的校验失败都会被收集起来一并报告。只报告第一个，会迫使
操作员每修一个缺失的键就要重跑一次检查，而这种来回折腾正是本层要消除的。
Raises
------
ContractError subclass
On contract violations, with actionable fix hints. When more than one
kind fails, a single aggregated :class:`ContractError` is raised whose
detail lists every problem.
ContractError 子类
出现契约违规时抛出，并附带可执行的修复提示。当不止一类设备失败时，
会抛出单个聚合的 :class:`ContractError`，其 detail 列出全部问题。
"""
registry = load_registry(registry_path)
site_file = site_file or paths.site_path()
@@ -114,7 +109,7 @@ def load_device_stack(
problems: List[ContractError] = []
for kind in (*REQUIRED_KINDS, "sim"):
# A kind that was not selected at all is optional unless required.
# 完全没有选型的类别，除非是必需类别，否则可选。
if kind not in registry.selections:
continue
@@ -156,11 +151,10 @@ def load_device_stack(
def _aggregate(problems: List[ContractError]) -> ContractError:
"""Combine several contract errors into one reportable error.
"""把多个契约错误合并成一个可上报的错误。
A single problem is returned unchanged so the common case keeps its focused
message. Multiple problems are merged, preserving each one's own detail and
fix hint.
只有单个问题时原样返回，让常见情形保持聚焦的消息。多个问题则合并，
并保留每个问题各自的 detail 与修复提示。
"""
if len(problems) == 1:
return problems[0]
@@ -188,16 +182,15 @@ def _aggregate(problems: List[ContractError]) -> ContractError:
def check_requirement(
stack: DeviceStack, requirement_name: str
) -> Dict[str, Set[str]]:
"""Verify a flow's capability needs are met by the selected drivers.
"""验证某个流程的能力需求是否被所选驱动满足。
Returns the capabilities that were satisfied, grouped by kind, so callers
can log what the flow is running against.
返回已满足的能力，按类别分组，便于调用方记录该流程运行在什么之上。
Raises
------
CapabilityError
When a needed capability is missing. The message names the capability,
the interface it maps to, the selected brand, and the manifest to edit.
缺少某项所需能力时抛出。消息会点出该能力、它映射到的接口、所选的
品牌，以及该编辑的清单。
"""
requirement = REQUIREMENTS.get(requirement_name)
if requirement is None:
src/arm_interfaces/arm_interfaces/manifest.py
+34 -36
View File
@@ -1,18 +1,17 @@
"""Load and validate a brand's ``manifest.yaml``.
"""加载并校验某个品牌的 ``manifest.yaml``。
A manifest is a brand's self-description. It answers four questions:
清单是品牌的自我介绍，回答四个问题：
1. **Who am I?** — ``id``, ``label``, ``vendor``, ``version``
2. **What vendor material must be on disk?** — ``assets`` (checked, with the
required format and naming convention recorded so operators can comply)
3. **What can I do?** — ``capabilities`` (subset of the vocabulary in
:mod:`arm_interfaces.capabilities`)
4. **What config do I need?** — ``config_schema``, whose ``site_path`` entries
point into ``config/site.yaml``
1. **我是谁？** —— ``id``、``label``、``vendor``、``version``
2. **磁盘上必须有哪些厂商物料？** —— ``assets``（会被检查，并记录必需的
格式与命名约定，方便操作员照做）
3. **我能做什么？** —— ``capabilities``（:mod:`arm_interfaces.capabilities`
词表的子集）
4. **我需要什么配置？** —— ``config_schema``，其中每项 ``site_path`` 都指向
``config/site.yaml`` 里的位置
Validation here is intentionally strict: a manifest that declares an unknown
capability or a malformed config entry is a bug that should surface at startup,
not a runtime surprise mid-motion.
这里的校验刻意从严：清单声明了未知能力或格式错误的配置项，属于应当在启动
时暴露的 bug，而不是运动中途才冒出来的运行时意外。
"""
from __future__ import annotations
@@ -27,7 +26,7 @@ from .capabilities import ALL_CAPABILITIES, capabilities_for_kind
from .errors import ManifestError, MissingAssetError
from .registry import KIND_FOLDERS
#: registry key (plural) -> manifest ``kind`` value (singular).
#: 选型中的键（复数）-> 清单里 ``kind`` 的取值（单数）。
KIND_SINGULAR: Dict[str, str] = {
"robots": "robot",
"grippers": "gripper",
@@ -36,15 +35,15 @@ KIND_SINGULAR: Dict[str, str] = {
"sim": "sim",
}
#: Config value types the contract layer knows how to check.
#: 契约层知道如何检查的配置值类型。
KNOWN_TYPES = {"str", "int", "float", "bool", "list", "dict"}
#: Manifest top-level keys that must be present.
#: 清单中必须存在的顶层键。
REQUIRED_KEYS = ("id", "label", "kind", "driver")
class Asset(NamedTuple):
"""One piece of vendor material a manifest expects on disk."""
"""清单期望磁盘上存在的一项厂商物料。"""
path: str
required: bool
@@ -56,7 +55,7 @@ class Asset(NamedTuple):
class DriverSpec(NamedTuple):
"""How to start this brand's adapter."""
"""如何启动该品牌的适配器。"""
package: str
executable: str
@@ -65,7 +64,7 @@ class DriverSpec(NamedTuple):
class ConfigField(NamedTuple):
"""One declared config key, pointing into ``config/site.yaml``."""
"""一个已声明的配置键，指向 ``config/site.yaml`` 中的位置。"""
key: str
site_path: str
@@ -77,7 +76,7 @@ class ConfigField(NamedTuple):
class Manifest(NamedTuple):
"""Parsed and validated ``manifest.yaml``."""
"""已解析并校验的 ``manifest.yaml``。"""
id: str
label: str
@@ -94,14 +93,14 @@ class Manifest(NamedTuple):
raw: Dict[str, Any]
def supports(self, capability: str) -> bool:
"""Whether this driver declares the capability."""
"""该驱动是否声明了这项能力。"""
return capability in self.capabilities
def model_capabilities(self, model_id: str) -> Optional[Tuple[str, ...]]:
"""Capabilities for a specific model id, if the manifest lists models.
"""某个具体型号 id 的能力（前提是清单列出了型号）。
Returns ``None`` when the manifest does not enumerate models, which
means the top-level ``capabilities`` list applies to all of them.
清单未逐一列举型号时返回 ``None``，表示顶层 ``capabilities`` 列表
对全部型号都适用。
"""
for model in self.models:
if model.get("id") == model_id:
@@ -113,19 +112,19 @@ class Manifest(NamedTuple):
def manifest_path(kind: str, brand: str) -> Path:
"""Return the expected manifest path for a kind/brand pair."""
"""返回某个“类别/品牌”组合预期的清单路径。"""
return paths.brand_dir(KIND_FOLDERS.get(kind, kind), brand) / "manifest.yaml"
def load_manifest(kind: str, brand: str, path: Path | None = None) -> Manifest:
"""Read, validate and return the manifest for a kind/brand pair.
"""读取、校验并返回某个“类别/品牌”组合的清单。
Parameters
----------
kind:
Registry key — ``robots``, ``grippers``, ``sensors``, ``cameras``, ``sim``.
选型中的键 —— ``robots``、``grippers``、``sensors``、``cameras``、``sim``。
brand:
Brand folder name under ``devices/<kind>/``.
``devices/<kind>/`` 下的品牌目录名。
"""
target = path or manifest_path(kind, brand)
@@ -187,11 +186,10 @@ def load_manifest(kind: str, brand: str, path: Path | None = None) -> Manifest:
def check_assets(manifest: Manifest, brand: str, kind: str) -> None:
"""Verify every ``required`` asset exists, with format hints in the error.
"""校验每个 ``required`` 物料都存在，并在错误里给出格式提示。
Only presence is checked. Format and naming cannot be verified
automatically, so they are reported back to the operator in the failure
message to make compliance possible.
这里只检查是否存在。格式与命名无法自动验证，因此在失败消息里回告给
操作员，让照做成为可能。
"""
base = paths.brand_dir(KIND_FOLDERS.get(kind, kind), brand)
missing: List[str] = []
@@ -225,7 +223,7 @@ def check_assets(manifest: Manifest, brand: str, kind: str) -> None:
# ---------------------------------------------------------------------------
# internal helpers
# 内部辅助函数
# ---------------------------------------------------------------------------
def _available_brands(kind: str) -> List[str]:
@@ -236,10 +234,10 @@ def _available_brands(kind: str) -> List[str]:
def _asset_present(target: Path) -> bool:
"""An asset counts as present if it is a file, a dir, or a symlink to either.
"""物料是文件、目录，或指向二者的符号链接，都算作存在。
A directory is accepted because vendors often ship a tree (``rdk/``) and the
operator should not be forced to add a sentinel file inside it.
接受目录，是因为厂商常以整棵目录树（``rdk/``）交付，不应该逼着操作员
在目录里额外放一个占位文件。
"""
if target.is_symlink():
return target.exists()
@@ -298,7 +296,7 @@ def _parse_capabilities(
detail=f"该清单里合法的能力: {', '.join(known_code) or '（无）'}",
)
# A capability declared twice is almost always a copy-paste slip.
# 同一项能力声明两次，几乎总是复制粘贴出的疏漏。
duplicates = sorted({c for c in raw if raw.count(c) > 1})
if duplicates:
raise ManifestError(
src/arm_interfaces/arm_interfaces/paths.py
+17 -17
View File
@@ -1,15 +1,15 @@
"""Single source of truth for workspace paths.
"""工作区路径的唯一权威来源。
The old codebase resolved the workspace root in three different ways:
``arm_application/site_io.py``, ``scripts/lib/export_site_env.py``, and several
nodes that read ``site.yaml`` directly with their own ``_site_path()`` helpers.
They did not always agree. This module is the one place that resolves paths, and
everything else — including ``site_io.py`` — should delegate here.
旧代码库用三种不同方式解析工作区根目录：
``arm_application/site_io.py``、``scripts/lib/export_site_env.py``，以及若干
用各自 ``_site_path()`` 辅助函数直接读 ``site.yaml`` 的节点。
它们并不总是一致。本模块是解析路径的唯一场所，其他所有地方
——包括 ``site_io.py``——都应委托到这里。
Resolution order (first non-empty wins):
解析顺序（第一个非空者胜出）：
1. Explicit environment override (``SITE_ROS2_ROOT`` / ``SITE_CONFIG``)
2. ``ROS2_WS`` (exported by ``scripts/env.sh``)
1. 显式环境变量覆盖（``SITE_ROS2_ROOT`` / ``SITE_CONFIG``）
2. ``ROS2_WS``（由 ``scripts/env.sh`` 导出）
3. ``~/ros2``
"""
@@ -20,7 +20,7 @@ from pathlib import Path
def ros2_root() -> Path:
"""Return the workspace root, honouring environment overrides."""
"""返回工作区根目录，并遵循环境变量覆盖。"""
for key in ("SITE_ROS2_ROOT", "ROS2_WS"):
value = os.environ.get(key, "").strip()
if value:
@@ -29,7 +29,7 @@ def ros2_root() -> Path:
def site_path() -> Path:
"""Return the path to the master config ``config/site.yaml``."""
"""返回主配置 ``config/site.yaml`` 的路径。"""
override = os.environ.get("SITE_CONFIG", "").strip()
if override:
return Path(override).expanduser()
@@ -37,7 +37,7 @@ def site_path() -> Path:
def devices_root() -> Path:
"""Return the root of the device folder tree."""
"""返回设备目录树的根目录。"""
override = os.environ.get("SITE_DEVICES_ROOT", "").strip()
if override:
return Path(override).expanduser()
@@ -45,19 +45,19 @@ def devices_root() -> Path:
def registry_path() -> Path:
"""Return the path to the device selection file ``devices/registry.yaml``."""
"""返回设备选型文件 ``devices/registry.yaml`` 的路径。"""
return devices_root() / "registry.yaml"
def handeye_root() -> Path:
"""Return the root of the hand-eye module directory (``handeye/``)."""
"""返回手眼模块目录（``handeye/``）的根目录。"""
return ros2_root() / "handeye"
def brand_dir(kind: str, brand: str) -> Path:
"""Return the directory holding one brand under a device kind.
"""返回某类设备下某个品牌所在的目录。
``kind`` is the plural folder name used in ``devices/`` (``robots``,
``grippers``, ``sensors``, ``cameras``, ``sim``).
``kind`` 是 ``devices/`` 中使用的复数目录名（``robots``、
``grippers``、``sensors``、``cameras``、``sim``）。
"""
return devices_root() / kind / brand
src/arm_interfaces/arm_interfaces/registry.py
+15 -17
View File
@@ -1,9 +1,8 @@
"""Load and validate ``devices/registry.yaml``.
"""加载并校验 ``devices/registry.yaml``。
The registry answers exactly one question: *which brand implements each device
kind right now?* It holds no field values by design — every speed, threshold and
serial number lives in ``config/site.yaml``. Keeping selection and values apart
is what stops the "I changed the value but nothing happened" class of bug.
选型文件只回答一个问题：*当前每一类设备由哪个品牌实现？* 按设计它不保存
任何字段值——所有速度、阈值和序列号都在 ``config/site.yaml`` 里。把“选型”
与“取值”分开，正是为了杜绝“改了值却毫无反应”这一类 bug。
"""
from __future__ import annotations
@@ -16,8 +15,8 @@ import yaml
from . import paths
from .errors import RegistryError
#: Device kind -> the folder under ``devices/`` holding its brands.
#: Also the key used in ``registry.yaml``.
#: 设备类别 -> ``devices/`` 下存放其品牌的目录。
#: 同时也是 ``registry.yaml`` 中使用的键。
KIND_FOLDERS: Dict[str, str] = {
"robots": "robots",
"grippers": "grippers",
@@ -26,25 +25,25 @@ KIND_FOLDERS: Dict[str, str] = {
"sim": "sim",
}
#: Every kind the registry must select. ``sim`` is optional (headless setups).
#: 选型文件必须选择的每一类设备。``sim`` 是可选的（无头环境）。
REQUIRED_KINDS = ("robots", "grippers", "sensors", "cameras")
OPTIONAL_KINDS = ("sim",)
class Registry(NamedTuple):
"""Parsed ``devices/registry.yaml``."""
"""解析后的 ``devices/registry.yaml``。"""
selections: Dict[str, str]
"""kind -> brand id, e.g. ``{"robots": "flexiv"}``."""
"""kind -> 品牌 id，例如 ``{"robots": "flexiv"}``。"""
path: Path
"""Where the registry was loaded from (for error messages)."""
"""选型文件的加载来源（用于错误消息）。"""
raw: Dict[str, Any]
"""Full parsed document, for forward compatibility."""
"""完整解析结果，便于向前兼容。"""
def brand(self, kind: str) -> str:
"""Return the selected brand for a kind, raising if unset."""
"""返回某类设备选中的品牌，未设置则抛错。"""
try:
return self.selections[kind]
except KeyError:
@@ -58,13 +57,12 @@ class Registry(NamedTuple):
def load_registry(path: Path | None = None) -> Registry:
"""Read and validate the registry file.
"""读取并校验选型文件。
Raises
------
RegistryError
If the file is missing, unparseable, is not a mapping, or omits a
required kind.
文件缺失、无法解析、顶层不是映射，或漏掉必需的设备类别时抛出。
"""
target = path or paths.registry_path()
@@ -119,7 +117,7 @@ def load_registry(path: Path | None = None) -> Registry:
def list_available_brands(kind: str) -> list[str]:
"""Return brand folder names present on disk for a device kind."""
"""返回到磁盘上某类设备现存的品牌目录名。"""
folder = paths.devices_root() / KIND_FOLDERS.get(kind, kind)
if not folder.is_dir():
return []
src/arm_interfaces/setup.cfg
+4 -4
View File
@@ -4,9 +4,9 @@ script_dir=$base/lib/arm_interfaces
install_scripts=$base/lib/arm_interfaces
[tool:pytest]
# ROS 2 Humble's two launch-testing pytest plugins use old pytest hook
# signatures that current pytest rejects, which aborts collection before our
# tests even run. Disabling them is safe: these are pure-Python unit tests and
# use none of launch_testing's fixtures.
# ROS 2 Humble 的两个 launch-testing pytest 插件使用的旧版 pytest hook
# 签名会被当前 pytest 拒绝，导致测试还没开始就被中止收集。
# 禁用它们是安全的：这些是纯 Python 单元测试，
# 完全没有用到 launch_testing 的任何 fixture。
addopts = -p no:launch_testing -p no:launch_ros
testpaths = test
src/arm_interfaces/setup.py
+1 -1
View File
@@ -18,7 +18,7 @@ setup(
license='Apache-2.0',
entry_points={
'console_scripts': [
# Validate the whole device stack: registry -> manifests -> site.yaml.
# 校验整条设备链：registry → 各 manifest → site.yaml。
'device_check = arm_interfaces.cli:main',
],
},
src/arm_interfaces/test/test_contract.py
+28 -29
View File
@@ -1,10 +1,9 @@
"""Unit tests for the device contract layer.
"""设备契约层的单元测试。
These build a throwaway workspace in ``tmp_path`` so they never touch the real
``devices/`` tree or ``config/site.yaml``. Every failure mode that the layer is
supposed to catch has a test asserting the *shape* of the error — in particular
that a ``fix_hint`` is present, because a contract error without a remediation
instruction is only half the job.
这些测试在 ``tmp_path`` 里搭一个用完即弃的工作区，因此绝不会碰真实的
``devices/`` 目录树或 ``config/site.yaml``。本层应当捕获的每一种失败模式
都有一个测试断言错误的*形态*——尤其是必须带 ``fix_hint``，因为一条没有
修复指示的契约错误只算完成了一半工作。
"""
from __future__ import annotations
@@ -36,7 +35,7 @@ from arm_interfaces.registry import list_available_brands
# ---------------------------------------------------------------------------
# fake workspace construction
# 构造假工作区
# ---------------------------------------------------------------------------
MINIMAL_REGISTRY = """\
@@ -143,7 +142,7 @@ cameras:
@pytest.fixture
def workspace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Dict[str, Path]:
"""Create a minimal, valid device workspace and point the layer at it."""
"""创建一个最小可用的设备工作区，并让契约层指向它。"""
devices = tmp_path / "devices"
for kind, brand, manifest in (
("robots", "acme", ACME_MANIFEST),
@@ -173,7 +172,7 @@ def write(path: Path, content: str) -> None:
# ---------------------------------------------------------------------------
# registry
# 选型
# ---------------------------------------------------------------------------
class TestRegistry:
@@ -217,7 +216,7 @@ class TestRegistry:
# ---------------------------------------------------------------------------
# manifest
# 清单
# ---------------------------------------------------------------------------
class TestManifest:
@@ -254,7 +253,7 @@ class TestManifest:
assert "get_arm_state" in (excinfo.value.fix_hint or "")
def test_capability_from_another_kind_is_rejected(self, workspace):
"""A robot must not be able to claim a gripper capability."""
"""机器人绝不能声称自己拥有夹爪能力。"""
write(
workspace["devices"] / "robots" / "acme" / "manifest.yaml",
ACME_MANIFEST.replace(" - get_arm_state", " - rotate\n - get_arm_state"),
@@ -276,7 +275,7 @@ class TestManifest:
assert "重复" in excinfo.value.what
def test_required_and_default_together_is_rejected(self, workspace):
"""Ambiguous intent: either the user must supply it, or we have a fallback."""
"""意图含糊：要么必须由用户提供，要么我们已有兜底值。"""
write(
workspace["devices"] / "robots" / "acme" / "manifest.yaml",
ACME_MANIFEST.replace(
@@ -334,7 +333,7 @@ class TestManifest:
# ---------------------------------------------------------------------------
# asset presence
# 物料是否存在
# ---------------------------------------------------------------------------
class TestAssets:
@@ -354,7 +353,7 @@ class TestAssets:
with pytest.raises(MissingAssetError) as excinfo:
check_assets(loaded, "acme", "robots")
message = str(excinfo.value)
# The operator must be told how to comply, not merely that it is missing.
# 必须告诉操作员该怎么照做，而不只是说“缺了”。
assert "厂商原始格式" in message
assert "保持原目录名" in message
@@ -368,7 +367,7 @@ class TestAssets:
write(workspace["devices"] / "robots" / "acme" / "manifest.yaml", manifest)
loaded = load_manifest("robots", "acme")
(workspace["devices"] / "robots" / "acme" / "rdk").mkdir()
check_assets(loaded, "acme", "robots") # must not raise
check_assets(loaded, "acme", "robots") # 不应抛错
def test_optional_asset_may_be_absent(self, workspace):
manifest = ACME_MANIFEST + textwrap.dedent("""\
@@ -379,11 +378,11 @@ class TestAssets:
""")
write(workspace["devices"] / "robots" / "acme" / "manifest.yaml", manifest)
loaded = load_manifest("robots", "acme")
check_assets(loaded, "acme", "robots") # must not raise
check_assets(loaded, "acme", "robots") # 不应抛错
# ---------------------------------------------------------------------------
# config_guard: path resolution and type checking
# config_guard：路径解析与类型检查
# ---------------------------------------------------------------------------
class TestResolvePath:
@@ -400,7 +399,7 @@ class TestResolvePath:
assert found[1].path == "robot.robots[1].sn"
def test_mapping_expansion(self):
"""site.yaml registers cameras as a mapping keyed by name, not a sequence."""
"""site.yaml 把相机登记为以名称作键的映射，而不是序列。"""
site = {"cameras": {"left": {"serial": "1"}, "right": {"serial": "2"}}}
found = resolve_path(site, "cameras[].serial")
assert [f.value for f in found] == ["1", "2"]
@@ -410,7 +409,7 @@ class TestResolvePath:
assert resolve_path({"a": {}}, "a.b.c") == []
def test_scalar_where_collection_expected_is_kept_for_type_check(self):
"""Better to report a type error than to silently claim 'missing'."""
"""报告类型错误，总好过静默声称“缺失”。"""
found = resolve_path({"cameras": 5}, "cameras[].serial")
assert len(found) == 1
assert found[0].value == 5
@@ -418,10 +417,10 @@ class TestResolvePath:
class TestCheckConfig:
def _manifest_with(self, workspace, extra: str):
"""Append extra ``config_schema`` entries to the Acme manifest.
"""把额外的 ``config_schema`` 条目追加到 Acme 清单上。
``extra`` is written flush-left and indented here, so call sites can
indent a YAML snippet naturally without counting spaces.
``extra`` 以顶格书写后在这里统一缩进，因此调用处可以自然地为 YAML
片段缩进，而无需数空格。
"""
body = textwrap.indent(textwrap.dedent(extra), " ")
write(workspace["devices"] / "robots" / "acme" / "manifest.yaml", ACME_MANIFEST + body)
@@ -483,7 +482,7 @@ class TestCheckConfig:
assert "fast" in (excinfo.value.detail or "")
def test_bool_is_not_accepted_as_float(self, workspace):
"""YAML ``true`` must not be silently used as 1.0."""
"""YAML 里的 ``true`` 绝不能被悄悄当作 1.0 使用。"""
manifest = self._manifest_with(
workspace,
"""\
@@ -499,7 +498,7 @@ class TestCheckConfig:
check_config(manifest, site)
def test_every_instance_must_supply_per_instance_value(self, workspace):
"""robot.robots[].sn requires all arms to be configured."""
"""robot.robots[].sn 要求所有机械臂都完成配置。"""
write(
workspace["devices"] / "robots" / "acme" / "manifest.yaml",
ACME_MANIFEST.replace("site_path: robot.sn", "site_path: robot.robots[].sn"),
@@ -509,12 +508,12 @@ class TestCheckConfig:
site["robot"]["robots"] = [{"sn": "A"}, {"sn": ""}]
with pytest.raises(MissingConfigError) as excinfo:
check_config(manifest, site)
# The offending instance must be identified by index.
# 出问题的那个实例必须按索引定位到。
assert "robots[1].sn" in str(excinfo.value)
# ---------------------------------------------------------------------------
# loader: aggregation and capabilities
# loader：聚合与能力
# ---------------------------------------------------------------------------
class TestLoader:
@@ -525,9 +524,9 @@ class TestLoader:
assert stack.capabilities("robots") == {"get_arm_state", "follow_joint_trajectory"}
def test_all_problems_reported_at_once(self, workspace):
"""One run must surface every broken kind, not just the first."""
"""一次运行必须暴露出每一类坏掉的设备，而不只是第一个。"""
site = workspace["site"]
write(site, "cameras:\n cam_a:\n serial: '1'\n") # robot.sn etc. missing
write(site, "cameras:\n cam_a:\n serial: '1'\n") # robot.sn 等缺失
with pytest.raises(Exception) as excinfo:
load_device_stack()
message = str(excinfo.value)
@@ -545,7 +544,7 @@ class TestLoader:
write(workspace["devices"] / "robots" / "acme" / "manifest.yaml", manifest)
with pytest.raises(MissingAssetError):
load_device_stack()
load_device_stack(verify_assets=False) # must not raise
load_device_stack(verify_assets=False) # 不应抛错
class TestEffectiveCapabilities:
src/arm_interfaces/test/test_real_devices.py
+18 -21
View File
@@ -1,11 +1,10 @@
"""Integration tests against the real workspace.
"""针对真实工作区的集成测试。
These are the tests that keep the shipped configuration honest. They load the
actual ``devices/`` tree and the actual ``config/site.yaml``, so a bad edit to
either fails here rather than on the robot.
这些测试用来让随仓库交付的配置保持诚实。它们加载真实的 ``devices/`` 目录树
和真实的 ``config/site.yaml``，所以对其中任何一处的糟糕改动都会在这里失败，
而不是在机器人上失败。
They are skipped when the real workspace is not present (for example when the
package is built in isolation), so they never turn into false failures.
真实工作区不存在时会跳过（例如包被单独构建时），因此它们绝不会变成假失败。
"""
from __future__ import annotations
@@ -31,26 +30,26 @@ pytestmark = pytest.mark.skipif(
class TestRealWorkspace:
def test_contract_satisfied(self, monkeypatch):
"""The shipped device tree and site.yaml must pass validation."""
# Guard against a leftover SITE_* override from another test leaking in.
"""随仓库交付的设备目录树和 site.yaml 必须通过校验。"""
# 防止另一个测试遗留的 SITE_* 覆盖泄漏进来。
for var in ("SITE_DEVICES_ROOT", "SITE_CONFIG", "SITE_ROS2_ROOT"):
monkeypatch.delenv(var, raising=False)
stack = load_device_stack()
# Every required kind is present and selected.
# 每个必需的设备类别都已存在并完成选型。
for kind in ("robots", "grippers", "sensors", "cameras"):
assert stack.manifest(kind) is not None, f"{kind} 未选型"
def test_selected_models_resolve(self, monkeypatch):
"""Model ids named in site.yaml must exist in the corresponding manifest."""
"""site.yaml 中写明的型号 id 必须存在于对应的清单里。"""
for var in ("SITE_DEVICES_ROOT", "SITE_CONFIG"):
monkeypatch.delenv(var, raising=False)
stack = load_device_stack()
site = load_site()
# Cameras: every entry's model must be a declared model id.
# 相机：每一项的 model 都必须是清单声明的型号 id。
camera_manifest = stack.manifest("cameras")
declared = {model["id"] for model in camera_manifest.models}
for name, camera in (site.get("cameras") or {}).items():
@@ -59,7 +58,7 @@ class TestRealWorkspace:
f"{camera_manifest.path} 声明的型号 {sorted(declared)} 里"
)
# Grippers: same check against the gripper manifest's models.
# 夹爪：对夹爪清单的型号做同样的检查。
gripper_manifest = stack.manifest("grippers")
gripper_models = {model["id"] for model in gripper_manifest.models}
for index, robot in enumerate((site.get("robot") or {}).get("robots") or []):
@@ -72,7 +71,7 @@ class TestRealWorkspace:
)
def test_sensor_models_folder_matches_config(self, monkeypatch):
"""tactile.model must match a subdirectory under sensors/orisys/models/."""
"""tactile.model 必须与 sensors/orisys/models/ 下的某个子目录对应。"""
for var in ("SITE_DEVICES_ROOT", "SITE_CONFIG"):
monkeypatch.delenv(var, raising=False)
@@ -95,7 +94,7 @@ class TestRealWorkspace:
)
def test_assembly_requirement_met_by_real_selection(self, monkeypatch):
"""The gauge assembly flow must be satisfiable with shipped hardware."""
"""随仓库交付的硬件必须能满足通止规装配流程。"""
for var in ("SITE_DEVICES_ROOT", "SITE_CONFIG"):
monkeypatch.delenv(var, raising=False)
@@ -115,13 +114,11 @@ class TestRealWorkspace:
class TestExampleParity:
"""``site.example.yaml`` is generated from ``site.yaml``; it must stay in sync.
"""``site.example.yaml`` 由 ``site.yaml`` 生成，必须保持同步。
The template used to be maintained by hand and had drifted 114 keys behind
the live config, so a fresh deployment following it would silently miss
``servo.*``, ``hole_detect.stage2.*`` and ``sim.*``. It is now generated by
``scripts/gen_site_example.py``, and this test fails the moment the two
diverge.
这份模板以前靠手工维护，曾比现场配置落后 114 个键，导致照着它做的新部署
会静默漏掉 ``servo.*``、``hole_detect.stage2.*`` 和 ``sim.*``。现在它由
``scripts/gen_site_example.py`` 生成，两者一旦分叉，本测试立即失败。
"""
def _example_path(self) -> Path:
@@ -144,7 +141,7 @@ class TestExampleParity:
)
def test_example_contains_no_live_values(self):
"""The template must not leak serial numbers, IPs or user paths."""
"""模板绝不能泄漏序列号、IP 或用户路径。"""
example = self._example_path()
if not example.is_file():
pytest.skip("config/site.example.yaml 不存在")