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

克隆后还缺仿真安装树和两边的编译目录。把 build、build_sim、install_sim 一并放进仓库；仿真入口不再写死本机 conda，build 里指向源码的链接改成相对路径。

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
arm_dev
2026-10-15 11:07:04 +08:00
parent 926a7f9796
commit 80983954a4
5804 changed files with 2640781 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
build/arm_interfaces/arm_interfaces
Symlink
+1
View File
@@ -0,0 +1 @@
../../src/arm_interfaces/arm_interfaces
build/arm_interfaces/arm_interfaces.egg-info/PKG-INFO
+14
View File
@@ -0,0 +1,14 @@
Metadata-Version: 2.4
Name: arm_interfaces
Version: 0.1.0
Summary: Device contract layer for the robotic arm stack.
Maintainer: arm_dev
Maintainer-email: dev@example.com
License: Apache-2.0
Requires-Dist: setuptools
Requires-Dist: PyYAML
Dynamic: license
Dynamic: maintainer
Dynamic: maintainer-email
Dynamic: requires-dist
Dynamic: summary
build/arm_interfaces/arm_interfaces.egg-info/SOURCES.txt
+27
View File
@@ -0,0 +1,27 @@
package.xml
setup.cfg
setup.py
../../build/arm_interfaces/arm_interfaces.egg-info/PKG-INFO
../../build/arm_interfaces/arm_interfaces.egg-info/SOURCES.txt
../../build/arm_interfaces/arm_interfaces.egg-info/dependency_links.txt
../../build/arm_interfaces/arm_interfaces.egg-info/entry_points.txt
../../build/arm_interfaces/arm_interfaces.egg-info/requires.txt
../../build/arm_interfaces/arm_interfaces.egg-info/top_level.txt
../../build/arm_interfaces/arm_interfaces.egg-info/zip-safe
arm_interfaces/__init__.py
arm_interfaces/capabilities.py
arm_interfaces/cli.py
arm_interfaces/config_guard.py
arm_interfaces/errors.py
arm_interfaces/loader.py
arm_interfaces/manifest.py
arm_interfaces/paths.py
arm_interfaces/registry.py
arm_interfaces.egg-info/PKG-INFO
arm_interfaces.egg-info/SOURCES.txt
arm_interfaces.egg-info/dependency_links.txt
arm_interfaces.egg-info/entry_points.txt
arm_interfaces.egg-info/requires.txt
arm_interfaces.egg-info/top_level.txt
arm_interfaces.egg-info/zip-safe
resource/arm_interfaces
build/arm_interfaces/arm_interfaces.egg-info/dependency_links.txt
+1
View File
@@ -0,0 +1 @@
build/arm_interfaces/arm_interfaces.egg-info/entry_points.txt
+2
View File
@@ -0,0 +1,2 @@
[console_scripts]
device_check = arm_interfaces.cli:main
build/arm_interfaces/arm_interfaces.egg-info/requires.txt
+2
View File
@@ -0,0 +1,2 @@
setuptools
PyYAML
build/arm_interfaces/arm_interfaces.egg-info/top_level.txt
+1
View File
@@ -0,0 +1 @@
arm_interfaces
build/arm_interfaces/arm_interfaces.egg-info/zip-safe
+1
View File
@@ -0,0 +1 @@
build/arm_interfaces/build/lib/arm_interfaces/__init__.py
+77
View File
@@ -0,0 +1,77 @@
"""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
Everything device-selection related funnels through here so that the rest of the
framework never reads ``devices/`` or guesses at ``site.yaml`` structure itself.
"""
from .capabilities import (
ALL_CAPABILITIES,
CAPABILITIES_BY_KIND,
REQUIREMENTS,
ROS_INTERFACE,
capabilities_for_kind,
describe,
)
from .config_guard import check_config, load_site, resolve_path
from .errors import (
BadConfigValueError,
CapabilityError,
ContractError,
ManifestError,
MissingAssetError,
MissingConfigError,
RegistryError,
format_error,
)
from .loader import (
DeviceStack,
check_requirement,
effective_capabilities,
load_device_stack,
)
from .manifest import Manifest, check_assets, load_manifest, manifest_path
from .registry import Registry, list_available_brands, load_registry
__version__ = "0.1.0"
__all__ = [
# loading
"load_device_stack",
"load_registry",
"load_manifest",
"load_site",
"check_config",
"check_assets",
"check_requirement",
"resolve_path",
"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",
"MissingAssetError",
"MissingConfigError",
"CapabilityError",
"BadConfigValueError",
"format_error",
]
build/arm_interfaces/build/lib/arm_interfaces/capabilities.py
+273
View File
@@ -0,0 +1,273 @@
"""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.
``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.
"""
from __future__ import annotations
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."""
summary: str
"""One-line description of what the capability does."""
# ---------------------------------------------------------------------------
# Robot capabilities
# ---------------------------------------------------------------------------
ROBOT_CAPABILITIES: Dict[str, CapabilityInfo] = {
"get_arm_state": CapabilityInfo(
"robot", "读关节角与末端位姿。手眼标定与状态记录依赖它。"
),
"follow_joint_trajectory": CapabilityInfo(
"robot", "关节空间轨迹（点到点与多路点）。手眼走点、装配移动依赖它。"
),
"set_named_pose": CapabilityInfo(
"robot", "移动到命名位姿（如 home）。启动归位与异常复位依赖它。"
),
"servo_joint": CapabilityInfo(
"robot", "高频关节伺服（~100Hz）。视觉伺服 servo_node 依赖它。"
),
"servo_cartesian": CapabilityInfo(
"robot", "笛卡尔位姿伺服。视觉跟踪与悬停依赖它。"
),
"compliant_descent": CapabilityInfo(
"robot", "柔顺下臂 —— 带接触力约束的下压。装配接触阶段（S07）依赖它。"
),
"switch_tool": CapabilityInfo(
"robot",
"切换到已保存的工具坐标系（如切到 tool1）。"
"RDK 只提供按名字切换（Tool.Switch），工具的参数（质量/质心/TCP）在 Studio 里预配好。"
"全局工具系锁定依赖它。",
),
"home": CapabilityInfo("robot", "回到初始位姿。每个装配循环开始与异常复位依赖它。"),
"enable": CapabilityInfo("robot", "伺服使能。启动自检依赖它。"),
"clear_fault": CapabilityInfo("robot", "清除故障。异常恢复依赖它。"),
"soft_drag": CapabilityInfo(
"robot", "软拖拽示教。手眼标定走点依赖它。"
),
}
# ---------------------------------------------------------------------------
# Gripper capabilities
# ---------------------------------------------------------------------------
GRIPPER_CAPABILITIES: Dict[str, CapabilityInfo] = {
"open": CapabilityInfo("gripper", "张开。"),
"close": CapabilityInfo("gripper", "夹紧（可带力与速度参数）。"),
"move": CapabilityInfo("gripper", "移动到指定开度。"),
"rotate": CapabilityInfo(
"gripper", "旋转到指定角度。通止规装配 S08 循环的核心原语。"
),
"get_status": CapabilityInfo("gripper", "读开度、力、是否夹持到位。"),
}
# ---------------------------------------------------------------------------
# Sensor capabilities
# ---------------------------------------------------------------------------
SENSOR_CAPABILITIES: Dict[str, CapabilityInfo] = {
"read_tangential": CapabilityInfo(
"sensor", "读切向力（两条正交通道）。接触判定与拧紧判定依赖它。"
),
"read_normal": CapabilityInfo("sensor", "读法向力。拧紧判定与安全上限依赖它。"),
"read_depth": CapabilityInfo("sensor", "读深度图。"),
"read_force_vector": CapabilityInfo("sensor", "读力矢量图。"),
"measure_baseline": CapabilityInfo(
"sensor", "测静止基线。所有力阈值都是相对基线的增量，因此这是必需能力。"
),
}
# ---------------------------------------------------------------------------
# Camera capabilities
# ---------------------------------------------------------------------------
CAMERA_CAPABILITIES: Dict[str, CapabilityInfo] = {
"color": CapabilityInfo("camera", "彩色图。推理与跟踪依赖它。"),
"depth": CapabilityInfo("camera", "深度图。"),
"aligned_depth_to_color": CapabilityInfo(
"camera", "对齐到彩图的深度。把像素换算成三维坐标依赖它。"
),
"infrared_left": CapabilityInfo("camera", "左红外。"),
"infrared_right": CapabilityInfo("camera", "右红外。"),
"query_intrinsics": CapabilityInfo(
"camera", "查询内参。手眼标定必需 —— 没有内参无法把像素反投成射线。"
),
"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,
**SENSOR_CAPABILITIES,
**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),
"sensor": frozenset(SENSOR_CAPABILITIES),
"camera": frozenset(CAMERA_CAPABILITIES),
}
# ---------------------------------------------------------------------------
# ROS 2 interface mapping
# ---------------------------------------------------------------------------
class RosInterface(NamedTuple):
"""The concrete ROS 2 endpoint implementing a capability."""
namespace: str
name: str
kind: str # topic | service | action
type: str
#: capability -> ROS 2 endpoint. Names are relative to the device namespace and
#: are resolved per-instance (two arms never share a namespace).
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"
),
"set_named_pose": RosInterface("", "set_named_pose", "service", "arm_msgs/srv/SetNamedPose"),
"servo_joint": RosInterface("", "joint_command", "topic", "sensor_msgs/msg/JointState"),
"servo_cartesian": RosInterface("servo", "target_pose", "topic", "geometry_msgs/msg/PoseStamped"),
"compliant_descent": RosInterface(
"", "compliant_descent", "action", "arm_msgs/action/CompliantDescent"
),
"switch_tool": RosInterface("", "switch_tool", "service", "arm_msgs/srv/SwitchTool"),
"home": RosInterface("", "home", "service", "std_srvs/srv/Trigger"),
"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(
"", "aligned_depth_to_color/image_raw", "topic", "sensor_msgs/msg/Image"
),
"infrared_left": RosInterface("", "infra1/image_raw", "topic", "sensor_msgs/msg/Image"),
"infrared_right": RosInterface("", "infra2/image_raw", "topic", "sensor_msgs/msg/Image"),
"query_intrinsics": RosInterface(
"", "color/camera_info", "topic", "sensor_msgs/msg/CameraInfo"
),
"set_exposure": RosInterface("", "set_exposure", "service", "std_srvs/srv/SetBool"),
}
# ---------------------------------------------------------------------------
# Requirement sets
# ---------------------------------------------------------------------------
class Requirement(NamedTuple):
"""A named flow and the capabilities it needs, grouped by device kind."""
summary: str
robot: Tuple[str, ...]
gripper: Tuple[str, ...]
sensor: Tuple[str, ...]
camera: Tuple[str, ...]
REQUIREMENTS: Dict[str, Requirement] = {
"tracking": Requirement(
summary="视觉跟踪 k0（腕部相机实时推理 + 锁高悬停）",
robot=("get_arm_state", "servo_joint", "servo_cartesian"),
gripper=(),
sensor=(),
camera=("color", "aligned_depth_to_color", "query_intrinsics"),
),
"handeye": Requirement(
summary="手眼标定（走点 + 拍照 + 解 AX=XB）",
robot=("get_arm_state", "follow_joint_trajectory", "soft_drag"),
gripper=(),
sensor=(),
camera=("color", "query_intrinsics"),
),
"coarse_locate": Requirement(
summary="手外粗定位（D435 拍照推理出工件坐标）",
robot=("follow_joint_trajectory",),
gripper=(),
sensor=(),
camera=("color", "depth", "query_intrinsics"),
),
"gauge_assembly": Requirement(
summary="通止规装配（接触 + 旋转拧入 + 力判据停止 + 拧出）",
robot=(
"get_arm_state",
"follow_joint_trajectory",
"set_named_pose",
"servo_cartesian",
"compliant_descent",
"switch_tool",
"home",
"enable",
"clear_fault",
),
gripper=("open", "close", "rotate", "get_status"),
sensor=("read_tangential", "read_normal", "measure_baseline"),
camera=("color", "query_intrinsics"),
),
"gripper_only": Requirement(
summary="仅夹爪控制",
robot=(),
gripper=("open", "close", "get_status"),
sensor=(),
camera=(),
),
"tactile_only": Requirement(
summary="仅触觉采集与图传（调阈值用）",
robot=(),
gripper=(),
sensor=("read_tangential", "read_normal", "read_depth", "read_force_vector"),
camera=(),
),
}
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}（框架未登记的能力名）"
return f"{capability} —— {info.summary}"
build/arm_interfaces/build/lib/arm_interfaces/cli.py
+176
View File
@@ -0,0 +1,176 @@
"""``device_check`` — validate the device stack from the command line.
Usage::
ros2 run arm_interfaces device_check # load everything, print summary
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
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)
=== ==========================================================
"""
from __future__ import annotations
import argparse
import json
import sys
from typing import Any, Dict, List, Optional
from . import paths
from .capabilities import REQUIREMENTS
from .errors import ContractError, format_error
from .loader import check_requirement, load_device_stack
from .registry import REQUIRED_KINDS, list_available_brands
EXIT_OK = 0
EXIT_CONTRACT = 2
EXIT_INTERNAL = 3
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="device_check",
description="校验设备契约：devices/registry.yaml -> manifest.yaml -> config/site.yaml",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"示例:\n"
" device_check # 全量校验并打印摘要\n"
" device_check --requirement handeye # 校验手眼标定所需能力\n"
" device_check --list-brands # 列出每类设备当前可选的品牌\n"
),
)
parser.add_argument(
"--requirement",
action="append",
default=[],
metavar="NAME",
help="额外校验某个流程的能力需求，可重复。可用: " + ", ".join(sorted(REQUIREMENTS)),
)
parser.add_argument(
"--list-brands", action="store_true", help="列出每类设备可选的品牌后退出"
)
parser.add_argument(
"--no-assets",
action="store_true",
help="跳过厂商物料存在性检查（物料还没放好时可用，方便先校验配置）",
)
parser.add_argument("--json", action="store_true", help="以 JSON 输出结果")
return parser
def main(argv: Optional[List[str]] = None) -> int:
args = _build_parser().parse_args(argv)
try:
if args.list_brands:
return _print_brands(as_json=args.json)
stack = load_device_stack(verify_assets=not args.no_assets)
report: Dict[str, Any] = {"ok": True, "devices": {}, "requirements": {}}
for kind in (*REQUIRED_KINDS, "sim"):
manifest = stack.manifests.get(kind)
if manifest is None:
continue
report["devices"][kind] = {
"brand": manifest.id,
"label": manifest.label,
"version": manifest.version,
"capabilities": sorted(stack.capabilities(kind)),
"manifest": str(manifest.path),
}
for name in args.requirement:
satisfied = check_requirement(stack, name)
report["requirements"][name] = {
detail: sorted(caps) for detail, caps in satisfied.items()
}
if args.json:
print(json.dumps(report, ensure_ascii=False, indent=2))
else:
_print_report(stack, report)
return EXIT_OK
except ContractError as err:
if args.json:
print(
json.dumps(
{
"ok": False,
"what": err.what,
"fix_hint": err.fix_hint,
"detail": err.detail,
},
ensure_ascii=False,
indent=2,
)
)
else:
print(format_error(err), file=sys.stderr)
return EXIT_CONTRACT
except Exception as exc: # noqa: BLE001 - the CLI is the last line of defence
print(
f"契约层内部错误（这本身是框架 bug，请上报）: {type(exc).__name__}: {exc}",
file=sys.stderr,
)
return EXIT_INTERNAL
def _print_brands(as_json: bool = False) -> int:
table: Dict[str, List[str]] = {
kind: list_available_brands(kind) for kind in (*REQUIRED_KINDS, "sim")
}
if as_json:
print(json.dumps(table, ensure_ascii=False, indent=2))
return EXIT_OK
print(f"设备目录: {paths.devices_root()}")
print(f"选型文件: {paths.registry_path()}")
print()
for kind, brands in table.items():
shown = ", ".join(brands) if brands else "（无可用品牌）"
print(f" {kind:10s} -> {shown}")
return EXIT_OK
def _print_report(stack, report: Dict[str, Any]) -> None:
print("=" * 68)
print("设备契约校验通过")
print("=" * 68)
print(f"选型文件: {stack.registry.path}")
print(f"现场配置: {stack.site_path}")
print()
for kind, info in report["devices"].items():
print(f"[{kind}] {info['brand']} — {info['label']}")
if info["version"]:
print(f" 版本: {info['version']}")
caps = info["capabilities"]
print(f" 能力({len(caps)}): {', '.join(caps) if caps else '（未声明）'}")
print()
if report["requirements"]:
print("流程能力校验:")
for name, detail in report["requirements"].items():
summary = REQUIREMENTS[name].summary
print(f" [OK] {name} — {summary}")
for kind, caps in detail.items():
print(f" {kind}: {', '.join(caps)}")
print()
print("提示: 加 --requirement <流程名> 可额外校验具体流程，例如 --requirement gauge_assembly")
if __name__ == "__main__":
sys.exit(main())
build/arm_interfaces/build/lib/arm_interfaces/config_guard.py
+246
View File
@@ -0,0 +1,246 @@
"""Validate ``config/site.yaml`` against a manifest's declared ``config_schema``.
Two classes of bug motivated this module:
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.
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.
``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.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict, List, NamedTuple, Optional
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.
_TYPE_CHECKS = {
"str": (str,),
"int": (int,),
"float": (int, float),
"bool": (bool,),
"list": (list,),
"dict": (dict,),
}
class ResolvedValue(NamedTuple):
"""One concrete value found in ``site.yaml`` for a declared field."""
path: str
"""Concrete path, e.g. ``robot.robots[1].ip``."""
value: Any
def load_site(path: Path | None = None) -> Dict[str, Any]:
"""Read ``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.
"""
from . import paths
target = path or paths.site_path()
if not target.is_file():
return {}
data = yaml.safe_load(target.read_text(encoding="utf-8")) or {}
return data if isinstance(data, dict) else {}
def resolve_path(site: Dict[str, Any], site_path: str) -> List[ResolvedValue]:
"""Resolve a (possibly list-expanding) dotted path against parsed 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``, ...
Mappings are supported because ``site.yaml`` registers cameras (and other
named devices) as mappings keyed by device name rather than as sequences.
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.
"""
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`."""
if not tokens:
return [ResolvedValue(path=prefix, value=node)]
token = tokens[0]
rest = tokens[1:]
expand = token.endswith("[]")
key = token[:-2] if expand else token
if not isinstance(node, dict) or key not in node:
return []
child = node[key]
child_path = f"{prefix}.{key}" if prefix else key
if not expand:
return _resolve(child, rest, child_path)
if isinstance(child, list):
out: List[ResolvedValue] = []
for index, item in enumerate(child):
out.extend(_resolve(item, rest, f"{child_path}[{index}]"))
return out
if isinstance(child, dict):
out = []
for name, item in child.items():
out.extend(_resolve(item, rest, f"{child_path}.{name}"))
return out
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)]
def check_config(
manifest: Manifest,
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.
Raises
------
MissingConfigError
A required field has no value in ``site.yaml``.
BadConfigValueError
A field is present but has the wrong YAML type.
"""
resolved: Dict[str, List[ResolvedValue]] = {}
problems: List[str] = []
for field in manifest.config_schema:
values = resolve_path(site, field.site_path)
resolved[field.key] = values
if not values:
if field.required:
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:
problems.append(_missing_message(field, blank_paths))
continue
for item in values:
if _is_blank(item.value):
continue
if not _matches_type(item.value, field.type):
raise BadConfigValueError(
f"config/site.yaml 的 '{item.path}' 类型不对"
f"（清单 '{manifest.id}' 的 '{field.key}' 要求 {field.type}）",
fix_hint=(
f"把 {item.path} 改成 {field.type} 类型的值"
+ (f"，单位是 {field.unit}" if field.unit else "")
+ (f"\n说明: {field.doc}" if field.doc else "")
),
detail=(
f"实际值: {item.value!r}（{type(item.value).__name__}）"
),
)
if problems:
target = site_file or _default_site_file()
raise MissingConfigError(
f"config/site.yaml 缺少 {len(problems)} 项必需配置"
f"（品牌 '{manifest.id}'）",
fix_hint=(
f"编辑 {target}，按下面每一项给出的 site_path 补上对应的键。"
"若整段缺失，可直接照抄 config/site.example.yaml 里的同名段"
),
detail="\n".join(problems),
)
return resolved
def _default_site_file() -> Path:
from . import paths
return paths.site_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():
return True
if isinstance(value, (list, dict)) and len(value) == 0:
return True
return False
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
return isinstance(value, expected)
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.
"""
lines = [f" 缺: {field.key}"]
if blank_paths:
lines.append(" 已有该键但值为空，需要补值的实例:")
for path in blank_paths:
lines.append(f" {path}")
lines.append(f" 去这里补: config/site.yaml -> {field.site_path}")
if field.unit:
lines.append(f" 单位: {field.unit}")
if field.default is not None:
lines.append(f" 建议默认值: {field.default}")
if field.doc:
lines.append(f" 说明: {field.doc}")
return "\n".join(lines)
build/arm_interfaces/build/lib/arm_interfaces/errors.py
+82
View File
@@ -0,0 +1,82 @@
"""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.
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.
"""
from __future__ import annotations
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.
detail:
Optional extra context (paths tried, values found, ...).
"""
def __init__(
self,
what: str,
fix_hint: Optional[str] = None,
detail: Optional[str] = None,
) -> None:
super().__init__(what)
self.what = what
self.fix_hint = fix_hint
self.detail = detail
def __str__(self) -> str: # pragma: no cover - exercised via format_error
return format_error(self)
class RegistryError(ContractError):
"""``devices/registry.yaml`` is missing, unreadable, or incomplete."""
class ManifestError(ContractError):
"""A per-brand ``manifest.yaml`` is missing, unreadable, or malformed."""
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."""
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():
lines.append(f" {line}")
if err.fix_hint:
lines.append(f" 怎么修: {err.fix_hint}")
else:
lines.append(" 怎么修: (契约层未给出修复建议，这本身是框架 bug，请上报)")
return "\n".join(lines)
build/arm_interfaces/build/lib/arm_interfaces/loader.py
+279
View File
@@ -0,0 +1,279 @@
"""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
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.
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
from pathlib import Path
from typing import Dict, List, NamedTuple, Optional, Sequence, Set
from . import paths
from .capabilities import REQUIREMENTS, ROS_INTERFACE, describe
from .config_guard import ResolvedValue, check_config, load_site
from .errors import CapabilityError, ContractError
from .manifest import KIND_SINGULAR, Manifest, check_assets, load_manifest
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>}``."""
config: Dict[str, List[ResolvedValue]]
"""``f"{kind}.{field.key}"`` -> resolved site.yaml values."""
site_path: Path
"""The ``config/site.yaml`` that was validated."""
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)."""
manifest = self.manifests.get(kind)
if manifest is None:
return []
selector = manifest.raw.get("model_selector")
if not selector:
return []
found: List[str] = []
for entry in self.config.get(f"{kind}.{selector}", []):
if isinstance(entry.value, str) and entry.value.strip():
found.append(entry.value.strip())
return found
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.
"""
caps: Set[str] = set(manifest.capabilities)
for model_id in selected_models:
model_caps = manifest.model_capabilities(model_id)
if model_caps:
caps |= set(model_caps)
return caps
def load_device_stack(
registry_path: Optional[Path] = None,
site_file: Optional[Path] = None,
*,
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.
"""
registry = load_registry(registry_path)
site_file = site_file or paths.site_path()
site = load_site(site_file)
manifests: Dict[str, Manifest] = {}
config: Dict[str, List[ResolvedValue]] = {}
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
brand = registry.brand(kind)
try:
manifest = load_manifest(kind, brand)
except ContractError as err:
problems.append(err)
continue
if verify_assets:
try:
check_assets(manifest, brand, kind)
except ContractError as err:
problems.append(err)
continue
manifests[kind] = manifest
try:
resolved = check_config(manifest, site, site_file)
except ContractError as err:
problems.append(err)
continue
for key, values in resolved.items():
config[f"{kind}.{key}"] = values
if problems:
raise _aggregate(problems)
return DeviceStack(
registry=registry,
manifests=manifests,
config=config,
site_path=site_file,
)
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.
"""
if len(problems) == 1:
return problems[0]
blocks: List[str] = []
for index, err in enumerate(problems, start=1):
block = [f"({index}) {err.what}"]
if err.detail:
for line in str(err.detail).splitlines():
block.append(f" {line}")
if err.fix_hint:
block.append(f" 怎么修: {err.fix_hint}")
blocks.append("\n".join(block))
return ContractError(
f"设备契约校验发现 {len(problems)} 类问题（已全部列出，可一次改完）",
fix_hint=(
"按下面每一项分别处理；改完后重新运行: "
"ros2 run arm_interfaces device_check"
),
detail="\n\n".join(blocks),
)
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:
known = ", ".join(sorted(REQUIREMENTS))
raise CapabilityError(
f"未知的流程需求名 '{requirement_name}'",
fix_hint=(
f"可用的流程名: {known}。"
"若需要新流程，先在 src/arm_interfaces/arm_interfaces/capabilities.py 的 "
"REQUIREMENTS 里登记"
),
)
needed_by_kind = {
"robots": requirement.robot,
"grippers": requirement.gripper,
"sensors": requirement.sensor,
"cameras": requirement.camera,
}
satisfied: Dict[str, Set[str]] = {}
missing_blocks: List[str] = []
for kind, needed in needed_by_kind.items():
if not needed:
continue
manifest = stack.manifests.get(kind)
if manifest is None:
missing_blocks.append(
"\n".join(
[
f" {kind}: 未在 devices/registry.yaml 中选型",
f" 该流程需要: {', '.join(needed)}",
" 怎么修: 在 devices/registry.yaml 里为这一类设备选一个品牌",
]
)
)
continue
have = stack.capabilities(kind)
missing = [cap for cap in needed if cap not in have]
satisfied[kind] = {cap for cap in needed if cap in have}
if missing:
missing_blocks.append(_missing_capability_block(manifest, kind, missing, have))
if missing_blocks:
raise CapabilityError(
f"流程 '{requirement_name}'（{requirement.summary}）所需能力不满足",
fix_hint=(
"按下面每一项给出的说明处理：若是选错了品牌/型号，改 "
"devices/registry.yaml 或 config/site.yaml 的型号；"
"若该品牌确实做不到，需要换驱动或换硬件"
),
detail="\n".join(missing_blocks),
)
return satisfied
def _missing_capability_block(
manifest: Manifest, kind: str, missing: Sequence[str], have: Set[str]
) -> str:
lines = [f" {kind}（品牌 '{manifest.id}'）缺少 {len(missing)} 项能力:"]
for cap in missing:
lines.append(f" - {describe(cap)}")
iface = ROS_INTERFACE.get(cap)
if iface is not None:
lines.append(
f" 需要实现的接口: {iface.kind} "
f"'{iface.name}' [{iface.type}]"
)
lines.append(f" 该品牌当前已声明: {', '.join(sorted(have)) or '（无）'}")
lines.append(
f" 怎么修: 编辑 {manifest.path} 的 capabilities；"
"但只有驱动真的实现了才能写进去，否则运行时会失败"
)
return "\n".join(lines)
build/arm_interfaces/build/lib/arm_interfaces/manifest.py
+490
View File
@@ -0,0 +1,490 @@
"""Load and validate a brand's ``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``
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.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict, List, NamedTuple, Optional, Tuple
import yaml
from . import paths
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_SINGULAR: Dict[str, str] = {
"robots": "robot",
"grippers": "gripper",
"sensors": "sensor",
"cameras": "camera",
"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
what: str
fmt: str
naming: str
where_used: str
notes: str
class DriverSpec(NamedTuple):
"""How to start this brand's adapter."""
package: str
executable: str
build: str
notes: str
class ConfigField(NamedTuple):
"""One declared config key, pointing into ``config/site.yaml``."""
key: str
site_path: str
type: str
required: bool
default: Any
unit: Optional[str]
doc: str
class Manifest(NamedTuple):
"""Parsed and validated ``manifest.yaml``."""
id: str
label: str
kind: str
vendor: str
version: str
description: str
driver: DriverSpec
capabilities: Tuple[str, ...]
assets: Tuple[Asset, ...]
config_schema: Tuple[ConfigField, ...]
models: Tuple[Dict[str, Any], ...]
path: Path
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.
Returns ``None`` when the manifest does not enumerate models, which
means the top-level ``capabilities`` list applies to all of them.
"""
for model in self.models:
if model.get("id") == model_id:
caps = model.get("capabilities")
if caps is None:
return None
return tuple(caps)
return None
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``.
brand:
Brand folder name under ``devices/<kind>/``.
"""
target = path or manifest_path(kind, brand)
if not target.is_file():
available = _available_brands(kind)
raise ManifestError(
f"品牌 '{brand}' 的清单文件不存在: {target}",
fix_hint=(
f"确认 devices/{KIND_FOLDERS.get(kind, kind)}/{brand}/manifest.yaml 存在；"
f"或改 devices/registry.yaml 把 '{kind}' 指向一个已存在的品牌"
),
detail=(
f"当前可用品牌: {', '.join(available) if available else '（无）'}"
),
)
try:
data = yaml.safe_load(target.read_text(encoding="utf-8")) or {}
except yaml.YAMLError as exc:
raise ManifestError(
f"{target} 不是合法的 YAML",
fix_hint="检查缩进（不要用 Tab）；多行文本要用 '>' 或 '|' 折叠",
detail=str(exc),
) from exc
if not isinstance(data, dict):
raise ManifestError(
f"{target} 顶层必须是一个映射",
fix_hint="manifest.yaml 应当是 'key: value' 结构",
detail=f"实际类型: {type(data).__name__}",
)
_check_required_keys(data, target)
_check_kind_matches(data, kind, target)
expected_kind = KIND_SINGULAR.get(kind, kind)
capabilities = _parse_capabilities(data, expected_kind, target)
assets = _parse_assets(data, target)
config_schema = _parse_config_schema(data, target)
driver = _parse_driver(data, target)
models = _parse_models(data, target)
_check_model_capabilities(models, expected_kind, target)
return Manifest(
id=str(data["id"]),
label=str(data["label"]),
kind=str(data["kind"]),
vendor=str(data.get("vendor", "")),
version=str(data.get("version", "")),
description=str(data.get("description", "")).strip(),
driver=driver,
capabilities=capabilities,
assets=assets,
config_schema=config_schema,
models=models,
path=target,
raw=data,
)
def check_assets(manifest: Manifest, brand: str, kind: str) -> None:
"""Verify every ``required`` asset exists, with format hints in the error.
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] = []
for asset in manifest.assets:
if not asset.required:
continue
target = base / asset.path
if not _asset_present(target):
missing.append(
"\n".join(
[
f" 缺少: {asset.path}",
f" 放什么: {asset.what or '（清单未说明）'}",
f" 格式要求: {asset.fmt or '（清单未说明）'}",
f" 命名要求: {asset.naming or '（清单未说明）'}",
f" 放好之后: {asset.where_used or '（清单未说明）'}",
]
)
)
if missing:
raise MissingAssetError(
f"品牌 '{brand}' 的必需物料缺失（{len(missing)} 项）",
fix_hint=(
f"把物料放进 devices/{KIND_FOLDERS.get(kind, kind)}/{brand}/ "
f"下对应路径；具体要求见该目录的 README.md"
),
detail="\n".join(missing),
)
# ---------------------------------------------------------------------------
# internal helpers
# ---------------------------------------------------------------------------
def _available_brands(kind: str) -> List[str]:
folder = paths.devices_root() / KIND_FOLDERS.get(kind, kind)
if not folder.is_dir():
return []
return sorted(p.name for p in folder.iterdir() if p.is_dir() and not p.name.startswith("."))
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.
"""
if target.is_symlink():
return target.exists()
return target.exists()
def _check_required_keys(data: Dict[str, Any], path: Path) -> None:
missing = [k for k in REQUIRED_KEYS if not data.get(k)]
if missing:
raise ManifestError(
f"{path} 缺少必需字段: {', '.join(missing)}",
fix_hint=(
"manifest.yaml 必须包含 id / label / kind / driver 四项；"
"driver 下要有 package 与 executable"
),
detail=f"已读到的顶层键: {', '.join(sorted(data)) or '（无）'}",
)
def _check_kind_matches(data: Dict[str, Any], registry_kind: str, path: Path) -> None:
expected = KIND_SINGULAR.get(registry_kind, registry_kind)
actual = str(data.get("kind", ""))
if actual != expected:
raise ManifestError(
f"{path} 的 kind 字段与所在目录不符",
fix_hint=(
f"把 'kind:' 改成 '{expected}'（devices/{KIND_FOLDERS.get(registry_kind, registry_kind)}/ "
f"下的清单都必须是这个值）"
),
detail=f"实际写的 kind: '{actual}'",
)
def _parse_capabilities(
data: Dict[str, Any], kind: str, path: Path
) -> Tuple[str, ...]:
raw = data.get("capabilities") or []
if not isinstance(raw, list):
raise ManifestError(
f"{path} 的 capabilities 必须是列表",
fix_hint="写成多行 '-' 列表，例如:\n capabilities:\n - open\n - close",
detail=f"实际类型: {type(raw).__name__}",
)
allowed = capabilities_for_kind(kind)
unknown = [c for c in raw if c not in allowed]
if unknown:
known_code = [c for c in raw if c in allowed]
raise ManifestError(
f"{path} 声明了框架不认识的 {kind} 能力: {', '.join(unknown)}",
fix_hint=(
f"{kind} 能声明的能力只有: {', '.join(sorted(allowed))}。"
"若确实需要新能力，先在 src/arm_interfaces/arm_interfaces/capabilities.py "
"里登记它，再写进清单"
),
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(
f"{path} 的能力声明有重复: {', '.join(duplicates)}",
fix_hint="删掉重复项（每项只写一次）",
)
return tuple(raw)
def _parse_assets(data: Dict[str, Any], path: Path) -> Tuple[Asset, ...]:
raw = data.get("assets") or []
if not isinstance(raw, list):
raise ManifestError(
f"{path} 的 assets 必须是列表",
fix_hint="写成多行 '-' 列表，每项含 path / required / what / format / naming",
detail=f"实际类型: {type(raw).__name__}",
)
assets: List[Asset] = []
for index, item in enumerate(raw):
if not isinstance(item, dict):
raise ManifestError(
f"{path} 的 assets 第 {index + 1} 项不是映射",
fix_hint="每一项都要写成 'path: rdk' 这样的键值对",
detail=f"实际值: {item!r}",
)
rel = item.get("path")
if not rel or not isinstance(rel, str):
raise ManifestError(
f"{path} 的 assets 第 {index + 1} 项缺少 path",
fix_hint="补上 'path:'，值是相对于品牌目录的相对路径（如 'rdk'）",
)
assets.append(
Asset(
path=rel.strip().rstrip("/"),
required=bool(item.get("required", False)),
what=str(item.get("what", "")).strip(),
fmt=str(item.get("format", "")).strip(),
naming=str(item.get("naming", "")).strip(),
where_used=str(item.get("where_used", "")).strip(),
notes=str(item.get("notes", "")).strip(),
)
)
return tuple(assets)
def _parse_config_schema(data: Dict[str, Any], path: Path) -> Tuple[ConfigField, ...]:
raw = data.get("config_schema") or []
if not isinstance(raw, list):
raise ManifestError(
f"{path} 的 config_schema 必须是列表",
fix_hint="每项形如:\n - key: port\n site_path: gripper.port\n type: str\n required: true",
detail=f"实际类型: {type(raw).__name__}",
)
fields: List[ConfigField] = []
seen: Dict[str, int] = {}
for index, item in enumerate(raw):
if not isinstance(item, dict):
raise ManifestError(
f"{path} 的 config_schema 第 {index + 1} 项不是映射",
fix_hint="每项都要是 'key: ... / site_path: ...' 形式的键值对",
detail=f"实际值: {item!r}",
)
key = item.get("key")
site_path = item.get("site_path")
if not key or not isinstance(key, str):
raise ManifestError(
f"{path} 的 config_schema 第 {index + 1} 项缺少 key",
fix_hint="补上 'key:'，它是这项配置在清单内部的标识名",
)
if not site_path or not isinstance(site_path, str):
raise ManifestError(
f"{path} 的 config_schema 的 '{key}' 缺少 site_path",
fix_hint=(
f"补上 'site_path:'，告诉框架去 config/site.yaml 的哪个键取值，"
f"例如 'gripper.port' 或 'robot.robots[].gripper.port'"
),
)
if key in seen:
raise ManifestError(
f"{path} 的 config_schema 里 '{key}' 重复声明（第 {seen[key]} 项与第 {index + 1} 项）",
fix_hint="同一 key 只能声明一次",
)
seen[key] = index + 1
value_type = str(item.get("type", "str"))
if value_type not in KNOWN_TYPES:
raise ManifestError(
f"{path} 的 config_schema 的 '{key}' 用了未知类型 '{value_type}'",
fix_hint=f"type 只能是: {', '.join(sorted(KNOWN_TYPES))}",
)
required = bool(item.get("required", False))
default = item.get("default")
if required and default is not None:
raise ManifestError(
f"{path} 的 config_schema 的 '{key}' 同时要求必填又给了默认值",
fix_hint=(
"二者只能留一个：真必填就删掉 default；有合理默认值就把 required 改成 false"
),
)
unit = item.get("unit")
fields.append(
ConfigField(
key=key,
site_path=site_path.strip(),
type=value_type,
required=required,
default=default,
unit=str(unit).strip() if unit else None,
doc=str(item.get("doc", "")).strip(),
)
)
return tuple(fields)
def _parse_driver(data: Dict[str, Any], path: Path) -> DriverSpec:
raw = data.get("driver")
if not isinstance(raw, dict):
raise ManifestError(
f"{path} 的 driver 必须是映射",
fix_hint="写成:\n driver:\n package: arm_driver_flexiv\n executable: flexiv_arm_driver",
detail=f"实际类型: {type(raw).__name__}",
)
package = raw.get("package")
executable = raw.get("executable")
if not package or not executable:
raise ManifestError(
f"{path} 的 driver 缺少 package 或 executable",
fix_hint="两个都要写：package 是 ROS 包名，executable 是节点可执行名",
)
return DriverSpec(
package=str(package),
executable=str(executable),
build=str(raw.get("build", "colcon")),
notes=str(raw.get("notes", "")).strip(),
)
def _parse_models(data: Dict[str, Any], path: Path) -> Tuple[Dict[str, Any], ...]:
raw = data.get("models") or []
if not isinstance(raw, list):
raise ManifestError(
f"{path} 的 models 必须是列表",
fix_hint="每个型号一项，形如:\n - id: d405\n label: D405 腕部相机",
detail=f"实际类型: {type(raw).__name__}",
)
ids: List[str] = []
for index, item in enumerate(raw):
if not isinstance(item, dict) or not item.get("id"):
raise ManifestError(
f"{path} 的 models 第 {index + 1} 项缺少 id",
fix_hint="每个型号都要有唯一的 'id:'（会被 config/site.yaml 引用）",
detail=f"实际值: {item!r}",
)
ids.append(str(item["id"]))
duplicates = sorted({i for i in ids if ids.count(i) > 1})
if duplicates:
raise ManifestError(
f"{path} 的 models 有重复 id: {', '.join(duplicates)}",
fix_hint="型号 id 必须唯一，因为它要被 site.yaml 引用",
)
return tuple(raw)
def _check_model_capabilities(
models: Tuple[Dict[str, Any], ...], kind: str, path: Path
) -> None:
allowed = capabilities_for_kind(kind)
for model in models:
caps = model.get("capabilities")
if caps is None:
continue
if not isinstance(caps, list):
raise ManifestError(
f"{path} 的型号 '{model.get('id')}' 的 capabilities 必须是列表",
fix_hint="写成多行 '-' 列表",
)
unknown = [c for c in caps if c not in allowed]
if unknown:
raise ManifestError(
f"{path} 的型号 '{model.get('id')}' 声明了未知能力: {', '.join(unknown)}",
fix_hint=f"{kind} 能声明的能力只有: {', '.join(sorted(allowed))}",
)
build/arm_interfaces/build/lib/arm_interfaces/paths.py
+71
View File
@@ -0,0 +1,71 @@
"""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.
Resolution order (first non-empty wins):
1. Explicit environment override (``SITE_ROS2_ROOT`` / ``SITE_CONFIG``)
2. ``ROS2_WS`` (exported by ``scripts/env.sh``)
3. ``~/ros2``
"""
from __future__ import annotations
import os
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:
return Path(value).expanduser()
return Path.home() / "ros2"
def site_path() -> Path:
"""Return the path to the master config ``config/site.yaml``."""
override = os.environ.get("SITE_CONFIG", "").strip()
if override:
return Path(override).expanduser()
return ros2_root() / "config" / "site.yaml"
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()
return ros2_root() / "devices"
def registry_path() -> Path:
"""Return the path to the device selection file ``devices/registry.yaml``."""
return devices_root() / "registry.yaml"
def assets_root() -> Path:
"""Return the root of the model-weight asset tree (``assets/``)."""
override = os.environ.get("SITE_ASSETS_ROOT", "").strip()
if override:
return Path(override).expanduser()
return ros2_root() / "assets"
def handeye_root() -> Path:
"""Return the root of the hand-eye module directory (``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``).
"""
return devices_root() / kind / brand
build/arm_interfaces/build/lib/arm_interfaces/registry.py
+130
View File
@@ -0,0 +1,130 @@
"""Load and validate ``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.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict, NamedTuple
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``.
KIND_FOLDERS: Dict[str, str] = {
"robots": "robots",
"grippers": "grippers",
"sensors": "sensors",
"cameras": "cameras",
"sim": "sim",
}
#: Every kind the registry must select. ``sim`` is optional (headless setups).
REQUIRED_KINDS = ("robots", "grippers", "sensors", "cameras")
OPTIONAL_KINDS = ("sim",)
class Registry(NamedTuple):
"""Parsed ``devices/registry.yaml``."""
selections: Dict[str, str]
"""kind -> brand id, e.g. ``{"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:
raise RegistryError(
f"{self.path} 里没有为 '{kind}' 选择驱动",
fix_hint=(
f"编辑 devices/registry.yaml，把 '{kind}:' 设成 "
f"devices/{KIND_FOLDERS[kind]}/ 下的目录名"
),
) from None
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()
if not target.is_file():
raise RegistryError(
f"找不到设备选型文件: {target}",
fix_hint=(
"确认 devices/registry.yaml 存在。若是新克隆的工作区，"
"检查 devices/ 目录是否完整"
),
detail=f"期望路径: {target}",
)
try:
data = yaml.safe_load(target.read_text(encoding="utf-8")) or {}
except yaml.YAMLError as exc:
raise RegistryError(
f"{target} 不是合法的 YAML",
fix_hint="检查缩进（YAML 不允许用 Tab）与冒号后的空格",
detail=str(exc),
) from exc
if not isinstance(data, dict):
raise RegistryError(
f"{target} 顶层必须是一个映射（key: value）",
fix_hint="文件开头应当形如 'robots: flexiv'，而不是列表或纯文本",
detail=f"实际读到的类型: {type(data).__name__}",
)
selections: Dict[str, str] = {}
for kind in (*REQUIRED_KINDS, *OPTIONAL_KINDS):
value = data.get(kind)
if value is None:
if kind in OPTIONAL_KINDS:
continue
raise RegistryError(
f"{target} 缺少 '{kind}' 选型",
fix_hint=(
f"在 devices/registry.yaml 里加一行 '{kind}: <品牌目录名>'，"
f"可选值为 devices/{KIND_FOLDERS[kind]}/ 下的目录名"
),
)
if not isinstance(value, str) or not value.strip():
raise RegistryError(
f"{target} 的 '{kind}' 必须是非空字符串",
fix_hint=f"把 '{kind}:' 后面改成一个品牌目录名（不要写成列表）",
detail=f"实际值: {value!r}",
)
selections[kind] = value.strip()
return Registry(selections=selections, path=target, raw=data)
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 []
return sorted(
p.name
for p in folder.iterdir()
if p.is_dir() and not p.name.startswith(".")
)
build/arm_interfaces/colcon_build.rc
+1
View File
@@ -0,0 +1 @@
0
build/arm_interfaces/colcon_command_prefix_setup_py.sh
+1
View File
@@ -0,0 +1 @@
# generated from colcon_core/shell/template/command_prefix.sh.em
build/arm_interfaces/colcon_command_prefix_setup_py.sh.env
+97
View File
File diff suppressed because one or more lines are too long
build/arm_interfaces/package.xml
Symlink
+1
View File
@@ -0,0 +1 @@
../../src/arm_interfaces/package.xml
build/arm_interfaces/prefix_override/sitecustomize.py
+4
View File
@@ -0,0 +1,4 @@
import sys
if sys.prefix == '/home/cc/anaconda3/envs/ros_flexiv':
sys.real_prefix = sys.prefix
sys.prefix = sys.exec_prefix = '/home/cc/ros2/install/arm_interfaces'
build/arm_interfaces/resource/arm_interfaces
Symlink
+1
View File
@@ -0,0 +1 @@
../../../src/arm_interfaces/resource/arm_interfaces
build/arm_interfaces/setup.cfg
Symlink
+1
View File
@@ -0,0 +1 @@
../../src/arm_interfaces/setup.cfg
build/arm_interfaces/setup.py
Symlink
+1
View File
@@ -0,0 +1 @@
../../src/arm_interfaces/setup.py
build/arm_interfaces/share/arm_interfaces/hook/pythonpath_develop.dsv
+1
View File
@@ -0,0 +1 @@
prepend-non-duplicate;PYTHONPATH;/home/cc/ros2/build/arm_interfaces
build/arm_interfaces/share/arm_interfaces/hook/pythonpath_develop.ps1
+3
View File
@@ -0,0 +1,3 @@
# generated from colcon_powershell/shell/template/hook_prepend_value.ps1.em
colcon_prepend_unique_value PYTHONPATH "$env:COLCON_CURRENT_PREFIX//home/cc/ros2/build/arm_interfaces"
build/arm_interfaces/share/arm_interfaces/hook/pythonpath_develop.sh
+3
View File
@@ -0,0 +1,3 @@
# generated from colcon_core/shell/template/hook_prepend_value.sh.em
_colcon_prepend_unique_value PYTHONPATH "/home/cc/ros2/build/arm_interfaces"