把当前控制栈收成一份可克隆的源码:腕部看不见时用顶视一次定到基座,二阶段对孔和手外标定跟着进同一份树。
现场 site.yaml、权重和编译产物仍不入库;模板只留占位符。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -14,6 +14,7 @@ from rclpy.node import Node
|
||||
|
||||
from arm_msgs.action import FollowJointTrajectory
|
||||
from arm_msgs.srv import SetNamedPose
|
||||
from std_srvs.srv import Trigger
|
||||
|
||||
DEFAULT_JOINTS = [
|
||||
'joint1', 'joint2', 'joint3', 'joint4', 'joint5', 'joint6', 'joint7',
|
||||
@@ -64,53 +65,77 @@ class NamedPoseServer(Node):
|
||||
self.create_service(
|
||||
SetNamedPose, 'set_named_pose', self._on_set, callback_group=self._cb
|
||||
)
|
||||
# ``/home``(std_srvs/Trigger):契约里的"回初始位"入口。
|
||||
#
|
||||
# **为什么这里也要有一个**:真机上 ``/home`` 原本由 ``arm_driver_flexiv``
|
||||
# 提供(它只是个包装,内部还是调 ``/set_named_pose``),但 real 模式的
|
||||
# bringup 起的是 ``flexiv_bringup + named_pose_server + arm_status_node``,
|
||||
# **没有任何 launch 文件会起 arm_driver_flexiv** —— 那个驱动还会自己开
|
||||
# RDK 连接,与 ros2_control 的 flexiv_hardware 抢独占(RDK 是单客户端)。
|
||||
# 于是真机上 ``/home`` 根本不存在,通止规装配的 S00 必检因此永远过不了
|
||||
# (``gauge_boot.py`` 的 ``gauge_home`` 项 target=/home,
|
||||
# ``gauge_assembly_node`` 的 ``_call_trigger("/home")`` 同)。
|
||||
# 放在这里最合适:本节点已经握着 ``/set_named_pose`` 的实现与 JTC 客户端,
|
||||
# 不需要再起一个进程、也不碰 RDK。
|
||||
self.declare_parameter('home_pose', 'home')
|
||||
self.declare_parameter('home_speed', 0.30)
|
||||
self.create_service(
|
||||
Trigger, 'home', self._on_home, callback_group=self._cb
|
||||
)
|
||||
self.get_logger().info(
|
||||
f'命名位姿 {list(self.poses)} joints={self.joint_names}'
|
||||
)
|
||||
|
||||
def _on_set(self, request: SetNamedPose.Request, response: SetNamedPose.Response):
|
||||
name = request.pose_name
|
||||
def _go_to_pose(self, name: str, velocity_scale: float):
|
||||
"""把 ``name`` 这个命名位姿走完。返回 ``(成功?, 说明)``。
|
||||
|
||||
两个服务(``/set_named_pose`` 与 ``/home``)共用这一段:分两份写的话,
|
||||
以后只改一处就会出现"手动回位能动、装配回位不能动"这类难查的偏差。
|
||||
"""
|
||||
if name not in self.poses:
|
||||
response.success = False
|
||||
response.message = f'未知位姿"{name}"。可用: {list(self.poses)}'
|
||||
return response
|
||||
return False, f'未知位姿"{name}"。可用: {list(self.poses)}'
|
||||
if not self._client.wait_for_server(timeout_sec=3.0):
|
||||
response.success = False
|
||||
response.message = 'follow_joint_trajectory action 不可用'
|
||||
return response
|
||||
return False, 'follow_joint_trajectory action 不可用'
|
||||
|
||||
goal = FollowJointTrajectory.Goal()
|
||||
goal.joint_names = self.joint_names
|
||||
goal.num_points = 1
|
||||
goal.positions = list(self.poses[name])
|
||||
duration = max(0.5, 1.0 / max(float(request.velocity_scale), 0.1))
|
||||
duration = max(0.5, 1.0 / max(float(velocity_scale), 0.1))
|
||||
goal.time_from_start = [duration]
|
||||
goal.velocity_scale = float(request.velocity_scale)
|
||||
goal.velocity_scale = float(velocity_scale)
|
||||
|
||||
self.get_logger().info(f'-> 位姿 [{name}] T={duration:.2f}s')
|
||||
send_fut = self._client.send_goal_async(goal)
|
||||
if not _wait_future(send_fut, 5.0):
|
||||
response.success = False
|
||||
response.message = '等待目标被接受超时'
|
||||
return response
|
||||
return False, '等待目标被接受超时'
|
||||
handle = send_fut.result()
|
||||
if handle is None or not handle.accepted:
|
||||
response.success = False
|
||||
response.message = '目标被拒绝'
|
||||
return response
|
||||
return False, '目标被拒绝'
|
||||
res_fut = handle.get_result_async()
|
||||
if not _wait_future(res_fut, 60.0):
|
||||
response.success = False
|
||||
response.message = '等待轨迹结果超时'
|
||||
return response
|
||||
return False, '等待轨迹结果超时'
|
||||
wrapped = res_fut.result()
|
||||
if wrapped is None:
|
||||
response.success = False
|
||||
response.message = '结果为空'
|
||||
return response
|
||||
return False, '结果为空'
|
||||
result = wrapped.result
|
||||
response.success = bool(result.success)
|
||||
response.message = result.message or f'已移动到位姿 {name}'
|
||||
return bool(result.success), (result.message or f'已移动到位姿 {name}')
|
||||
|
||||
def _on_set(self, request: SetNamedPose.Request, response: SetNamedPose.Response):
|
||||
response.success, response.message = self._go_to_pose(
|
||||
request.pose_name, float(request.velocity_scale)
|
||||
)
|
||||
return response
|
||||
|
||||
def _on_home(self, _request: Trigger.Request, response: Trigger.Response):
|
||||
"""``/home``:回到 ``home_pose`` 指定的命名位姿(默认 ``home``)。"""
|
||||
pose = str(self.get_parameter('home_pose').value)
|
||||
speed = float(self.get_parameter('home_speed').value)
|
||||
ok, message = self._go_to_pose(pose, speed)
|
||||
if not ok:
|
||||
self.get_logger().error(f'回初始位失败(pose={pose}): {message}')
|
||||
response.success = ok
|
||||
response.message = message
|
||||
return response
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user