Compare commits
9 Commits
c064cdb191
..
main
| Author | SHA1 | Date | |
|---|---|---|---|
| eaf54420ad | |||
| 2a3fcf868f | |||
| 33578fa83e | |||
| 3507a2d313 | |||
| 0662033f0d | |||
| 543d8a5ccc | |||
| d948883e93 | |||
| 8be8a80cff | |||
| 6393be9518 |
@@ -1 +1,2 @@
|
||||
release/*.zip filter=lfs diff=lfs merge=lfs -text
|
||||
release/*-aarch64-*.tar.gz filter=lfs diff=lfs merge=lfs -text
|
||||
|
||||
@@ -43,7 +43,6 @@ prof/
|
||||
*.log
|
||||
log/
|
||||
debug.txt
|
||||
config/gui_layout.json
|
||||
config/gui_layout.json.tmp
|
||||
test.json
|
||||
|
||||
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 2.5 MiB After Width: | Height: | Size: 2.0 MiB |
+49
-37
@@ -6,31 +6,32 @@
|
||||
"sensor_fps": 30.0,
|
||||
"motion_threshold": 0.0,
|
||||
"sensor_cpu": false,
|
||||
"port": "COM10",
|
||||
"port": "COM5",
|
||||
"slave_id": 1,
|
||||
"baudrate": 115200,
|
||||
"serial_timeout": 0.2,
|
||||
"open_pos": 0,
|
||||
"close_pos": 9000,
|
||||
"speed": 55,
|
||||
"speed": 30,
|
||||
"accel": 70,
|
||||
"decel": 70,
|
||||
"release_open_speed": 55,
|
||||
"release_open_accel": 70,
|
||||
"release_open_decel": 70,
|
||||
"open_force": 25,
|
||||
"close_force": 33,
|
||||
"grip_start_force": 25,
|
||||
"hold_force": 33,
|
||||
"close_force": 70,
|
||||
"grip_start_force": 70,
|
||||
"hold_force": 70,
|
||||
"initial_force": 25,
|
||||
"force_min": 20,
|
||||
"force_max": 99,
|
||||
"force_max": 100,
|
||||
"control_hz": 30.0,
|
||||
"baseline_seconds": 1.0,
|
||||
"filter_alpha": 0.42,
|
||||
"shear_filter_alpha": 0.52,
|
||||
"trigger_normal_force": 0.06,
|
||||
"trigger_shear_force": 0.06,
|
||||
"shear_control_axes": "y",
|
||||
"trigger_normal_force": 0.12,
|
||||
"trigger_shear_force": 0.12,
|
||||
"trigger_clear_stable_seconds": 0.15,
|
||||
"open_wait_stable_samples": 8,
|
||||
"open_wait_stable_range_n": 0.025,
|
||||
@@ -39,47 +40,52 @@
|
||||
"grip_requires_both": true,
|
||||
"close_timeout_seconds": 3.5,
|
||||
"closing_command_interval": 0.2,
|
||||
"force_ramp_step": 1,
|
||||
"force_ramp_interval": 0.1,
|
||||
"force_ramp_step": 5,
|
||||
"force_ramp_interval": 0.03,
|
||||
"grip_settle_seconds": 0.2,
|
||||
"grip_behavior_mode": "stable",
|
||||
"adaptive_grip_enabled": true,
|
||||
"adaptive_shear_start_force": 0.1,
|
||||
"adaptive_shear_full_force": 0.6,
|
||||
"adaptive_force_min": 33,
|
||||
"adaptive_force_max": 80,
|
||||
"adaptive_force_min": 70,
|
||||
"adaptive_force_max": 70,
|
||||
"adaptive_force_step": 5,
|
||||
"adaptive_force_interval": 0.08,
|
||||
"release_arm_delay_seconds": 0.28,
|
||||
"release_min_hold_seconds": 0.65,
|
||||
"release_settle_seconds": 0.12,
|
||||
"release_min_change": 0.08,
|
||||
"release_touch_window_seconds": 0.14,
|
||||
"release_side_touch_change": 0.12,
|
||||
"release_confirm_window_seconds": 0.14,
|
||||
"release_sample_timeout_seconds": 0.2,
|
||||
"release_confirm_samples": 4,
|
||||
"release_slow_confirm_samples": 8,
|
||||
"release_normal_rate": 2.6,
|
||||
"release_shear_rate": 2.4,
|
||||
"release_slow_rate": 0.22,
|
||||
"release_normal_change": 0.42,
|
||||
"release_shear_change": 0.35,
|
||||
"release_curve_history": 30,
|
||||
"release_slope_window_seconds": 0.8,
|
||||
"release_slope_settle_rate": 0.22,
|
||||
"release_slope_min_points": 10,
|
||||
"release_require_slope_settle": true,
|
||||
"release_slope_hold_seconds": 0.45,
|
||||
"adaptive_force_interval": 0.03,
|
||||
"release_arm_delay_seconds": 0.55,
|
||||
"release_min_hold_seconds": 2.0,
|
||||
"release_settle_seconds": 0.22,
|
||||
"release_min_change": 0.12,
|
||||
"release_touch_window_seconds": 0.22,
|
||||
"release_side_touch_change": 0.24,
|
||||
"release_confirm_window_seconds": 0.28,
|
||||
"release_sample_timeout_seconds": 0.3,
|
||||
"release_confirm_samples": 8,
|
||||
"release_slow_confirm_samples": 12,
|
||||
"release_normal_rate": 4.0,
|
||||
"release_shear_rate": 3.6,
|
||||
"release_slow_rate": 0.14,
|
||||
"release_normal_change": 0.7,
|
||||
"release_shear_change": 0.6,
|
||||
"release_curve_history": 42,
|
||||
"release_slope_window_seconds": 1.2,
|
||||
"release_slope_settle_rate": 0.12,
|
||||
"release_slope_min_points": 14,
|
||||
"release_require_slope_settle": false,
|
||||
"release_slope_hold_seconds": 0.9,
|
||||
"release_settle_normal_only": true,
|
||||
"release_shear_oscillation_open": false,
|
||||
"release_shear_oscillation_window_seconds": 0.6,
|
||||
"release_shear_oscillation_min_flips": 5,
|
||||
"release_shear_oscillation_amplitude": 0.3,
|
||||
"release_shear_oscillation_confirm_samples": 4,
|
||||
"release_side_shear_enabled": true,
|
||||
"release_side_normal_enabled": false,
|
||||
"release_shear_onset_enabled": true,
|
||||
"release_normal_onset_enabled": false,
|
||||
"rearm_seconds": 0.55,
|
||||
"rearm_seconds": 0.9,
|
||||
"recover_normal_force": 0.02,
|
||||
"recover_shear_force": 0.02,
|
||||
"recover_stable_seconds": 0.55,
|
||||
"recover_stable_seconds": 0.9,
|
||||
"open_position_recover_enabled": true,
|
||||
"open_position_tolerance": 100,
|
||||
"open_position_stable_seconds": 0.1,
|
||||
@@ -133,5 +139,11 @@
|
||||
"force_n": 2.06
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"release_guard_fast_paths": true,
|
||||
"grip_contact_hold_position": false,
|
||||
"release_shear_axes": "y",
|
||||
"release_shear_displacement": 0.6,
|
||||
"release_shear_displacement_samples": 6,
|
||||
"release_object_lost_open": true
|
||||
}
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
{
|
||||
"version": 1,
|
||||
"window": {
|
||||
"geometry": {
|
||||
"x": 640,
|
||||
"y": 270,
|
||||
"width": 640,
|
||||
"height": 480
|
||||
},
|
||||
"fullscreen": true,
|
||||
"maximized": false
|
||||
},
|
||||
"workspace": {
|
||||
"overlays": {
|
||||
"left_flow": {
|
||||
"x": 0.172414,
|
||||
"y": 0.0,
|
||||
"width": 0.106142,
|
||||
"height": 0.245536
|
||||
},
|
||||
"right_flow": {
|
||||
"x": 0.573815,
|
||||
"y": 0.0,
|
||||
"width": 0.114224,
|
||||
"height": 0.257812
|
||||
},
|
||||
"plot": {
|
||||
"x": 0.746228,
|
||||
"y": 0.0,
|
||||
"width": 0.253772,
|
||||
"height": 0.768973
|
||||
},
|
||||
"gripper": {
|
||||
"x": 0.0,
|
||||
"y": 0.723214,
|
||||
"width": 0.250539,
|
||||
"height": 0.276786
|
||||
},
|
||||
"sensor": {
|
||||
"x": 0.723599,
|
||||
"y": 0.746652,
|
||||
"width": 0.276401,
|
||||
"height": 0.253348
|
||||
}
|
||||
},
|
||||
"camera": {
|
||||
"center": [
|
||||
-0.016087427735328674,
|
||||
0.4006982445716858,
|
||||
0.0
|
||||
],
|
||||
"distance": 1.3952865865983717,
|
||||
"elevation": 9.0,
|
||||
"azimuth": -81.0,
|
||||
"fov": 60.0
|
||||
}
|
||||
},
|
||||
"visual_transforms": {
|
||||
"left": {
|
||||
"rotation_angle": 270,
|
||||
"flip_x": false,
|
||||
"flip_y": false
|
||||
},
|
||||
"right": {
|
||||
"rotation_angle": 270,
|
||||
"flip_x": false,
|
||||
"flip_y": false
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
# Radxa Dragon Q6A(aarch64)运行说明
|
||||
|
||||
适用系统:`radxa-dragon-q6a_noble_gnome`(Ubuntu 24.04 GNOME,aarch64)。
|
||||
|
||||
发布包:`release/GripperDemo02-ubuntu24.04-aarch64-lite-<日期>.tar.gz`
|
||||
|
||||
## 1. 拷贝到板子
|
||||
|
||||
任选一种方式:
|
||||
|
||||
- U 盘拷贝。
|
||||
- `scp`:
|
||||
|
||||
```bash
|
||||
scp GripperDemo02-ubuntu24.04-aarch64-lite-<日期>.tar.gz radxa@<板子IP>:~/
|
||||
```
|
||||
|
||||
- 在板子上 `git clone` 本仓库。发布包使用 Git LFS,需要先安装:
|
||||
|
||||
```bash
|
||||
sudo apt install git-lfs
|
||||
git lfs install
|
||||
```
|
||||
|
||||
可选校验(`.sha256` 文件与压缩包放在同一目录):
|
||||
|
||||
```bash
|
||||
sha256sum -c GripperDemo02-ubuntu24.04-aarch64-lite-<日期>.tar.gz.sha256
|
||||
```
|
||||
|
||||
## 2. 解压
|
||||
|
||||
```bash
|
||||
tar xzf GripperDemo02-ubuntu24.04-aarch64-lite-<日期>.tar.gz
|
||||
```
|
||||
|
||||
解压后的 `GripperDemo02/` 目录结构:
|
||||
|
||||
```
|
||||
GripperDemo02/
|
||||
├─ GripperDemo02 # 可执行程序(ELF aarch64)
|
||||
├─ _internal/ # 运行库,不要删除
|
||||
├─ config/ # gripper.json、传感器配置
|
||||
├─ sdk/
|
||||
├─ crt_ctag2f90d_gripper_visualization/
|
||||
└─ gripper/ # logo
|
||||
```
|
||||
|
||||
整个目录必须一起保留,不能只拷贝 `GripperDemo02` 单个文件。
|
||||
|
||||
## 3. 串口和摄像头权限(首次)
|
||||
|
||||
```bash
|
||||
sudo usermod -aG dialout,video $USER
|
||||
```
|
||||
|
||||
执行后注销并重新登录一次。
|
||||
|
||||
## 4. 运行
|
||||
|
||||
在板子桌面上打开终端:
|
||||
|
||||
```bash
|
||||
cd GripperDemo02
|
||||
./GripperDemo02
|
||||
```
|
||||
|
||||
通过 SSH 远程启动、界面显示在板子屏幕上时:
|
||||
|
||||
```bash
|
||||
export DISPLAY=:0
|
||||
cd GripperDemo02
|
||||
./GripperDemo02
|
||||
```
|
||||
|
||||
## 5. 设备配置
|
||||
|
||||
- 串口:默认 `/dev/ttyUSB0`。可在界面中搜索串口,或修改 `config/gripper.json` 的 `port`。查看板子上的串口:
|
||||
|
||||
```bash
|
||||
ls /dev/ttyUSB* /dev/ttyACM*
|
||||
```
|
||||
|
||||
- 摄像头:程序自动扫描 `/dev/video*`。Linux 下一个 USB 摄像头可能出现两个节点,只有其中一个能出图,两路传感器可能是 `/dev/video0` 和 `/dev/video2`。画面异常时在界面中切换摄像头。
|
||||
|
||||
## 6. 日志
|
||||
|
||||
每次运行的日志在 `GripperDemo02/logs/<时间>_<pid>/main.log`,排查问题时提供该文件。
|
||||
@@ -1,15 +1,29 @@
|
||||
# 释放控制与回放验证
|
||||
|
||||
## 行为模式(迅速开合 / 夹持稳定)
|
||||
## 行为模式(灵敏 / 较稳定 / 十分稳定)
|
||||
|
||||
设置页「行为模式」与主界面「迅速开合 / 夹持稳定」按钮共用 `grip_behavior_mode`:
|
||||
设置里可切换三套预设(`grip_behavior_mode`),切换后改写**夹持力**与**释放/开合**相关参数;
|
||||
闭合触发(`trigger_*`、等待再武装稳定门槛)一律固定为灵敏档,闭合速度各档相同。
|
||||
|
||||
| 模式 | 目标 | 典型倾向 |
|
||||
| 模式 | 夹持力 | 释放 |
|
||||
| --- | --- | --- |
|
||||
| `rapid`(迅速开合) | 跟手开合 | `release_min_hold=0`、更低侧触/确认门槛、更快再武装、关闭自适应加力 |
|
||||
| `stable`(夹持稳定) | 夹得住、少误开 | 更长接触保护、更高确认帧数与速率门槛、开启自适应;**长历史斜率趋零后再允许侧触/监视** |
|
||||
| `sensitive`(灵敏) | 起步 25%,按切向负载自适应 33~80% | 跟手;保护期 0.4 s(`release_min_hold_seconds=0.4`),振荡/侧触在保护期内也生效,法向也可张开;张开速度 80% |
|
||||
| `stable`(较稳定) | 闭合指令即 70%,接触后不锁位置(`grip_contact_hold_position=false`),继续以 70% 朝闭合方向顶住,不再加力 | 接触后 0.7 s 保护期内不触发任何张开(`release_min_hold_seconds=0.7`,不再额外等法向夹稳);张开只看 Y 切向拟合偏移:保护期结束时记下两侧带符号 Y 的拟合值作基准,之后任一侧拟合值偏离基准 ≥ 0.6 N 连续 4 帧即张开(`release_shear_displacement=0.6`,`release_shear_displacement_samples=4`,张开速度 80%,不取绝对值、回头不取消,基准只在接近时缓慢跟随漂移);法向、切向振荡和侧触/突变/触碰路径都不参与。夹持中夹爪走到闭合极限(物体已脱出)立即张开(`release_object_lost_open=true`,不受保护期限制) |
|
||||
| `firm`(十分稳定) | 同较稳定档的恒力方式,但闭合指令即 100% | 与较稳定档相同的张开方式,保护期 1.0 s(`release_min_hold_seconds=1.0`),Y 拟合偏移阈值 0.8 N 连续 6 帧(`release_shear_displacement=0.8`,`release_shear_displacement_samples=6`),张开速度 80%;同样支持物体脱出张开 |
|
||||
|
||||
预设定义见 `gripper/config/behavior_profiles.py`。切换模式会改写释放/滤波/爬力相关字段;可在设置里再微调后保存。运行中切换会即时写入 live `config` 并重置 `ReleaseWatch`。
|
||||
满力需要 `force_max` ≥ 100。
|
||||
|
||||
预设见 `gripper/config/behavior_profiles.py`。旧配置里的 `rapid` /「迅速开合」映射为 `sensitive`。
|
||||
|
||||
### 稳定度配置条与自定义挡位
|
||||
|
||||
设置 →「行为模式」页的稳定度条(`grip_behavior_level`,0~100)可整体拖动调整,三挡是固定刻度点:灵敏 = 0,较稳定 = 50,十分稳定 = 100,刻度点上的参数与上表完全一致。
|
||||
|
||||
- 0~49:灵敏算法(`grip_behavior_mode=sensitive`)。数值参数(速度、自适应夹持力、保护期、确认帧数、门槛等)按比例从灵敏值过渡到较稳定值;开关类参数和 Y 偏移张开阈值保持灵敏档设置。
|
||||
- 50~100:稳定算法(50~99 为 `stable`,100 为 `firm`)。恒力、保护期、Y 拟合偏移阈值按比例从较稳定过渡到十分稳定(例如 75 = 85% / 0.85 s / 0.7 N / 5 帧)。
|
||||
- 闭合触发参数在任何位置都固定为灵敏档。
|
||||
|
||||
拖到合适位置后点「保存为新挡位」并命名,挡位立即写入 `gripper.json` 的 `grip_behavior_presets`(`[{name, level}]`),出现在预设下拉框和刻度条上;选中自定义挡位后可删除,默认三挡不能删除。拖动或选择挡位后需点「保存到 gripper.json」才会应用到运行中的控制。
|
||||
|
||||
## 会话录制(优化张开用)
|
||||
|
||||
@@ -38,7 +52,7 @@
|
||||
具体处理:
|
||||
|
||||
1. 闭合阶段保留最近几帧,接触时继承各通道的趋势,不把接触帧当作静止起点。
|
||||
2. 每个通道保留可配置长度的历史(`release_curve_history`;rapid 约 6 点,stable 约 30 点)。
|
||||
2. 每个通道保留可配置长度的历史(`release_curve_history`,stable 约 30 点)。
|
||||
**起始检测**仍只用最近约 6 点做局部二次拟合;**夹稳斜率**用更长窗口(`release_slope_window_seconds`)
|
||||
做一次拟合,得到 `fit_slope`(N/s)。夹住加力过程表现为该斜率逐渐变小并趋近于 0。
|
||||
3. 自然减速、变平、相近速度的斜率变号是曲线延续;它们不因累计位移达到门槛而启动释放候选。
|
||||
@@ -52,12 +66,15 @@
|
||||
后续自然变化。已达幅度的候选若明显回落则取消,记录 `settling-turn`;回落判定容许
|
||||
当前噪声范围内的波动,避免真实触碰后的轻微抖动被误拒绝。
|
||||
同一套起始判断覆盖 `side_touch`、`touch`、`sudden` 和弱路径,避免其他分支绕过。
|
||||
7. **`release_require_slope_settle`(stable 默认开)**:仅看法向通道长窗 `|fit_slope|`
|
||||
7. **`release_require_slope_settle`(默认开)**:仅看法向通道长窗 `|fit_slope|`
|
||||
(`release_settle_normal_only`)低于 `release_slope_settle_rate`,并连续保持
|
||||
`release_slope_hold_seconds`,**首次**达到后闩锁为已夹稳。之后自适应加力或主动
|
||||
张开使斜率再升高,**不会**撤销释放武装。
|
||||
8. **stable 干净分工**:夹稳只看法向;张开只走切向(`release_normal_onset_enabled=false`、
|
||||
`release_side_normal_enabled=false`,切向 onset/侧触开启)。rapid 仍法向+切向都能开。
|
||||
8. **通道开关**:默认法向与切向均允许 onset 与侧触;可用
|
||||
`release_*_onset_enabled` / `release_side_*_enabled` 单独关闭。夹稳判定仍可只看法向
|
||||
(`release_settle_normal_only`)。
|
||||
9. **切向振荡旁路(默认开)**:短窗内切向反复换向且峰峰值够大时,走
|
||||
`shear_oscillation`,**不要求**已夹稳闩锁,可在 `force-settling` 期间直接确认开合。
|
||||
|
||||
没有增加固定夹稳等待,也没有恢复“四通道同时安静”的全局条件。以下是幅度与确认路径:
|
||||
|
||||
@@ -66,6 +83,7 @@
|
||||
| 即时侧触 `side_touch` | 继承接触前左右力差曲线,检测偏离该曲线的新侧触 | 默认两次左右均更新的传感器帧;不等 0.28s 夹稳计时 |
|
||||
| 普通轻触 `touch` | 出现新起始变化后,实际位移和相对原曲线的偏移均达到 0.04N | 默认两帧;不要求瞬时速率达到 1.6N/s |
|
||||
| 突变 `sudden` / 弱路径 `slow_external` | 新起始变化叠加大速率 / 大幅偏移 | 默认两帧 / 五帧 |
|
||||
| 切向振荡 `shear_oscillation` | 切向短窗内多次换向 + 峰峰值 | 默认两帧;绕过斜率夹稳门控 |
|
||||
|
||||
即时侧触使用左减右的法向/切向力差的变化,抵消两侧共同压紧或负载变化。命令排队、
|
||||
执行及短响应尾部内,法向差分仍受保护,切向差分继续检测。普通共同法向卸载仍经过
|
||||
@@ -113,17 +131,22 @@
|
||||
| `release_slow_confirm_samples` | 5 | 弱路径确认帧数,不少于突变路径 |
|
||||
| `release_confirm_window_seconds` | 0.2s | 候选最长确认时间;超时丢弃 |
|
||||
| `release_sample_timeout_seconds` | 0.2s | 任一侧断流或时间戳间隔过大时,丢弃已有证据 |
|
||||
| `release_curve_history` | 6 / stable 30 | 力曲线保留点数;onset 仍用最近约 6 点,斜率门控用更长窗 |
|
||||
| `release_curve_history` | 30 | 力曲线保留点数;onset 仍用最近约 6 点,斜率门控用更长窗 |
|
||||
| `release_slope_window_seconds` | 0.8s | 夹稳斜率估计时间窗 |
|
||||
| `release_slope_settle_rate` | 0.2 N/s | `|fit_slope|` 低于此值才视为加力趋稳 |
|
||||
| `release_slope_min_points` | 8 | 斜率门控生效前最少历史点数 |
|
||||
| `release_require_slope_settle` | false / stable true | 是否在侧触与退出 settle 前要求斜率趋零 |
|
||||
| `release_slope_hold_seconds` | 0 / stable 0.45s | 斜率达标后还需保持多久(滤掉短刺) |
|
||||
| `release_require_slope_settle` | true | 是否在侧触与退出 settle 前要求斜率趋零 |
|
||||
| `release_slope_hold_seconds` | 0.45s | 斜率达标后还需保持多久(滤掉短刺) |
|
||||
| `release_settle_normal_only` | true | 夹稳只看法向,不看切向 |
|
||||
| `release_shear_oscillation_open` | true | 切向振荡时不待夹稳也开合 |
|
||||
| `release_shear_oscillation_window_seconds` | 0.4 | 切向振荡统计窗口 |
|
||||
| `release_shear_oscillation_min_flips` | 3 | 窗口内最少换向次数 |
|
||||
| `release_shear_oscillation_amplitude` | 0.15 N | 切向峰峰值门槛 |
|
||||
| `release_shear_oscillation_confirm_samples` | 2 | 振荡确认帧数 |
|
||||
| `release_side_shear_enabled` | true | 是否允许切向左右差侧触 |
|
||||
| `release_side_normal_enabled` | true / stable false | 是否允许法向左右差侧触 |
|
||||
| `release_side_normal_enabled` | true | 是否允许法向左右差侧触 |
|
||||
| `release_shear_onset_enabled` | true | 是否允许切向触发普通释放路径 |
|
||||
| `release_normal_onset_enabled` | true / stable false | 是否允许法向触发普通释放路径 |
|
||||
| `release_normal_onset_enabled` | true | 是否允许法向触发普通释放路径 |
|
||||
|
||||
N/N/s 适用于当前标定配置。缓存帧不重复滤波或投票,提高控制频率不能代替真实传感器帧。
|
||||
`grip_settle_seconds` 只控制外围状态进入 `hold_check` 的最早时间。
|
||||
|
||||
+331
-119
@@ -1,31 +1,51 @@
|
||||
"""Grip behavior presets: extreme rapid open/close vs stable hold."""
|
||||
"""Grip behavior presets: sensitive / stable / firm (grip force + release)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
|
||||
|
||||
# rapid = 迅速开合:优先跟手释放 / 再武装
|
||||
# stable = 夹持稳定:优先防误开、夹得住
|
||||
GRIP_BEHAVIOR_MODES = ("rapid", "stable")
|
||||
# sensitive = 灵敏:跟手释放,自适应夹持力,0.4 s 保护期
|
||||
# stable = 较稳定:接触即 70% 恒力夹持,0.7 s 保护期,Y 拟合偏移 0.6 N 连续 4 帧张开
|
||||
# firm = 十分稳定:接触即 100% 恒力夹持,1.0 s 保护期,Y 拟合偏移 0.8 N 连续 6 帧张开
|
||||
# 三挡张开速度统一 80%。
|
||||
# 稳定度 0..100:三挡分别位于 0 / 50 / 100,中间值按比例插值(见 grip_behavior_level_profile)。
|
||||
GRIP_BEHAVIOR_MODES = ("sensitive", "stable", "firm")
|
||||
|
||||
GRIP_BEHAVIOR_LABELS = {
|
||||
"rapid": "迅速开合",
|
||||
"stable": "夹持稳定",
|
||||
"sensitive": "灵敏",
|
||||
"stable": "较稳定",
|
||||
"firm": "十分稳定",
|
||||
}
|
||||
|
||||
# Keys owned by a behavior profile. Selecting a mode rewrites these fields.
|
||||
GRIP_BEHAVIOR_KEYS = (
|
||||
"filter_alpha",
|
||||
"shear_filter_alpha",
|
||||
# Close/re-arm trigger always uses the sensitive values, regardless of mode.
|
||||
GRIP_TRIGGER_KEYS = (
|
||||
"trigger_normal_force",
|
||||
"trigger_shear_force",
|
||||
"open_wait_stable_samples",
|
||||
"open_wait_stable_range_n",
|
||||
"open_wait_stable_trend_n",
|
||||
"grip_settle_seconds",
|
||||
)
|
||||
|
||||
# Keys rewritten by the behavior mode. Trigger params are not included —
|
||||
# they stay locked to the sensitive profile via GRIP_TRIGGER_KEYS.
|
||||
GRIP_BEHAVIOR_KEYS = (
|
||||
"speed",
|
||||
"grip_contact_hold_position",
|
||||
"close_force",
|
||||
"grip_start_force",
|
||||
"hold_force",
|
||||
"force_ramp_step",
|
||||
"force_ramp_interval",
|
||||
"adaptive_grip_enabled",
|
||||
"adaptive_force_min",
|
||||
"adaptive_force_max",
|
||||
"adaptive_force_step",
|
||||
"adaptive_force_interval",
|
||||
"release_guard_fast_paths",
|
||||
"release_shear_axes",
|
||||
"release_shear_displacement",
|
||||
"release_shear_displacement_samples",
|
||||
"release_object_lost_open",
|
||||
"release_open_speed",
|
||||
"release_open_accel",
|
||||
"release_open_decel",
|
||||
@@ -51,6 +71,12 @@ GRIP_BEHAVIOR_KEYS = (
|
||||
"release_require_slope_settle",
|
||||
"release_slope_hold_seconds",
|
||||
"release_settle_normal_only",
|
||||
"release_shear_oscillation_open",
|
||||
"release_shear_oscillation_window_seconds",
|
||||
"release_shear_oscillation_min_flips",
|
||||
"release_shear_oscillation_amplitude",
|
||||
"release_shear_oscillation_confirm_samples",
|
||||
"release_shear_oscillation_min_seconds",
|
||||
"release_side_shear_enabled",
|
||||
"release_side_normal_enabled",
|
||||
"release_shear_onset_enabled",
|
||||
@@ -59,136 +85,322 @@ GRIP_BEHAVIOR_KEYS = (
|
||||
"recover_stable_seconds",
|
||||
)
|
||||
|
||||
_SENSITIVE_TRIGGER = {
|
||||
# Close trigger must stay above force-display noise (~0.05N).
|
||||
"trigger_normal_force": 0.12,
|
||||
"trigger_shear_force": 0.12,
|
||||
"open_wait_stable_samples": 8,
|
||||
"open_wait_stable_range_n": 0.025,
|
||||
"open_wait_stable_trend_n": 0.015,
|
||||
}
|
||||
|
||||
# Adaptive clamp: start light, climb with payload shear.
|
||||
_ADAPTIVE_FORCE = {
|
||||
"grip_contact_hold_position": True,
|
||||
"close_force": 33,
|
||||
"grip_start_force": 25,
|
||||
"hold_force": 33,
|
||||
"force_ramp_step": 1,
|
||||
"force_ramp_interval": 0.1,
|
||||
"adaptive_force_min": 33,
|
||||
"adaptive_force_max": 80,
|
||||
"adaptive_force_step": 5,
|
||||
"adaptive_force_interval": 0.08,
|
||||
}
|
||||
|
||||
def _constant_force(pct):
|
||||
"""Close at *pct* and keep that close command squeezing after contact
|
||||
(no position hold, no later ramp). Close speed stays the global speed."""
|
||||
return {
|
||||
"grip_contact_hold_position": False,
|
||||
"close_force": pct,
|
||||
"grip_start_force": pct,
|
||||
"hold_force": pct,
|
||||
"force_ramp_step": 5,
|
||||
"force_ramp_interval": 0.03,
|
||||
"adaptive_force_min": pct,
|
||||
"adaptive_force_max": pct,
|
||||
"adaptive_force_step": 5,
|
||||
"adaptive_force_interval": 0.03,
|
||||
}
|
||||
|
||||
_SENSITIVE_PROFILE = {
|
||||
**_SENSITIVE_TRIGGER,
|
||||
**_ADAPTIVE_FORCE,
|
||||
"speed": 55,
|
||||
"release_guard_fast_paths": False,
|
||||
"release_shear_axes": "",
|
||||
"release_shear_displacement": 0.0,
|
||||
"release_shear_displacement_samples": 3,
|
||||
"release_object_lost_open": False,
|
||||
"release_open_speed": 80,
|
||||
"release_open_accel": 70,
|
||||
"release_open_decel": 70,
|
||||
"release_arm_delay_seconds": 0.28,
|
||||
"release_min_hold_seconds": 0.4,
|
||||
"release_settle_seconds": 0.12,
|
||||
"release_min_change": 0.08,
|
||||
"release_touch_window_seconds": 0.14,
|
||||
"release_side_touch_change": 0.12,
|
||||
"release_confirm_window_seconds": 0.14,
|
||||
"release_sample_timeout_seconds": 0.2,
|
||||
"release_confirm_samples": 3,
|
||||
"release_slow_confirm_samples": 8,
|
||||
"release_normal_rate": 2.6,
|
||||
"release_shear_rate": 2.4,
|
||||
"release_slow_rate": 0.22,
|
||||
"release_normal_change": 0.42,
|
||||
"release_shear_change": 0.35,
|
||||
"release_curve_history": 30,
|
||||
"release_slope_window_seconds": 0.8,
|
||||
"release_slope_settle_rate": 0.22,
|
||||
"release_slope_min_points": 10,
|
||||
"release_require_slope_settle": True,
|
||||
"release_slope_hold_seconds": 0.45,
|
||||
"release_settle_normal_only": True,
|
||||
# Intentional rub/twist: oscillating shear opens even before plateau latch.
|
||||
"release_shear_oscillation_open": True,
|
||||
"release_shear_oscillation_window_seconds": 0.5,
|
||||
"release_shear_oscillation_min_flips": 4,
|
||||
"release_shear_oscillation_amplitude": 0.15,
|
||||
"release_shear_oscillation_confirm_samples": 2,
|
||||
"release_shear_oscillation_min_seconds": 0.15,
|
||||
"release_side_shear_enabled": True,
|
||||
"release_side_normal_enabled": True,
|
||||
"release_shear_onset_enabled": True,
|
||||
"release_normal_onset_enabled": True,
|
||||
"rearm_seconds": 0.55,
|
||||
"recover_stable_seconds": 0.55,
|
||||
}
|
||||
|
||||
_STABLE_PROFILE = {
|
||||
**_constant_force(70),
|
||||
"speed": 30,
|
||||
"release_guard_fast_paths": True,
|
||||
# Release looks at Y shear only, regardless of the global control axes.
|
||||
"release_shear_axes": "y",
|
||||
# Full clamp swings Y by several N and flips sign, so onset confirmation
|
||||
# keeps cancelling; open on sustained fitted Y displacement instead.
|
||||
"release_shear_displacement": 0.6,
|
||||
"release_shear_displacement_samples": 4,
|
||||
# The close command keeps pushing: reaching close_pos means the object is gone.
|
||||
"release_object_lost_open": True,
|
||||
"release_open_speed": 80,
|
||||
"release_open_accel": 70,
|
||||
"release_open_decel": 70,
|
||||
"release_arm_delay_seconds": 0.55,
|
||||
# Protection: no release of any kind within 0.7 s of contact.
|
||||
"release_min_hold_seconds": 0.7,
|
||||
"release_settle_seconds": 0.22,
|
||||
"release_min_change": 0.12,
|
||||
"release_touch_window_seconds": 0.22,
|
||||
"release_side_touch_change": 0.24,
|
||||
"release_confirm_window_seconds": 0.28,
|
||||
"release_sample_timeout_seconds": 0.3,
|
||||
# Unused by the displacement release; kept equal to its frame count so the
|
||||
# stability bar scales the sensitive confirm 3 -> 4 instead of toward 8.
|
||||
"release_confirm_samples": 4,
|
||||
"release_slow_confirm_samples": 12,
|
||||
"release_normal_rate": 4.0,
|
||||
"release_shear_rate": 3.6,
|
||||
"release_slow_rate": 0.14,
|
||||
"release_normal_change": 0.7,
|
||||
"release_shear_change": 0.6,
|
||||
"release_curve_history": 42,
|
||||
"release_slope_window_seconds": 1.2,
|
||||
"release_slope_settle_rate": 0.12,
|
||||
"release_slope_min_points": 14,
|
||||
# Full constant clamp: the protection window is the only arming gate.
|
||||
"release_require_slope_settle": False,
|
||||
"release_slope_hold_seconds": 0.9,
|
||||
"release_settle_normal_only": True,
|
||||
# Open by fitted Y shear displacement only, no oscillation path.
|
||||
"release_shear_oscillation_open": False,
|
||||
"release_shear_oscillation_window_seconds": 0.6,
|
||||
"release_shear_oscillation_min_flips": 5,
|
||||
"release_shear_oscillation_amplitude": 0.3,
|
||||
"release_shear_oscillation_confirm_samples": 4,
|
||||
# Normal only judges the clamp plateau; opening is by Y shear only.
|
||||
"release_side_shear_enabled": True,
|
||||
"release_side_normal_enabled": False,
|
||||
"release_shear_onset_enabled": True,
|
||||
"release_normal_onset_enabled": False,
|
||||
"rearm_seconds": 0.9,
|
||||
"recover_stable_seconds": 0.9,
|
||||
}
|
||||
|
||||
# Same release scheme as stable, with a harder clamp and larger margins.
|
||||
_FIRM_PROFILE = {
|
||||
**_STABLE_PROFILE,
|
||||
**_constant_force(100),
|
||||
"release_min_hold_seconds": 1.0,
|
||||
"release_shear_displacement": 0.8,
|
||||
"release_shear_displacement_samples": 6,
|
||||
"release_confirm_samples": 6,
|
||||
}
|
||||
|
||||
GRIP_BEHAVIOR_PROFILES = {
|
||||
"rapid": {
|
||||
"filter_alpha": 0.72,
|
||||
"shear_filter_alpha": 0.78,
|
||||
"open_wait_stable_samples": 3,
|
||||
"open_wait_stable_range_n": 0.04,
|
||||
"open_wait_stable_trend_n": 0.03,
|
||||
"grip_settle_seconds": 0.05,
|
||||
"force_ramp_step": 3,
|
||||
"force_ramp_interval": 0.04,
|
||||
"adaptive_grip_enabled": False,
|
||||
"adaptive_force_interval": 0.05,
|
||||
"release_open_speed": 85,
|
||||
"release_open_accel": 90,
|
||||
"release_open_decel": 90,
|
||||
# Almost no normal-contact lock; rely on curve/side-touch rejection.
|
||||
"release_arm_delay_seconds": 0.04,
|
||||
"release_min_hold_seconds": 0.0,
|
||||
"release_settle_seconds": 0.03,
|
||||
"release_min_change": 0.03,
|
||||
"release_touch_window_seconds": 0.28,
|
||||
"release_side_touch_change": 0.04,
|
||||
"release_confirm_window_seconds": 0.28,
|
||||
"release_sample_timeout_seconds": 0.25,
|
||||
"release_confirm_samples": 2,
|
||||
"release_slow_confirm_samples": 3,
|
||||
"release_normal_rate": 1.0,
|
||||
"release_shear_rate": 1.0,
|
||||
"release_slow_rate": 0.55,
|
||||
"release_normal_change": 0.14,
|
||||
"release_shear_change": 0.14,
|
||||
"release_curve_history": 6,
|
||||
"release_slope_window_seconds": 0.35,
|
||||
"release_slope_settle_rate": 0.8,
|
||||
"release_slope_min_points": 4,
|
||||
"release_require_slope_settle": False,
|
||||
"release_slope_hold_seconds": 0.0,
|
||||
"release_settle_normal_only": False,
|
||||
"release_side_shear_enabled": True,
|
||||
"release_side_normal_enabled": True,
|
||||
"release_shear_onset_enabled": True,
|
||||
"release_normal_onset_enabled": True,
|
||||
"rearm_seconds": 0.12,
|
||||
"recover_stable_seconds": 0.12,
|
||||
},
|
||||
"stable": {
|
||||
"filter_alpha": 0.42,
|
||||
"shear_filter_alpha": 0.52,
|
||||
"open_wait_stable_samples": 8,
|
||||
"open_wait_stable_range_n": 0.025,
|
||||
"open_wait_stable_trend_n": 0.015,
|
||||
"grip_settle_seconds": 0.2,
|
||||
"force_ramp_step": 1,
|
||||
"force_ramp_interval": 0.1,
|
||||
"adaptive_grip_enabled": True,
|
||||
"adaptive_force_interval": 0.08,
|
||||
"release_open_speed": 55,
|
||||
"release_open_accel": 70,
|
||||
"release_open_decel": 70,
|
||||
"release_arm_delay_seconds": 0.28,
|
||||
"release_min_hold_seconds": 0.65,
|
||||
"release_settle_seconds": 0.12,
|
||||
"release_min_change": 0.08,
|
||||
"release_touch_window_seconds": 0.14,
|
||||
"release_side_touch_change": 0.12,
|
||||
"release_confirm_window_seconds": 0.14,
|
||||
"release_sample_timeout_seconds": 0.2,
|
||||
"release_confirm_samples": 4,
|
||||
"release_slow_confirm_samples": 8,
|
||||
"release_normal_rate": 2.6,
|
||||
"release_shear_rate": 2.4,
|
||||
"release_slow_rate": 0.22,
|
||||
"release_normal_change": 0.42,
|
||||
"release_shear_change": 0.35,
|
||||
# Clean split: settle on normal plateau; open only via shear.
|
||||
"release_curve_history": 30,
|
||||
"release_slope_window_seconds": 0.8,
|
||||
"release_slope_settle_rate": 0.22,
|
||||
"release_slope_min_points": 10,
|
||||
"release_require_slope_settle": True,
|
||||
"release_slope_hold_seconds": 0.45,
|
||||
"release_settle_normal_only": True,
|
||||
"release_side_shear_enabled": True,
|
||||
"release_side_normal_enabled": False,
|
||||
"release_shear_onset_enabled": True,
|
||||
"release_normal_onset_enabled": False,
|
||||
"rearm_seconds": 0.55,
|
||||
"recover_stable_seconds": 0.55,
|
||||
},
|
||||
"sensitive": _SENSITIVE_PROFILE,
|
||||
"stable": _STABLE_PROFILE,
|
||||
"firm": _FIRM_PROFILE,
|
||||
}
|
||||
|
||||
|
||||
def normalize_grip_behavior_mode(mode) -> str:
|
||||
"""Resolve aliases to sensitive / stable / firm; default sensitive."""
|
||||
text = str(mode or "").strip().lower()
|
||||
aliases = {
|
||||
"fast": "rapid",
|
||||
"quick": "rapid",
|
||||
"snap": "rapid",
|
||||
"迅速": "rapid",
|
||||
"迅速开合": "rapid",
|
||||
"hold": "stable",
|
||||
"firm": "stable",
|
||||
"secure": "stable",
|
||||
"fast": "sensitive",
|
||||
"quick": "sensitive",
|
||||
"snap": "sensitive",
|
||||
"rapid": "sensitive",
|
||||
"responsive": "sensitive",
|
||||
"灵敏": "sensitive",
|
||||
"灵敏开合": "sensitive",
|
||||
"迅速": "sensitive",
|
||||
"迅速开合": "sensitive",
|
||||
"sensitive": "sensitive",
|
||||
"稳定": "stable",
|
||||
"较稳定": "stable",
|
||||
"夹持稳定": "stable",
|
||||
"stable": "stable",
|
||||
"hold": "firm",
|
||||
"firm": "firm",
|
||||
"secure": "firm",
|
||||
"very_stable": "firm",
|
||||
"verystable": "firm",
|
||||
"十分稳定": "firm",
|
||||
"极度稳定": "firm",
|
||||
}
|
||||
text = aliases.get(text, text)
|
||||
if text in aliases:
|
||||
return aliases[text]
|
||||
if text not in GRIP_BEHAVIOR_PROFILES:
|
||||
raise ValueError(
|
||||
f"unsupported grip_behavior_mode: {mode!r}; "
|
||||
f"expected one of {list(GRIP_BEHAVIOR_PROFILES)}"
|
||||
)
|
||||
return "sensitive"
|
||||
return text
|
||||
|
||||
|
||||
def grip_behavior_profile(mode: str) -> dict:
|
||||
def _merged_profile(mode: str) -> dict:
|
||||
"""Release/open params from *mode*, trigger params always from sensitive."""
|
||||
mode = normalize_grip_behavior_mode(mode)
|
||||
return copy.deepcopy(GRIP_BEHAVIOR_PROFILES[mode])
|
||||
profile = {
|
||||
key: copy.deepcopy(GRIP_BEHAVIOR_PROFILES[mode][key])
|
||||
for key in GRIP_BEHAVIOR_KEYS
|
||||
if key in GRIP_BEHAVIOR_PROFILES[mode]
|
||||
}
|
||||
for key in GRIP_TRIGGER_KEYS:
|
||||
profile[key] = copy.deepcopy(_SENSITIVE_TRIGGER[key])
|
||||
return profile
|
||||
|
||||
|
||||
def apply_grip_behavior_profile(target, mode: str) -> str:
|
||||
"""Write profile fields onto a config object or dict. Returns normalized mode."""
|
||||
def grip_behavior_profile(mode: str = "sensitive") -> dict:
|
||||
return _merged_profile(mode)
|
||||
|
||||
|
||||
# Stability level 0..100: the three built-in modes sit at fixed anchors.
|
||||
# Below the stable anchor the sensitive algorithm runs with its numeric params
|
||||
# scaled toward the stable values; from the stable anchor on, the held
|
||||
# algorithm runs, scaled from stable to firm.
|
||||
GRIP_BEHAVIOR_LEVEL_MIN = 0
|
||||
GRIP_BEHAVIOR_LEVEL_MAX = 100
|
||||
GRIP_BEHAVIOR_LEVELS = {"sensitive": 0, "stable": 50, "firm": 100}
|
||||
# Only read by the held (stable/firm) release; stays off on the sensitive side.
|
||||
_HELD_ONLY_KEYS = ("release_shear_displacement",)
|
||||
|
||||
|
||||
def clamp_grip_behavior_level(level) -> int:
|
||||
try:
|
||||
value = int(round(float(level)))
|
||||
except (TypeError, ValueError):
|
||||
value = GRIP_BEHAVIOR_LEVEL_MIN
|
||||
return max(GRIP_BEHAVIOR_LEVEL_MIN, min(GRIP_BEHAVIOR_LEVEL_MAX, value))
|
||||
|
||||
|
||||
def grip_behavior_mode_for_level(level) -> str:
|
||||
level = clamp_grip_behavior_level(level)
|
||||
if level >= GRIP_BEHAVIOR_LEVELS["firm"]:
|
||||
return "firm"
|
||||
if level >= GRIP_BEHAVIOR_LEVELS["stable"]:
|
||||
return "stable"
|
||||
return "sensitive"
|
||||
|
||||
|
||||
def default_grip_behavior_level(mode) -> int:
|
||||
return GRIP_BEHAVIOR_LEVELS[normalize_grip_behavior_mode(mode)]
|
||||
|
||||
|
||||
def builtin_grip_behavior_presets() -> list:
|
||||
return [
|
||||
{
|
||||
"name": GRIP_BEHAVIOR_LABELS[mode],
|
||||
"level": GRIP_BEHAVIOR_LEVELS[mode],
|
||||
"speed": GRIP_BEHAVIOR_PROFILES[mode]["speed"],
|
||||
"builtin": True,
|
||||
}
|
||||
for mode in GRIP_BEHAVIOR_MODES
|
||||
]
|
||||
|
||||
|
||||
def normalize_grip_behavior_presets(presets) -> list:
|
||||
"""User presets as [{name, level[, speed]}], without duplicates of built-in names."""
|
||||
reserved = {GRIP_BEHAVIOR_LABELS[mode] for mode in GRIP_BEHAVIOR_MODES}
|
||||
result = []
|
||||
seen = set()
|
||||
for item in presets if isinstance(presets, list) else []:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
name = str(item.get("name") or "").strip()
|
||||
if not name or name in reserved or name in seen:
|
||||
continue
|
||||
seen.add(name)
|
||||
preset = {"name": name, "level": clamp_grip_behavior_level(item.get("level"))}
|
||||
if item.get("speed") is not None:
|
||||
preset["speed"] = clamp_grip_behavior_level(item.get("speed"))
|
||||
result.append(preset)
|
||||
return result
|
||||
|
||||
|
||||
def _lerp_value(a, b, t):
|
||||
if isinstance(a, bool) or isinstance(b, bool):
|
||||
return copy.deepcopy(a)
|
||||
if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):
|
||||
return copy.deepcopy(a)
|
||||
value = a + (b - a) * t
|
||||
if isinstance(a, int) and isinstance(b, int):
|
||||
return int(round(value))
|
||||
return round(float(value), 4)
|
||||
|
||||
|
||||
def grip_behavior_level_profile(level) -> dict:
|
||||
"""Merged params for stability *level*; anchors equal the built-in modes."""
|
||||
level = clamp_grip_behavior_level(level)
|
||||
for mode, anchor in GRIP_BEHAVIOR_LEVELS.items():
|
||||
if level == anchor:
|
||||
return _merged_profile(mode)
|
||||
stable_at = GRIP_BEHAVIOR_LEVELS["stable"]
|
||||
if level < stable_at:
|
||||
low, high = _merged_profile("sensitive"), _merged_profile("stable")
|
||||
t = (level - GRIP_BEHAVIOR_LEVELS["sensitive"]) / float(stable_at - GRIP_BEHAVIOR_LEVELS["sensitive"])
|
||||
else:
|
||||
low, high = _merged_profile("stable"), _merged_profile("firm")
|
||||
t = (level - stable_at) / float(GRIP_BEHAVIOR_LEVELS["firm"] - stable_at)
|
||||
fixed = GRIP_TRIGGER_KEYS + (_HELD_ONLY_KEYS if level < stable_at else ())
|
||||
profile = {}
|
||||
for key, value in low.items():
|
||||
if key in fixed:
|
||||
profile[key] = copy.deepcopy(value)
|
||||
else:
|
||||
profile[key] = _lerp_value(value, high.get(key, value), t)
|
||||
return profile
|
||||
|
||||
|
||||
def apply_grip_behavior_profile(target, mode: str = "sensitive") -> str:
|
||||
mode = normalize_grip_behavior_mode(mode)
|
||||
profile = GRIP_BEHAVIOR_PROFILES[mode]
|
||||
profile = _merged_profile(mode)
|
||||
if isinstance(target, dict):
|
||||
target["grip_behavior_mode"] = mode
|
||||
for key, value in profile.items():
|
||||
target[key] = copy.deepcopy(value)
|
||||
return mode
|
||||
|
||||
setattr(target, "grip_behavior_mode", mode)
|
||||
for key, value in profile.items():
|
||||
if hasattr(target, key):
|
||||
|
||||
@@ -68,6 +68,8 @@ class Demo02Config:
|
||||
baseline_seconds: float = 1.0
|
||||
filter_alpha: float = 0.55
|
||||
shear_filter_alpha: float = 0.65
|
||||
# Shear magnitude used for trigger/adaptive/release: "xy" | "x" | "y".
|
||||
shear_control_axes: str = "y"
|
||||
|
||||
trigger_normal_force: float = 0.04
|
||||
trigger_shear_force: float = 0.04
|
||||
@@ -82,7 +84,10 @@ class Demo02Config:
|
||||
force_ramp_step: int = 1
|
||||
force_ramp_interval: float = 0.1
|
||||
grip_settle_seconds: float = 0.1
|
||||
grip_behavior_mode: str = "stable"
|
||||
# On contact: True = hold at the contact position; False = keep the close
|
||||
# command squeezing toward close_pos at full commanded force.
|
||||
grip_contact_hold_position: bool = True
|
||||
grip_behavior_mode: str = "sensitive"
|
||||
adaptive_grip_enabled: bool = True
|
||||
adaptive_shear_start_force: float = 0.10
|
||||
adaptive_shear_full_force: float = 0.60
|
||||
@@ -112,6 +117,23 @@ class Demo02Config:
|
||||
release_require_slope_settle: bool = False
|
||||
release_slope_hold_seconds: float = 0.0
|
||||
release_settle_normal_only: bool = True
|
||||
release_shear_oscillation_open: bool = False
|
||||
release_shear_oscillation_window_seconds: float = 0.4
|
||||
release_shear_oscillation_min_flips: int = 3
|
||||
release_shear_oscillation_amplitude: float = 0.15
|
||||
release_shear_oscillation_confirm_samples: int = 2
|
||||
# Oscillation must persist this long (in addition to confirm samples) before opening.
|
||||
release_shear_oscillation_min_seconds: float = 0.0
|
||||
# Block oscillation/side-touch release during contact hold and servo motion.
|
||||
release_guard_fast_paths: bool = False
|
||||
# Shear axes for release only: "" follows shear_control_axes; "y" | "x" | "xy".
|
||||
release_shear_axes: str = ""
|
||||
# >0: open when curve-fitted signed Y shear leaves its post-protection
|
||||
# baseline by this much (N) for N consecutive samples; replaces onset paths.
|
||||
release_shear_displacement: float = 0.0
|
||||
release_shear_displacement_samples: int = 6
|
||||
# Open when the gripper travels to close_pos while holding (object gone).
|
||||
release_object_lost_open: bool = False
|
||||
release_side_shear_enabled: bool = True
|
||||
release_side_normal_enabled: bool = True
|
||||
release_shear_onset_enabled: bool = True
|
||||
@@ -184,6 +206,12 @@ class Demo02Config:
|
||||
add("--baseline-seconds", type=float, default=get("baseline_seconds", cls.baseline_seconds))
|
||||
add("--filter-alpha", type=float, default=get("filter_alpha", cls.filter_alpha))
|
||||
add("--shear-filter-alpha", type=float, default=get("shear_filter_alpha", cls.shear_filter_alpha))
|
||||
add(
|
||||
"--shear-control-axes",
|
||||
type=str,
|
||||
choices=["xy", "x", "y"],
|
||||
default=str(get("shear_control_axes", cls.shear_control_axes)).strip().lower() or "y",
|
||||
)
|
||||
|
||||
add("--trigger-normal-force", type=float, default=get("trigger_normal_force", cls.trigger_normal_force))
|
||||
add("--trigger-shear-force", type=float, default=get("trigger_shear_force", cls.trigger_shear_force))
|
||||
@@ -199,10 +227,13 @@ class Demo02Config:
|
||||
add("--force-ramp-step", type=int, default=get("force_ramp_step", cls.force_ramp_step))
|
||||
add("--force-ramp-interval", type=float, default=get("force_ramp_interval", cls.force_ramp_interval))
|
||||
add("--grip-settle-seconds", type=float, default=get("grip_settle_seconds", cls.grip_settle_seconds))
|
||||
add("--grip-contact-hold-position", dest="grip_contact_hold_position", action="store_true",
|
||||
default=get("grip_contact_hold_position", cls.grip_contact_hold_position))
|
||||
add("--no-grip-contact-hold-position", dest="grip_contact_hold_position", action="store_false")
|
||||
add(
|
||||
"--grip-behavior-mode",
|
||||
type=str,
|
||||
choices=["rapid", "stable"],
|
||||
choices=["sensitive", "stable", "firm"],
|
||||
default=normalize_grip_behavior_mode(
|
||||
get("grip_behavior_mode", cls.grip_behavior_mode)
|
||||
),
|
||||
@@ -241,6 +272,31 @@ class Demo02Config:
|
||||
add("--release-settle-normal-only", dest="release_settle_normal_only", action="store_true",
|
||||
default=get("release_settle_normal_only", cls.release_settle_normal_only))
|
||||
add("--release-settle-all-channels", dest="release_settle_normal_only", action="store_false")
|
||||
add("--release-shear-oscillation-open", dest="release_shear_oscillation_open", action="store_true",
|
||||
default=get("release_shear_oscillation_open", cls.release_shear_oscillation_open))
|
||||
add("--no-release-shear-oscillation-open", dest="release_shear_oscillation_open", action="store_false")
|
||||
add("--release-shear-oscillation-window-seconds", type=float,
|
||||
default=get("release_shear_oscillation_window_seconds", cls.release_shear_oscillation_window_seconds))
|
||||
add("--release-shear-oscillation-min-flips", type=int,
|
||||
default=get("release_shear_oscillation_min_flips", cls.release_shear_oscillation_min_flips))
|
||||
add("--release-shear-oscillation-amplitude", type=float,
|
||||
default=get("release_shear_oscillation_amplitude", cls.release_shear_oscillation_amplitude))
|
||||
add("--release-shear-oscillation-confirm-samples", type=int,
|
||||
default=get("release_shear_oscillation_confirm_samples", cls.release_shear_oscillation_confirm_samples))
|
||||
add("--release-shear-oscillation-min-seconds", type=float,
|
||||
default=get("release_shear_oscillation_min_seconds", cls.release_shear_oscillation_min_seconds))
|
||||
add("--release-guard-fast-paths", dest="release_guard_fast_paths", action="store_true",
|
||||
default=get("release_guard_fast_paths", cls.release_guard_fast_paths))
|
||||
add("--no-release-guard-fast-paths", dest="release_guard_fast_paths", action="store_false")
|
||||
add("--release-shear-axes", type=str, choices=["", "xy", "x", "y"],
|
||||
default=str(get("release_shear_axes", cls.release_shear_axes) or "").strip().lower())
|
||||
add("--release-shear-displacement", type=float,
|
||||
default=get("release_shear_displacement", cls.release_shear_displacement))
|
||||
add("--release-shear-displacement-samples", type=int,
|
||||
default=get("release_shear_displacement_samples", cls.release_shear_displacement_samples))
|
||||
add("--release-object-lost-open", dest="release_object_lost_open", action="store_true",
|
||||
default=get("release_object_lost_open", cls.release_object_lost_open))
|
||||
add("--no-release-object-lost-open", dest="release_object_lost_open", action="store_false")
|
||||
add("--release-side-shear-enabled", dest="release_side_shear_enabled", action="store_true",
|
||||
default=get("release_side_shear_enabled", cls.release_side_shear_enabled))
|
||||
add("--no-release-side-shear", dest="release_side_shear_enabled", action="store_false")
|
||||
|
||||
@@ -26,7 +26,7 @@ _MOTION_KINDS = frozenset({"open", "close", "move", "hold"})
|
||||
class ActuatorWorker:
|
||||
"""Runs all GripperClient I/O on one thread; control loop only submits."""
|
||||
|
||||
def __init__(self, gripper, position_poll_hz: float = 10.0):
|
||||
def __init__(self, gripper, position_poll_hz: float = 30.0):
|
||||
self.gripper = gripper
|
||||
self.position_poll_hz = max(1.0, float(position_poll_hz))
|
||||
self._stop_event = threading.Event()
|
||||
@@ -40,6 +40,10 @@ class ActuatorWorker:
|
||||
self._motion_completed_at: Optional[float] = None
|
||||
self._wake = threading.Event()
|
||||
self._busy = threading.Event()
|
||||
# After a motion command, poll faster so the 3D view tracks travel.
|
||||
self._boost_poll_until = 0.0
|
||||
self._boost_poll_hz = 50.0
|
||||
self._boost_poll_seconds = 2.5
|
||||
|
||||
self._state_lock = threading.Lock()
|
||||
self._position: Optional[int] = None
|
||||
@@ -225,7 +229,6 @@ class ActuatorWorker:
|
||||
return
|
||||
|
||||
self._started.set()
|
||||
poll_interval = 1.0 / self.position_poll_hz
|
||||
next_poll = time.perf_counter()
|
||||
self._perf_last_log = time.perf_counter()
|
||||
|
||||
@@ -251,7 +254,7 @@ class ActuatorWorker:
|
||||
now = time.perf_counter()
|
||||
if now >= next_poll:
|
||||
self._poll_position()
|
||||
next_poll = now + poll_interval
|
||||
next_poll = now + self._current_poll_interval(now)
|
||||
self._maybe_log_perf(now)
|
||||
continue
|
||||
|
||||
@@ -259,6 +262,16 @@ class ActuatorWorker:
|
||||
self._wake.wait(timeout=wait_s)
|
||||
self._wake.clear()
|
||||
|
||||
def _current_poll_interval(self, now: float) -> float:
|
||||
if now < self._boost_poll_until:
|
||||
return 1.0 / self._boost_poll_hz
|
||||
return 1.0 / self.position_poll_hz
|
||||
|
||||
def _note_motion_command(self) -> None:
|
||||
"""Prefer measured travel over commanded target for position cache."""
|
||||
self._boost_poll_until = time.perf_counter() + self._boost_poll_seconds
|
||||
self._poll_position()
|
||||
|
||||
def _execute(self, cmd: GripCommand) -> None:
|
||||
t0 = time.perf_counter()
|
||||
try:
|
||||
@@ -270,7 +283,7 @@ class ActuatorWorker:
|
||||
accel=cmd.accel,
|
||||
decel=cmd.decel,
|
||||
)
|
||||
self._update_position(self.gripper.last_commanded_position)
|
||||
self._note_motion_command()
|
||||
elif cmd.kind == "close":
|
||||
self.gripper.close(
|
||||
cmd.force_pct,
|
||||
@@ -279,7 +292,7 @@ class ActuatorWorker:
|
||||
accel=cmd.accel,
|
||||
decel=cmd.decel,
|
||||
)
|
||||
self._update_position(self.gripper.last_commanded_position)
|
||||
self._note_motion_command()
|
||||
elif cmd.kind == "move":
|
||||
self.gripper.move(
|
||||
cmd.position,
|
||||
@@ -289,7 +302,7 @@ class ActuatorWorker:
|
||||
accel=cmd.accel,
|
||||
decel=cmd.decel,
|
||||
)
|
||||
self._update_position(cmd.position)
|
||||
self._note_motion_command()
|
||||
elif cmd.kind == "set_force":
|
||||
self.gripper.set_force(cmd.force_pct)
|
||||
elif cmd.kind == "hold":
|
||||
@@ -303,9 +316,10 @@ class ActuatorWorker:
|
||||
self._hold_pos = hold_pos
|
||||
if hold_pos is not None:
|
||||
self._position = int(hold_pos)
|
||||
self._note_motion_command()
|
||||
elif cmd.kind == "final_open":
|
||||
self.gripper.final_open_if_needed()
|
||||
self._update_position(self.gripper.last_commanded_position)
|
||||
self._note_motion_command()
|
||||
else:
|
||||
self._set_error(f"unknown command kind: {cmd.kind}")
|
||||
except Exception as exc:
|
||||
|
||||
@@ -2,12 +2,13 @@ import time
|
||||
import threading
|
||||
import json
|
||||
from collections import deque
|
||||
from dataclasses import asdict
|
||||
from dataclasses import asdict, replace
|
||||
|
||||
from ..common.filters import clamp
|
||||
from ..common.paths import REPO_ROOT
|
||||
from .actuator import ActuatorWorker
|
||||
from .release_watch import ReleaseWatch
|
||||
from .release_dispatch import ModeReleaseWatch
|
||||
from .release_watch import ReleasePhase
|
||||
from .session_recorder import SessionRecorder
|
||||
|
||||
|
||||
@@ -28,7 +29,7 @@ class SideTriggerGripController:
|
||||
self.shear_unit = feedback_processor.shear_converter.unit
|
||||
self.status_callback = status_callback
|
||||
self.actuator = None
|
||||
self.release_watch = ReleaseWatch(config)
|
||||
self.release_watch = ModeReleaseWatch(config)
|
||||
self._release_decision = None
|
||||
self._force_trace = deque(maxlen=120)
|
||||
self.release_trace_path = REPO_ROOT / "logs" / "release-last.json"
|
||||
@@ -54,6 +55,7 @@ class SideTriggerGripController:
|
||||
"move_position",
|
||||
"restart_left_sensor",
|
||||
"restart_right_sensor",
|
||||
"reset_sensor_reference",
|
||||
}:
|
||||
raise ValueError(f"unsupported manual command: {command}")
|
||||
payload = {
|
||||
@@ -78,7 +80,7 @@ class SideTriggerGripController:
|
||||
self._control_mode = mode
|
||||
|
||||
def set_grip_behavior_mode(self, mode):
|
||||
"""Apply rapid/stable preset onto live config and reset release watch."""
|
||||
"""Apply sensitive/stable/firm grip preset onto live config and reset release watch."""
|
||||
applied = self.config.apply_grip_behavior_mode(mode)
|
||||
# Filters should match the new profile immediately.
|
||||
try:
|
||||
@@ -91,6 +93,9 @@ class SideTriggerGripController:
|
||||
pass
|
||||
print(
|
||||
f"[behavior] grip_behavior_mode={applied} "
|
||||
f"force=close{self.config.close_force}/start{self.config.grip_start_force}/"
|
||||
f"hold{self.config.hold_force}/adaptive{self.config.adaptive_force_min}-{self.config.adaptive_force_max} "
|
||||
f"guard={getattr(self.config, 'release_guard_fast_paths', False)} "
|
||||
f"min_hold={self.config.release_min_hold_seconds:.2f}s "
|
||||
f"side_touch={self.config.release_side_touch_change:.3f} "
|
||||
f"confirm={self.config.release_confirm_samples} "
|
||||
@@ -112,12 +117,20 @@ class SideTriggerGripController:
|
||||
"control_hz",
|
||||
"filter_alpha",
|
||||
"shear_filter_alpha",
|
||||
"shear_control_axes",
|
||||
"trigger_normal_force",
|
||||
"trigger_shear_force",
|
||||
"grip_normal_force",
|
||||
"grip_requires_both",
|
||||
"grip_behavior_mode",
|
||||
"adaptive_grip_enabled",
|
||||
"close_force",
|
||||
"grip_start_force",
|
||||
"hold_force",
|
||||
"grip_contact_hold_position",
|
||||
"force_max",
|
||||
"adaptive_force_min",
|
||||
"adaptive_force_max",
|
||||
}
|
||||
},
|
||||
}
|
||||
@@ -143,7 +156,7 @@ class SideTriggerGripController:
|
||||
def run(self):
|
||||
self._stop_event.clear()
|
||||
self.reader.start()
|
||||
self.actuator = ActuatorWorker(self.gripper, position_poll_hz=10.0)
|
||||
self.actuator = ActuatorWorker(self.gripper, position_poll_hz=30.0)
|
||||
|
||||
try:
|
||||
self.actuator.start()
|
||||
@@ -253,6 +266,7 @@ class SideTriggerGripController:
|
||||
recent_feedback = deque(maxlen=6)
|
||||
recent_pair_stamps = None
|
||||
hold_pos = None
|
||||
grip_contact_pos = None
|
||||
actuator = self.actuator
|
||||
|
||||
def reset_open_wait_stability():
|
||||
@@ -275,6 +289,19 @@ class SideTriggerGripController:
|
||||
def reset_release_watch():
|
||||
release_watch.reset()
|
||||
|
||||
def object_lost():
|
||||
# The close command keeps pushing: arriving at close_pos from a
|
||||
# contact position well short of it means the fingers are empty.
|
||||
if not getattr(config, "release_object_lost_open", False) or grip_contact_pos is None:
|
||||
return False
|
||||
position = current_position()
|
||||
if position is None:
|
||||
return False
|
||||
tolerance = max(1, int(config.open_position_tolerance))
|
||||
close_pos = int(config.close_pos)
|
||||
return (abs(position - close_pos) <= tolerance
|
||||
and abs(grip_contact_pos - close_pos) > 2 * tolerance)
|
||||
|
||||
def try_release_watch(now_ts, feedback_now):
|
||||
# The release FSM returns effects; only this adapter submits I/O.
|
||||
nonlocal state, state_start, force_pct, hold_pos
|
||||
@@ -285,6 +312,9 @@ class SideTriggerGripController:
|
||||
now_ts, feedback_now,
|
||||
motion_pending=pending, motion_completed_at=completed_at,
|
||||
)
|
||||
if not decision.release and object_lost():
|
||||
decision = replace(decision, phase=ReleasePhase.RELEASED,
|
||||
reason="object-lost", release=True)
|
||||
self._release_decision = decision
|
||||
action = decision.action
|
||||
if decision.release:
|
||||
@@ -441,6 +471,14 @@ class SideTriggerGripController:
|
||||
manual_action = "right-sensor-restart"
|
||||
else:
|
||||
manual_action = "right-sensor-restart-failed"
|
||||
elif manual_command == "reset_sensor_reference":
|
||||
if self.reader.reset_reference_maps():
|
||||
release_watch.suspend()
|
||||
reset_open_wait_stability()
|
||||
trigger_armed = False
|
||||
manual_action = "sensor-reference-reset"
|
||||
else:
|
||||
manual_action = "sensor-reference-reset-failed"
|
||||
|
||||
# Non-blocking sample read: never stall the FSM on sensor wait.
|
||||
feedback = self.feedback_processor.read(self.reader, timeout=0.0)
|
||||
@@ -538,14 +576,22 @@ class SideTriggerGripController:
|
||||
|
||||
elif state == "closing":
|
||||
if grip_contact:
|
||||
force_pct = int(clamp(
|
||||
start_force = int(clamp(
|
||||
config.grip_start_force,
|
||||
config.force_min,
|
||||
config.force_max,
|
||||
))
|
||||
actuator.set_force(force_pct)
|
||||
actuator.hold(force_pct)
|
||||
hold_pos = refresh_hold_pos() or current_position()
|
||||
if getattr(config, "grip_contact_hold_position", True):
|
||||
force_pct = start_force
|
||||
actuator.set_force(force_pct)
|
||||
actuator.hold(force_pct)
|
||||
elif start_force != force_pct:
|
||||
# Keep the close command pushing toward close_pos; a hold
|
||||
# would retarget to the contact position and drop squeeze.
|
||||
force_pct = start_force
|
||||
actuator.close(force_pct, "contact-squeeze")
|
||||
grip_contact_pos = current_position()
|
||||
hold_pos = refresh_hold_pos() or grip_contact_pos
|
||||
state = "gripping"
|
||||
state_start = now
|
||||
last_force_ramp_time = now
|
||||
@@ -812,6 +858,11 @@ class SideTriggerGripController:
|
||||
return stable
|
||||
|
||||
def _open_wait_delta_triggered(self, sample, stable_values):
|
||||
"""Arm close only when a channel rises enough above its quiet baseline.
|
||||
|
||||
Stable mode previously used 0.06N which sits inside typical optical-flow
|
||||
residual noise and caused spontaneous side-trigger-close.
|
||||
"""
|
||||
thresholds = {
|
||||
"left_normal": self.config.trigger_normal_force,
|
||||
"right_normal": self.config.trigger_normal_force,
|
||||
@@ -844,6 +895,16 @@ class SideTriggerGripController:
|
||||
<= int(self.config.open_position_tolerance)
|
||||
)
|
||||
|
||||
def _load_shear(self, feedback):
|
||||
"""Payload/gravity estimate uses shear X; trigger may stay on Y."""
|
||||
left = abs(float(getattr(feedback, "left_shear_x", 0.0) or 0.0))
|
||||
right = abs(float(getattr(feedback, "right_shear_x", 0.0) or 0.0))
|
||||
magnitude = max(left, right)
|
||||
if magnitude > 0.0:
|
||||
return magnitude
|
||||
# Magnitude-mode fallback (no per-axis components).
|
||||
return max(0.0, float(getattr(feedback, "max_shear", 0.0) or 0.0))
|
||||
|
||||
def _adaptive_force_target(self, feedback):
|
||||
config = self.config
|
||||
if not config.adaptive_grip_enabled:
|
||||
@@ -864,7 +925,7 @@ class SideTriggerGripController:
|
||||
|
||||
shear_start = max(0.0, float(config.adaptive_shear_start_force))
|
||||
shear_full = max(shear_start + 1e-6, float(config.adaptive_shear_full_force))
|
||||
shear = max(0.0, float(feedback.max_shear))
|
||||
shear = self._load_shear(feedback)
|
||||
if shear <= shear_start:
|
||||
return force_min
|
||||
if shear >= shear_full:
|
||||
@@ -1108,7 +1169,7 @@ class SideTriggerGripController:
|
||||
"release_channel": self._release_decision.channel if self._release_decision else "",
|
||||
"manual_hold_active": state == "manual_hold",
|
||||
"control_mode": control_mode,
|
||||
"grip_behavior_mode": getattr(self.config, "grip_behavior_mode", "stable"),
|
||||
"grip_behavior_mode": getattr(self.config, "grip_behavior_mode", "sensitive"),
|
||||
"speed_pct": speed_pct,
|
||||
"gripper_position": gripper_position,
|
||||
"open_pos": self.config.open_pos,
|
||||
|
||||
@@ -4,8 +4,33 @@ import time
|
||||
from ..common.filters import ExponentialFilter
|
||||
|
||||
|
||||
def shear_magnitude(sample):
|
||||
return (float(sample["fshearx"]) ** 2 + float(sample["fsheary"]) ** 2) ** 0.5
|
||||
def _normalize_shear_axes(axes):
|
||||
value = str(axes or "xy").strip().lower()
|
||||
if value in {"y", "fy", "sheary", "fsheary"}:
|
||||
return "y"
|
||||
if value in {"x", "fx", "shearx", "fshearx"}:
|
||||
return "x"
|
||||
return "xy"
|
||||
|
||||
|
||||
def shear_magnitude(sample, axes="xy"):
|
||||
axes = _normalize_shear_axes(axes)
|
||||
shear_x = float(sample["fshearx"])
|
||||
shear_y = float(sample["fsheary"])
|
||||
if axes == "y":
|
||||
return abs(shear_y)
|
||||
if axes == "x":
|
||||
return abs(shear_x)
|
||||
return (shear_x ** 2 + shear_y ** 2) ** 0.5
|
||||
|
||||
|
||||
def combine_shear_components(shear_x, shear_y, axes="xy"):
|
||||
axes = _normalize_shear_axes(axes)
|
||||
if axes == "y":
|
||||
return abs(float(shear_y))
|
||||
if axes == "x":
|
||||
return abs(float(shear_x))
|
||||
return (float(shear_x) ** 2 + float(shear_y) ** 2) ** 0.5
|
||||
|
||||
|
||||
def _component_shear_enabled(converter):
|
||||
@@ -70,12 +95,16 @@ class TactileFeedbackProcessor:
|
||||
return filter_.update(value)
|
||||
return filter_.value
|
||||
|
||||
def _shear_axes(self):
|
||||
return _normalize_shear_axes(getattr(self.config, "shear_control_axes", "xy"))
|
||||
|
||||
def _convert_shear(self, sample):
|
||||
axes = self._shear_axes()
|
||||
if _component_shear_enabled(self.shear_converter):
|
||||
shear_x = self.shear_converter.convert_component(sample["fshearx"])
|
||||
shear_y = self.shear_converter.convert_component(sample["fsheary"])
|
||||
return (shear_x ** 2 + shear_y ** 2) ** 0.5, shear_x, shear_y
|
||||
return self.shear_converter.convert(shear_magnitude(sample)), 0.0, 0.0
|
||||
return combine_shear_components(shear_x, shear_y, axes), shear_x, shear_y
|
||||
return self.shear_converter.convert(shear_magnitude(sample, axes)), 0.0, 0.0
|
||||
|
||||
def calibrate_baseline(self, reader):
|
||||
seconds = self.config.baseline_seconds
|
||||
@@ -133,6 +162,7 @@ class TactileFeedbackProcessor:
|
||||
0.0,
|
||||
self.normal_converter.convert(right["fnormal"]) - self.baseline.right_normal,
|
||||
)
|
||||
axes = self._shear_axes()
|
||||
if _component_shear_enabled(self.shear_converter):
|
||||
_, raw_left_shear_x, raw_left_shear_y = self._convert_shear(left)
|
||||
_, raw_right_shear_x, raw_right_shear_y = self._convert_shear(right)
|
||||
@@ -140,17 +170,17 @@ class TactileFeedbackProcessor:
|
||||
left_shear_y = raw_left_shear_y - self.baseline.left_shear_y
|
||||
right_shear_x = raw_right_shear_x - self.baseline.right_shear_x
|
||||
right_shear_y = raw_right_shear_y - self.baseline.right_shear_y
|
||||
left_shear = (left_shear_x ** 2 + left_shear_y ** 2) ** 0.5
|
||||
right_shear = (right_shear_x ** 2 + right_shear_y ** 2) ** 0.5
|
||||
left_shear = combine_shear_components(left_shear_x, left_shear_y, axes)
|
||||
right_shear = combine_shear_components(right_shear_x, right_shear_y, axes)
|
||||
else:
|
||||
left_shear_x = left_shear_y = right_shear_x = right_shear_y = 0.0
|
||||
left_shear = max(
|
||||
0.0,
|
||||
self.shear_converter.convert(shear_magnitude(left)) - self.baseline.left_shear,
|
||||
self.shear_converter.convert(shear_magnitude(left, axes)) - self.baseline.left_shear,
|
||||
)
|
||||
right_shear = max(
|
||||
0.0,
|
||||
self.shear_converter.convert(shear_magnitude(right)) - self.baseline.right_shear,
|
||||
self.shear_converter.convert(shear_magnitude(right, axes)) - self.baseline.right_shear,
|
||||
)
|
||||
|
||||
left_timestamp, right_timestamp = left.get("timestamp"), right.get("timestamp")
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
"""Route release decisions to the FSM owned by the current grip behavior mode.
|
||||
|
||||
sensitive -> release_watch_sensitive (frozen copy of the tuned sensitive logic)
|
||||
stable / firm -> release_watch (constant clamp, fitted Y displacement)
|
||||
"""
|
||||
|
||||
from . import release_watch as held_release
|
||||
from . import release_watch_sensitive as sensitive_release
|
||||
|
||||
|
||||
class ModeReleaseWatch:
|
||||
def __init__(self, config):
|
||||
self.config = config
|
||||
self._sensitive = sensitive_release.ReleaseWatch(config)
|
||||
self._held = held_release.ReleaseWatch(config)
|
||||
|
||||
@property
|
||||
def active(self):
|
||||
mode = getattr(self.config, "grip_behavior_mode", "sensitive")
|
||||
return self._sensitive if mode == "sensitive" else self._held
|
||||
|
||||
def reset(self):
|
||||
self._sensitive.reset()
|
||||
self._held.reset()
|
||||
|
||||
def __getattr__(self, name):
|
||||
return getattr(self.active, name)
|
||||
@@ -1,13 +1,101 @@
|
||||
"""Release FSM. No clock, serial I/O or callbacks: step() returns a decision."""
|
||||
|
||||
from collections import deque
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
import math
|
||||
|
||||
from .force_curve import ForceCurve, ForceOnset
|
||||
from .feedback import combine_shear_components
|
||||
from .force_curve import ForceCurve, ForceOnset, _fit_curve
|
||||
|
||||
|
||||
CHANNELS = ("left_normal", "right_normal", "left_shear", "right_shear")
|
||||
_SHEAR_COMPONENTS = ("left_shear_x", "left_shear_y", "right_shear_x", "right_shear_y")
|
||||
|
||||
|
||||
def channel_values(config, feedback):
|
||||
"""Release channels; shear may use its own axes independent of the trigger."""
|
||||
values = {key: float(getattr(feedback, key)) for key in CHANNELS}
|
||||
axes = str(getattr(config, "release_shear_axes", "") or "").strip().lower()
|
||||
components = [float(getattr(feedback, key, 0.0) or 0.0) for key in _SHEAR_COMPONENTS]
|
||||
# Magnitude-mode calibration has no per-axis components; keep feedback shear.
|
||||
if axes and any(components):
|
||||
values["left_shear"] = combine_shear_components(components[0], components[1], axes)
|
||||
values["right_shear"] = combine_shear_components(components[2], components[3], axes)
|
||||
return values
|
||||
|
||||
|
||||
def signed_shear_y(feedback, key):
|
||||
"""Signed Y for *key* ("left_shear"/"right_shear"); magnitude if no components."""
|
||||
side = key.split("_", 1)[0]
|
||||
x = float(getattr(feedback, f"{side}_shear_x", 0.0) or 0.0)
|
||||
y = float(getattr(feedback, f"{side}_shear_y", 0.0) or 0.0)
|
||||
if x or y:
|
||||
return y
|
||||
return float(getattr(feedback, key))
|
||||
|
||||
|
||||
class _ShearDisplacement:
|
||||
"""Sustained signed-Y departure from the post-protection baseline.
|
||||
|
||||
Each side is smoothed by a local least-squares line (no abs fold, so a
|
||||
sign flip under full clamp is ordinary travel). A turn never cancels:
|
||||
only falling back under the threshold resets the consecutive count. The
|
||||
baseline follows slow drift while the fitted value stays near it.
|
||||
"""
|
||||
|
||||
KEYS = ("left_shear", "right_shear")
|
||||
FIT_POINTS = 6
|
||||
DRIFT_GAIN = 0.02
|
||||
|
||||
def __init__(self, config):
|
||||
self.config = config
|
||||
self.reset()
|
||||
|
||||
def reset(self):
|
||||
self.interrupt()
|
||||
self.baseline = {}
|
||||
|
||||
def interrupt(self):
|
||||
self.history = {key: deque(maxlen=self.FIT_POINTS) for key in self.KEYS}
|
||||
self.fitted = {}
|
||||
self.count = 0
|
||||
self.channel = ""
|
||||
self.deviation = 0.0
|
||||
|
||||
def observe(self, key, stamp, value):
|
||||
history = self.history[key]
|
||||
if history and stamp <= history[-1][0]:
|
||||
if stamp == history[-1][0]:
|
||||
return
|
||||
history.clear()
|
||||
history.append((stamp, value))
|
||||
if len(history) >= 3:
|
||||
noise_floor = max(1e-6, float(self.config.release_min_change) * 0.025)
|
||||
self.fitted[key] = _fit_curve(list(history), 1, noise_floor).predict(stamp)
|
||||
else:
|
||||
self.fitted[key] = value
|
||||
|
||||
def step(self, threshold):
|
||||
"""Update after fresh samples; returns True once the baseline is armed."""
|
||||
if len(self.fitted) < len(self.KEYS):
|
||||
return False
|
||||
if not self.baseline:
|
||||
self.baseline = dict(self.fitted)
|
||||
return False
|
||||
deviations = {key: self.fitted[key] - self.baseline[key] for key in self.KEYS}
|
||||
key = max(deviations, key=lambda k: abs(deviations[k]))
|
||||
self.deviation = deviations[key]
|
||||
if abs(self.deviation) + 1e-9 >= threshold:
|
||||
self.count += 1
|
||||
self.channel = key
|
||||
return True
|
||||
self.count = 0
|
||||
self.channel = ""
|
||||
for name, deviation in deviations.items():
|
||||
if abs(deviation) < 0.5 * threshold:
|
||||
self.baseline[name] += self.DRIFT_GAIN * deviation
|
||||
return True
|
||||
|
||||
|
||||
class ReleasePhase(str, Enum):
|
||||
@@ -226,6 +314,7 @@ class ReleaseWatch:
|
||||
def __init__(self, config):
|
||||
self.config = config
|
||||
self._side_touch = _SideTouch(config)
|
||||
self._shear_disp = _ShearDisplacement(config)
|
||||
self.reset()
|
||||
|
||||
def reset(self):
|
||||
@@ -241,10 +330,13 @@ class ReleaseWatch:
|
||||
self._tails = {}
|
||||
self._watch_since = None
|
||||
self._candidate = None
|
||||
self._osc_count = 0
|
||||
self._osc_channel = ""
|
||||
self._last_fresh = {}
|
||||
self._rates = {}
|
||||
self._curves = {key: ForceCurve(self.config) for key in CHANNELS}
|
||||
self._side_touch.reset()
|
||||
self._shear_disp.reset()
|
||||
|
||||
@property
|
||||
def armed(self):
|
||||
@@ -261,17 +353,18 @@ class ReleaseWatch:
|
||||
self._contact_at = now
|
||||
self.phase = ReleasePhase.GRIP_SETTLE
|
||||
for frame_time, frame in history:
|
||||
values = {key: float(getattr(frame, key)) for key in CHANNELS}
|
||||
values = channel_values(self.config, frame)
|
||||
for key, value in values.items():
|
||||
stamp = getattr(frame, f"{key.split('_', 1)[0]}_timestamp", None)
|
||||
self._curves[key].observe(frame_time if stamp is None else stamp, value)
|
||||
self._curves[key].discard_onset()
|
||||
self._side_touch.prime(self._pair_time(frame_time, frame), values)
|
||||
if feedback is not None:
|
||||
self._side_touch.seed(self._pair_time(now, feedback), {key: float(getattr(feedback, key)) for key in CHANNELS})
|
||||
values = channel_values(self.config, feedback)
|
||||
self._side_touch.seed(self._pair_time(now, feedback), values)
|
||||
for key in CHANNELS:
|
||||
stamp = getattr(feedback, f"{key.split('_', 1)[0]}_timestamp", None)
|
||||
self._curves[key].observe(now if stamp is None else stamp, float(getattr(feedback, key)))
|
||||
self._curves[key].observe(now if stamp is None else stamp, values[key])
|
||||
self._curves[key].discard_onset()
|
||||
|
||||
@staticmethod
|
||||
@@ -285,6 +378,7 @@ class ReleaseWatch:
|
||||
return
|
||||
self._protect("sample-gap", {})
|
||||
self._side_touch.reset()
|
||||
self._shear_disp.interrupt()
|
||||
self._reference.clear()
|
||||
self._previous.clear()
|
||||
self._rates.clear()
|
||||
@@ -330,6 +424,74 @@ class ReleaseWatch:
|
||||
return True
|
||||
return False
|
||||
|
||||
def _shear_oscillation(self, now):
|
||||
"""Detect repeated shear direction flips with enough peak-to-peak swing.
|
||||
|
||||
Intentional open/close often shows as left/right shear oscillating while
|
||||
normals stay flat and the grip has not yet latched a plateau. Returns
|
||||
the strongest oscillating shear channel name, or "".
|
||||
"""
|
||||
if not getattr(self.config, "release_shear_oscillation_open", False):
|
||||
return ""
|
||||
if not getattr(self.config, "release_shear_onset_enabled", True):
|
||||
return ""
|
||||
window = max(0.12, float(getattr(self.config, "release_shear_oscillation_window_seconds", 0.4) or 0.4))
|
||||
min_flips = max(2, int(getattr(self.config, "release_shear_oscillation_min_flips", 3) or 3))
|
||||
amplitude = max(0.04, float(getattr(self.config, "release_shear_oscillation_amplitude", 0.15) or 0.15))
|
||||
step_floor = max(0.02, amplitude * 0.2)
|
||||
best_key, best_flips, best_span = "", 0, 0.0
|
||||
cutoff = now - window
|
||||
for key in CHANNELS[2:]:
|
||||
points = [(t, v) for t, v in self._curves[key].history if t >= cutoff]
|
||||
if len(points) < min_flips + 2:
|
||||
continue
|
||||
values = [v for _, v in points]
|
||||
span = max(values) - min(values)
|
||||
if span + 1e-9 < amplitude:
|
||||
continue
|
||||
flips = 0
|
||||
prev_sign = 0
|
||||
for i in range(1, len(points)):
|
||||
delta = points[i][1] - points[i - 1][1]
|
||||
if abs(delta) < step_floor:
|
||||
continue
|
||||
sign = 1 if delta > 0 else -1
|
||||
if prev_sign and sign != prev_sign:
|
||||
flips += 1
|
||||
prev_sign = sign
|
||||
if flips >= min_flips and (flips > best_flips or (flips == best_flips and span > best_span)):
|
||||
best_key, best_flips, best_span = key, flips, span
|
||||
return best_key
|
||||
|
||||
def _shear_oscillation_release(self, now, metrics):
|
||||
"""Confirm oscillating shear and release without waiting for plateau."""
|
||||
channel = self._shear_oscillation(now)
|
||||
if not channel:
|
||||
self._osc_count = 0
|
||||
self._osc_channel = ""
|
||||
return None
|
||||
needed = max(2, int(getattr(self.config, "release_shear_oscillation_confirm_samples", 2) or 2))
|
||||
if self._osc_channel != channel:
|
||||
self._osc_channel = channel
|
||||
self._osc_count = 1
|
||||
else:
|
||||
self._osc_count += 1
|
||||
released = self._osc_count >= needed
|
||||
if released:
|
||||
self.phase = ReleasePhase.RELEASED
|
||||
else:
|
||||
self.phase = ReleasePhase.CONFIRMING
|
||||
self._candidate = None
|
||||
return ReleaseDecision(
|
||||
self.phase,
|
||||
"shear_oscillation",
|
||||
release=released,
|
||||
confirm_count=self._osc_count,
|
||||
confirm_needed=needed,
|
||||
channel=channel,
|
||||
**dict(metrics, curve_motion="shear_oscillation"),
|
||||
)
|
||||
|
||||
def _protect(self, reason, values, **metrics):
|
||||
self.phase = (
|
||||
ReleasePhase.SELF_SETTLE if self._grip_settled
|
||||
@@ -337,6 +499,8 @@ class ReleaseWatch:
|
||||
)
|
||||
self._reference.update(values)
|
||||
self._candidate = None
|
||||
self._osc_count = 0
|
||||
self._osc_channel = ""
|
||||
self._watch_since = None
|
||||
self._settle_seen.clear()
|
||||
for curve in self._curves.values():
|
||||
@@ -349,11 +513,12 @@ class ReleaseWatch:
|
||||
|
||||
cfg = self.config
|
||||
timeout = max(0.001, cfg.release_sample_timeout_seconds)
|
||||
values = {key: float(getattr(feedback, key)) for key in CHANNELS}
|
||||
values = channel_values(cfg, feedback)
|
||||
if not all(math.isfinite(value) for value in values.values()):
|
||||
self._previous.clear()
|
||||
self._rates.clear()
|
||||
self._side_touch.reset()
|
||||
self._shear_disp.interrupt()
|
||||
self._curves = {key: ForceCurve(cfg) for key in CHANNELS}
|
||||
self._recovery_until = now + cfg.release_settle_seconds
|
||||
return self._protect("invalid-sample", {})
|
||||
@@ -387,6 +552,7 @@ class ReleaseWatch:
|
||||
stale = any(now - self._last_fresh.get(k, now) > timeout for k in CHANNELS)
|
||||
if stale or discontinuity:
|
||||
self._side_touch.reset()
|
||||
self._shear_disp.interrupt()
|
||||
self._curves = {key: ForceCurve(cfg) for key in CHANNELS}
|
||||
self._recovery_until = now + cfg.release_settle_seconds
|
||||
return self._protect("sample-gap", values, **metrics)
|
||||
@@ -403,17 +569,38 @@ class ReleaseWatch:
|
||||
and now - motion_completed_at < max(0.0, cfg.release_arm_delay_seconds)
|
||||
)
|
||||
contact_guard = now - self._contact_at < max(0.0, cfg.release_min_hold_seconds)
|
||||
displacement = float(getattr(cfg, "release_shear_displacement", 0.0) or 0.0)
|
||||
if displacement > 0:
|
||||
return self._displacement_step(feedback, fresh, contact_guard, displacement, metrics)
|
||||
# Guarded modes: a full-force clamp shakes the shear channels, so the
|
||||
# fast paths (oscillation / side touch) wait for contact hold + servo idle.
|
||||
guarded = bool(getattr(cfg, "release_guard_fast_paths", False))
|
||||
fast_blocked = guarded and (motion_pending or self_tail or contact_guard)
|
||||
|
||||
# Shear oscillation = intentional open/close: bypass plateau / settle wait.
|
||||
if not fast_blocked:
|
||||
osc_decision = self._shear_oscillation_release(now, metrics)
|
||||
if osc_decision is not None:
|
||||
return osc_decision
|
||||
|
||||
protected = motion_pending or self_tail or contact_guard or self.phase in {
|
||||
ReleasePhase.GRIP_SETTLE, ReleasePhase.SELF_SETTLE,
|
||||
}
|
||||
force_settled = self._force_settled(now)
|
||||
if not force_settled:
|
||||
metrics["curve_motion"] = "force_settling"
|
||||
if guarded:
|
||||
side_enabled = not fast_blocked and (
|
||||
self.phase == ReleasePhase.WATCHING
|
||||
or self._side_touch.candidate is not None
|
||||
)
|
||||
else:
|
||||
side_enabled = protected or self._side_touch.candidate is not None
|
||||
side = self._side_touch.step(now, values, fresh,
|
||||
enabled=protected or self._side_touch.candidate is not None,
|
||||
normal_masked=motion_pending or self_tail,
|
||||
enabled=side_enabled,
|
||||
normal_masked=motion_pending or self_tail or fast_blocked,
|
||||
sample_time=self._pair_time(now, feedback),
|
||||
onset_allowed=force_settled)
|
||||
onset_allowed=force_settled and not fast_blocked)
|
||||
if side in ("rejected", "settling-turn"):
|
||||
reason = "settling-turn" if side == "settling-turn" else "transient-rejected"
|
||||
return self._cancel(now, values, reason, metrics, reset_side=False)
|
||||
@@ -583,6 +770,38 @@ class ReleaseWatch:
|
||||
return ReleaseDecision(self.phase, reason,
|
||||
allow_force_update=allow_force, **metrics)
|
||||
|
||||
def _displacement_step(self, feedback, fresh, contact_guard, threshold, metrics):
|
||||
"""Sole release path when displacement is enabled (onset paths off)."""
|
||||
disp = self._shear_disp
|
||||
for key in disp.KEYS:
|
||||
if key in fresh:
|
||||
disp.observe(key, self._previous[key][0], signed_shear_y(feedback, key))
|
||||
metrics = dict(metrics, curve_motion="shear_displacement")
|
||||
if contact_guard:
|
||||
disp.baseline.clear()
|
||||
disp.count = 0
|
||||
self.phase = ReleasePhase.GRIP_SETTLE
|
||||
return ReleaseDecision(self.phase, "grip-settle", **metrics)
|
||||
if not any(key in fresh for key in disp.KEYS):
|
||||
return ReleaseDecision(self.phase, "waiting-sample", allow_force_update=True, **metrics)
|
||||
if not disp.step(threshold):
|
||||
self._grip_settled = True
|
||||
self.phase = ReleasePhase.WATCHING
|
||||
return ReleaseDecision(self.phase, "ready", allow_force_update=True, **metrics)
|
||||
needed = max(1, int(getattr(self.config, "release_shear_displacement_samples", 6) or 6))
|
||||
metrics["shear_change"] = abs(disp.deviation)
|
||||
if disp.count >= needed:
|
||||
self.phase = ReleasePhase.RELEASED
|
||||
else:
|
||||
self.phase = ReleasePhase.CONFIRMING if disp.count else ReleasePhase.WATCHING
|
||||
return ReleaseDecision(
|
||||
self.phase, "shear_displacement" if disp.count else "watch",
|
||||
release=self.phase == ReleasePhase.RELEASED,
|
||||
allow_force_update=not disp.count,
|
||||
confirm_count=disp.count, confirm_needed=needed, channel=disp.channel,
|
||||
**metrics,
|
||||
)
|
||||
|
||||
def _cancel(self, now, values, reason, metrics, *, reset_side=True):
|
||||
# Wait for the rejected pulse's tail; otherwise its falling edge can
|
||||
# become a fresh negative-direction candidate and release on rebound.
|
||||
|
||||
@@ -0,0 +1,680 @@
|
||||
"""Release FSM. No clock, serial I/O or callbacks: step() returns a decision."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
import math
|
||||
|
||||
from .force_curve import ForceCurve, ForceOnset
|
||||
|
||||
|
||||
CHANNELS = ("left_normal", "right_normal", "left_shear", "right_shear")
|
||||
|
||||
|
||||
class ReleasePhase(str, Enum):
|
||||
DISARMED = "disarmed"
|
||||
GRIP_SETTLE = "grip_settle"
|
||||
SELF_SETTLE = "self_settle"
|
||||
WATCHING = "watching"
|
||||
CONFIRMING = "confirming"
|
||||
RELEASED = "released"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReleaseDecision:
|
||||
phase: ReleasePhase
|
||||
reason: str
|
||||
release: bool = False
|
||||
allow_force_update: bool = False
|
||||
normal_change: float = 0.0
|
||||
shear_change: float = 0.0
|
||||
normal_rate: float = 0.0
|
||||
shear_rate: float = 0.0
|
||||
confirm_count: int = 0
|
||||
confirm_needed: int = 0
|
||||
channel: str = ""
|
||||
innovation: float = 0.0
|
||||
curve_motion: str = ""
|
||||
|
||||
@property
|
||||
def action(self):
|
||||
return (
|
||||
f"release-{self.reason} phase={self.phase.value} "
|
||||
f"confirm={self.confirm_count}/{self.confirm_needed} "
|
||||
f"rN={self.normal_rate:.2f} rS={self.shear_rate:.2f}"
|
||||
f" channel={self.channel or '-'} innovation={self.innovation:.4f} curve={self.curve_motion or '-'}"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class _Candidate:
|
||||
kind: str
|
||||
channel: str
|
||||
origin: float
|
||||
direction: float
|
||||
peak: float
|
||||
started_at: float
|
||||
needed: int
|
||||
floor: float
|
||||
count: int = 1
|
||||
onset: ForceOnset | None = None
|
||||
last_displacement: float = 0.0
|
||||
last_sample_time: float = 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class _Rebound:
|
||||
origin: float
|
||||
direction: float
|
||||
remaining: float
|
||||
|
||||
|
||||
def _returning_tail(tails, key, value, floor):
|
||||
"""Suppress only a rejected pulse returning towards its original baseline.
|
||||
|
||||
This is channel-local and ends on return or renewed movement. It does not
|
||||
wait for unrelated channels to become quiet or impose a new hold timer.
|
||||
"""
|
||||
tail = tails.get(key)
|
||||
if tail is None:
|
||||
return False
|
||||
remaining = (value - tail.origin) * tail.direction
|
||||
if abs(remaining) <= floor * 0.5:
|
||||
del tails[key]
|
||||
return True
|
||||
if remaining < 0 or remaining > tail.remaining + floor * 0.25:
|
||||
del tails[key]
|
||||
return False
|
||||
tail.remaining = min(tail.remaining, remaining)
|
||||
return True
|
||||
|
||||
|
||||
class _SideTouch:
|
||||
"""Observe a new lateral touch even during clamp protection.
|
||||
|
||||
Bilateral common-mode force is cancelled. A vote needs new frames from
|
||||
both sides so asynchronous camera updates cannot fabricate a difference.
|
||||
Normal force stays masked during an actual force command and its tail;
|
||||
differential shear remains available without a contact timer.
|
||||
"""
|
||||
|
||||
def __init__(self, config):
|
||||
self.config = config
|
||||
self.reset()
|
||||
|
||||
def reset(self):
|
||||
self.last_pair = None
|
||||
self.seen = set()
|
||||
self.candidate = None
|
||||
self.recover_until = 0.0
|
||||
self.tails = {}
|
||||
self.curves = {index: ForceCurve(self.config) for index in (1, 2)}
|
||||
|
||||
def seed(self, now, values):
|
||||
self.last_pair = (now, values["left_normal"] - values["right_normal"],
|
||||
values["left_shear"] - values["right_shear"])
|
||||
for index in (1, 2):
|
||||
if not self.curves[index].history:
|
||||
self.curves[index].observe(now, self.last_pair[index])
|
||||
|
||||
def prime(self, now, values):
|
||||
for index, field in ((1, "normal"), (2, "shear")):
|
||||
self.curves[index].observe(now, values[f"left_{field}"] - values[f"right_{field}"])
|
||||
self.curves[index].discard_onset()
|
||||
|
||||
def step(self, now, values, fresh, enabled, normal_masked, sample_time,
|
||||
onset_allowed=True):
|
||||
self.seen.update(fresh)
|
||||
if not all(key in self.seen for key in CHANNELS):
|
||||
return None
|
||||
self.seen.clear()
|
||||
sample = (sample_time, values["left_normal"] - values["right_normal"],
|
||||
values["left_shear"] - values["right_shear"])
|
||||
for index in (1, 2):
|
||||
self.curves[index].observe(sample_time, sample[index])
|
||||
if normal_masked:
|
||||
self.curves[1].discard_onset()
|
||||
if not enabled or now < self.recover_until:
|
||||
self.seed(sample_time, values)
|
||||
self.candidate = None
|
||||
for curve in self.curves.values():
|
||||
curve.discard_onset()
|
||||
return None
|
||||
self.last_pair = sample
|
||||
# Stable grip: keep learning the long ramp curve, but do not arm side
|
||||
# touch while the settling slope is still decaying toward zero.
|
||||
if not onset_allowed:
|
||||
self.candidate = None
|
||||
for curve in self.curves.values():
|
||||
curve.discard_onset()
|
||||
return None
|
||||
# Drop leftover candidates that the profile has disabled.
|
||||
if self.candidate is not None:
|
||||
if (self.candidate.channel == "side_shear"
|
||||
and not getattr(self.config, "release_side_shear_enabled", True)):
|
||||
self.candidate = None
|
||||
self.curves[2].discard_onset()
|
||||
elif (self.candidate.channel == "side_normal"
|
||||
and not getattr(self.config, "release_side_normal_enabled", True)):
|
||||
self.candidate = None
|
||||
self.curves[1].discard_onset()
|
||||
returning = {index for index in (1, 2) if _returning_tail(
|
||||
self.tails, index, sample[index], self.config.release_side_touch_change)}
|
||||
if returning:
|
||||
for index in returning:
|
||||
self.curves[index].discard_onset()
|
||||
candidate = self.candidate
|
||||
if candidate is not None:
|
||||
index = 1 if candidate.channel == "side_normal" else 2
|
||||
baseline = candidate.onset.baseline_at(sample_time)
|
||||
delta = (sample[index] - baseline) * candidate.direction
|
||||
return_margin = max(self.config.release_min_change * 0.1,
|
||||
candidate.onset.model.change_uncertainty(candidate.last_sample_time, sample_time))
|
||||
turning = delta < candidate.last_displacement - return_margin
|
||||
if (now - candidate.started_at > self.config.release_confirm_window_seconds
|
||||
or delta + 1e-9 < max(candidate.floor, 0.6 * candidate.peak)
|
||||
or not candidate.onset.significant_at(sample_time, sample[index])
|
||||
or (normal_masked and index == 1) or turning):
|
||||
self.candidate = None
|
||||
self.tails[index] = _Rebound(baseline, candidate.direction, delta)
|
||||
self.recover_until = now + self.config.release_settle_seconds
|
||||
self.seed(sample_time, values)
|
||||
self.curves[index].discard_onset()
|
||||
return "settling-turn" if turning else "rejected"
|
||||
candidate.peak = max(candidate.peak, delta)
|
||||
candidate.last_displacement = delta
|
||||
candidate.last_sample_time = sample_time
|
||||
candidate.count += 1
|
||||
else:
|
||||
shear_ok = getattr(self.config, "release_side_shear_enabled", True)
|
||||
normal_ok = getattr(self.config, "release_side_normal_enabled", True)
|
||||
channels = []
|
||||
if shear_ok:
|
||||
channels.append((2, "side_shear"))
|
||||
if normal_ok:
|
||||
channels.append((1, "side_normal"))
|
||||
for index, channel in channels:
|
||||
if index in returning or (normal_masked and index == 1):
|
||||
continue
|
||||
onset = self.curves[index].onset
|
||||
if onset is None:
|
||||
continue
|
||||
delta = sample[index] - onset.baseline_at(sample_time)
|
||||
if (abs(delta) + 1e-9 >= self.config.release_side_touch_change
|
||||
and (sample[index] - onset.origin) * onset.direction + 1e-9 >= self.config.release_side_touch_change
|
||||
and onset.significant_at(sample_time, sample[index])
|
||||
and delta * onset.direction > 0):
|
||||
candidate = _Candidate("side_touch", channel, onset.origin,
|
||||
1.0 if delta > 0 else -1.0, abs(delta), now,
|
||||
max(2, self.config.release_confirm_samples),
|
||||
self.config.release_side_touch_change,
|
||||
onset=onset, last_displacement=abs(delta), last_sample_time=sample_time)
|
||||
self.candidate = candidate
|
||||
break
|
||||
return candidate
|
||||
|
||||
|
||||
class ReleaseWatch:
|
||||
"""Immediate side touch + bounded clamp protection + persistent displacement.
|
||||
|
||||
A force command masks normal changes through execution and a short tail.
|
||||
Lateral shear remains observable; global quiet never gates release.
|
||||
New force commands require an observation window, so repeated adaptive
|
||||
commands cannot continuously re-arm a timer or erase release evidence.
|
||||
Sensor timestamps are optional for replay; production supplies both sides.
|
||||
"""
|
||||
|
||||
def __init__(self, config):
|
||||
self.config = config
|
||||
self._side_touch = _SideTouch(config)
|
||||
self.reset()
|
||||
|
||||
def reset(self):
|
||||
self.phase = ReleasePhase.DISARMED
|
||||
self._contact_at = None
|
||||
self._grip_settled = False
|
||||
self._slope_settle_since = None
|
||||
self._plateau_latched = False
|
||||
self._reference = {}
|
||||
self._previous = {}
|
||||
self._recovery_until = 0.0
|
||||
self._settle_seen = set()
|
||||
self._tails = {}
|
||||
self._watch_since = None
|
||||
self._candidate = None
|
||||
self._osc_count = 0
|
||||
self._osc_channel = ""
|
||||
self._osc_since = None
|
||||
self._last_fresh = {}
|
||||
self._rates = {}
|
||||
self._curves = {key: ForceCurve(self.config) for key in CHANNELS}
|
||||
self._side_touch.reset()
|
||||
|
||||
@property
|
||||
def armed(self):
|
||||
return (self.phase in {ReleasePhase.WATCHING, ReleasePhase.CONFIRMING}
|
||||
or self._side_touch.candidate is not None)
|
||||
|
||||
@property
|
||||
def side_watch_active(self):
|
||||
return (self.phase not in {ReleasePhase.DISARMED, ReleasePhase.RELEASED}
|
||||
and self._side_touch.last_pair is not None and len(self._previous) == len(CHANNELS))
|
||||
|
||||
def contact(self, now, feedback=None, history=()):
|
||||
self.reset()
|
||||
self._contact_at = now
|
||||
self.phase = ReleasePhase.GRIP_SETTLE
|
||||
for frame_time, frame in history:
|
||||
values = {key: float(getattr(frame, key)) for key in CHANNELS}
|
||||
for key, value in values.items():
|
||||
stamp = getattr(frame, f"{key.split('_', 1)[0]}_timestamp", None)
|
||||
self._curves[key].observe(frame_time if stamp is None else stamp, value)
|
||||
self._curves[key].discard_onset()
|
||||
self._side_touch.prime(self._pair_time(frame_time, frame), values)
|
||||
if feedback is not None:
|
||||
self._side_touch.seed(self._pair_time(now, feedback), {key: float(getattr(feedback, key)) for key in CHANNELS})
|
||||
for key in CHANNELS:
|
||||
stamp = getattr(feedback, f"{key.split('_', 1)[0]}_timestamp", None)
|
||||
self._curves[key].observe(now if stamp is None else stamp, float(getattr(feedback, key)))
|
||||
self._curves[key].discard_onset()
|
||||
|
||||
@staticmethod
|
||||
def _pair_time(now, feedback):
|
||||
stamps = (getattr(feedback, "left_timestamp", None), getattr(feedback, "right_timestamp", None))
|
||||
return now if None in stamps else max(stamps)
|
||||
|
||||
def suspend(self):
|
||||
"""Discard evidence during missing feedback, sensor restart or manual mode."""
|
||||
if self.phase in {ReleasePhase.DISARMED, ReleasePhase.RELEASED}:
|
||||
return
|
||||
self._protect("sample-gap", {})
|
||||
self._side_touch.reset()
|
||||
self._reference.clear()
|
||||
self._previous.clear()
|
||||
self._rates.clear()
|
||||
self._last_fresh.clear()
|
||||
self._tails.clear()
|
||||
self._curves = {key: ForceCurve(self.config) for key in CHANNELS}
|
||||
|
||||
def _normal_slopes_live_ok(self, now):
|
||||
"""Live check: normal long-window slopes are near zero long enough."""
|
||||
if not getattr(self.config, "release_require_slope_settle", False):
|
||||
return True
|
||||
threshold = max(0.0, float(getattr(self.config, "release_slope_settle_rate", 0.2) or 0.2))
|
||||
min_points = max(3, int(getattr(self.config, "release_slope_min_points", 8) or 8))
|
||||
hold = max(0.0, float(getattr(self.config, "release_slope_hold_seconds", 0.0) or 0.0))
|
||||
normal_only = getattr(self.config, "release_settle_normal_only", True)
|
||||
keys = CHANNELS[:2] if normal_only else CHANNELS
|
||||
for key in keys:
|
||||
curve = self._curves[key]
|
||||
if len(curve.history) < min_points or abs(curve.fit_slope) > threshold:
|
||||
self._slope_settle_since = None
|
||||
return False
|
||||
if self._slope_settle_since is None:
|
||||
self._slope_settle_since = now
|
||||
if hold > 0 and now - self._slope_settle_since < hold:
|
||||
return False
|
||||
return True
|
||||
|
||||
def _force_settled(self, now):
|
||||
"""Release-armed after the first relative normal plateau (latched).
|
||||
|
||||
The live slope near zero is only required to *enter* the armed state.
|
||||
Once latched, further adaptive 加力 or an intentional open push may make
|
||||
the slope non-zero again — that must not revoke release permission, or
|
||||
the open gesture is mistaken for "still settling".
|
||||
"""
|
||||
if not getattr(self.config, "release_require_slope_settle", False):
|
||||
self._plateau_latched = True
|
||||
return True
|
||||
if self._plateau_latched:
|
||||
return True
|
||||
if self._normal_slopes_live_ok(now):
|
||||
self._plateau_latched = True
|
||||
return True
|
||||
return False
|
||||
|
||||
def _shear_oscillation(self, now):
|
||||
"""Detect repeated shear direction flips with enough peak-to-peak swing.
|
||||
|
||||
Intentional open/close often shows as left/right shear oscillating while
|
||||
normals stay flat and the grip has not yet latched a plateau. Returns
|
||||
the strongest oscillating shear channel name, or "".
|
||||
"""
|
||||
if not getattr(self.config, "release_shear_oscillation_open", False):
|
||||
return ""
|
||||
if not getattr(self.config, "release_shear_onset_enabled", True):
|
||||
return ""
|
||||
window = max(0.12, float(getattr(self.config, "release_shear_oscillation_window_seconds", 0.4) or 0.4))
|
||||
min_flips = max(2, int(getattr(self.config, "release_shear_oscillation_min_flips", 3) or 3))
|
||||
amplitude = max(0.04, float(getattr(self.config, "release_shear_oscillation_amplitude", 0.15) or 0.15))
|
||||
step_floor = max(0.02, amplitude * 0.2)
|
||||
best_key, best_flips, best_span = "", 0, 0.0
|
||||
cutoff = now - window
|
||||
for key in CHANNELS[2:]:
|
||||
points = [(t, v) for t, v in self._curves[key].history if t >= cutoff]
|
||||
if len(points) < min_flips + 2:
|
||||
continue
|
||||
values = [v for _, v in points]
|
||||
span = max(values) - min(values)
|
||||
if span + 1e-9 < amplitude:
|
||||
continue
|
||||
flips = 0
|
||||
prev_sign = 0
|
||||
for i in range(1, len(points)):
|
||||
delta = points[i][1] - points[i - 1][1]
|
||||
if abs(delta) < step_floor:
|
||||
continue
|
||||
sign = 1 if delta > 0 else -1
|
||||
if prev_sign and sign != prev_sign:
|
||||
flips += 1
|
||||
prev_sign = sign
|
||||
if flips >= min_flips and (flips > best_flips or (flips == best_flips and span > best_span)):
|
||||
best_key, best_flips, best_span = key, flips, span
|
||||
return best_key
|
||||
|
||||
def _shear_oscillation_release(self, now, metrics):
|
||||
"""Confirm oscillating shear and release without waiting for plateau."""
|
||||
channel = self._shear_oscillation(now)
|
||||
if not channel:
|
||||
self._osc_count = 0
|
||||
self._osc_channel = ""
|
||||
self._osc_since = None
|
||||
return None
|
||||
needed = max(2, int(getattr(self.config, "release_shear_oscillation_confirm_samples", 2) or 2))
|
||||
min_seconds = max(0.0, float(getattr(self.config, "release_shear_oscillation_min_seconds", 0.0) or 0.0))
|
||||
if self._osc_channel != channel:
|
||||
self._osc_channel = channel
|
||||
self._osc_count = 1
|
||||
self._osc_since = now
|
||||
else:
|
||||
self._osc_count += 1
|
||||
released = self._osc_count >= needed and now - self._osc_since >= min_seconds
|
||||
if released:
|
||||
self.phase = ReleasePhase.RELEASED
|
||||
else:
|
||||
self.phase = ReleasePhase.CONFIRMING
|
||||
self._candidate = None
|
||||
return ReleaseDecision(
|
||||
self.phase,
|
||||
"shear_oscillation",
|
||||
release=released,
|
||||
confirm_count=self._osc_count,
|
||||
confirm_needed=needed,
|
||||
channel=channel,
|
||||
**dict(metrics, curve_motion="shear_oscillation"),
|
||||
)
|
||||
|
||||
def _protect(self, reason, values, **metrics):
|
||||
self.phase = (
|
||||
ReleasePhase.SELF_SETTLE if self._grip_settled
|
||||
else ReleasePhase.GRIP_SETTLE
|
||||
)
|
||||
self._reference.update(values)
|
||||
self._candidate = None
|
||||
self._osc_count = 0
|
||||
self._osc_channel = ""
|
||||
self._osc_since = None
|
||||
self._watch_since = None
|
||||
self._settle_seen.clear()
|
||||
for curve in self._curves.values():
|
||||
curve.discard_onset()
|
||||
return ReleaseDecision(self.phase, reason, **metrics)
|
||||
|
||||
def step(self, now, feedback, *, motion_pending=False, motion_completed_at=None):
|
||||
if self.phase in {ReleasePhase.DISARMED, ReleasePhase.RELEASED}:
|
||||
return ReleaseDecision(self.phase, self.phase.value)
|
||||
|
||||
cfg = self.config
|
||||
timeout = max(0.001, cfg.release_sample_timeout_seconds)
|
||||
values = {key: float(getattr(feedback, key)) for key in CHANNELS}
|
||||
if not all(math.isfinite(value) for value in values.values()):
|
||||
self._previous.clear()
|
||||
self._rates.clear()
|
||||
self._side_touch.reset()
|
||||
self._curves = {key: ForceCurve(cfg) for key in CHANNELS}
|
||||
self._recovery_until = now + cfg.release_settle_seconds
|
||||
return self._protect("invalid-sample", {})
|
||||
|
||||
rates, fresh = {}, set()
|
||||
discontinuity = False
|
||||
for key, value in values.items():
|
||||
side = key.split("_", 1)[0]
|
||||
stamp = getattr(feedback, f"{side}_timestamp", None)
|
||||
stamp = now if stamp is None else stamp
|
||||
previous = self._previous.get(key)
|
||||
if previous is not None and stamp == previous[0]:
|
||||
continue
|
||||
fresh.add(key)
|
||||
self._last_fresh[key] = now
|
||||
if previous is not None:
|
||||
dt = stamp - previous[0]
|
||||
if 0 < dt <= timeout:
|
||||
rates[key] = (value - previous[1]) / dt
|
||||
else:
|
||||
discontinuity = True
|
||||
self._previous[key] = (stamp, value)
|
||||
self._rates.update(rates)
|
||||
|
||||
normal_rate = max((abs(rates.get(k, 0.0)) for k in CHANNELS[:2]), default=0.0)
|
||||
shear_rate = max((abs(rates.get(k, 0.0)) for k in CHANNELS[2:]), default=0.0)
|
||||
normal_change = max(abs(values[k] - self._reference.get(k, values[k])) for k in CHANNELS[:2])
|
||||
shear_change = max(abs(values[k] - self._reference.get(k, values[k])) for k in CHANNELS[2:])
|
||||
metrics = dict(normal_change=normal_change, shear_change=shear_change,
|
||||
normal_rate=normal_rate, shear_rate=shear_rate)
|
||||
stale = any(now - self._last_fresh.get(k, now) > timeout for k in CHANNELS)
|
||||
if stale or discontinuity:
|
||||
self._side_touch.reset()
|
||||
self._curves = {key: ForceCurve(cfg) for key in CHANNELS}
|
||||
self._recovery_until = now + cfg.release_settle_seconds
|
||||
return self._protect("sample-gap", values, **metrics)
|
||||
|
||||
for key in fresh:
|
||||
self._curves[key].observe(self._previous[key][0], values[key])
|
||||
motions = {curve.motion for curve in self._curves.values()}
|
||||
metrics["curve_motion"] = ("new_onset" if "new_onset" in motions else
|
||||
"settling_turn" if "settling_turn" in motions else
|
||||
"settling_flatten" if "settling_flatten" in motions else "curve_follow")
|
||||
|
||||
# Shear oscillation = intentional open/close: bypass plateau / settle wait.
|
||||
osc_decision = self._shear_oscillation_release(now, metrics)
|
||||
if osc_decision is not None:
|
||||
return osc_decision
|
||||
|
||||
self_tail = (
|
||||
motion_completed_at is not None
|
||||
and now - motion_completed_at < max(0.0, cfg.release_arm_delay_seconds)
|
||||
)
|
||||
contact_guard = now - self._contact_at < max(0.0, cfg.release_min_hold_seconds)
|
||||
protected = motion_pending or self_tail or contact_guard or self.phase in {
|
||||
ReleasePhase.GRIP_SETTLE, ReleasePhase.SELF_SETTLE,
|
||||
}
|
||||
force_settled = self._force_settled(now)
|
||||
if not force_settled:
|
||||
metrics["curve_motion"] = "force_settling"
|
||||
side = self._side_touch.step(now, values, fresh,
|
||||
enabled=protected or self._side_touch.candidate is not None,
|
||||
normal_masked=motion_pending or self_tail,
|
||||
sample_time=self._pair_time(now, feedback),
|
||||
onset_allowed=force_settled)
|
||||
if side in ("rejected", "settling-turn"):
|
||||
reason = "settling-turn" if side == "settling-turn" else "transient-rejected"
|
||||
return self._cancel(now, values, reason, metrics, reset_side=False)
|
||||
if side is not None:
|
||||
self._candidate = None
|
||||
released = side.count >= side.needed
|
||||
if released:
|
||||
self.phase = ReleasePhase.RELEASED
|
||||
return ReleaseDecision(ReleasePhase.RELEASED if released else ReleasePhase.CONFIRMING,
|
||||
"side_touch", release=released, confirm_count=side.count,
|
||||
confirm_needed=side.needed, channel=side.channel,
|
||||
innovation=side.onset.innovation,
|
||||
**dict(metrics, curve_motion="onset_confirm"))
|
||||
|
||||
slope_gate = getattr(cfg, "release_require_slope_settle", False)
|
||||
# Contact min-hold always clamps. Self-motion clamps until the first
|
||||
# normal plateau is latched (stable) — or always when slope gate is off.
|
||||
if contact_guard or ((motion_pending or self_tail) and (not slope_gate or not force_settled)):
|
||||
reason = ("force-settling" if slope_gate and not force_settled
|
||||
else "self-settle" if self._grip_settled else "grip-settle")
|
||||
return self._protect(reason, values, **metrics)
|
||||
if slope_gate and force_settled and (motion_pending or self_tail):
|
||||
# Adaptive 加力 after plateau: keep watching for intentional open.
|
||||
if self.phase in {ReleasePhase.GRIP_SETTLE, ReleasePhase.SELF_SETTLE}:
|
||||
self._grip_settled = True
|
||||
self.phase = ReleasePhase.WATCHING
|
||||
if self._watch_since is None:
|
||||
self._watch_since = now
|
||||
for curve in self._curves.values():
|
||||
curve.discard_onset()
|
||||
|
||||
if not fresh:
|
||||
return ReleaseDecision(self.phase, "waiting-sample", **metrics)
|
||||
|
||||
# Quiet is diagnostic only. No global "all four rates quiet" arming
|
||||
# gate: even small continuous camera noise can keep that gate shut.
|
||||
quiet = (
|
||||
len(self._rates) == len(CHANNELS)
|
||||
and max(abs(rate) for rate in self._rates.values()) <= cfg.release_slow_rate
|
||||
)
|
||||
if self.phase in {ReleasePhase.GRIP_SETTLE, ReleasePhase.SELF_SETTLE}:
|
||||
self._settle_seen.update(fresh)
|
||||
self._reference.update(values)
|
||||
if now < self._recovery_until or len(self._settle_seen) < len(CHANNELS):
|
||||
reason = "self-settle" if self._grip_settled else "grip-settle"
|
||||
return ReleaseDecision(self.phase, reason, **metrics)
|
||||
# Stable mode waits until the long-history slope has decayed near
|
||||
# zero before ordinary watching / release paths arm.
|
||||
if not force_settled:
|
||||
return ReleaseDecision(self.phase, "force-settling", **metrics)
|
||||
self._grip_settled = True
|
||||
self.phase = ReleasePhase.WATCHING
|
||||
self._watch_since = now
|
||||
for curve in self._curves.values():
|
||||
curve.discard_onset()
|
||||
return ReleaseDecision(self.phase, "ready", **metrics)
|
||||
returning = {key for key in fresh if _returning_tail(
|
||||
self._tails, key, values[key], cfg.release_min_change)}
|
||||
for key in returning:
|
||||
self._reference[key] = values[key]
|
||||
self._curves[key].discard_onset()
|
||||
|
||||
# Drop watching candidates on channel families the profile disabled.
|
||||
if (self._candidate is not None
|
||||
and self._candidate.channel.endswith("normal")
|
||||
and not getattr(cfg, "release_normal_onset_enabled", True)):
|
||||
self._candidate = None
|
||||
if self._grip_settled:
|
||||
self.phase = ReleasePhase.WATCHING
|
||||
if (self._candidate is not None
|
||||
and self._candidate.channel.endswith("shear")
|
||||
and not getattr(cfg, "release_shear_onset_enabled", True)):
|
||||
self._candidate = None
|
||||
if self._grip_settled:
|
||||
self.phase = ReleasePhase.WATCHING
|
||||
|
||||
candidate = self._candidate
|
||||
if candidate is not None:
|
||||
if now - candidate.started_at > cfg.release_confirm_window_seconds:
|
||||
return self._cancel(now, values, "confirm-expired", metrics)
|
||||
if candidate.channel not in fresh:
|
||||
return ReleaseDecision(self.phase, "waiting-sample", **metrics)
|
||||
stamp = self._previous[candidate.channel][0]
|
||||
displacement = (values[candidate.channel] - candidate.onset.baseline_at(stamp)) * candidate.direction
|
||||
return_margin = max(cfg.release_min_change * 0.1,
|
||||
candidate.onset.model.change_uncertainty(candidate.last_sample_time, stamp))
|
||||
if displacement < candidate.last_displacement - return_margin:
|
||||
return self._cancel(now, values, "settling-turn", metrics)
|
||||
# A spike returning to baseline, reversal, or channel swap is not a
|
||||
# second vote. A step that stays displaced is, even with dF/dt=0.
|
||||
if displacement + 1e-9 < max(candidate.floor, 0.6 * candidate.peak):
|
||||
return self._cancel(now, values, "transient-rejected", metrics)
|
||||
if not candidate.onset.significant_at(stamp, values[candidate.channel]):
|
||||
return self._cancel(now, values, "transient-rejected", metrics)
|
||||
candidate.peak = max(candidate.peak, displacement)
|
||||
candidate.last_displacement = displacement
|
||||
candidate.last_sample_time = stamp
|
||||
candidate.count += 1
|
||||
else:
|
||||
choices = []
|
||||
for key in fresh:
|
||||
if key in returning:
|
||||
continue
|
||||
is_normal = key.endswith("normal")
|
||||
if is_normal and not getattr(cfg, "release_normal_onset_enabled", True):
|
||||
continue
|
||||
if (not is_normal
|
||||
and not getattr(cfg, "release_shear_onset_enabled", True)):
|
||||
continue
|
||||
onset = self._curves[key].onset
|
||||
if onset is None or not onset.significant_at(self._previous[key][0], values[key]):
|
||||
continue
|
||||
rate_limit = cfg.release_normal_rate if is_normal else cfg.release_shear_rate
|
||||
weak_limit = cfg.release_normal_change if is_normal else cfg.release_shear_change
|
||||
delta = values[key] - onset.baseline_at(self._previous[key][0])
|
||||
rate = rates.get(key, 0.0)
|
||||
kind = None
|
||||
origin = onset.origin
|
||||
measured = (values[key] - origin) * onset.direction
|
||||
if (abs(rate) >= rate_limit and abs(delta) >= cfg.release_min_change
|
||||
and measured + 1e-9 >= cfg.release_min_change and rate * delta > 0):
|
||||
kind, needed, floor = "sudden", max(2, cfg.release_confirm_samples), cfg.release_min_change
|
||||
elif (abs(delta) + 1e-9 >= cfg.release_min_change
|
||||
and measured + 1e-9 >= cfg.release_min_change and delta * onset.direction > 0):
|
||||
kind, needed, floor = "touch", max(2, cfg.release_confirm_samples), cfg.release_min_change
|
||||
elif abs(delta) >= weak_limit and measured >= weak_limit:
|
||||
kind = "slow_external"
|
||||
needed = max(2, cfg.release_confirm_samples, cfg.release_slow_confirm_samples)
|
||||
floor = weak_limit
|
||||
if kind:
|
||||
choices.append((kind == "sudden", abs(delta), key, kind, needed, floor, delta, origin, onset))
|
||||
if choices:
|
||||
_, _, key, kind, needed, floor, delta, origin, onset = max(choices)
|
||||
candidate = _Candidate(kind, key, origin,
|
||||
1.0 if delta > 0 else -1.0, abs(delta), now, needed, floor,
|
||||
onset=onset, last_displacement=abs(delta),
|
||||
last_sample_time=self._previous[key][0])
|
||||
self._candidate = candidate
|
||||
self.phase = ReleasePhase.CONFIRMING
|
||||
|
||||
if candidate is not None:
|
||||
released = candidate.count >= candidate.needed
|
||||
if released:
|
||||
self.phase = ReleasePhase.RELEASED
|
||||
return ReleaseDecision(self.phase, candidate.kind, release=released,
|
||||
confirm_count=candidate.count,
|
||||
confirm_needed=candidate.needed, channel=candidate.channel,
|
||||
innovation=candidate.onset.innovation,
|
||||
**dict(metrics, curve_motion="onset_confirm"))
|
||||
|
||||
# Follow curve continuation, including large smooth turns. Displacement
|
||||
# without a new onset must never become a weak-path release later.
|
||||
for key in fresh:
|
||||
if self._curves[key].onset is None:
|
||||
self._reference[key] = values[key]
|
||||
observation = max(0.1, max(2, cfg.release_confirm_samples) / max(1.0, cfg.control_hz))
|
||||
# Ordinary noise must not prevent clamp force from reaching its target.
|
||||
# Meaningful coherent movement reserves the next frame for release.
|
||||
touch_pending = any(
|
||||
self._curves[key].onset is not None
|
||||
and abs(values[key] - self._curves[key].onset.baseline_at(self._previous[key][0])) >= cfg.release_min_change * 0.5
|
||||
for key in fresh
|
||||
)
|
||||
allow_force = (not touch_pending and self._watch_since is not None
|
||||
and now - self._watch_since >= observation)
|
||||
reason = "settling-turn" if metrics["curve_motion"] == "settling_turn" else "slow_internal" if quiet else "watch"
|
||||
return ReleaseDecision(self.phase, reason,
|
||||
allow_force_update=allow_force, **metrics)
|
||||
|
||||
def _cancel(self, now, values, reason, metrics, *, reset_side=True):
|
||||
# Wait for the rejected pulse's tail; otherwise its falling edge can
|
||||
# become a fresh negative-direction candidate and release on rebound.
|
||||
self._recovery_until = now + self.config.release_settle_seconds
|
||||
if self._candidate is not None:
|
||||
candidate = self._candidate
|
||||
baseline = candidate.onset.baseline_at(self._previous[candidate.channel][0])
|
||||
remaining = (values[candidate.channel] - baseline) * candidate.direction
|
||||
self._tails[candidate.channel] = _Rebound(baseline, candidate.direction, remaining)
|
||||
if reset_side:
|
||||
self._side_touch.reset()
|
||||
self._side_touch.recover_until = self._recovery_until
|
||||
return self._protect(reason, values, **metrics)
|
||||
@@ -3,10 +3,13 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import dataclass, replace
|
||||
|
||||
import cv2
|
||||
|
||||
@@ -15,6 +18,12 @@ import cv2
|
||||
# prevents a search from spending seconds probing empty DirectShow devices.
|
||||
DEFAULT_MAX_CAMERA_INDEX = 3
|
||||
|
||||
# Gripper tactile sensors on this project: SunplusIT UVC "USB Camera".
|
||||
SENSOR_FRIENDLY_NAME = "USB Camera"
|
||||
SENSOR_VENDOR_NAME = "SunplusIT Inc"
|
||||
SENSOR_NAME_NEEDLE = "usb camera"
|
||||
SENSOR_VENDOR_NEEDLES = ("sunplusit", "1bcf", "vid_1bcf")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CameraInfo:
|
||||
@@ -23,6 +32,11 @@ class CameraInfo:
|
||||
backend_api: int
|
||||
width: int
|
||||
height: int
|
||||
friendly_name: str = ""
|
||||
manufacturer: str = ""
|
||||
vid_pid: str = ""
|
||||
location: str = ""
|
||||
instance_id: str = ""
|
||||
|
||||
@property
|
||||
def source(self):
|
||||
@@ -31,9 +45,15 @@ class CameraInfo:
|
||||
return str(self.index)
|
||||
|
||||
def label(self):
|
||||
device = self.source if sys.platform.startswith("linux") else f"相机 {self.index}"
|
||||
device = self.source if sys.platform.startswith("linux") else f"ID {self.index}"
|
||||
name = self.friendly_name or "Camera"
|
||||
vendor = self.manufacturer or self.vid_pid
|
||||
meta = f"{name}"
|
||||
if vendor:
|
||||
meta = f"{meta} / {vendor}"
|
||||
resolution = f" {self.width}x{self.height}" if self.width and self.height else ""
|
||||
return f"{device}{resolution} [{self.backend_name}]"
|
||||
where = f" {self.location}" if self.location else ""
|
||||
return f"{device} {meta}{resolution}{where} [{self.backend_name}]"
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
@@ -80,10 +100,11 @@ def _suppress_opencv_noise():
|
||||
def _backend_candidates():
|
||||
candidates = []
|
||||
if sys.platform == "win32":
|
||||
if hasattr(cv2, "CAP_DSHOW"):
|
||||
candidates.append(("DSHOW", cv2.CAP_DSHOW))
|
||||
# MSMF currently opens the SunplusIT sensors more reliably than DSHOW.
|
||||
if hasattr(cv2, "CAP_MSMF"):
|
||||
candidates.append(("MSMF", cv2.CAP_MSMF))
|
||||
if hasattr(cv2, "CAP_DSHOW"):
|
||||
candidates.append(("DSHOW", cv2.CAP_DSHOW))
|
||||
elif sys.platform.startswith("linux") and hasattr(cv2, "CAP_V4L2"):
|
||||
candidates.append(("V4L2", cv2.CAP_V4L2))
|
||||
return candidates or [("ANY", cv2.CAP_ANY)]
|
||||
@@ -113,16 +134,122 @@ def _candidate_indices(max_index):
|
||||
return list(range(max_index + 1))
|
||||
|
||||
|
||||
def _parse_vid_pid(text):
|
||||
match = re.search(r"VID_([0-9A-Fa-f]{4}).*PID_([0-9A-Fa-f]{4})", str(text or ""))
|
||||
if not match:
|
||||
match = re.search(r"\b([0-9A-Fa-f]{4}):([0-9A-Fa-f]{4})\b", str(text or ""))
|
||||
if not match:
|
||||
return ""
|
||||
return f"{match.group(1).upper()}:{match.group(2).upper()}"
|
||||
|
||||
|
||||
def _is_sensor_device(friendly_name="", manufacturer="", hardware_ids="", bus_reported=""):
|
||||
blob = " ".join(
|
||||
str(part or "") for part in (friendly_name, manufacturer, hardware_ids, bus_reported)
|
||||
).lower()
|
||||
name_ok = SENSOR_NAME_NEEDLE in str(friendly_name or bus_reported or "").lower()
|
||||
vendor_ok = any(needle in blob for needle in SENSOR_VENDOR_NEEDLES)
|
||||
return name_ok and vendor_ok
|
||||
|
||||
|
||||
def list_windows_sensor_devices():
|
||||
"""Return Status=OK Windows Camera devices matching the tactile sensors."""
|
||||
if sys.platform != "win32":
|
||||
return []
|
||||
# Keep this query narrow and property-light: startup calls it on the UI thread
|
||||
# via a short timer, so multi-second usbipd / full-property scans freeze the GUI.
|
||||
script = r"""
|
||||
$ErrorActionPreference = 'SilentlyContinue'
|
||||
$items = @()
|
||||
Get-PnpDevice -Class Camera -Status OK |
|
||||
Where-Object {
|
||||
$_.FriendlyName -eq 'USB Camera' -and
|
||||
$_.InstanceId -like '*VID_1BCF&PID_2281*'
|
||||
} |
|
||||
ForEach-Object {
|
||||
$id = $_.InstanceId
|
||||
$loc = (Get-PnpDeviceProperty -InstanceId $id -KeyName 'DEVPKEY_Device_LocationInfo').Data
|
||||
$addr = (Get-PnpDeviceProperty -InstanceId $id -KeyName 'DEVPKEY_Device_Address').Data
|
||||
$items += [pscustomobject]@{
|
||||
FriendlyName = [string]$_.FriendlyName
|
||||
Location = [string]$loc
|
||||
Address = [string]$addr
|
||||
InstanceId = [string]$id
|
||||
Status = [string]$_.Status
|
||||
}
|
||||
}
|
||||
$items | ConvertTo-Json -Depth 4 -Compress
|
||||
"""
|
||||
try:
|
||||
raw = subprocess.check_output(
|
||||
["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", script],
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
timeout=20,
|
||||
)
|
||||
devices = json.loads(raw) if raw.strip() else []
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
if isinstance(devices, dict):
|
||||
devices = [devices]
|
||||
|
||||
matched = []
|
||||
for item in devices:
|
||||
friendly = str(item.get("FriendlyName") or SENSOR_FRIENDLY_NAME)
|
||||
instance_id = str(item.get("InstanceId") or "")
|
||||
if SENSOR_NAME_NEEDLE not in friendly.lower():
|
||||
continue
|
||||
if "VID_1BCF" not in instance_id.upper() or "PID_2281" not in instance_id.upper():
|
||||
continue
|
||||
address = item.get("Address")
|
||||
try:
|
||||
address_value = int(str(address))
|
||||
except (TypeError, ValueError):
|
||||
address_value = 10**9
|
||||
matched.append(
|
||||
{
|
||||
"friendly_name": friendly or SENSOR_FRIENDLY_NAME,
|
||||
"manufacturer": SENSOR_VENDOR_NAME,
|
||||
"vid_pid": "1BCF:2281",
|
||||
"location": str(item.get("Location") or ""),
|
||||
"instance_id": instance_id,
|
||||
"address": address_value,
|
||||
}
|
||||
)
|
||||
matched.sort(key=lambda row: (row["address"], row["location"], row["instance_id"]))
|
||||
return matched
|
||||
|
||||
|
||||
def list_camera_ids(max_index=DEFAULT_MAX_CAMERA_INDEX):
|
||||
"""List selectable camera IDs without opening devices or reading frames."""
|
||||
backend_name, backend_api = _backend_candidates()[0]
|
||||
sensors = list_windows_sensor_devices()
|
||||
if sensors:
|
||||
return [
|
||||
CameraInfo(
|
||||
index=index,
|
||||
backend_name=backend_name,
|
||||
backend_api=backend_api,
|
||||
width=0,
|
||||
height=0,
|
||||
friendly_name=str(sensor.get("friendly_name") or SENSOR_FRIENDLY_NAME),
|
||||
manufacturer=str(sensor.get("manufacturer") or SENSOR_VENDOR_NAME),
|
||||
vid_pid=str(sensor.get("vid_pid") or ""),
|
||||
location=str(sensor.get("location") or ""),
|
||||
instance_id=str(sensor.get("instance_id") or ""),
|
||||
)
|
||||
for index, sensor in enumerate(sensors)
|
||||
if index <= max_index
|
||||
]
|
||||
return [
|
||||
CameraInfo(index, backend_name, backend_api, 0, 0)
|
||||
for index in _candidate_indices(max_index)
|
||||
]
|
||||
|
||||
|
||||
def _try_open(index, backend_name, backend_api):
|
||||
def _try_open(index, backend_name, backend_api, meta=None):
|
||||
capture = cv2.VideoCapture(index, backend_api)
|
||||
try:
|
||||
if not capture.isOpened():
|
||||
@@ -136,7 +263,17 @@ def _try_open(index, backend_name, backend_api):
|
||||
height = height or int(frame_height)
|
||||
elif width <= 0 or height <= 0:
|
||||
return None
|
||||
return CameraInfo(index, backend_name, backend_api, width, height)
|
||||
info = CameraInfo(index, backend_name, backend_api, width, height)
|
||||
if meta:
|
||||
info = replace(
|
||||
info,
|
||||
friendly_name=str(meta.get("friendly_name") or ""),
|
||||
manufacturer=str(meta.get("manufacturer") or ""),
|
||||
vid_pid=str(meta.get("vid_pid") or ""),
|
||||
location=str(meta.get("location") or ""),
|
||||
instance_id=str(meta.get("instance_id") or ""),
|
||||
)
|
||||
return info
|
||||
finally:
|
||||
capture.release()
|
||||
|
||||
@@ -170,3 +307,65 @@ def list_cameras(
|
||||
seen_indices.add(index)
|
||||
found.sort(key=lambda camera: camera.index)
|
||||
return found
|
||||
|
||||
|
||||
def list_sensor_cameras(
|
||||
max_index=DEFAULT_MAX_CAMERA_INDEX,
|
||||
skip_indices=None,
|
||||
fallback_backends=True,
|
||||
):
|
||||
"""Return OpenCV IDs for SunplusIT ``USB Camera`` tactile sensors only."""
|
||||
sensors = list_windows_sensor_devices()
|
||||
if sys.platform == "win32" and not sensors:
|
||||
return []
|
||||
|
||||
backends = _backend_candidates()
|
||||
if not fallback_backends:
|
||||
backends = backends[:1]
|
||||
|
||||
skip = set()
|
||||
for source in skip_indices or ():
|
||||
index = camera_source_index(source)
|
||||
if index is not None:
|
||||
skip.add(index)
|
||||
|
||||
found = []
|
||||
with _suppress_opencv_noise():
|
||||
for backend_name, backend_api in backends:
|
||||
# Prefer default IDs 0/1 first, then any remaining openable indices.
|
||||
preferred = list(range(min(2, max_index + 1))) + [
|
||||
index for index in _candidate_indices(max_index) if index >= 2
|
||||
]
|
||||
for index in preferred:
|
||||
if index in skip or any(camera.index == index for camera in found):
|
||||
continue
|
||||
meta = sensors[len(found)] if len(found) < len(sensors) else {
|
||||
"friendly_name": SENSOR_FRIENDLY_NAME,
|
||||
"manufacturer": SENSOR_VENDOR_NAME,
|
||||
"vid_pid": "1BCF:2281",
|
||||
}
|
||||
info = _try_open(index, backend_name, backend_api, meta=meta)
|
||||
if info is None:
|
||||
continue
|
||||
found.append(info)
|
||||
if sensors and len(found) >= len(sensors):
|
||||
break
|
||||
if sensors and len(found) >= len(sensors):
|
||||
break
|
||||
if found:
|
||||
break
|
||||
found.sort(key=lambda camera: camera.index)
|
||||
return found
|
||||
|
||||
|
||||
def default_sensor_camera_sources(cameras=None):
|
||||
"""Prefer OpenCV IDs 0/1 when both sensor cameras are available."""
|
||||
cameras = list(cameras if cameras is not None else list_sensor_cameras())
|
||||
sources = [camera.source for camera in cameras]
|
||||
if "0" in sources and "1" in sources:
|
||||
return "0", "1"
|
||||
if len(sources) >= 2:
|
||||
return sources[0], sources[1]
|
||||
if len(sources) == 1:
|
||||
return sources[0], ""
|
||||
return "", ""
|
||||
|
||||
@@ -155,6 +155,29 @@ def _make_visuals(orisys_module, img, flow, visual_size):
|
||||
return arrows, magnitude, float(flow_mag.mean()), float(flow_mag.max())
|
||||
|
||||
|
||||
def _drain_commands(command_queue, sensor):
|
||||
"""Apply pending worker commands. Returns True if a reference reset ran."""
|
||||
if command_queue is None:
|
||||
return False
|
||||
reset = False
|
||||
while True:
|
||||
try:
|
||||
command = command_queue.get_nowait()
|
||||
except queue.Empty:
|
||||
break
|
||||
except Exception:
|
||||
break
|
||||
if command == "reset":
|
||||
try:
|
||||
if sensor.get_img() is not None:
|
||||
sensor.reset()
|
||||
reset = True
|
||||
except Exception as exc:
|
||||
print(f"sensor reference reset failed: {exc}")
|
||||
traceback.print_exc()
|
||||
return reset
|
||||
|
||||
|
||||
def _sensor_force_worker(
|
||||
sensor_id,
|
||||
vid_src,
|
||||
@@ -169,6 +192,7 @@ def _sensor_force_worker(
|
||||
motion_threshold,
|
||||
include_visuals,
|
||||
visual_size,
|
||||
command_queue=None,
|
||||
):
|
||||
ensure_project_paths()
|
||||
import orisys
|
||||
@@ -190,6 +214,7 @@ def _sensor_force_worker(
|
||||
|
||||
while not stop_event.is_set():
|
||||
t_start = time.perf_counter()
|
||||
_drain_commands(command_queue, sensor)
|
||||
img = sensor.get_img()
|
||||
if img is None:
|
||||
time.sleep(0.01)
|
||||
@@ -309,6 +334,7 @@ class TwoSensorForceReader:
|
||||
self.visual_size = visual_size
|
||||
|
||||
self.result_queues = []
|
||||
self.command_queues = []
|
||||
self.stop_events = []
|
||||
self.processes = []
|
||||
self.latest = [None, None]
|
||||
@@ -318,6 +344,7 @@ class TwoSensorForceReader:
|
||||
return
|
||||
|
||||
self.result_queues = [Queue(maxsize=self.queue_size) for _ in self.sensor_configs]
|
||||
self.command_queues = [Queue(maxsize=8) for _ in self.sensor_configs]
|
||||
self.stop_events = [mp.Event() for _ in self.sensor_configs]
|
||||
self.processes = [None] * len(self.sensor_configs)
|
||||
self.latest = [None] * len(self.sensor_configs)
|
||||
@@ -343,6 +370,7 @@ class TwoSensorForceReader:
|
||||
self.motion_threshold,
|
||||
self.include_visuals,
|
||||
self.visual_size,
|
||||
self.command_queues[idx],
|
||||
),
|
||||
)
|
||||
process.start()
|
||||
@@ -364,11 +392,36 @@ class TwoSensorForceReader:
|
||||
old_process.join(timeout=1.0)
|
||||
|
||||
self.result_queues[idx] = Queue(maxsize=self.queue_size)
|
||||
self.command_queues[idx] = Queue(maxsize=8)
|
||||
self.stop_events[idx] = mp.Event()
|
||||
self.latest[idx] = None
|
||||
self._start_sensor_process(idx)
|
||||
return True
|
||||
|
||||
def reset_reference_maps(self):
|
||||
"""Ask both sensor workers to replace the optical-flow reference frame."""
|
||||
if not self.processes or not self.command_queues:
|
||||
return False
|
||||
sent = 0
|
||||
for idx, command_queue in enumerate(self.command_queues):
|
||||
process = self.processes[idx] if idx < len(self.processes) else None
|
||||
if process is None or not process.is_alive():
|
||||
continue
|
||||
try:
|
||||
command_queue.put_nowait("reset")
|
||||
sent += 1
|
||||
except queue.Full:
|
||||
try:
|
||||
command_queue.get_nowait()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
command_queue.put_nowait("reset")
|
||||
sent += 1
|
||||
except queue.Full:
|
||||
pass
|
||||
return sent > 0
|
||||
|
||||
def update(self):
|
||||
for idx, result_queue in enumerate(self.result_queues):
|
||||
data = None
|
||||
|
||||
+1163
-381
File diff suppressed because it is too large
Load Diff
Binary file not shown.
@@ -1 +0,0 @@
|
||||
5675c6f6400b3d9b3d27602c4639c307cb35d0357e658ec5c344aebdc7eba7c0 GripperDemo02-ubuntu22.04-x86_64-lite-20260922.tar.gz
|
||||
Binary file not shown.
@@ -0,0 +1 @@
|
||||
afdec27bff2e0e4d5dc43f1678272eae8456336fd8cf4f0c6e0aee260abd8fa7 GripperDemo02-ubuntu24.04-aarch64-lite-20261008.tar.gz
|
||||
Binary file not shown.
@@ -1 +0,0 @@
|
||||
8307c8fe39a850275c09b7a4092584216b8ee1bf7891ec0da12737604e7905d0 GripperDemo02-ubuntu26.04-x86_64-lite-20260922.tar.gz
|
||||
Binary file not shown.
@@ -1 +0,0 @@
|
||||
5f66fb92f4f3f17d129106fd436aeb56d5019f3d36f9dd81ae8a99ce847d6a69 GripperDemo02-windows-x64-lite-20260922.zip
|
||||
Reference in New Issue
Block a user