Compare commits

...

10 Commits

Author SHA1 Message Date
cheng c609c3901f first commit 2026-06-29 17:14:42 +08:00
cheng 9ef1c8782a v 2.1 2026-06-09 16:25:40 +08:00
cheng c700f95fb0 v 2.0 2026-06-09 14:59:43 +08:00
cheng 9a3a397e2c 风格调节 2026-06-09 14:35:43 +08:00
cheng 1f1ee6d8fe v 1.2 2026-06-09 14:29:44 +08:00
cheng 74d2bc4853 v 1.1 2026-06-09 13:45:08 +08:00
cheng 886e5df5ee v 1.0 2026-06-09 11:23:14 +08:00
cheng a2c820bce5 qt折线图修改 2026-06-09 10:39:17 +08:00
cheng 66a00a8b5b qt第一版可视化 2026-06-09 09:31:44 +08:00
cheng cbd8683429 01 2026-06-08 17:56:09 +08:00
48 changed files with 1811 additions and 161 deletions
+4
View File
@@ -0,0 +1,4 @@
{
"python-envs.defaultEnvManager": "ms-python.python:conda",
"python-envs.defaultPackageManager": "ms-python.python:conda"
}
BIN
View File
Binary file not shown.

After

Width:  |  Height:  |  Size: 17 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 2.3 MiB

After

Width:  |  Height:  |  Size: 2.3 MiB

