Compare commits

..

9 Commits

Author SHA1 Message Date
cheng eaf54420ad Tune grip guard/frames per stability level, unify open speed 80%, rebuild aarch64 package
- Guard (release_min_hold_seconds): sensitive 0.4 s / stable 0.7 s / firm 1.0 s, label clarified
- Y-displacement samples 3-4-6, release_confirm_samples aligned 3-4-6 so the bar shows consistent frames
- release_open_speed 80% for all modes; level summary shows frames and open speed
- Rebuild ubuntu24.04 aarch64 lite 20261008 with stable-level-50 config; drop old release packages

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-08 17:21:25 +08:00
cheng 2a3fcf868f 新增 aarch64 lite 发布包 20261008(Radxa Dragon Q6A / Ubuntu 24.04):包含稳定度配置条、独立速度、状态文字简写与平滑、灵敏挡振荡判据收紧、重置底图按钮。
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-08 15:25:46 +08:00
cheng 33578fa83e 稳定度配置条与自定义挡位、独立速度调节;状态文字中文简写与平滑;灵敏挡切向振荡判据收紧;新增重置底图按钮。
- 设置框新增 0~100 稳定度配置条(灵敏 0 / 较稳定 50 / 十分稳定 100 为内置挡位,中间按比例插值),支持滚轮/左键拖动,可另存/删除自定义挡位;速度改为独立滑块。
- 状态栏改为中文简写(悬停显示原文);夹持保护期文字按挡位区分;合并同义子状态并加最短停留,避免逐帧跳动。
- 灵敏挡切向振荡:0.5 s 内换向 4 次,并须持续至少 0.15 s 才张开(新参数 release_shear_oscillation_min_seconds)。
- 传感器控制区新增「重置底图」:左右传感器一起用当前画面更换光流参考底图(orisys Sensor.reset),不断开相机。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-08 13:34:30 +08:00
cheng 3507a2d313 纳入 config/gui_layout.json(界面布局:全屏、浮动面板位置、3D 视角、画面旋转 270°)。
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-29 11:09:37 +08:00
cheng 0662033f0d 新增 Radxa Dragon Q6A(aarch64)运行说明文档。
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-28 17:48:31 +08:00
cheng 543d8a5ccc 新增 Radxa Dragon Q6A(Ubuntu 24.04 aarch64)lite 发布包,aarch64 tar.gz 走 Git LFS。
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-28 17:45:01 +08:00
cheng d948883e93 灵敏挡与稳定挡代码分离:灵敏挡使用 6393be9 原版 release_watch(release_watch_sensitive.py),较稳定/十分稳定使用 release_watch.py,按挡位分发;速度按挡位设置(灵敏 55%,较稳定/十分稳定 30%);切换挡位后界面速度/最大力滑块同步。
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-28 16:47:21 +08:00
cheng 8be8a80cff 较稳定/十分稳定挡改为恒力夹持:接触后不锁位置持续以闭合力顶住;张开改为 Y 切向带符号拟合偏移(保护期后记基准,连续 6 帧超阈值),夹爪合到底判定物体脱出即张开。较稳定 70%/2s/0.6N,十分稳定 100%/2.5s/0.8N。
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-28 16:30:54 +08:00
cheng 6393be9518 开合控制调整:切向触发只看 Y 轴,三挡位只控制释放、触发固定灵敏参数;自适应负载改看切向 X;恢复原灵敏档张开逻辑与排查前夹持力参数。
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-28 15:36:05 +08:00
26 changed files with 3126 additions and 602 deletions
+1
View File
@@ -1 +1,2 @@
release/*.zip filter=lfs diff=lfs merge=lfs -text
release/*-aarch64-*.tar.gz filter=lfs diff=lfs merge=lfs -text
-1
View File
@@ -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
View File
@@ -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
}
+70
View File
@@ -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
}
}
}
+88
View File
@@ -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`,排查问题时提供该文件。
+38 -15
View File
@@ -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
View File
@@ -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):
+58 -2
View File
@@ -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")
+21 -7
View File
@@ -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:
+72 -11
View File
@@ -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,
+38 -8
View File
@@ -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")
+27
View File
@@ -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)
+227 -8
View File
@@ -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.
+680
View File
@@ -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)
+206 -7
View File
@@ -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 "", ""
+53
View File
@@ -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
View File
File diff suppressed because it is too large Load Diff
@@ -1 +0,0 @@
5675c6f6400b3d9b3d27602c4639c307cb35d0357e658ec5c344aebdc7eba7c0 GripperDemo02-ubuntu22.04-x86_64-lite-20260922.tar.gz
@@ -0,0 +1 @@
afdec27bff2e0e4d5dc43f1678272eae8456336fd8cf4f0c6e0aee260abd8fa7 GripperDemo02-ubuntu24.04-aarch64-lite-20261008.tar.gz
@@ -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