forked from YikaiFu-cart/acRealman_xr
refactor launch_ui.py and update README.md
This commit is contained in:
@ -1,7 +1,6 @@
|
||||
"""xr_rm_teleop 包安装配置。
|
||||
|
||||
该包提供基于 XR 相对位移的 RM75 笛卡尔速度遥操作节点,以及把
|
||||
手柄 trigger 映射到夹爪力控命令的桥接节点。
|
||||
该包提供基于 XR 相对位移的 RM75 笛卡尔速度遥操作节点。
|
||||
"""
|
||||
|
||||
from setuptools import setup
|
||||
@ -25,7 +24,6 @@ setup(
|
||||
tests_require=["pytest"],
|
||||
entry_points={
|
||||
"console_scripts": [
|
||||
"gripper_trigger_bridge = xr_rm_teleop.gripper_trigger_bridge:main",
|
||||
"single_arm_velocity_teleop = xr_rm_teleop.single_arm_velocity_teleop:main",
|
||||
],
|
||||
},
|
||||
|
||||
@ -1,119 +0,0 @@
|
||||
"""手柄 trigger 到夹爪力控指令的桥接节点。
|
||||
|
||||
订阅 XR 手柄消息,将 trigger 压力映射为 OmniPicker 可使用的归一化
|
||||
闭合力比例,并在消息超时或未握持时自动回到开爪指令。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import rclpy
|
||||
from rclpy.node import Node
|
||||
from std_msgs.msg import Float32
|
||||
|
||||
from xr_rm_interfaces.msg import XrController
|
||||
|
||||
|
||||
def _clamp(value: float, low: float, high: float) -> float:
|
||||
return min(max(value, low), high)
|
||||
|
||||
|
||||
class GripperTriggerBridge(Node):
|
||||
"""将 PICO 扳机键映射为 OmniPicker 可订阅的归一化力控指令。"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__("gripper_trigger_bridge")
|
||||
|
||||
self.declare_parameter("controller_topic", "/xr/right_controller")
|
||||
self.declare_parameter("command_topic", "/omnipicker/right/force_ratio")
|
||||
self.declare_parameter("publish_rate_hz", 30.0)
|
||||
self.declare_parameter("stale_timeout_sec", 0.25)
|
||||
self.declare_parameter("require_grip", True)
|
||||
# 扳机键输入小于死区时保持开爪,避免手柄轻微误触导致夹爪闭合。
|
||||
self.declare_parameter("trigger_deadband", 0.05)
|
||||
self.declare_parameter("open_command", 0.0)
|
||||
# 阶段一以保护番茄为先,闭合力控上限应从低值开始实测。
|
||||
self.declare_parameter("min_close_command", 0.15)
|
||||
self.declare_parameter("max_close_command", 0.45)
|
||||
self.declare_parameter("low_pass_alpha", 0.5)
|
||||
|
||||
controller_topic = str(self.get_parameter("controller_topic").value)
|
||||
command_topic = str(self.get_parameter("command_topic").value)
|
||||
publish_rate_hz = float(self.get_parameter("publish_rate_hz").value)
|
||||
|
||||
self._stale_timeout_sec = float(self.get_parameter("stale_timeout_sec").value)
|
||||
self._require_grip = bool(self.get_parameter("require_grip").value)
|
||||
self._trigger_deadband = float(self.get_parameter("trigger_deadband").value)
|
||||
self._open_command = float(self.get_parameter("open_command").value)
|
||||
self._min_close_command = float(self.get_parameter("min_close_command").value)
|
||||
self._max_close_command = float(self.get_parameter("max_close_command").value)
|
||||
self._low_pass_alpha = float(self.get_parameter("low_pass_alpha").value)
|
||||
self._validate_parameters()
|
||||
|
||||
self._last_msg: XrController | None = None
|
||||
self._last_msg_time = None
|
||||
self._filtered_command = self._open_command
|
||||
|
||||
self._publisher = self.create_publisher(Float32, command_topic, 10)
|
||||
self.create_subscription(XrController, controller_topic, self._on_controller, 10)
|
||||
self.create_timer(1.0 / publish_rate_hz, self._publish_command)
|
||||
|
||||
self.get_logger().info(
|
||||
f"夹爪 trigger 桥接已启动:{controller_topic} -> {command_topic}"
|
||||
)
|
||||
|
||||
def _on_controller(self, msg: XrController) -> None:
|
||||
self._last_msg = msg
|
||||
self._last_msg_time = self.get_clock().now()
|
||||
|
||||
def _publish_command(self) -> None:
|
||||
command = self._open_command
|
||||
|
||||
# 手柄消息超时后立即退回开爪,避免夹爪在通信中断时保持闭合力。
|
||||
if self._last_msg is not None and self._last_msg_time is not None:
|
||||
age = (self.get_clock().now() - self._last_msg_time).nanoseconds * 1e-9
|
||||
if age <= self._stale_timeout_sec:
|
||||
command = self._command_from_trigger(self._last_msg)
|
||||
|
||||
# 对夹爪指令做一阶平滑,避免扳机键抖动直接传到力控命令。
|
||||
alpha = self._low_pass_alpha
|
||||
self._filtered_command = alpha * command + (1.0 - alpha) * self._filtered_command
|
||||
msg = Float32()
|
||||
msg.data = float(self._filtered_command)
|
||||
self._publisher.publish(msg)
|
||||
|
||||
def _command_from_trigger(self, msg: XrController) -> float:
|
||||
if self._require_grip and not msg.grip:
|
||||
return self._open_command
|
||||
|
||||
trigger = _clamp(float(msg.trigger), 0.0, 1.0)
|
||||
if trigger <= self._trigger_deadband:
|
||||
return self._open_command
|
||||
|
||||
ratio = (trigger - self._trigger_deadband) / (1.0 - self._trigger_deadband)
|
||||
return self._min_close_command + ratio * (self._max_close_command - self._min_close_command)
|
||||
|
||||
def _validate_parameters(self) -> None:
|
||||
if not 0.0 <= self._trigger_deadband < 1.0:
|
||||
raise ValueError("trigger_deadband must be in [0, 1)")
|
||||
if self._min_close_command < self._open_command:
|
||||
raise ValueError("min_close_command must be >= open_command")
|
||||
if self._max_close_command < self._min_close_command:
|
||||
raise ValueError("max_close_command must be >= min_close_command")
|
||||
if not 0.0 <= self._low_pass_alpha <= 1.0:
|
||||
raise ValueError("low_pass_alpha must be in [0, 1]")
|
||||
|
||||
|
||||
def main(args: list[str] | None = None) -> None:
|
||||
rclpy.init(args=args)
|
||||
node = GripperTriggerBridge()
|
||||
try:
|
||||
rclpy.spin(node)
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
finally:
|
||||
node.destroy_node()
|
||||
rclpy.shutdown()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user