Binary file not shown.
+17
View File
@@ -0,0 +1,17 @@
"""Compatibility entry point for gripper demo 02 PyQt viewer."""
import multiprocessing as mp
import sys
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
from gripper_control_02.visualizer import main
if __name__ == "__main__":
mp.freeze_support()
raise SystemExit(main(sys.argv[1:]))
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+11 -1
View File
@@ -47,6 +47,8 @@ class ForceConverter:
enabled: bool = True enabled: bool = True
extrapolate: bool = True extrapolate: bool = True
clamp_output_min: float | None = 0.0 clamp_output_min: float | None = 0.0
input_mode: str = "magnitude"
signed: bool = False
@classmethod @classmethod
def from_config(cls, calibration, config_name): def from_config(cls, calibration, config_name):
@@ -71,6 +73,8 @@ class ForceConverter:
enabled=True, enabled=True,
extrapolate=bool(calibration.get("extrapolate", True)), extrapolate=bool(calibration.get("extrapolate", True)),
clamp_output_min=calibration.get("clamp_output_min", 0.0), clamp_output_min=calibration.get("clamp_output_min", 0.0),
input_mode=str(calibration.get("input", "magnitude")).lower(),
signed=bool(calibration.get("signed", False)),
) )
def convert(self, raw): def convert(self, raw):
@@ -104,8 +108,14 @@ class ForceConverter:
value = n0 + ratio * (n1 - n0) value = n0 + ratio * (n1 - n0)
return self._clamp(value) return self._clamp(value)
def convert_component(self, raw):
if not self.signed:
return self.convert(raw)
raw = float(raw)
sign = -1.0 if raw < 0.0 else 1.0
return sign * self.convert(abs(raw))
def _clamp(self, value): def _clamp(self, value):
if self.clamp_output_min is not None: if self.clamp_output_min is not None:
value = max(float(self.clamp_output_min), value) value = max(float(self.clamp_output_min), value)
return value return value
+58 -3
View File
@@ -8,12 +8,20 @@ def shear_magnitude(sample):
return (float(sample["fshearx"]) ** 2 + float(sample["fsheary"]) ** 2) ** 0.5 return (float(sample["fshearx"]) ** 2 + float(sample["fsheary"]) ** 2) ** 0.5
def _component_shear_enabled(converter):
return getattr(converter, "input_mode", "magnitude") == "components"
@dataclass @dataclass
class ForceBaseline: class ForceBaseline:
left_normal: float = 0.0 left_normal: float = 0.0
right_normal: float = 0.0 right_normal: float = 0.0
left_shear: float = 0.0 left_shear: float = 0.0
right_shear: float = 0.0 right_shear: float = 0.0
left_shear_x: float = 0.0
left_shear_y: float = 0.0
right_shear_x: float = 0.0
right_shear_y: float = 0.0
@dataclass @dataclass
@@ -27,6 +35,10 @@ class ForceFeedback:
right_shear: float right_shear: float
max_shear: float max_shear: float
shear_diff: float shear_diff: float
left_shear_x: float = 0.0
left_shear_y: float = 0.0
right_shear_x: float = 0.0
right_shear_y: float = 0.0
class TactileFeedbackProcessor: class TactileFeedbackProcessor:
@@ -42,6 +54,17 @@ class TactileFeedbackProcessor:
self.right_normal_filter = ExponentialFilter(self.config.filter_alpha) self.right_normal_filter = ExponentialFilter(self.config.filter_alpha)
self.left_shear_filter = ExponentialFilter(self.config.shear_filter_alpha) self.left_shear_filter = ExponentialFilter(self.config.shear_filter_alpha)
self.right_shear_filter = ExponentialFilter(self.config.shear_filter_alpha) self.right_shear_filter = ExponentialFilter(self.config.shear_filter_alpha)
self.left_shear_x_filter = ExponentialFilter(self.config.shear_filter_alpha)
self.left_shear_y_filter = ExponentialFilter(self.config.shear_filter_alpha)
self.right_shear_x_filter = ExponentialFilter(self.config.shear_filter_alpha)
self.right_shear_y_filter = ExponentialFilter(self.config.shear_filter_alpha)
def _convert_shear(self, sample):
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
def calibrate_baseline(self, reader): def calibrate_baseline(self, reader):
seconds = self.config.baseline_seconds seconds = self.config.baseline_seconds
@@ -53,15 +76,25 @@ class TactileFeedbackProcessor:
right_values = [] right_values = []
left_shear_values = [] left_shear_values = []
right_shear_values = [] right_shear_values = []
left_shear_x_values = []
left_shear_y_values = []
right_shear_x_values = []
right_shear_y_values = []
deadline = time.perf_counter() + seconds deadline = time.perf_counter() + seconds
while time.perf_counter() < deadline: while time.perf_counter() < deadline:
left, right = reader.get_samples(timeout=0.2) left, right = reader.get_samples(timeout=0.2)
if left is not None: if left is not None:
left_values.append(self.normal_converter.convert(left["fnormal"])) left_values.append(self.normal_converter.convert(left["fnormal"]))
left_shear_values.append(self.shear_converter.convert(shear_magnitude(left))) shear, shear_x, shear_y = self._convert_shear(left)
left_shear_values.append(shear)
left_shear_x_values.append(shear_x)
left_shear_y_values.append(shear_y)
if right is not None: if right is not None:
right_values.append(self.normal_converter.convert(right["fnormal"])) right_values.append(self.normal_converter.convert(right["fnormal"]))
right_shear_values.append(self.shear_converter.convert(shear_magnitude(right))) shear, shear_x, shear_y = self._convert_shear(right)
right_shear_values.append(shear)
right_shear_x_values.append(shear_x)
right_shear_y_values.append(shear_y)
time.sleep(0.01) time.sleep(0.01)
self.baseline = ForceBaseline( self.baseline = ForceBaseline(
@@ -69,6 +102,10 @@ class TactileFeedbackProcessor:
right_normal=sum(right_values) / len(right_values) if right_values else 0.0, right_normal=sum(right_values) / len(right_values) if right_values else 0.0,
left_shear=sum(left_shear_values) / len(left_shear_values) if left_shear_values else 0.0, left_shear=sum(left_shear_values) / len(left_shear_values) if left_shear_values else 0.0,
right_shear=sum(right_shear_values) / len(right_shear_values) if right_shear_values else 0.0, right_shear=sum(right_shear_values) / len(right_shear_values) if right_shear_values else 0.0,
left_shear_x=sum(left_shear_x_values) / len(left_shear_x_values) if left_shear_x_values else 0.0,
left_shear_y=sum(left_shear_y_values) / len(left_shear_y_values) if left_shear_y_values else 0.0,
right_shear_x=sum(right_shear_x_values) / len(right_shear_x_values) if right_shear_x_values else 0.0,
right_shear_y=sum(right_shear_y_values) / len(right_shear_y_values) if right_shear_y_values else 0.0,
) )
return self.baseline return self.baseline
@@ -85,6 +122,17 @@ class TactileFeedbackProcessor:
0.0, 0.0,
self.normal_converter.convert(right["fnormal"]) - self.baseline.right_normal, self.normal_converter.convert(right["fnormal"]) - self.baseline.right_normal,
) )
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)
left_shear_x = raw_left_shear_x - self.baseline.left_shear_x
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
else:
left_shear_x = left_shear_y = right_shear_x = right_shear_y = 0.0
left_shear = max( left_shear = max(
0.0, 0.0,
self.shear_converter.convert(shear_magnitude(left)) - self.baseline.left_shear, self.shear_converter.convert(shear_magnitude(left)) - self.baseline.left_shear,
@@ -98,6 +146,10 @@ class TactileFeedbackProcessor:
right_normal = self.right_normal_filter.update(right_force) right_normal = self.right_normal_filter.update(right_force)
left_shear_filtered = self.left_shear_filter.update(left_shear) left_shear_filtered = self.left_shear_filter.update(left_shear)
right_shear_filtered = self.right_shear_filter.update(right_shear) right_shear_filtered = self.right_shear_filter.update(right_shear)
left_shear_x_filtered = self.left_shear_x_filter.update(left_shear_x)
left_shear_y_filtered = self.left_shear_y_filter.update(left_shear_y)
right_shear_x_filtered = self.right_shear_x_filter.update(right_shear_x)
right_shear_y_filtered = self.right_shear_y_filter.update(right_shear_y)
return ForceFeedback( return ForceFeedback(
left_normal=left_normal, left_normal=left_normal,
@@ -109,5 +161,8 @@ class TactileFeedbackProcessor:
right_shear=right_shear_filtered, right_shear=right_shear_filtered,
max_shear=max(left_shear_filtered, right_shear_filtered), max_shear=max(left_shear_filtered, right_shear_filtered),
shear_diff=left_shear_filtered - right_shear_filtered, shear_diff=left_shear_filtered - right_shear_filtered,
left_shear_x=left_shear_x_filtered,
left_shear_y=left_shear_y_filtered,
right_shear_x=right_shear_x_filtered,
right_shear_y=right_shear_y_filtered,
) )
+103 -5
View File
@@ -7,6 +7,9 @@ import queue
import time import time
from multiprocessing import Process, Queue from multiprocessing import Process, Queue
import cv2
import numpy as np
from .paths import ensure_project_paths from .paths import ensure_project_paths
ensure_project_paths() ensure_project_paths()
@@ -38,6 +41,69 @@ def _put_latest(result_queue, data):
pass pass
def _draw_magnitude_map(flow, visual_size, threshold=1.5):
magnitude = np.sqrt(flow[:, :, 0] ** 2 + flow[:, :, 1] ** 2)
h, w = magnitude.shape
if h <= 0 or w <= 0:
return None
downsample_factor = 10
magnitude_downsampled = magnitude[::downsample_factor, ::downsample_factor]
magnitude_smoothed = cv2.resize(
magnitude_downsampled,
(w, h),
interpolation=cv2.INTER_CUBIC,
)
magnitude_thresholded = magnitude_smoothed.copy()
magnitude_thresholded[magnitude_smoothed < threshold] = 0
mag_min = float(magnitude_thresholded.min())
mag_max = float(magnitude_thresholded.max())
if mag_max > mag_min:
magnitude_norm = (
(magnitude_thresholded - mag_min) / (mag_max - mag_min) * 255
).astype(np.uint8)
else:
magnitude_norm = np.zeros_like(magnitude_thresholded, dtype=np.uint8)
magnitude_colored = cv2.applyColorMap(magnitude_norm, cv2.COLORMAP_JET)
return cv2.resize(
magnitude_colored,
(visual_size, visual_size),
interpolation=cv2.INTER_CUBIC,
)
def _make_visuals(orisys_module, img, flow, visual_size):
if img is None or flow is None or np.asarray(flow).size == 0:
return None, None, 0.0, 0.0
if len(img.shape) == 2:
img_bgr = cv2.cvtColor(img, cv2.COLOR_GRAY2BGR)
else:
img_bgr = img
h, w = img_bgr.shape[:2]
bg = np.empty((h, w, 3), dtype=img_bgr.dtype)
bg[:, :, 0] = 22
bg[:, :, 1] = 16
bg[:, :, 2] = 11
arrows = orisys_module.util.draw_arrows(
bg,
flow,
threshold=2,
grid_spacing=20,
arrow_scale=0.5,
below_threshold_color=(160, 160, 160),
)
arrows = cv2.resize(arrows, (visual_size, visual_size), interpolation=cv2.INTER_CUBIC)
magnitude = _draw_magnitude_map(flow, visual_size)
flow_mag = np.sqrt(flow[:, :, 0] ** 2 + flow[:, :, 1] ** 2)
return arrows, magnitude, float(flow_mag.mean()), float(flow_mag.max())
def _sensor_force_worker( def _sensor_force_worker(
sensor_id, sensor_id,
vid_src, vid_src,
@@ -50,6 +116,8 @@ def _sensor_force_worker(
isstitch, isstitch,
backend, backend,
motion_threshold, motion_threshold,
include_visuals,
visual_size,
): ):
ensure_project_paths() ensure_project_paths()
import orisys import orisys
@@ -79,16 +147,34 @@ def _sensor_force_worker(
check_motion=True, check_motion=True,
threshold=motion_threshold, threshold=motion_threshold,
) )
if include_visuals:
fps, fnormal, fshearx, fsheary, flow, img_view = sensor.read_info(
sensor.info.FPS,
sensor.info.FNORMAL,
sensor.info.FSHEARX,
sensor.info.FSHEARY,
sensor.info.VRAW,
sensor.info.IMG,
)
flow_view, magnitude_view, flow_mean, flow_max = _make_visuals(
orisys,
img_view,
flow,
int(visual_size),
)
else:
fps, fnormal, fshearx, fsheary = sensor.read_info( fps, fnormal, fshearx, fsheary = sensor.read_info(
sensor.info.FPS, sensor.info.FPS,
sensor.info.FNORMAL, sensor.info.FNORMAL,
sensor.info.FSHEARX, sensor.info.FSHEARX,
sensor.info.FSHEARY, sensor.info.FSHEARY,
) )
flow_view = None
magnitude_view = None
flow_mean = 0.0
flow_max = 0.0
_put_latest( data = {
result_queue,
{
"sensor_id": sensor_id, "sensor_id": sensor_id,
"fnormal": float(fnormal), "fnormal": float(fnormal),
"fshearx": float(fshearx), "fshearx": float(fshearx),
@@ -96,8 +182,14 @@ def _sensor_force_worker(
"fps": float(fps), "fps": float(fps),
"is_contact": bool(is_contact), "is_contact": bool(is_contact),
"timestamp": time.time(), "timestamp": time.time(),
}, "flow_mean": flow_mean,
) "flow_max": flow_max,
}
if include_visuals:
data["flow_view"] = flow_view
data["magnitude_view"] = magnitude_view
_put_latest(result_queue, data)
if frame_interval > 0: if frame_interval > 0:
sleep_time = frame_interval - (time.perf_counter() - t_start) sleep_time = frame_interval - (time.perf_counter() - t_start)
@@ -136,6 +228,8 @@ class TwoSensorForceReader:
backend="auto", backend="auto",
motion_threshold=0, motion_threshold=0,
queue_size=5, queue_size=5,
include_visuals=False,
visual_size=320,
): ):
self.sensor_configs = [ self.sensor_configs = [
{ {
@@ -157,6 +251,8 @@ class TwoSensorForceReader:
self.backend = backend self.backend = backend
self.motion_threshold = motion_threshold self.motion_threshold = motion_threshold
self.queue_size = queue_size self.queue_size = queue_size
self.include_visuals = include_visuals
self.visual_size = visual_size
self.result_queues = [] self.result_queues = []
self.stop_events = [] self.stop_events = []
@@ -186,6 +282,8 @@ class TwoSensorForceReader:
self.isstitch, self.isstitch,
self.backend, self.backend,
self.motion_threshold, self.motion_threshold,
self.include_visuals,
self.visual_size,
), ),
) )
process.start() process.start()
+26 -9
View File
@@ -22,27 +22,43 @@ class GripperClient:
) )
return self return self
def move(self, position, force_pct, label): def move(self, position, force_pct, label, speed_pct=None, accel=None, decel=None):
config = self.config config = self.config
speed_pct = config.speed if speed_pct is None else speed_pct
accel = config.accel if accel is None else accel
decel = config.decel if decel is None else decel
if config.dry_run: if config.dry_run:
print( print(
f"[dry-run] {label}: temp_move " f"[dry-run] {label}: temp_move "
f"position={position}, speed={config.speed}, force={force_pct}, " f"position={position}, speed={speed_pct}, force={force_pct}, "
f"accel={config.accel}, decel={config.decel}" f"accel={accel}, decel={decel}"
) )
return return
if self.motor is None:
raise RuntimeError(
"gripper motor is not connected; check serial port, conda environment, "
"and changingtek_p_rtu_Servo.py"
)
self.motor.temp_move( self.motor.temp_move(
position_mm=int(position), position_mm=int(position),
speed_pct=int(config.speed), speed_pct=int(speed_pct),
force_pct=int(force_pct), force_pct=int(force_pct),
accel=int(config.accel), accel=int(accel),
decel=int(config.decel), decel=int(decel),
trigger=True, trigger=True,
) )
def open(self, force_pct, label="open"): def open(self, force_pct, label="open", speed_pct=None, accel=None, decel=None):
self.move(self.config.open_pos, force_pct, label) self.move(
self.config.open_pos,
force_pct,
label,
speed_pct=speed_pct,
accel=accel,
decel=decel,
)
def close(self, force_pct, label="close"): def close(self, force_pct, label="close"):
self.move(self.config.close_pos, force_pct, label) self.move(self.config.close_pos, force_pct, label)
@@ -84,6 +100,8 @@ class GripperClient:
def final_open_if_needed(self): def final_open_if_needed(self):
if not self.config.open_at_end: if not self.config.open_at_end:
return return
if not self.config.dry_run and self.motor is None:
return
final_force = int(clamp( final_force = int(clamp(
self.config.initial_force, self.config.initial_force,
self.config.force_min, self.config.force_min,
@@ -99,4 +117,3 @@ class GripperClient:
self.motor.trigger_motion() self.motor.trigger_motion()
except Exception as exc: except Exception as exc:
print(f"stop by speed=0 failed: {exc}") print(f"stop by speed=0 failed: {exc}")
+72 -18
View File
@@ -5,15 +5,17 @@
## 展示流程 ## 展示流程
1. 程序启动后,夹爪先张开。 1. 程序启动后,夹爪先张开。
2. 程序等待任意一侧触觉传感器感受到法向力或切向力;进入等待后会先记录张开时的安静参考值 2. 程序等待任意一侧触觉传感器感受到法向力或切向力。
3. 展示时,把物体从某个侧边扫一下,触发单侧力变化。 3. 展示时,把物体从某个侧边扫一下,触发单侧力变化。
4. 触发后夹爪开始闭合,你把物体移动到中间。 4. 触发后夹爪开始闭合,你把物体移动到中间。
5. 闭合过程中检测到物体后,切到低力并保持当前位置。 5. 闭合过程中检测到物体后,切到低力并保持当前位置。
6. 像 01 一样进入 `gripping`,慢慢加力到 `HOLD_FORCE` 6. 像 01 一样进入 `gripping`慢慢加力到基础 `HOLD_FORCE`
7. 进入 `hold_check`记录夹持参考力。 7. 进入 `hold_check`,根据切向力估计物体负载;切向力越大,目标夹紧力越高,并按小步进继续补力。
8. 后续法向力或切向力变化超过阈值,夹爪直接张开到 `OPEN_POS` 8. 自适应补力停止并稳定 `RELEASE_ARM_DELAY_SECONDS` 后,重新记录释放参考力,才开始检测人取物松开
9. 张开后进入 `open_recover`,保持张开并等待传感器力清空 9. 后续法向力或切向力变化超过阈值,夹爪直接张开到 `OPEN_POS`
10. 法向和切向读数都低于恢复阈值并稳定后,才回到 `open_wait` 等待下一次单侧扫过触发 10. 张开后进入 `open_recover`,保持张开并等待恢复条件
11. 夹爪回到 `OPEN_POS` 初始点位,或法向/切向读数都低于恢复阈值并稳定后,继续休息至少 `REARM_SECONDS`
12. 休息结束后才回到 `open_wait` 并重新武装;休息期间不管有没有物体扫过,都不会闭合。
## 运行 ## 运行
@@ -30,6 +32,24 @@ python -m gripper_control_02.main
python -m gripper_control_02.main --dry-run python -m gripper_control_02.main --dry-run
``` ```
PyQt 可视化展示:
```powershell
python -m gripper_control_02.visualizer
```
窗口模式运行:
```powershell
python -m gripper_control_02.visualizer --windowed
```
兼容入口:
```powershell
python examples\gripper_demo_02_viewer.py
```
默认配置文件: 默认配置文件:
```text ```text
@@ -40,8 +60,7 @@ gripper_control_02\config\gripper_demo_02.py
- `TRIGGER_NORMAL_FORCE`: 单侧法向力触发阈值。 - `TRIGGER_NORMAL_FORCE`: 单侧法向力触发阈值。
- `TRIGGER_SHEAR_FORCE`: 单侧切向力触发阈值。 - `TRIGGER_SHEAR_FORCE`: 单侧切向力触发阈值。
- `TRIGGER_NORMAL_CHANGE`: 单侧法向力相对张开参考值的变化触发阈值 - `TRIGGER_CLEAR_STABLE_SECONDS`: 张开恢复后,触发力必须清掉并稳定多久,才允许下一次扫过触发
- `TRIGGER_SHEAR_CHANGE`: 单侧切向力相对张开参考值的变化触发阈值。
- `GRIP_NORMAL_FORCE`: 夹住时左右接触阈值。 - `GRIP_NORMAL_FORCE`: 夹住时左右接触阈值。
- `GRIP_REQUIRES_BOTH`: 是否必须左右都接触才算夹住。 - `GRIP_REQUIRES_BOTH`: 是否必须左右都接触才算夹住。
- `GRIP_START_FORCE`: 检测到物体后先切到的低力。 - `GRIP_START_FORCE`: 检测到物体后先切到的低力。
@@ -49,15 +68,27 @@ gripper_control_02\config\gripper_demo_02.py
- `FORCE_RAMP_INTERVAL`: 慢慢加力时两次加力的间隔。 - `FORCE_RAMP_INTERVAL`: 慢慢加力时两次加力的间隔。
- `HOLD_FORCE`: 慢慢加力到这个值后进入 `hold_check` - `HOLD_FORCE`: 慢慢加力到这个值后进入 `hold_check`
- `GRIP_SETTLE_SECONDS`: 夹住后等待稳定多久再记录参考力。 - `GRIP_SETTLE_SECONDS`: 夹住后等待稳定多久再记录参考力。
- `ADAPTIVE_GRIP_ENABLED`: 是否启用按切向力估计负载并自适应补夹紧力。
- `ADAPTIVE_SHEAR_START_FORCE`: 切向力低于这个值时,认为是轻物体,只使用最小目标力。
- `ADAPTIVE_SHEAR_FULL_FORCE`: 切向力达到这个值时,认为负载较大,目标力加到最大目标力。
- `ADAPTIVE_FORCE_MIN`: 自适应夹持的最小目标力。
- `ADAPTIVE_FORCE_MAX`: 自适应夹持的最大目标力;重物夹不住优先加这个,但不要超过 `FORCE_MAX`
- `ADAPTIVE_FORCE_STEP`: 自适应补力时每次增加多少 `force_pct`
- `ADAPTIVE_FORCE_INTERVAL`: 自适应补力两次之间的间隔,越大越柔和。
- `RELEASE_ARM_DELAY_SECONDS`: 自适应补力停止后,等待多久再启用“人取物松开”检测。
- `RELEASE_NORMAL_CHANGE`: 夹住后法向力变化超过它就松开。 - `RELEASE_NORMAL_CHANGE`: 夹住后法向力变化超过它就松开。
- `RELEASE_SHEAR_CHANGE`: 夹住后切向力变化超过它就松开。 - `RELEASE_SHEAR_CHANGE`: 夹住后切向力变化超过它就松开。
- `RELEASE_CONTACT_LOST_NORMAL_FORCE`: 夹住后判断接触丢失的最大法向力 - `RELEASE_OPEN_SPEED`: 检测到松开后,张开到 `OPEN_POS` 使用的速度
- `RELEASE_CONTACT_LOST_SHEAR_FORCE`: 夹住后判断接触丢失的最大切向力 - `RELEASE_OPEN_ACCEL`: release 张开的加速度
- `RELEASE_CONTACT_LOST_SECONDS`: 接触丢失持续多久后松开 - `RELEASE_OPEN_DECEL`: release 张开的减速度
- `REARM_SECONDS`: 张开后最短等待时间,防止刚松开马上又误触发 - `REARM_SECONDS`: 夹爪回到 `OPEN_POS` 或恢复条件成立后,至少休息多久才重新等待触发;休息期间扫过也不会闭合
- `RECOVER_NORMAL_FORCE`: 张开恢复时允许重新触发的最大法向残余力。 - `RECOVER_NORMAL_FORCE`: 张开恢复时允许重新触发的最大法向残余力。
- `RECOVER_SHEAR_FORCE`: 张开恢复时允许重新触发的最大切向残余力。 - `RECOVER_SHEAR_FORCE`: 张开恢复时允许重新触发的最大切向残余力。
- `RECOVER_STABLE_SECONDS`: 力清空后需要连续稳定多久,才重新等待触发。 - `RECOVER_STABLE_SECONDS`: 力清空后需要连续稳定多久,才重新等待触发。
- `OPEN_POSITION_RECOVER_ENABLED`: 是否允许夹爪回到 `OPEN_POS` 后直接恢复等待。
- `OPEN_POSITION_TOLERANCE`: 当前点位和 `OPEN_POS` 差值小于它,就认为已经回到初始点位。
- `OPEN_POSITION_STABLE_SECONDS`: 回到初始点位后需要稳定多久,才重新等待触发。
- `OPEN_POSITION_CHECK_INTERVAL`: `open_recover` 中读取夹爪当前位置的间隔。
## 日志字段 ## 日志字段
@@ -65,27 +96,50 @@ gripper_control_02\config\gripper_demo_02.py
- `maxN`: 左右法向力最大值,单位 N。 - `maxN`: 左右法向力最大值,单位 N。
- `shear`: 左右切向力最大值,单位 N。 - `shear`: 左右切向力最大值,单位 N。
- `trigger`: 是否满足单侧触发条件。 - `trigger`: 是否满足单侧触发条件。
- `armed`: 是否已经重新武装;只有 `trigger=1 armed=1` 且是新的一次触发上升沿才会闭合。
- `grip_contact`: 是否满足夹住接触条件。 - `grip_contact`: 是否满足夹住接触条件。
- `dN`: 夹住后当前法向力相对参考值的最大变化,单位 N。 - `dN`: 夹住后当前法向力相对参考值的最大变化,单位 N。
- `dShear`: 夹住后当前切向力相对参考值的变化,单位 N。 - `dShear`: 夹住后当前切向力相对参考值的变化,单位 N。
- `state`: 当前状态。 - `state`: 当前状态。
- `hold_pos`: 夹住后保持的位置。 - `hold_pos`: 夹住后保持的位置。
- `force_pct`: 当前夹爪目标力百分比。 - `force_pct`: 当前夹爪目标力百分比。
- `target`: 当前根据切向力计算出来的自适应目标力。
- `release_armed`: 是否已经启用释放检测;`0` 表示还在补力/等待稳定,`1` 表示法向或切向变化会触发张开。
- `action`: 当前动作。 - `action`: 当前动作。
## 可视化展示
- 左右两边的光流箭头图。
- 左右两边的光流幅值图。
- 左右法向力、左切向力、右切向力。
- 当前 `state``action``trigger/armed``grip_contact`
- 当前 `force_pct`、自适应目标力、速度 `speed_pct``hold_pos`
- 顶部有 `重启` 按钮,会先停止当前控制线程,释放相机与夹爪后再重新启动。
- 当前 `release_armed`,用于判断系统是在补夹紧力,还是已经开始检测人取物松开。
- 折线图带图例,默认固定 y 轴为 `0~3 N`,范围在 `visualizer.py` 顶部的 `PLOT_FORCE_Y_MIN_N / PLOT_FORCE_Y_MAX_N` 调整。
- 左右传感器 FPS、接触标志、光流均值/最大值。
## 状态说明 ## 状态说明
- `open_wait`: 夹爪张开,等待任意单侧触发。 - `open_wait`: 夹爪张开,等待任意单侧触发。
- `closing`: 单侧触发后正在闭合,等待检测到物体。 - `closing`: 单侧触发后正在闭合,等待检测到物体。
- `gripping`: 检测到物体后低力保持当前位置,并慢慢加力。 - `gripping`: 检测到物体后低力保持当前位置,并慢慢加力。
- `hold_check`: 已夹住并记录参考力,监测法向/切向变化;变化超过阈值会直接张开。 - `hold_check`: 已夹住;先根据切向力自适应补夹紧力,补力稳定后再监测法向/切向变化;变化超过阈值会直接张开。
- `open_recover`: 已经张开到初始位置,正在等待法向/切向残余力清空;这个状态不会闭合夹爪。 - `open_recover`: 已经发出张开命令,正在等待回到初始点位或等待法向/切向残余力清空;这个状态不会闭合夹爪。
## 常见动作说明 ## 常见动作说明
- `open-wait-armed`: 已经记录张开状态参考值,下一次单侧力变化可以触发闭合。
- `force-change-release-open`: 检测到夹住后的法向或切向变化,已经发出张开命令。 - `force-change-release-open`: 检测到夹住后的法向或切向变化,已经发出张开命令。
- `contact-lost-release-open`: 夹住后接触力消失了一小段时间,已经发出张开命令 - `adaptive-grip-ramp`: 切向力显示负载较大,正在按小步进增加夹紧力
- `adaptive-grip-wait`: 已经需要更高目标力,但还没到下一次加力时间。
- `release-arm-delay`: 补力刚停止,正在等待力稳定,暂时不触发松开。
- `release-reference-armed`: 已重新记录释放参考力,后续法向/切向变化会触发张开。
- `open-recover-position`: 夹爪当前位置已经接近 `OPEN_POS`,正在等待到位稳定。
- `open-ready-position`: 夹爪已经回到初始点位,并且休息时间结束,重新进入等待下一次扫过触发;此时会直接武装,但残余触发保持不变不会自动闭合。
- `open-wait-clear-trigger`: 已经回到等待状态,但当前力还超过触发阈值,暂时不会闭合。
- `open-wait-trigger-clearing`: 触发力已经低于阈值,正在等待清零稳定时间。
- `open-wait-armed`: 触发已经清掉并稳定,下一次扫过可以闭合。
- `open-wait-trigger-held`: 已经武装,但当前触发不是新上升沿,等待力先降下去再下一次扫过。
- `open-recover-wait-force-clear`: 夹爪保持张开,但传感器还有残余力,不允许重新闭合。 - `open-recover-wait-force-clear`: 夹爪保持张开,但传感器还有残余力,不允许重新闭合。
- `open-recover-stabilizing`: 力已经低于恢复阈值,正在等待稳定时间 - `open-recover-stabilizing`: 力已经低于恢复阈值,正在等待稳定和休息时间结束
- `open-ready`: 力已经清空并稳定,重新进入等待下一次扫过触发。 - `open-ready`: 力已经清空并稳定,并且休息时间结束,重新进入等待下一次扫过触发。
Binary file not shown.
Binary file not shown.
+69 -36
View File
@@ -29,7 +29,7 @@ SENSOR_CPU = False
# 夹爪串口配置。 # 夹爪串口配置。
PORT = "COM5" PORT = "COM6"
SLAVE_ID = 1 SLAVE_ID = 1
BAUDRATE = 115200 BAUDRATE = 115200
SERIAL_TIMEOUT = 0.2 SERIAL_TIMEOUT = 0.2
@@ -43,18 +43,21 @@ CLOSE_POS = 9000
SPEED = 25 SPEED = 25
ACCEL = 80 ACCEL = 80
DECEL = 80 DECEL = 80
RELEASE_OPEN_SPEED = 50 # 检测到人取物/力变化后,张开使用的速度。
RELEASE_OPEN_ACCEL = 120 # release 张开的加速度。
RELEASE_OPEN_DECEL = 120 # release 张开的减速度。
# 夹爪力度百分比。 # 夹爪力度百分比。
OPEN_FORCE = 15 # 张开时使用的力。 OPEN_FORCE = 15 # 张开时使用的力。
CLOSE_FORCE = 15 # 单侧触发后闭合时使用的小力。 CLOSE_FORCE = 15 # 单侧触发后闭合时使用的小力。
GRIP_START_FORCE = 10 # 闭合检测到物体后,先切到这个低力保持。 GRIP_START_FORCE = 10 # 闭合检测到物体后,先切到这个低力保持。
HOLD_FORCE = 15 # 慢慢加到这个值后进入 hold_check HOLD_FORCE = 15 # 基础夹持力;轻物体默认慢慢加到这个
FORCE_MIN = 10 FORCE_MIN = 10
FORCE_MAX = 30 FORCE_MAX = 80 # 允许自适应加力的最大总上限;重物夹不住时可以适当加大。
# 控制循环频率。展示时调高一点,单侧快速扫过不容易漏检 # 控制循环频率。越高检测释放越快
CONTROL_HZ = 30.0 CONTROL_HZ = 30.0
@@ -67,43 +70,57 @@ SHEAR_FILTER_ALPHA = 0.65
# 等待触发:任意一侧超过这些阈值,就开始闭合。 # 等待触发:任意一侧超过这些阈值,就开始闭合。
TRIGGER_NORMAL_FORCE = 0.04 # 单侧法向力触发阈值,单位 N。 TRIGGER_NORMAL_FORCE = 0.1 # 单侧法向力触发阈值,单位 N。
TRIGGER_SHEAR_FORCE = 0.04 # 单侧切向力触发阈值,单位 N。 TRIGGER_SHEAR_FORCE = 0.1 # 单侧切向力触发阈值,单位 N。
TRIGGER_CLEAR_STABLE_SECONDS = 0.2 # 张开恢复后,必须先看到触发力清掉并稳定这么久,才允许下一次扫过触发。
# 等待触发:如果已经记录了张开时的安静参考值,任意一侧变化超过这些阈值也会闭合 OPEN_WAIT_STABLE_SAMPLES = 12 # 张开等待时,用多少个近期样本判断当前稳定基准
# 这样现场轻扫一下也能触发,不完全依赖绝对力超过 TRIGGER_*_FORCE OPEN_WAIT_STABLE_RANGE_N = 0.03 # 近期样本最大-最小值小于该值,认为该通道趋于稳定,单位 N
TRIGGER_NORMAL_CHANGE = 0.04 # 单侧法向力相对张开参考值的变化触发阈值,单位 N。 OPEN_WAIT_STABLE_TREND_N = 0.015 # 近期样本首尾变化小于该值,认为该通道无明显漂移,单位 N。
TRIGGER_SHEAR_CHANGE = 0.04 # 单侧切向力相对张开参考值的变化触发阈值,单位 N。
# 闭合夹取:物体移动到中间后,左右两边都超过该阈值就认为夹住。 # 闭合夹取:物体移动到中间后,左右两边都超过该阈值就认为夹住。
GRIP_NORMAL_FORCE = 0.05 # 双侧接触阈值,单位 N。 GRIP_NORMAL_FORCE = 0.2 # 双侧接触阈值,单位 N。
GRIP_REQUIRES_BOTH = True # True 表示必须左右两侧都接触才算夹住。 GRIP_REQUIRES_BOTH = True # True 表示必须左右两侧都接触才算夹住。
CLOSE_TIMEOUT_SECONDS = 5.0 # 触发闭合后这么久还没夹住,就重新张开等待。 CLOSE_TIMEOUT_SECONDS = 5.0 # 触发闭合后这么久还没夹住,就重新张开等待。
CLOSING_COMMAND_INTERVAL = 0.2 # 闭合阶段还没双侧夹上时,按该间隔续发闭合命令,避免单侧扫过后停住。
# 检测到物体后慢慢加力,类似 01 的 gripping 状态。 # 检测到物体后慢慢加力,类似 01 的 gripping 状态。
FORCE_RAMP_STEP = 1 FORCE_RAMP_STEP = 1
FORCE_RAMP_INTERVAL = 0.2 FORCE_RAMP_INTERVAL = 0.1
GRIP_SETTLE_SECONDS = 0.3 # 加到 HOLD_FORCE 后再稳定这么久,进入 hold_check。 GRIP_SETTLE_SECONDS = 0.1 # 加到 HOLD_FORCE 后再稳定这么久,进入 hold_check。
# 根据切向力估计物体负载,并慢慢补夹紧力。
# 逻辑:切向力越大,说明物体越重或越容易下滑,目标夹紧力越高。
ADAPTIVE_GRIP_ENABLED = True # 是否启用切向力自适应加力。
ADAPTIVE_SHEAR_START_FORCE = 0.10 # 切向力低于这个值时,认为是轻物体,只用 ADAPTIVE_FORCE_MIN。
ADAPTIVE_SHEAR_FULL_FORCE = 0.60 # 切向力达到这个值时,认为负载较大,目标力加到 ADAPTIVE_FORCE_MAX。
ADAPTIVE_FORCE_MIN = 15 # 自适应夹持的最小目标力,通常等于 HOLD_FORCE。
ADAPTIVE_FORCE_MAX = 45 # 自适应夹持的最大目标力;重物夹不住可以加大,但不要超过 FORCE_MAX。
ADAPTIVE_FORCE_STEP = 1 # 自适应加力每次增加多少 force_pct。
ADAPTIVE_FORCE_INTERVAL = 0.15 # 自适应加力两次之间的间隔,单位秒;越大加力越柔和。
RELEASE_ARM_DELAY_SECONDS = 0.8 # 自适应加力停止后,等待这么久再启用“人取物松开”检测。
# 松开触发阈值。夹住后,法向或切向变化超过其中任意一个,就张开。 # 松开触发阈值。夹住后,法向或切向变化超过其中任意一个,就张开。
RELEASE_NORMAL_CHANGE = 0.03 # 左/右法向力相对夹住参考值的变化阈值,单位 N。 RELEASE_NORMAL_CHANGE = 0.2 # 左/右法向力相对夹住参考值的变化阈值,单位 N。
RELEASE_SHEAR_CHANGE = 0.03 # 左/右切向力相对夹住参考值的变化阈值,单位 N。 RELEASE_SHEAR_CHANGE = 0.2 # 左/右切向力相对夹住参考值的变化阈值,单位 N。
# 夹住后如果物体被拿走,法向和切向都降到这些阈值以下并持续一小段时间,也直接张开 # 张开到位或恢复条件成立后,至少休息多久才重新进入等待触发
RELEASE_CONTACT_LOST_NORMAL_FORCE = 0.025 # 判断接触丢失的最大法向力,单位 N # 休息期间即使有物体扫过,也不会触发闭合
RELEASE_CONTACT_LOST_SHEAR_FORCE = 0.025 # 判断接触丢失的最大切向力,单位 N。
RELEASE_CONTACT_LOST_SECONDS = 0.15 # 接触丢失持续这么久后张开。
# 松开后至少等待多久重新进入等待触发,避免刚张开时马上再次触发。
REARM_SECONDS = 0.5 REARM_SECONDS = 0.5
# 松开/超时张开后,必须等传感器读数低于这些阈值并稳定,才允许下一次触发。 # 松开/超时张开后,必须等传感器读数低于这些阈值并稳定,才允许下一次触发。
# 这几个值应小于 TRIGGER_NORMAL_FORCE / TRIGGER_SHEAR_FORCE。 # 这几个值应小于 TRIGGER_NORMAL_FORCE / TRIGGER_SHEAR_FORCE。
RECOVER_NORMAL_FORCE = 0.035 # 张开恢复时允许重新触发的最大法向残余力,单位 N。 RECOVER_NORMAL_FORCE = 0.02 # 张开恢复时允许重新触发的最大法向残余力,单位 N。
RECOVER_SHEAR_FORCE = 0.035 # 张开恢复时允许重新触发的最大切向残余力,单位 N。 RECOVER_SHEAR_FORCE = 0.02 # 张开恢复时允许重新触发的最大切向残余力,单位 N。
RECOVER_STABLE_SECONDS = 0.25 # 力清空后需要连续稳定这么久,才回到 open_wait。 RECOVER_STABLE_SECONDS = 0.4 # 力清空后需要连续稳定这么久,才回到 open_wait。
# 松开/超时张开后,如果夹爪已经回到 OPEN_POS 初始点位,也允许重新触发。
# 这样传感器上还有一点残余力时,不会一直卡在 open_recover。
OPEN_POSITION_RECOVER_ENABLED = True
OPEN_POSITION_TOLERANCE = 100 # 当前位置和 OPEN_POS 的差值小于它,就认为回到初始点位。
OPEN_POSITION_STABLE_SECONDS = 0.1 # 回到初始点位后稳定这么久,才回到 open_wait。
OPEN_POSITION_CHECK_INTERVAL = 0.1 # open_recover 状态下读取夹爪位置的间隔,单位秒。
# 如果读取当前位置失败,保持当前位置时使用这个备用位置;None 表示不使用。 # 如果读取当前位置失败,保持当前位置时使用这个备用位置;None 表示不使用。
@@ -131,18 +148,19 @@ NORMAL_FORCE_CALIBRATION = {
} }
# 切向力标定:sqrt(FSHEARX^2 + FSHEARY^2) 原始幅值 -> N。 # 切向力标定:FSHEARX / FSHEARY 原始分量 -> N。
# 目前临时沿用 N_.jpg 的法向标定;有切向力标定后替换 points # X/Y 分量分别按同一条曲线标定;控制逻辑再使用标定后分量的合力
SHEAR_FORCE_CALIBRATION = { SHEAR_FORCE_CALIBRATION = {
"enabled": True, "enabled": True,
"method": "piecewise_linear", "method": "piecewise_linear",
"input": "components",
"signed": True,
"extrapolate": True, "extrapolate": True,
"clamp_output_min": 0.0, "clamp_output_min": 0.0,
"points": [ "points": [
{"raw": 1000.0, "force_n": 0.0}, {"raw": 1000.0, "force_n": 0.0},
{"raw": 10000.0, "force_n": 0.377}, {"raw": 36000.0, "force_n": 1.19},
{"raw": 30000.0, "force_n": 1.377}, {"raw": 68000.0, "force_n": 2.06},
{"raw": 62000.0, "force_n": 2.377},
], ],
} }
@@ -168,6 +186,9 @@ CONFIG = {
"speed": SPEED, "speed": SPEED,
"accel": ACCEL, "accel": ACCEL,
"decel": DECEL, "decel": DECEL,
"release_open_speed": RELEASE_OPEN_SPEED,
"release_open_accel": RELEASE_OPEN_ACCEL,
"release_open_decel": RELEASE_OPEN_DECEL,
"open_force": OPEN_FORCE, "open_force": OPEN_FORCE,
"close_force": CLOSE_FORCE, "close_force": CLOSE_FORCE,
@@ -184,23 +205,35 @@ CONFIG = {
"trigger_normal_force": TRIGGER_NORMAL_FORCE, "trigger_normal_force": TRIGGER_NORMAL_FORCE,
"trigger_shear_force": TRIGGER_SHEAR_FORCE, "trigger_shear_force": TRIGGER_SHEAR_FORCE,
"trigger_normal_change": TRIGGER_NORMAL_CHANGE, "trigger_clear_stable_seconds": TRIGGER_CLEAR_STABLE_SECONDS,
"trigger_shear_change": TRIGGER_SHEAR_CHANGE, "open_wait_stable_samples": OPEN_WAIT_STABLE_SAMPLES,
"open_wait_stable_range_n": OPEN_WAIT_STABLE_RANGE_N,
"open_wait_stable_trend_n": OPEN_WAIT_STABLE_TREND_N,
"grip_normal_force": GRIP_NORMAL_FORCE, "grip_normal_force": GRIP_NORMAL_FORCE,
"grip_requires_both": GRIP_REQUIRES_BOTH, "grip_requires_both": GRIP_REQUIRES_BOTH,
"close_timeout_seconds": CLOSE_TIMEOUT_SECONDS, "close_timeout_seconds": CLOSE_TIMEOUT_SECONDS,
"closing_command_interval": CLOSING_COMMAND_INTERVAL,
"force_ramp_step": FORCE_RAMP_STEP, "force_ramp_step": FORCE_RAMP_STEP,
"force_ramp_interval": FORCE_RAMP_INTERVAL, "force_ramp_interval": FORCE_RAMP_INTERVAL,
"grip_settle_seconds": GRIP_SETTLE_SECONDS, "grip_settle_seconds": GRIP_SETTLE_SECONDS,
"adaptive_grip_enabled": ADAPTIVE_GRIP_ENABLED,
"adaptive_shear_start_force": ADAPTIVE_SHEAR_START_FORCE,
"adaptive_shear_full_force": ADAPTIVE_SHEAR_FULL_FORCE,
"adaptive_force_min": ADAPTIVE_FORCE_MIN,
"adaptive_force_max": ADAPTIVE_FORCE_MAX,
"adaptive_force_step": ADAPTIVE_FORCE_STEP,
"adaptive_force_interval": ADAPTIVE_FORCE_INTERVAL,
"release_arm_delay_seconds": RELEASE_ARM_DELAY_SECONDS,
"release_normal_change": RELEASE_NORMAL_CHANGE, "release_normal_change": RELEASE_NORMAL_CHANGE,
"release_shear_change": RELEASE_SHEAR_CHANGE, "release_shear_change": RELEASE_SHEAR_CHANGE,
"release_contact_lost_normal_force": RELEASE_CONTACT_LOST_NORMAL_FORCE,
"release_contact_lost_shear_force": RELEASE_CONTACT_LOST_SHEAR_FORCE,
"release_contact_lost_seconds": RELEASE_CONTACT_LOST_SECONDS,
"rearm_seconds": REARM_SECONDS, "rearm_seconds": REARM_SECONDS,
"recover_normal_force": RECOVER_NORMAL_FORCE, "recover_normal_force": RECOVER_NORMAL_FORCE,
"recover_shear_force": RECOVER_SHEAR_FORCE, "recover_shear_force": RECOVER_SHEAR_FORCE,
"recover_stable_seconds": RECOVER_STABLE_SECONDS, "recover_stable_seconds": RECOVER_STABLE_SECONDS,
"open_position_recover_enabled": OPEN_POSITION_RECOVER_ENABLED,
"open_position_tolerance": OPEN_POSITION_TOLERANCE,
"open_position_stable_seconds": OPEN_POSITION_STABLE_SECONDS,
"open_position_check_interval": OPEN_POSITION_CHECK_INTERVAL,
"hold_fallback_pos": HOLD_FALLBACK_POS, "hold_fallback_pos": HOLD_FALLBACK_POS,
"trigger_force_update": TRIGGER_FORCE_UPDATE, "trigger_force_update": TRIGGER_FORCE_UPDATE,
+501 -34
View File
@@ -1,18 +1,46 @@
import time import time
import threading
from gripper_control.filters import clamp from gripper_control.filters import clamp
class SideTriggerGripController: class SideTriggerGripController:
def __init__(self, config, reader, gripper, feedback_processor): def __init__(
self,
config,
reader,
gripper,
feedback_processor,
status_callback=None,
):
self.config = config self.config = config
self.reader = reader self.reader = reader
self.gripper = gripper self.gripper = gripper
self.feedback_processor = feedback_processor self.feedback_processor = feedback_processor
self.normal_unit = feedback_processor.normal_converter.unit self.normal_unit = feedback_processor.normal_converter.unit
self.shear_unit = feedback_processor.shear_converter.unit self.shear_unit = feedback_processor.shear_converter.unit
self.status_callback = status_callback
self._stop_event = threading.Event()
self._manual_lock = threading.Lock()
self._manual_command = None
def stop(self):
self._stop_event.set()
def request_manual_command(self, command):
if command not in {"open", "close", "hold", "release"}:
raise ValueError(f"unsupported manual command: {command}")
with self._manual_lock:
self._manual_command = command
def _pop_manual_command(self):
with self._manual_lock:
command = self._manual_command
self._manual_command = None
return command
def run(self): def run(self):
self._stop_event.clear()
self.reader.start() self.reader.start()
try: try:
@@ -59,6 +87,30 @@ class SideTriggerGripController:
f"release_normal_change={self.config.release_normal_change:.3f}{self.normal_unit}, " f"release_normal_change={self.config.release_normal_change:.3f}{self.normal_unit}, "
f"release_shear_change={self.config.release_shear_change:.3f}{self.shear_unit}" f"release_shear_change={self.config.release_shear_change:.3f}{self.shear_unit}"
) )
print(
"Adaptive grip: "
f"enabled={int(self.config.adaptive_grip_enabled)}, "
f"shear_start={self.config.adaptive_shear_start_force:.3f}{self.shear_unit}, "
f"shear_full={self.config.adaptive_shear_full_force:.3f}{self.shear_unit}, "
f"force={self.config.adaptive_force_min}-{self.config.adaptive_force_max}, "
f"step={self.config.adaptive_force_step}, "
f"interval={self.config.adaptive_force_interval:.2f}s, "
f"release_arm_delay={self.config.release_arm_delay_seconds:.2f}s"
)
print(
"Release open: "
f"speed={self.config.release_open_speed}, "
f"accel={self.config.release_open_accel}, "
f"decel={self.config.release_open_decel}"
)
print(
"Recover: "
f"force_normal={self.config.recover_normal_force:.3f}{self.normal_unit}, "
f"force_shear={self.config.recover_shear_force:.3f}{self.shear_unit}, "
f"open_pos={self.config.open_pos}, "
f"pos_enabled={int(self.config.open_position_recover_enabled)}, "
f"pos_tol={self.config.open_position_tolerance}"
)
def _run_event_monitor(self): def _run_event_monitor(self):
config = self.config config = self.config
@@ -67,18 +119,125 @@ class SideTriggerGripController:
state = "open_wait" state = "open_wait"
state_start = time.perf_counter() state_start = time.perf_counter()
last_force_ramp_time = state_start last_force_ramp_time = state_start
last_open_position_check_time = 0.0
last_open_position = None
last_open_position_ready = False
recover_quiet_since = None recover_quiet_since = None
recover_position_since = None
trigger_clear_since = None
trigger_armed = False
last_trigger = False
open_wait_samples = []
open_wait_stable_values = {}
force_pct = int(clamp(config.open_force, config.force_min, config.force_max)) force_pct = int(clamp(config.open_force, config.force_min, config.force_max))
adaptive_target_force = int(clamp(config.hold_force, config.force_min, config.force_max))
release_armed = False
hold_pos = None hold_pos = None
reference = None reference = None
while True: def reset_open_wait_stability():
nonlocal open_wait_samples, open_wait_stable_values
open_wait_samples = []
open_wait_stable_values = {}
while not self._stop_event.is_set():
tick_start = time.perf_counter() tick_start = time.perf_counter()
now = time.perf_counter() now = time.perf_counter()
manual_action = None
manual_command = self._pop_manual_command()
if manual_command == "open":
force_pct = int(clamp(
config.open_force,
config.force_min,
config.force_max,
))
self.gripper.open(force_pct, "manual-open")
self.feedback_processor.reset_filters()
state = "open_recover"
state_start = now
last_open_position_check_time = 0.0
last_open_position = None
last_open_position_ready = False
recover_quiet_since = None
recover_position_since = None
trigger_clear_since = None
trigger_armed = False
reset_open_wait_stability()
release_armed = False
hold_pos = None
reference = None
manual_action = "manual-open"
elif manual_command == "close":
force_pct = int(clamp(
config.close_force,
config.force_min,
config.force_max,
))
self.gripper.close(force_pct, "manual-close")
state = "manual_close"
state_start = now
last_force_ramp_time = now
last_open_position_check_time = 0.0
last_open_position = None
last_open_position_ready = False
recover_quiet_since = None
recover_position_since = None
trigger_clear_since = None
trigger_armed = False
reset_open_wait_stability()
release_armed = False
hold_pos = None
reference = None
manual_action = "manual-close"
elif manual_command == "hold":
force_pct = int(clamp(
config.hold_force,
config.force_min,
config.force_max,
))
self.gripper.set_force(force_pct)
hold_pos = self.gripper.hold_current_position(force_pct)
state = "manual_hold"
state_start = now
last_force_ramp_time = now
trigger_clear_since = None
trigger_armed = False
reset_open_wait_stability()
release_armed = False
reference = None
manual_action = "manual-hold"
elif manual_command == "release":
if state == "manual_hold":
state = "open_wait"
state_start = now
trigger_clear_since = None
trigger_armed = False
reset_open_wait_stability()
release_armed = False
hold_pos = None
reference = None
manual_action = "manual-release"
feedback = self.feedback_processor.read(self.reader, timeout=0.3) feedback = self.feedback_processor.read(self.reader, timeout=0.3)
if feedback is None: if feedback is None:
print(f"state={state} action=waiting-for-sensor-samples") action = manual_action or "waiting-for-sensor-samples"
print(f"state={state} action={action}")
self._emit_status(
{
"timestamp": time.time(),
"state": state,
"action": action,
"force_pct": force_pct,
"adaptive_target_force": adaptive_target_force,
"adaptive_grip_enabled": config.adaptive_grip_enabled,
"release_armed": release_armed,
"manual_hold_active": state == "manual_hold",
"speed_pct": config.speed,
"normal_unit": self.normal_unit,
"shear_unit": self.shear_unit,
}
)
time.sleep(0.05) time.sleep(0.05)
continue continue
@@ -86,10 +245,27 @@ class SideTriggerGripController:
grip_contact = self._grip_contact(feedback) grip_contact = self._grip_contact(feedback)
normal_change = 0.0 normal_change = 0.0
shear_change = 0.0 shear_change = 0.0
action = "wait" action = manual_action or "wait"
if state == "open_wait": if manual_action is not None:
if trigger: pass
elif state == "open_wait":
open_wait_sample = self._open_wait_sample(feedback)
self._append_open_wait_sample(open_wait_samples, open_wait_sample)
stable_values = self._open_wait_stable_values(open_wait_samples)
trigger = self._open_wait_delta_triggered(
open_wait_sample,
open_wait_stable_values,
)
if not trigger:
open_wait_stable_values.update(stable_values)
trigger_armed = bool(open_wait_stable_values)
if not trigger_armed:
action = "open-wait-stabilizing"
elif not trigger:
action = "open-wait-armed"
else:
force_pct = int(clamp( force_pct = int(clamp(
config.close_force, config.close_force,
config.force_min, config.force_min,
@@ -99,12 +275,19 @@ class SideTriggerGripController:
state = "closing" state = "closing"
state_start = now state_start = now
last_force_ramp_time = now last_force_ramp_time = now
last_open_position_check_time = 0.0
last_open_position = None
last_open_position_ready = False
recover_quiet_since = None recover_quiet_since = None
recover_position_since = None
trigger_clear_since = None
trigger_armed = False
reset_open_wait_stability()
adaptive_target_force = self._adaptive_force_target(feedback)
release_armed = False
hold_pos = None hold_pos = None
reference = None reference = None
action = "side-trigger-close" action = "side-trigger-close"
else:
action = "open-wait-touch"
elif state == "closing": elif state == "closing":
if grip_contact: if grip_contact:
@@ -118,6 +301,8 @@ class SideTriggerGripController:
state = "gripping" state = "gripping"
state_start = now state_start = now
last_force_ramp_time = now last_force_ramp_time = now
adaptive_target_force = self._adaptive_force_target(feedback)
release_armed = False
reference = None reference = None
action = "object-detected-low-force" action = "object-detected-low-force"
elif now - state_start >= config.close_timeout_seconds: elif now - state_start >= config.close_timeout_seconds:
@@ -130,10 +315,24 @@ class SideTriggerGripController:
self.feedback_processor.reset_filters() self.feedback_processor.reset_filters()
state = "open_recover" state = "open_recover"
state_start = now state_start = now
last_open_position_check_time = 0.0
last_open_position = None
last_open_position_ready = False
recover_quiet_since = None recover_quiet_since = None
recover_position_since = None
reset_open_wait_stability()
release_armed = False
hold_pos = None hold_pos = None
reference = None reference = None
action = "close-timeout-open" action = "close-timeout-open"
else:
if (
config.closing_command_interval > 0
and now - last_force_ramp_time >= config.closing_command_interval
):
self.gripper.close(force_pct, "closing-continue")
last_force_ramp_time = now
action = "closing-continue"
else: else:
action = "closing-to-grip" action = "closing-to-grip"
@@ -157,11 +356,49 @@ class SideTriggerGripController:
reference = self._make_reference(feedback) reference = self._make_reference(feedback)
state = "hold_check" state = "hold_check"
state_start = now state_start = now
adaptive_target_force = self._adaptive_force_target(feedback)
release_armed = False
action = "hold-reference-armed" action = "hold-reference-armed"
else: else:
action = "grip-hold" action = "grip-hold"
elif state == "hold_check": elif state == "hold_check":
adaptive_target_force = self._adaptive_force_target(feedback)
should_adapt_force = (
config.adaptive_grip_enabled
and not release_armed
and force_pct < adaptive_target_force
)
if (
should_adapt_force
and now - last_force_ramp_time >= config.adaptive_force_interval
):
force_pct = int(clamp(
force_pct + config.adaptive_force_step,
config.force_min,
adaptive_target_force,
))
self.gripper.set_force(force_pct)
last_force_ramp_time = now
state_start = now
reference = self._make_reference(feedback)
release_armed = False
action = "adaptive-grip-ramp"
elif should_adapt_force:
release_armed = False
action = "adaptive-grip-wait"
elif (
not release_armed
and now - max(state_start, last_force_ramp_time)
< config.release_arm_delay_seconds
):
action = "release-arm-delay"
elif not release_armed:
reference = self._make_reference(feedback)
release_armed = True
action = "release-reference-armed"
else:
normal_change, shear_change = self._release_changes(feedback, reference) normal_change, shear_change = self._release_changes(feedback, reference)
should_release = ( should_release = (
normal_change >= config.release_normal_change normal_change >= config.release_normal_change
@@ -173,11 +410,23 @@ class SideTriggerGripController:
config.force_min, config.force_min,
config.force_max, config.force_max,
)) ))
self.gripper.open(force_pct, "force-change-release-open") self.gripper.open(
force_pct,
"force-change-release-open",
speed_pct=config.release_open_speed,
accel=config.release_open_accel,
decel=config.release_open_decel,
)
self.feedback_processor.reset_filters() self.feedback_processor.reset_filters()
state = "open_recover" state = "open_recover"
state_start = now state_start = now
last_open_position_check_time = 0.0
last_open_position = None
last_open_position_ready = False
recover_quiet_since = None recover_quiet_since = None
recover_position_since = None
reset_open_wait_stability()
release_armed = False
hold_pos = None hold_pos = None
reference = None reference = None
action = "force-change-release-open" action = "force-change-release-open"
@@ -185,17 +434,62 @@ class SideTriggerGripController:
action = "hold-check" action = "hold-check"
elif state == "open_recover": elif state == "open_recover":
if self._recover_quiet(feedback): current_pos = last_open_position
if recover_quiet_since is None: position_ready = last_open_position_ready
recover_quiet_since = now if (
action = "open-recover-quiet" config.open_position_recover_enabled
and now - last_open_position_check_time
>= config.open_position_check_interval
):
last_open_position_check_time = now
current_pos = self.gripper.read_position()
position_ready = self._open_position_ready(current_pos)
last_open_position = current_pos
last_open_position_ready = position_ready
if position_ready:
if recover_position_since is None:
recover_position_since = now
action = f"open-recover-position pos={current_pos}"
elif ( elif (
now - recover_quiet_since >= config.recover_stable_seconds now - recover_position_since
and now - state_start >= config.rearm_seconds >= max(
config.open_position_stable_seconds,
config.rearm_seconds,
)
): ):
state = "open_wait" state = "open_wait"
state_start = now state_start = now
recover_quiet_since = None recover_quiet_since = None
recover_position_since = None
trigger_clear_since = None
trigger_armed = True
reset_open_wait_stability()
last_trigger = trigger
release_armed = False
hold_pos = None
reference = None
action = f"open-ready-position pos={current_pos}"
else:
action = f"open-recover-position-stabilizing pos={current_pos}"
elif self._recover_quiet(feedback):
recover_position_since = None
if recover_quiet_since is None:
recover_quiet_since = now
action = "open-recover-quiet"
elif (
now - recover_quiet_since
>= max(config.recover_stable_seconds, config.rearm_seconds)
):
state = "open_wait"
state_start = now
recover_quiet_since = None
recover_position_since = None
trigger_clear_since = None
trigger_armed = True
reset_open_wait_stability()
last_trigger = trigger
release_armed = False
hold_pos = None hold_pos = None
reference = None reference = None
action = "open-ready" action = "open-ready"
@@ -203,8 +497,15 @@ class SideTriggerGripController:
action = "open-recover-stabilizing" action = "open-recover-stabilizing"
else: else:
recover_quiet_since = None recover_quiet_since = None
recover_position_since = None
action = "open-recover-wait-force-clear" action = "open-recover-wait-force-clear"
elif state == "manual_hold":
action = "manual-hold"
elif state == "manual_close":
action = "manual-close"
self._log_tick( self._log_tick(
feedback=feedback, feedback=feedback,
state=state, state=state,
@@ -214,27 +515,86 @@ class SideTriggerGripController:
shear_change=shear_change, shear_change=shear_change,
hold_pos=hold_pos, hold_pos=hold_pos,
force_pct=force_pct, force_pct=force_pct,
adaptive_target_force=adaptive_target_force,
release_armed=release_armed,
action=action, action=action,
trigger_armed=trigger_armed,
) )
self._emit_status(
self._make_status(
feedback=feedback,
state=state,
trigger=trigger,
grip_contact=grip_contact,
normal_change=normal_change,
shear_change=shear_change,
hold_pos=hold_pos,
force_pct=force_pct,
adaptive_target_force=adaptive_target_force,
release_armed=release_armed,
action=action,
trigger_armed=trigger_armed,
)
)
last_trigger = trigger
sleep_time = interval - (time.perf_counter() - tick_start) sleep_time = interval - (time.perf_counter() - tick_start)
if sleep_time > 0: if sleep_time > 0:
time.sleep(sleep_time) time.sleep(sleep_time)
def _side_triggered(self, feedback, reference=None): print("demo02 control loop stopped.")
if reference is not None:
normal_change, shear_change = self._release_changes(feedback, reference)
if (
normal_change >= self.config.trigger_normal_change
or shear_change >= self.config.trigger_shear_change
):
return True
def _side_triggered(self, feedback):
return ( return (
feedback.max_normal >= self.config.trigger_normal_force feedback.max_normal >= self.config.trigger_normal_force
or feedback.max_shear >= self.config.trigger_shear_force or feedback.max_shear >= self.config.trigger_shear_force
) )
def _open_wait_sample(self, feedback):
return {
"left_normal": feedback.left_normal,
"right_normal": feedback.right_normal,
"left_shear": feedback.left_shear,
"right_shear": feedback.right_shear,
}
def _append_open_wait_sample(self, samples, sample):
samples.append(sample)
stable_samples = max(2, int(self.config.open_wait_stable_samples))
if len(samples) > stable_samples:
del samples[0]
def _open_wait_stable_values(self, samples):
stable_samples = max(2, int(self.config.open_wait_stable_samples))
if len(samples) < stable_samples:
return {}
stable = {}
first = samples[0]
last = samples[-1]
for key in first:
values = [sample[key] for sample in samples]
value_range = max(values) - min(values)
trend = abs(last[key] - first[key])
if (
value_range <= self.config.open_wait_stable_range_n
and trend <= self.config.open_wait_stable_trend_n
):
stable[key] = sum(values) / len(values)
return stable
def _open_wait_delta_triggered(self, sample, stable_values):
thresholds = {
"left_normal": self.config.trigger_normal_force,
"right_normal": self.config.trigger_normal_force,
"left_shear": self.config.trigger_shear_force,
"right_shear": self.config.trigger_shear_force,
}
for key, stable_value in stable_values.items():
if sample[key] - stable_value >= thresholds[key]:
return True
return False
def _grip_contact(self, feedback): def _grip_contact(self, feedback):
left_contact = feedback.left_normal >= self.config.grip_normal_force left_contact = feedback.left_normal >= self.config.grip_normal_force
right_contact = feedback.right_normal >= self.config.grip_normal_force right_contact = feedback.right_normal >= self.config.grip_normal_force
@@ -248,11 +608,48 @@ class SideTriggerGripController:
and feedback.max_shear <= self.config.recover_shear_force and feedback.max_shear <= self.config.recover_shear_force
) )
def _open_position_ready(self, current_pos):
if current_pos is None:
return False
return (
abs(int(current_pos) - int(self.config.open_pos))
<= int(self.config.open_position_tolerance)
)
def _adaptive_force_target(self, feedback):
config = self.config
if not config.adaptive_grip_enabled:
return int(clamp(config.hold_force, config.force_min, config.force_max))
force_min = int(clamp(
config.adaptive_force_min,
config.force_min,
config.force_max,
))
force_max = int(clamp(
config.adaptive_force_max,
config.force_min,
config.force_max,
))
if force_max < force_min:
force_max = force_min
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))
if shear <= shear_start:
return force_min
if shear >= shear_full:
return force_max
ratio = (shear - shear_start) / (shear_full - shear_start)
target = round(force_min + ratio * (force_max - force_min))
return int(clamp(target, force_min, force_max))
def _make_reference(self, feedback): def _make_reference(self, feedback):
return { return {
"left_normal": feedback.left_normal, "left_normal": feedback.left_normal,
"right_normal": feedback.right_normal, "right_normal": feedback.right_normal,
"max_normal": feedback.max_normal,
"left_shear": feedback.left_shear, "left_shear": feedback.left_shear,
"right_shear": feedback.right_shear, "right_shear": feedback.right_shear,
"max_shear": feedback.max_shear, "max_shear": feedback.max_shear,
@@ -264,21 +661,15 @@ class SideTriggerGripController:
normal_change = max( normal_change = max(
abs(feedback.left_normal - reference["left_normal"]), abs(feedback.left_normal - reference["left_normal"]),
abs(feedback.right_normal - reference["right_normal"]), abs(feedback.right_normal - reference["right_normal"]),
abs(feedback.max_normal - reference["max_normal"]),
) )
shear_change = abs(feedback.max_shear - reference["max_shear"])
shear_change = max( shear_change = max(
shear_change,
abs(feedback.left_shear - reference["left_shear"]), abs(feedback.left_shear - reference["left_shear"]),
abs(feedback.right_shear - reference["right_shear"]), abs(feedback.right_shear - reference["right_shear"]),
abs(feedback.max_shear - reference["max_shear"]),
) )
return normal_change, shear_change return normal_change, shear_change
def _release_contact_lost(self, feedback):
return (
feedback.max_normal <= self.config.release_contact_lost_normal_force
and feedback.max_shear <= self.config.release_contact_lost_shear_force
)
def _log_tick( def _log_tick(
self, self,
feedback, feedback,
@@ -289,16 +680,92 @@ class SideTriggerGripController:
shear_change, shear_change,
hold_pos, hold_pos,
force_pct, force_pct,
adaptive_target_force,
release_armed,
action, action,
trigger_armed,
): ):
print( print(
f"L={feedback.left_normal:7.3f}{self.normal_unit} " f"L={feedback.left_normal:7.3f}{self.normal_unit} "
f"R={feedback.right_normal:7.3f}{self.normal_unit} " f"R={feedback.right_normal:7.3f}{self.normal_unit} "
f"maxN={feedback.max_normal:7.3f}{self.normal_unit} " f"maxN={feedback.max_normal:7.3f}{self.normal_unit} "
f"shear={feedback.max_shear:7.3f}{self.shear_unit} " f"shear={feedback.max_shear:7.3f}{self.shear_unit} "
f"trigger={int(trigger)} grip_contact={int(grip_contact)} " f"trigger={int(trigger)} armed={int(trigger_armed)} "
f"grip_contact={int(grip_contact)} "
f"dN={normal_change:7.3f}{self.normal_unit} " f"dN={normal_change:7.3f}{self.normal_unit} "
f"dShear={shear_change:7.3f}{self.shear_unit} " f"dShear={shear_change:7.3f}{self.shear_unit} "
f"state={state} hold_pos={hold_pos} " f"state={state} hold_pos={hold_pos} "
f"force_pct={force_pct:3d} action={action}" f"force_pct={force_pct:3d} target={adaptive_target_force:3d} "
f"release_armed={int(release_armed)} action={action}"
) )
def _make_status(
self,
feedback,
state,
trigger,
grip_contact,
normal_change,
shear_change,
hold_pos,
force_pct,
adaptive_target_force,
release_armed,
action,
trigger_armed,
):
speed_pct = self.config.speed
if action == "force-change-release-open":
speed_pct = self.config.release_open_speed
left_sample = None
right_sample = None
try:
if len(self.reader.latest) >= 2:
left_sample = self.reader.latest[0]
right_sample = self.reader.latest[1]
except Exception:
pass
return {
"timestamp": time.time(),
"state": state,
"action": action,
"trigger": bool(trigger),
"trigger_armed": bool(trigger_armed),
"grip_contact": bool(grip_contact),
"left_normal": feedback.left_normal,
"right_normal": feedback.right_normal,
"min_normal": feedback.min_normal,
"max_normal": feedback.max_normal,
"normal_diff": feedback.normal_diff,
"left_shear": feedback.left_shear,
"right_shear": feedback.right_shear,
"left_shear_x": feedback.left_shear_x,
"left_shear_y": feedback.left_shear_y,
"right_shear_x": feedback.right_shear_x,
"right_shear_y": feedback.right_shear_y,
"max_shear": feedback.max_shear,
"shear_diff": feedback.shear_diff,
"normal_change": normal_change,
"shear_change": shear_change,
"hold_pos": hold_pos,
"force_pct": force_pct,
"adaptive_target_force": adaptive_target_force,
"adaptive_grip_enabled": self.config.adaptive_grip_enabled,
"release_armed": bool(release_armed),
"manual_hold_active": state == "manual_hold",
"speed_pct": speed_pct,
"normal_unit": self.normal_unit,
"shear_unit": self.shear_unit,
"left_sample": left_sample,
"right_sample": right_sample,
}
def _emit_status(self, status):
if self.status_callback is None:
return
try:
self.status_callback(status)
except Exception as exc:
print(f"status callback failed: {exc}")
+15 -2
View File
@@ -10,7 +10,12 @@ from .controller import SideTriggerGripController
from .settings import Demo02Config from .settings import Demo02Config
def build_controller(argv=None): def build_controller(
argv=None,
status_callback=None,
include_visuals=False,
visual_size=320,
):
ensure_project_paths() ensure_project_paths()
config = Demo02Config.from_args(argv) config = Demo02Config.from_args(argv)
@@ -33,10 +38,18 @@ def build_controller(argv=None):
target_fps=config.sensor_fps, target_fps=config.sensor_fps,
cuda=not config.sensor_cpu, cuda=not config.sensor_cpu,
motion_threshold=config.motion_threshold, motion_threshold=config.motion_threshold,
include_visuals=include_visuals,
visual_size=visual_size,
) )
gripper = GripperClient(config) gripper = GripperClient(config)
feedback = TactileFeedbackProcessor(config, normal_converter, shear_converter) feedback = TactileFeedbackProcessor(config, normal_converter, shear_converter)
return SideTriggerGripController(config, reader, gripper, feedback) return SideTriggerGripController(
config,
reader,
gripper,
feedback,
status_callback=status_callback,
)
def main(argv=None): def main(argv=None):
+48 -16
View File
@@ -43,6 +43,9 @@ class Demo02Config:
speed: int = 25 speed: int = 25
accel: int = 80 accel: int = 80
decel: int = 80 decel: int = 80
release_open_speed: int = 50
release_open_accel: int = 120
release_open_decel: int = 120
open_force: int = 10 open_force: int = 10
close_force: int = 10 close_force: int = 10
@@ -50,7 +53,7 @@ class Demo02Config:
hold_force: int = 15 hold_force: int = 15
initial_force: int = 10 initial_force: int = 10
force_min: int = 10 force_min: int = 10
force_max: int = 30 force_max: int = 60
control_hz: float = 30.0 control_hz: float = 30.0
baseline_seconds: float = 1.0 baseline_seconds: float = 1.0
@@ -59,23 +62,35 @@ class Demo02Config:
trigger_normal_force: float = 0.04 trigger_normal_force: float = 0.04
trigger_shear_force: float = 0.04 trigger_shear_force: float = 0.04
trigger_normal_change: float = 0.02 trigger_clear_stable_seconds: float = 0.2
trigger_shear_change: float = 0.02 open_wait_stable_samples: int = 12
open_wait_stable_range_n: float = 0.03
open_wait_stable_trend_n: float = 0.015
grip_normal_force: float = 0.05 grip_normal_force: float = 0.05
grip_requires_both: bool = True grip_requires_both: bool = True
close_timeout_seconds: float = 5.0 close_timeout_seconds: float = 5.0
closing_command_interval: float = 0.2
force_ramp_step: int = 1 force_ramp_step: int = 1
force_ramp_interval: float = 0.2 force_ramp_interval: float = 0.1
grip_settle_seconds: float = 0.3 grip_settle_seconds: float = 0.1
adaptive_grip_enabled: bool = True
adaptive_shear_start_force: float = 0.10
adaptive_shear_full_force: float = 0.60
adaptive_force_min: int = 15
adaptive_force_max: int = 45
adaptive_force_step: int = 1
adaptive_force_interval: float = 0.15
release_arm_delay_seconds: float = 0.8
release_normal_change: float = 0.03 release_normal_change: float = 0.03
release_shear_change: float = 0.03 release_shear_change: float = 0.03
release_contact_lost_normal_force: float = 0.025
release_contact_lost_shear_force: float = 0.025
release_contact_lost_seconds: float = 0.15
rearm_seconds: float = 0.5 rearm_seconds: float = 0.5
recover_normal_force: float = 0.035 recover_normal_force: float = 0.02
recover_shear_force: float = 0.035 recover_shear_force: float = 0.02
recover_stable_seconds: float = 0.25 recover_stable_seconds: float = 0.4
open_position_recover_enabled: bool = True
open_position_tolerance: int = 100
open_position_stable_seconds: float = 0.1
open_position_check_interval: float = 0.1
hold_fallback_pos: int | None = None hold_fallback_pos: int | None = None
trigger_force_update: bool = False trigger_force_update: bool = False
@@ -121,6 +136,9 @@ class Demo02Config:
add("--speed", type=int, default=get("speed", cls.speed)) add("--speed", type=int, default=get("speed", cls.speed))
add("--accel", type=int, default=get("accel", cls.accel)) add("--accel", type=int, default=get("accel", cls.accel))
add("--decel", type=int, default=get("decel", cls.decel)) add("--decel", type=int, default=get("decel", cls.decel))
add("--release-open-speed", type=int, default=get("release_open_speed", cls.release_open_speed))
add("--release-open-accel", type=int, default=get("release_open_accel", cls.release_open_accel))
add("--release-open-decel", type=int, default=get("release_open_decel", cls.release_open_decel))
add("--open-force", type=int, default=get("open_force", cls.open_force)) add("--open-force", type=int, default=get("open_force", cls.open_force))
add("--close-force", type=int, default=get("close_force", cls.close_force)) add("--close-force", type=int, default=get("close_force", cls.close_force))
@@ -136,24 +154,38 @@ class Demo02Config:
add("--trigger-normal-force", type=float, default=get("trigger_normal_force", cls.trigger_normal_force)) 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)) add("--trigger-shear-force", type=float, default=get("trigger_shear_force", cls.trigger_shear_force))
add("--trigger-normal-change", type=float, default=get("trigger_normal_change", cls.trigger_normal_change)) add("--trigger-clear-stable-seconds", type=float, default=get("trigger_clear_stable_seconds", cls.trigger_clear_stable_seconds))
add("--trigger-shear-change", type=float, default=get("trigger_shear_change", cls.trigger_shear_change)) add("--open-wait-stable-samples", type=int, default=get("open_wait_stable_samples", cls.open_wait_stable_samples))
add("--open-wait-stable-range-n", type=float, default=get("open_wait_stable_range_n", cls.open_wait_stable_range_n))
add("--open-wait-stable-trend-n", type=float, default=get("open_wait_stable_trend_n", cls.open_wait_stable_trend_n))
add("--grip-normal-force", type=float, default=get("grip_normal_force", cls.grip_normal_force)) add("--grip-normal-force", type=float, default=get("grip_normal_force", cls.grip_normal_force))
add("--grip-requires-both", dest="grip_requires_both", action="store_true", default=get("grip_requires_both", cls.grip_requires_both)) add("--grip-requires-both", dest="grip_requires_both", action="store_true", default=get("grip_requires_both", cls.grip_requires_both))
add("--allow-single-side-grip", dest="grip_requires_both", action="store_false") add("--allow-single-side-grip", dest="grip_requires_both", action="store_false")
add("--close-timeout-seconds", type=float, default=get("close_timeout_seconds", cls.close_timeout_seconds)) add("--close-timeout-seconds", type=float, default=get("close_timeout_seconds", cls.close_timeout_seconds))
add("--closing-command-interval", type=float, default=get("closing_command_interval", cls.closing_command_interval))
add("--force-ramp-step", type=int, default=get("force_ramp_step", cls.force_ramp_step)) 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("--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-settle-seconds", type=float, default=get("grip_settle_seconds", cls.grip_settle_seconds))
add("--enable-adaptive-grip", dest="adaptive_grip_enabled", action="store_true", default=get("adaptive_grip_enabled", cls.adaptive_grip_enabled))
add("--disable-adaptive-grip", dest="adaptive_grip_enabled", action="store_false")
add("--adaptive-shear-start-force", type=float, default=get("adaptive_shear_start_force", cls.adaptive_shear_start_force))
add("--adaptive-shear-full-force", type=float, default=get("adaptive_shear_full_force", cls.adaptive_shear_full_force))
add("--adaptive-force-min", type=int, default=get("adaptive_force_min", cls.adaptive_force_min))
add("--adaptive-force-max", type=int, default=get("adaptive_force_max", cls.adaptive_force_max))
add("--adaptive-force-step", type=int, default=get("adaptive_force_step", cls.adaptive_force_step))
add("--adaptive-force-interval", type=float, default=get("adaptive_force_interval", cls.adaptive_force_interval))
add("--release-arm-delay-seconds", type=float, default=get("release_arm_delay_seconds", cls.release_arm_delay_seconds))
add("--release-normal-change", type=float, default=get("release_normal_change", cls.release_normal_change)) add("--release-normal-change", type=float, default=get("release_normal_change", cls.release_normal_change))
add("--release-shear-change", type=float, default=get("release_shear_change", cls.release_shear_change)) add("--release-shear-change", type=float, default=get("release_shear_change", cls.release_shear_change))
add("--release-contact-lost-normal-force", type=float, default=get("release_contact_lost_normal_force", cls.release_contact_lost_normal_force))
add("--release-contact-lost-shear-force", type=float, default=get("release_contact_lost_shear_force", cls.release_contact_lost_shear_force))
add("--release-contact-lost-seconds", type=float, default=get("release_contact_lost_seconds", cls.release_contact_lost_seconds))
add("--rearm-seconds", type=float, default=get("rearm_seconds", cls.rearm_seconds)) add("--rearm-seconds", type=float, default=get("rearm_seconds", cls.rearm_seconds))
add("--recover-normal-force", type=float, default=get("recover_normal_force", cls.recover_normal_force)) add("--recover-normal-force", type=float, default=get("recover_normal_force", cls.recover_normal_force))
add("--recover-shear-force", type=float, default=get("recover_shear_force", cls.recover_shear_force)) add("--recover-shear-force", type=float, default=get("recover_shear_force", cls.recover_shear_force))
add("--recover-stable-seconds", type=float, default=get("recover_stable_seconds", cls.recover_stable_seconds)) add("--recover-stable-seconds", type=float, default=get("recover_stable_seconds", cls.recover_stable_seconds))
add("--open-position-recover-enabled", action="store_true", default=get("open_position_recover_enabled", cls.open_position_recover_enabled))
add("--disable-open-position-recover", dest="open_position_recover_enabled", action="store_false")
add("--open-position-tolerance", type=int, default=get("open_position_tolerance", cls.open_position_tolerance))
add("--open-position-stable-seconds", type=float, default=get("open_position_stable_seconds", cls.open_position_stable_seconds))
add("--open-position-check-interval", type=float, default=get("open_position_check_interval", cls.open_position_check_interval))
add("--hold-fallback-pos", type=int, default=get("hold_fallback_pos", cls.hold_fallback_pos)) add("--hold-fallback-pos", type=int, default=get("hold_fallback_pos", cls.hold_fallback_pos))
add("--trigger-force-update", action="store_true", default=get("trigger_force_update", cls.trigger_force_update)) add("--trigger-force-update", action="store_true", default=get("trigger_force_update", cls.trigger_force_update))
+793
View File
@@ -0,0 +1,793 @@
import argparse
import multiprocessing as mp
import sys
import threading
import time
import cv2
import numpy as np
def _fix_qt_plugin_path():
try:
from PyQt5.QtCore import QLibraryInfo
import os
os.environ["QT_QPA_PLATFORM_PLUGIN_PATH"] = QLibraryInfo.location(
QLibraryInfo.PluginsPath
)
except Exception:
pass
_fix_qt_plugin_path()
_QT_IMPORT_ERROR = None
try:
from PyQt5.QtCore import QEasingCurve, QPropertyAnimation, Qt, QTimer
from PyQt5.QtGui import QFont, QImage, QPixmap
from PyQt5.QtWidgets import (
QApplication,
QFrame,
QGraphicsOpacityEffect,
QGridLayout,
QHBoxLayout,
QLabel,
QMainWindow,
QPushButton,
QSizePolicy,
QSplitter,
QVBoxLayout,
QWidget,
)
except Exception as exc:
_QT_IMPORT_ERROR = exc
Qt = QTimer = QPropertyAnimation = QEasingCurve = QFont = QImage = QPixmap = QApplication = None
QGridLayout = QHBoxLayout = QLabel = QPushButton = QSizePolicy = QSplitter = None
QVBoxLayout = QWidget = None
QFrame = QGraphicsOpacityEffect = QMainWindow = object
try:
import pyqtgraph as pg
except Exception as exc:
pg = None
_PYQTGRAPH_IMPORT_ERROR = exc
else:
_PYQTGRAPH_IMPORT_ERROR = None
from .main import build_controller
PLOT_POINTS = 180
PLOT_FORCE_Y_MIN_N = 0.0
PLOT_FORCE_Y_MAX_N = 3.0
PLOT_SHEAR_Y_MIN_N = -3.0
PLOT_SHEAR_Y_MAX_N = 3.0
FORCE_DISPLAY_ZERO_THRESHOLD_N = 0.05
CURVE_COLOR = "#00E5FF"
def cv2_to_pixmap(img_bgr):
if img_bgr is None:
return QPixmap()
if len(img_bgr.shape) == 2:
img_bgr = cv2.cvtColor(img_bgr, cv2.COLOR_GRAY2BGR)
rgb = cv2.cvtColor(np.ascontiguousarray(img_bgr), cv2.COLOR_BGR2RGB)
h, w, ch = rgb.shape
qimg = QImage(rgb.data, w, h, ch * w, QImage.Format_RGB888).copy()
return QPixmap.fromImage(qimg)
def force_display_value(value):
value = float(value)
if abs(value) < FORCE_DISPLAY_ZERO_THRESHOLD_N:
return 0.0
return value
def fmt(value, unit="", width=0, precision=3):
if value is None:
text = "--"
else:
value = force_display_value(value)
text = "0" if value == 0.0 else f"{value:.{precision}f}"
if width:
text = text.rjust(width)
return f"{text}{unit}"
STATE_TEXT = {
"open_wait": "等待触发",
"closing": "闭合中",
"gripping": "夹持加力",
"hold_check": "夹持检测",
"open_recover": "张开恢复",
"manual_close": "手动闭合",
"manual_hold": "手动保持",
"error": "错误",
}
ACTION_TEXT = {
"wait": "等待",
"waiting-for-sensor-samples": "等待传感器数据",
"open-wait-stabilizing": "等待稳定基准",
"open-wait-armed": "已就绪",
"side-trigger-close": "侧边触发闭合",
"closing-continue": "继续闭合",
"closing-to-grip": "闭合寻找物体",
"object-detected-low-force": "检测到物体",
"close-timeout-open": "闭合超时张开",
"grip-ramp-up": "夹持加力",
"grip-hold": "夹持保持",
"hold-reference-armed": "记录夹持基准",
"adaptive-grip-ramp": "自适应加力",
"adaptive-grip-wait": "等待自适应加力",
"release-arm-delay": "释放检测延时",
"release-reference-armed": "释放检测就绪",
"hold-check": "保持检测",
"force-change-release-open": "力变化张开",
"open-recover-position": "张开位置恢复",
"open-recover-position-stabilizing": "张开位置稳定中",
"open-ready-position": "张开到位",
"open-recover-quiet": "力稳定恢复",
"open-recover-stabilizing": "恢复稳定中",
"open-recover-wait-force-clear": "等待力恢复",
"open-ready": "恢复完成",
"manual-open": "手动张开",
"manual-close": "手动闭合",
"manual-hold": "手动保持",
"manual-release": "手动释放",
}
def label_with_cn(value, mapping):
value = str(value or "--")
cn = mapping.get(value)
if cn is None:
for prefix, text in mapping.items():
if value.startswith(f"{prefix} "):
cn = text
break
if cn is None:
return value
return f"{value}{cn}"
STYLE = """
QMainWindow {
background-color: #000000;
}
QWidget {
background-color: #000000;
color: #9a9a9a;
font-family: "Microsoft YaHei", "SimHei";
}
QLabel#Title {
color: #d7d7d7;
font-size: 18px;
font-weight: 600;
}
QLabel#Subtle {
color: #777777;
font-size: 12px;
}
QLabel#MetricLabel {
color: #888888;
font-size: 12px;
}
QLabel#MetricValue {
color: #f2f2f2;
font-family: "Cascadia Code", "Consolas";
font-size: 22px;
font-weight: 600;
}
QLabel#SmallValue {
color: #d8d8d8;
font-family: "Cascadia Code", "Consolas";
font-size: 15px;
}
QLabel#StatusChip {
background-color: #101010;
border: 1px solid #303030;
border-radius: 4px;
color: #00e5ff;
font-family: "Cascadia Code", "Consolas";
font-size: 18px;
font-weight: 600;
padding: 8px 12px;
}
QFrame#StatusPanel {
background-color: #101010;
border: 1px solid #303030;
border-radius: 4px;
}
QLabel#StatusMain {
color: #00e5ff;
font-family: "Cascadia Code", "Consolas";
font-size: 17px;
font-weight: 600;
}
QLabel#StatusSub {
color: #9a9a9a;
font-family: "Cascadia Code", "Consolas";
font-size: 12px;
}
QLabel#ActionChip {
background-color: #0b0b0b;
border: 1px solid #262626;
border-radius: 4px;
color: #c8c8c8;
font-family: "Cascadia Code", "Consolas";
font-size: 13px;
padding: 7px 10px;
}
QLabel#ImagePanel {
background-color: #0b0b0b;
border: 1px solid #262626;
border-radius: 4px;
}
QFrame#Panel {
background-color: #050505;
border: 1px solid #222222;
border-radius: 4px;
}
QPushButton {
background-color: #171717;
color: #d8d8d8;
border: 1px solid #363636;
border-radius: 4px;
padding: 7px 16px;
font-size: 13px;
}
QPushButton:hover {
background-color: #242424;
color: #ffffff;
border: 1px solid #5a5a5a;
}
QPushButton:pressed {
background-color: #0f0f0f;
border: 1px solid #8a8a8a;
padding-top: 8px;
padding-bottom: 6px;
}
QPushButton:disabled {
background-color: #101010;
color: #606060;
border: 1px solid #242424;
}
QPushButton#StartButton {
background-color: #171717;
color: #d8d8d8;
border: 1px solid #363636;
}
QPushButton#StopButton {
background-color: #171717;
color: #d8d8d8;
border: 1px solid #363636;
}
QPushButton#RestartButton {
background-color: #171717;
color: #d8d8d8;
border: 1px solid #363636;
}
QPushButton#ManualButton {
background-color: #171717;
color: #d8d8d8;
border: 1px solid #363636;
}
QPushButton#HoldButton {
background-color: #171717;
color: #d8d8d8;
border: 1px solid #363636;
}
QSplitter::handle {
background-color: #151515;
}
QSplitter::handle:hover {
background-color: #2a2a2a;
}
QSplitter::handle:horizontal {
width: 8px;
}
QSplitter::handle:vertical {
height: 8px;
}
"""
if QPushButton is not None:
class AnimatedButton(QPushButton):
def __init__(self, text=""):
super().__init__(text)
self._opacity = QGraphicsOpacityEffect(self)
self._opacity.setOpacity(1.0)
self.setGraphicsEffect(self._opacity)
self._animation = QPropertyAnimation(self._opacity, b"opacity", self)
self._animation.setEasingCurve(QEasingCurve.OutCubic)
def _animate_opacity(self, target, duration):
self._animation.stop()
self._animation.setDuration(duration)
self._animation.setStartValue(self._opacity.opacity())
self._animation.setEndValue(target)
self._animation.start()
def mousePressEvent(self, event):
if self.isEnabled():
self._animate_opacity(0.72, 70)
super().mousePressEvent(event)
def mouseReleaseEvent(self, event):
if self.isEnabled():
self._animate_opacity(1.0, 140)
super().mouseReleaseEvent(event)
else:
AnimatedButton = None
class MetricBox(QFrame):
def __init__(self, title, value="--"):
super().__init__()
self.setObjectName("Panel")
layout = QVBoxLayout(self)
layout.setContentsMargins(10, 8, 10, 8)
layout.setSpacing(4)
self.title_label = QLabel(title)
self.title_label.setObjectName("MetricLabel")
self.value_label = QLabel(value)
self.value_label.setObjectName("MetricValue")
layout.addWidget(self.title_label)
layout.addWidget(self.value_label)
def set_value(self, value, sub=None):
self.value_label.setText(value)
class ImageBox(QFrame):
def __init__(self, title):
super().__init__()
self.setObjectName("Panel")
layout = QVBoxLayout(self)
layout.setContentsMargins(8, 8, 8, 8)
layout.setSpacing(6)
self.title_label = QLabel(title)
self.title_label.setObjectName("MetricLabel")
self.image_label = QLabel()
self.image_label.setObjectName("ImagePanel")
self.image_label.setAlignment(Qt.AlignCenter)
self.image_label.setMinimumSize(180, 180)
self.image_label.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding)
self.image_label.setScaledContents(True)
layout.addWidget(self.title_label)
layout.addWidget(self.image_label, 1)
def set_image(self, img):
self.image_label.setPixmap(cv2_to_pixmap(img))
class GripperDemo02Viewer(QMainWindow):
def __init__(self, argv=None):
super().__init__()
if _QT_IMPORT_ERROR is not None:
raise RuntimeError(f"PyQt5 import failed: {_QT_IMPORT_ERROR}")
if pg is None:
raise RuntimeError(f"pyqtgraph import failed: {_PYQTGRAPH_IMPORT_ERROR}")
self.setWindowTitle("Gripper Demo 02")
self.setStyleSheet(STYLE)
self.controller = None
self.control_thread = None
self.latest_status = None
self.status_lock = threading.Lock()
self.controller_finished = False
self.restart_pending = False
self.button_state = "idle"
self.start_time = time.time()
self.force_history = []
self.argv = argv
self._build_ui()
self._configure_plot()
self.timer = QTimer(self)
self.timer.timeout.connect(self._refresh)
self.timer.start(33)
def _build_ui(self):
root = QWidget()
self.setCentralWidget(root)
outer = QVBoxLayout(root)
outer.setContentsMargins(8, 8, 8, 8)
outer.setSpacing(8)
top = QHBoxLayout()
top.setSpacing(8)
title = QLabel("Orisys 夹爪")
title.setObjectName("Title")
self.status_panel = QFrame()
self.status_panel.setObjectName("StatusPanel")
status_layout = QVBoxLayout(self.status_panel)
status_layout.setContentsMargins(10, 6, 10, 6)
status_layout.setSpacing(1)
self.status_chip = QLabel("state=--")
self.status_chip.setObjectName("StatusMain")
self.action_label = QLabel("action=--")
self.action_label.setObjectName("StatusSub")
status_layout.addWidget(self.status_chip)
status_layout.addWidget(self.action_label)
self.start_button = AnimatedButton("启动")
self.start_button.setObjectName("StartButton")
self.start_button.clicked.connect(self.start_control)
self.stop_button = AnimatedButton("停止")
self.stop_button.setObjectName("StopButton")
self.stop_button.clicked.connect(self.stop_control)
self.stop_button.setEnabled(False)
self.restart_button = AnimatedButton("重启")
self.restart_button.setObjectName("RestartButton")
self.restart_button.clicked.connect(self.restart_control)
self.manual_open_button = AnimatedButton("夹爪张开")
self.manual_open_button.setObjectName("ManualButton")
self.manual_open_button.clicked.connect(lambda: self.send_manual_command("open"))
self.manual_close_button = AnimatedButton("夹爪闭合")
self.manual_close_button.setObjectName("ManualButton")
self.manual_close_button.clicked.connect(lambda: self.send_manual_command("close"))
self.manual_hold_button = AnimatedButton("保持")
self.manual_hold_button.setObjectName("HoldButton")
self.manual_hold_button.clicked.connect(self.toggle_manual_hold)
self._set_button_state("idle")
top.addWidget(title)
top.addWidget(self.status_panel)
top.addStretch(1)
top.addWidget(self.start_button)
top.addWidget(self.stop_button)
top.addWidget(self.restart_button)
top.addWidget(self.manual_open_button)
top.addWidget(self.manual_close_button)
top.addWidget(self.manual_hold_button)
outer.addLayout(top)
main_splitter = QSplitter(Qt.Horizontal)
main_splitter.setChildrenCollapsible(False)
outer.addWidget(main_splitter, 1)
flow_splitter = QSplitter(Qt.Vertical)
flow_splitter.setChildrenCollapsible(False)
self.left_flow = ImageBox("左侧光流")
self.right_flow = ImageBox("右侧光流")
flow_splitter.addWidget(self.left_flow)
flow_splitter.addWidget(self.right_flow)
flow_splitter.setStretchFactor(0, 1)
flow_splitter.setStretchFactor(1, 1)
flow_splitter.setSizes([360, 360])
main_splitter.addWidget(flow_splitter)
side_widget = QWidget()
side = QVBoxLayout(side_widget)
side.setContentsMargins(0, 0, 0, 0)
side.setSpacing(8)
main_splitter.addWidget(side_widget)
main_splitter.setStretchFactor(0, 2)
main_splitter.setStretchFactor(1, 4)
main_splitter.setSizes([520, 980])
metrics_grid = QGridLayout()
metrics_grid.setSpacing(8)
side.addLayout(metrics_grid)
self.left_force = MetricBox("左法向力", "--")
self.right_force = MetricBox("右法向力", "--")
self.left_shear_box = MetricBox("左切向力", "--")
self.right_shear_box = MetricBox("右切向力", "--")
self.force_pct_box = MetricBox("夹爪当前力比例(%", "--")
self.target_force_box = MetricBox("夹爪目标力比例(%", "--")
self.speed_box = MetricBox("速度", "--")
self.trigger_box = MetricBox("是否触发/系统是否允许触发", "--")
self.contact_box = MetricBox("接触", "--")
self.release_box = MetricBox("释放检测", "--")
boxes = [
self.left_force,
self.right_force,
self.left_shear_box,
self.right_shear_box,
self.force_pct_box,
self.target_force_box,
self.speed_box,
self.trigger_box,
self.contact_box,
self.release_box,
]
for idx, box in enumerate(boxes):
metrics_grid.addWidget(box, idx // 2, idx % 2)
pg.setConfigOption("background", "#000000")
pg.setConfigOption("foreground", "#777777")
pg.setConfigOptions(antialias=True)
self.plot_widget = pg.GraphicsLayoutWidget()
self.plot_widget.setMinimumHeight(240)
side.addWidget(self.plot_widget, 1)
def _configure_plot(self):
self.plots = []
self.curves = {}
plot_specs = [
("left_normal", "左法向力", "#00E5FF", PLOT_FORCE_Y_MIN_N, PLOT_FORCE_Y_MAX_N),
("right_normal", "右法向力", "#ffb000", PLOT_FORCE_Y_MIN_N, PLOT_FORCE_Y_MAX_N),
("left_shear_x", "左切向 X", "#61d394", PLOT_SHEAR_Y_MIN_N, PLOT_SHEAR_Y_MAX_N),
("left_shear_y", "左切向 Y", "#f07178", PLOT_SHEAR_Y_MIN_N, PLOT_SHEAR_Y_MAX_N),
("right_shear_x", "右切向 X", "#8ab4f8", PLOT_SHEAR_Y_MIN_N, PLOT_SHEAR_Y_MAX_N),
("right_shear_y", "右切向 Y", "#d19a66", PLOT_SHEAR_Y_MIN_N, PLOT_SHEAR_Y_MAX_N),
]
for idx, (key, title, color, y_min, y_max) in enumerate(plot_specs):
row = idx // 2
col = idx % 2
plot = self.plot_widget.addPlot(row=row, col=col)
plot.setTitle(title, color="#999999", size="11pt")
plot.titleLabel.setFont(QFont("SimHei", 11))
plot.showAxis("top", False)
plot.showAxis("right", False)
plot.showGrid(x=False, y=True, alpha=0.12)
plot.setLabel("bottom", "时间", units="s")
plot.setLabel("left", "", units="N")
plot.getAxis("bottom").setTickSpacing(major=1.0, minor=1.0)
plot.setYRange(y_min, y_max, padding=0)
plot.setLimits(yMin=y_min, yMax=y_max)
plot.enableAutoRange(axis=pg.ViewBox.YAxis, enable=False)
self.plots.append(plot)
self.curves[key] = plot.plot(pen=pg.mkPen(color, width=2))
def _set_button_state(self, state):
labels = {
"idle": ("启动", "停止", "重启"),
"starting": ("启动中", "停止", "重启"),
"running": ("启动", "停止", "重启"),
"stopping": ("启动", "停止中", "重启"),
"restarting": ("启动", "停止", "重启中"),
}
enabled = {
"idle": (True, False, True),
"starting": (False, True, False),
"running": (False, True, True),
"stopping": (False, False, False),
"restarting": (False, False, False),
}
self.button_state = state
start_text, stop_text, restart_text = labels[state]
start_enabled, stop_enabled, restart_enabled = enabled[state]
self.start_button.setText(start_text)
self.stop_button.setText(stop_text)
self.restart_button.setText(restart_text)
self.start_button.setEnabled(start_enabled)
self.stop_button.setEnabled(stop_enabled)
self.restart_button.setEnabled(restart_enabled)
manual_enabled = state == "running"
self.manual_open_button.setEnabled(manual_enabled)
self.manual_close_button.setEnabled(manual_enabled)
self.manual_hold_button.setEnabled(manual_enabled)
def send_manual_command(self, command):
if self.controller is None or self.control_thread is None or not self.control_thread.is_alive():
return
try:
self.controller.request_manual_command(command)
except Exception as exc:
print(f"manual command failed: {exc}")
def toggle_manual_hold(self):
command = "release" if self.manual_hold_button.text() == "释放" else "hold"
self.send_manual_command(command)
def start_control(self, from_restart=False):
if self.control_thread is not None and self.control_thread.is_alive():
return
self.force_history.clear()
self.start_time = time.time()
self._set_button_state("restarting" if from_restart else "starting")
self.controller_finished = False
self.controller = build_controller(
self.argv,
status_callback=self._on_status,
include_visuals=True,
visual_size=320,
)
self.control_thread = threading.Thread(
target=self._run_controller,
daemon=True,
)
self.control_thread.start()
def stop_control(self):
if self.controller is not None:
self.controller.stop()
self._set_button_state("stopping")
def restart_control(self):
if self.control_thread is not None and self.control_thread.is_alive():
self.restart_pending = True
self._set_button_state("restarting")
if self.controller is not None:
self.controller.stop()
return
self.restart_pending = False
self.start_control(from_restart=True)
def _run_controller(self):
try:
self.controller.run()
except Exception as exc:
self._on_status({"state": "error", "action": str(exc), "timestamp": time.time()})
finally:
self.controller_finished = True
def _on_status(self, status):
with self.status_lock:
self.latest_status = status
def _refresh(self):
with self.status_lock:
status = self.latest_status
if not status:
self._update_thread_buttons()
return
self._update_thread_buttons()
if (
self.control_thread is not None
and self.control_thread.is_alive()
and not self.restart_pending
and self.button_state in ("starting", "restarting")
):
self._set_button_state("running")
state = status.get("state", "--")
action = status.get("action", "--")
unit = status.get("normal_unit", "N")
shear_unit = status.get("shear_unit", "N")
self.status_chip.setText(f"state={label_with_cn(state, STATE_TEXT)}")
self.action_label.setText(f"action={label_with_cn(action, ACTION_TEXT)}")
self.manual_hold_button.setText(
"释放" if bool(status.get("manual_hold_active")) else "保持"
)
self.left_force.set_value(
fmt(status.get("left_normal"), unit),
)
self.right_force.set_value(
fmt(status.get("right_normal"), unit),
)
self.left_shear_box.set_value(
fmt(status.get("left_shear"), shear_unit),
)
self.right_shear_box.set_value(
fmt(status.get("right_shear"), shear_unit),
)
self.force_pct_box.set_value(
str(status.get("force_pct", "--")),
)
self.target_force_box.set_value(
str(status.get("adaptive_target_force", "--")),
)
self.speed_box.set_value(
str(status.get("speed_pct", "--")),
)
self.trigger_box.set_value(
f"{int(bool(status.get('trigger')))} / {int(bool(status.get('trigger_armed')))}",
)
self.contact_box.set_value(
str(int(bool(status.get("grip_contact")))),
)
self.release_box.set_value(
str(int(bool(status.get("release_armed")))),
)
left_sample = status.get("left_sample") or {}
right_sample = status.get("right_sample") or {}
self.left_flow.set_image(left_sample.get("flow_view"))
self.right_flow.set_image(right_sample.get("flow_view"))
elapsed = time.time() - self.start_time
self.force_history.append(
(
elapsed,
force_display_value(status.get("left_normal", 0.0) or 0.0),
force_display_value(status.get("right_normal", 0.0) or 0.0),
force_display_value(status.get("left_shear_x", 0.0) or 0.0),
force_display_value(status.get("left_shear_y", 0.0) or 0.0),
force_display_value(status.get("right_shear_x", 0.0) or 0.0),
force_display_value(status.get("right_shear_y", 0.0) or 0.0),
)
)
if len(self.force_history) > PLOT_POINTS:
self.force_history = self.force_history[-PLOT_POINTS:]
self._update_plot()
def _update_plot(self):
if not self.force_history:
return
data = np.asarray(self.force_history, dtype=float)
x = data[:, 0]
keys = [
"left_normal",
"right_normal",
"left_shear_x",
"left_shear_y",
"right_shear_x",
"right_shear_y",
]
for idx, key in enumerate(keys, start=1):
self.curves[key].setData(x, data[:, idx])
if len(x) >= 2:
x_max = max(6.0, float(np.ceil(x[-1])))
x_min = max(0.0, x_max - 6.0)
for plot in self.plots:
plot.setXRange(x_min, x_max, padding=0)
def _update_thread_buttons(self):
if not self.controller_finished:
return
self.controller_finished = False
if self.control_thread is not None:
self.control_thread.join(timeout=0.1)
self.control_thread = None
if self.restart_pending:
self.restart_pending = False
self.start_control(from_restart=True)
return
self._set_button_state("idle")
def keyPressEvent(self, event):
if event.key() == Qt.Key_Q:
self.close()
return
if event.key() == Qt.Key_Escape:
if self.isFullScreen():
self.showNormal()
else:
self.showFullScreen()
return
super().keyPressEvent(event)
def closeEvent(self, event):
self.timer.stop()
self.restart_pending = False
self.stop_control()
if self.control_thread is not None:
self.control_thread.join(timeout=3.0)
event.accept()
def main(argv=None):
parser = argparse.ArgumentParser()
parser.add_argument("--windowed", action="store_true")
ui_args, controller_argv = parser.parse_known_args(argv)
if _QT_IMPORT_ERROR is not None:
print(
"PyQt5 is required for the visualizer. "
"Install it in the active environment, for example: pip install PyQt5 pyqtgraph"
)
print(f"Import error: {_QT_IMPORT_ERROR}")
return 1
if pg is None:
print(
"pyqtgraph is required for the visualizer. "
"Install it in the active environment, for example: pip install pyqtgraph"
)
print(f"Import error: {_PYQTGRAPH_IMPORT_ERROR}")
return 1
QApplication.setAttribute(Qt.AA_EnableHighDpiScaling, True)
QApplication.setAttribute(Qt.AA_UseHighDpiPixmaps, True)
app = QApplication(sys.argv[:1])
window = GripperDemo02Viewer(controller_argv)
if ui_args.windowed:
window.resize(1500, 900)
window.show()
else:
window.showFullScreen()
window.start_control()
return app.exec_()
if __name__ == "__main__":
mp.freeze_support()
raise SystemExit(main(sys.argv[1:]))
+32
View File
@@ -14,3 +14,35 @@ conda activate py311
1. 某个单侧有东西扫过时开始闭合夹住 1. 某个单侧有东西扫过时开始闭合夹住
2. 闭合的时候要是检测到有物体,就像01那样几个状态,夹住它,不同的是后续有法向力,切向力输入的时候就松开夹爪(不要卡顿),直接张开到最开始的位置。 2. 闭合的时候要是检测到有物体,就像01那样几个状态,夹住它,不同的是后续有法向力,切向力输入的时候就松开夹爪(不要卡顿),直接张开到最开始的位置。
3. 这就是整套流程,后续单侧再检测到有东西扫过的时候再执行这一整个过程 3. 这就是整套流程,后续单侧再检测到有东西扫过的时候再执行这一整个过程
# 展示03
现在把hold_check状态抽离出来,这里后续需要扩写,这里我夹住时 物体的力可能受到
# 可视化展示 qt界面
目前夹住程序还是ok,现在需要可视化展示,用pyqt写,
1. 展示两边的光流 和 赋值图
2. 展示两边作用力的大小
3. 展示当前状态 state=open_wait这种
4. force_pct与速度 展示
5. 你看看还有其他什么可以展示的
1. 夹爪程序编写与调试
目前程序只是常驻监听单侧触发,每次触发后执行一次“闭合夹住 -> 监测变化 -> 直接张开”的流程。
## 流程
1.1. 程序启动后,夹爪先张开。
1.2. 程序等待任意一侧触觉传感器感受到法向力或切向力。
1.3. 展示时,把物体从某个侧边扫一下,触发单侧力变化。
1.4. 触发后夹爪开始闭合,你把物体移动到中间。
1.5. 闭合过程中检测到物体后,切到低力并保持当前位置。
1.6. 进入 `gripping`,先慢慢加力到基础 `HOLD_FORCE`
1.7. 进入 `hold_check` 后,根据切向力估计物体负载;切向力越大,目标夹紧力越高,并按小步进继续补力。
1.8. 自适应补力停止并稳定 `RELEASE_ARM_DELAY_SECONDS` 后,重新记录释放参考力,才开始检测人取物松开。
1.9. 后续法向力或切向力变化超过阈值,夹爪直接张开到 `OPEN_POS`
1.10. 张开后进入 `open_recover`,保持张开并等待恢复条件。
1.11. 夹爪回到 `OPEN_POS` 初始点位,或法向/切向读数都低于恢复阈值并稳定后,继续休息至少 `REARM_SECONDS`
1.12. 休息结束后才回到 `open_wait` 状态;休息期间不管有没有物体扫过,都不会闭合。
1.13. 如果达到没有夹持到物体自动闭合时会自动张开。
2. 参加展会
3. 01 02 3D展示效果探索
4. 02效果目前比较差,在优化中,但是一直没有实物都是使用视频来处理
+25
View File
@@ -0,0 +1,25 @@
@echo off
setlocal
cd /d "%~dp0"
set "CONDA_ENV=py311"
call conda activate "%CONDA_ENV%"
if errorlevel 1 (
echo Failed to activate conda environment: %CONDA_ENV%
echo Please check that conda is initialized and the environment exists.
pause
exit /b 1
)
python -m gripper_control_02.visualizer %*
set EXIT_CODE=%ERRORLEVEL%
if not "%EXIT_CODE%"=="0" (
echo.
echo Gripper Demo 02 exited with code %EXIT_CODE%.
pause
)
exit /b %EXIT_CODE%