完善释放控制：稳定/迅速行为预设、力曲线夹稳门控与会话记录。
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
cheng
@@ -0,0 +1,57 @@
|
||||
import unittest
|
||||
|
||||
from gripper.config.behavior_profiles import (
|
||||
GRIP_BEHAVIOR_PROFILES,
|
||||
apply_grip_behavior_profile,
|
||||
normalize_grip_behavior_mode,
|
||||
)
|
||||
from gripper.config.settings import Demo02Config
|
||||
|
||||
|
||||
class BehaviorProfileTests(unittest.TestCase):
|
||||
def test_normalize_aliases(self):
|
||||
self.assertEqual(normalize_grip_behavior_mode("迅速开合"), "rapid")
|
||||
self.assertEqual(normalize_grip_behavior_mode("夹持稳定"), "stable")
|
||||
|
||||
def test_apply_to_config_object(self):
|
||||
config = Demo02Config()
|
||||
apply_grip_behavior_profile(config, "rapid")
|
||||
self.assertEqual(config.grip_behavior_mode, "rapid")
|
||||
self.assertEqual(config.release_min_hold_seconds, 0.0)
|
||||
self.assertFalse(config.adaptive_grip_enabled)
|
||||
|
||||
apply_grip_behavior_profile(config, "stable")
|
||||
self.assertEqual(config.grip_behavior_mode, "stable")
|
||||
self.assertGreater(config.release_min_hold_seconds, 0.5)
|
||||
self.assertTrue(config.adaptive_grip_enabled)
|
||||
self.assertTrue(config.release_require_slope_settle)
|
||||
self.assertTrue(config.release_settle_normal_only)
|
||||
self.assertTrue(config.release_side_shear_enabled)
|
||||
self.assertFalse(config.release_side_normal_enabled)
|
||||
self.assertTrue(config.release_shear_onset_enabled)
|
||||
self.assertFalse(config.release_normal_onset_enabled)
|
||||
self.assertGreater(config.release_slope_hold_seconds, 0.2)
|
||||
self.assertGreaterEqual(config.release_curve_history, 24)
|
||||
self.assertGreater(
|
||||
config.release_confirm_samples,
|
||||
GRIP_BEHAVIOR_PROFILES["rapid"]["release_confirm_samples"],
|
||||
)
|
||||
|
||||
apply_grip_behavior_profile(config, "rapid")
|
||||
self.assertFalse(config.release_require_slope_settle)
|
||||
self.assertTrue(config.release_side_shear_enabled)
|
||||
self.assertTrue(config.release_side_normal_enabled)
|
||||
self.assertTrue(config.release_shear_onset_enabled)
|
||||
self.assertTrue(config.release_normal_onset_enabled)
|
||||
self.assertLessEqual(config.release_curve_history, 8)
|
||||
|
||||
def test_apply_to_dict(self):
|
||||
data = {"port": "COM1"}
|
||||
apply_grip_behavior_profile(data, "rapid")
|
||||
self.assertEqual(data["grip_behavior_mode"], "rapid")
|
||||
self.assertEqual(data["release_side_touch_change"], 0.04)
|
||||
self.assertEqual(data["port"], "COM1")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,273 @@
|
||||
import contextlib
|
||||
import io
|
||||
import threading
|
||||
import time
|
||||
import tempfile
|
||||
import json
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from gripper.config.settings import Demo02Config
|
||||
from gripper.control.actuator import ActuatorWorker
|
||||
from gripper.control.controller import SideTriggerGripController
|
||||
from gripper.control.feedback import TactileFeedbackProcessor
|
||||
from test_release_watch import feedback
|
||||
|
||||
|
||||
class Clock:
|
||||
now = 10.0
|
||||
|
||||
def sleep(self, seconds):
|
||||
self.now += seconds
|
||||
|
||||
|
||||
class ReplayActuator:
|
||||
position = 0
|
||||
hold_pos = None
|
||||
completed_at = None
|
||||
|
||||
def __init__(self, clock):
|
||||
self.clock = clock
|
||||
self.commands = []
|
||||
self.force_steps = 0
|
||||
|
||||
@property
|
||||
def motion_status(self):
|
||||
return False, self.completed_at
|
||||
|
||||
def record(self, kind, *args, **kwargs):
|
||||
self.commands.append((self.clock.now, kind, args, kwargs))
|
||||
self.completed_at = self.clock.now
|
||||
|
||||
def open(self, *args, **kwargs):
|
||||
self.record("open", *args, **kwargs)
|
||||
self.position = 0
|
||||
|
||||
def close(self, *args, **kwargs):
|
||||
self.record("close", *args, **kwargs)
|
||||
self.position = 500
|
||||
|
||||
def hold(self, *args, **kwargs):
|
||||
self.record("hold", *args, **kwargs)
|
||||
self.hold_pos = self.position
|
||||
|
||||
def set_force(self, *args, **kwargs):
|
||||
self.record("set_force", *args, **kwargs)
|
||||
self.force_steps += 1
|
||||
|
||||
def move(self, position, *args, **kwargs):
|
||||
self.record("move", position, *args, **kwargs)
|
||||
self.position = position
|
||||
|
||||
|
||||
class ControllerReplayTests(unittest.TestCase):
|
||||
def run_replay(self, curve, *, config=None, duration=1.5, manual=None):
|
||||
config = config or Demo02Config(open_wait_stable_samples=2, grip_start_force=30,
|
||||
hold_force=30, adaptive_force_min=30)
|
||||
clock = Clock()
|
||||
actuator = ReplayActuator(clock)
|
||||
reader = SimpleNamespace(latest=[])
|
||||
processor = SimpleNamespace(
|
||||
normal_converter=SimpleNamespace(unit="N"), shear_converter=SimpleNamespace(unit="N"),
|
||||
reset_filters=lambda: None,
|
||||
read=lambda *_args, **_kwargs: curve(clock.now - 10, actuator),
|
||||
)
|
||||
statuses = []
|
||||
controller = SideTriggerGripController(config, reader, None, processor)
|
||||
controller.actuator = actuator
|
||||
trace_directory = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(trace_directory.cleanup)
|
||||
controller.release_trace_path = Path(trace_directory.name) / "release-last.json"
|
||||
|
||||
def on_status(status):
|
||||
statuses.append((clock.now - 10, status))
|
||||
if manual:
|
||||
manual(controller, len(statuses))
|
||||
if clock.now - 10 >= duration:
|
||||
controller.stop()
|
||||
|
||||
controller.status_callback = on_status
|
||||
if manual:
|
||||
controller.set_control_mode("manual")
|
||||
manual(controller, 0)
|
||||
output = io.StringIO()
|
||||
with patch("gripper.control.controller.time.perf_counter", side_effect=lambda: clock.now), \
|
||||
patch("gripper.control.controller.time.sleep", side_effect=clock.sleep), \
|
||||
contextlib.redirect_stdout(output):
|
||||
controller._run_event_monitor()
|
||||
self.last_release_trace = (json.loads(controller.release_trace_path.read_text(encoding="utf-8"))
|
||||
if controller.release_trace_path.exists() else None)
|
||||
return actuator.commands, statuses, output.getvalue()
|
||||
|
||||
@staticmethod
|
||||
def contact_curve(t, *, shear=0.05, normal=0.5):
|
||||
if t < 0.1:
|
||||
return feedback(normal=0, shear=0, stamp=t)
|
||||
if t < 0.14:
|
||||
return feedback(normal=0.2, right_normal=0, shear=0, stamp=t)
|
||||
return feedback(normal=normal, shear=shear, stamp=t)
|
||||
|
||||
def test_release_wins_over_adaptive_force_request(self):
|
||||
commands, statuses, logs = self.run_replay(
|
||||
lambda t, _: self.contact_curve(t, shear=0.35 if t >= 0.8 else 0.05))
|
||||
releases = [entry for entry in commands if entry[1] == "open"]
|
||||
self.assertEqual(len(releases), 1)
|
||||
self.assertIn("force-change-release-open", releases[0][2])
|
||||
self.assertLessEqual(releases[0][0] - 10 - 0.8, 0.2)
|
||||
self.assertEqual(len([c for c in commands if c[1] == "set_force"]), 1)
|
||||
self.assertIn("release-sudden", logs, "short confirmation must survive log throttling")
|
||||
released_status = next(s for _, s in statuses if s["action"].startswith("force-change-release"))
|
||||
self.assertEqual(released_status["speed_pct"], Demo02Config.release_open_speed)
|
||||
|
||||
def test_release_in_gripping_before_ramp_finishes(self):
|
||||
config = Demo02Config(open_wait_stable_samples=2, grip_start_force=15, hold_force=50)
|
||||
commands, statuses, _ = self.run_replay(
|
||||
lambda t, _: self.contact_curve(t, shear=0.35 if t >= 0.5 else 0.05), config=config)
|
||||
self.assertTrue(any(c[1] == "open" for c in commands))
|
||||
pre_release = [s["state"] for t, s in statuses if t < 0.5]
|
||||
self.assertIn("gripping", pre_release)
|
||||
self.assertNotIn("hold_check", pre_release)
|
||||
|
||||
def test_side_touch_opens_during_initial_contact_guard(self):
|
||||
def curve(t, _):
|
||||
sample = self.contact_curve(t)
|
||||
if 0.2 <= t < 0.4:
|
||||
sample.left_shear = 0.15
|
||||
sample.max_shear = 0.15
|
||||
sample.shear_diff = 0.1
|
||||
return sample
|
||||
|
||||
commands, statuses, _ = self.run_replay(curve, duration=0.5)
|
||||
contact_time = next(t for t, s in statuses if s["action"] == "object-detected-low-force")
|
||||
opens = [c for c in commands if c[1] == "open"]
|
||||
self.assertEqual(len(opens), 1)
|
||||
self.assertLess(opens[0][0] - 10 - contact_time, 0.1)
|
||||
self.assertTrue(any(s["action"] == "force-change-release-open(side_touch)" for _, s in statuses))
|
||||
trace = self.last_release_trace
|
||||
self.assertIsNotNone(trace)
|
||||
self.assertEqual(trace["frames"][-1]["release"]["channel"], "side_shear")
|
||||
self.assertTrue(trace["frames"][-1]["release"]["release"])
|
||||
self.assertGreater(len(trace["frames"]), 2)
|
||||
self.assertIn("release_min_change", trace["config"])
|
||||
|
||||
def test_contact_and_adaptive_spikes_do_not_open(self):
|
||||
def curve(t, actuator):
|
||||
return self.contact_curve(t, normal=0.5 + actuator.force_steps * 0.2, shear=0.6)
|
||||
commands, _, logs = self.run_replay(curve, duration=3)
|
||||
self.assertFalse(any(c[1] == "open" for c in commands))
|
||||
self.assertGreater(len([c for c in commands if c[1] == "set_force"]), 3)
|
||||
self.assertIn("release-grip-settle", logs)
|
||||
self.assertIn("release-self-settle", logs)
|
||||
self.assertIn("release-slow_internal", logs)
|
||||
|
||||
def test_manual_command_paths_remain_manual(self):
|
||||
sequence = {0: ("close", {}), 3: ("hold", {}), 6: ("release", {}),
|
||||
9: ("move_position", {"position": 123}), 12: ("open", {})}
|
||||
|
||||
def manual(controller, count):
|
||||
if count in sequence:
|
||||
command, args = sequence[count]
|
||||
controller.request_manual_command(command, **args)
|
||||
|
||||
commands, statuses, _ = self.run_replay(
|
||||
lambda t, _: feedback(normal=0.5 if round(t * 30) % 2 else 0, shear=t, stamp=t),
|
||||
manual=manual, duration=0.7)
|
||||
self.assertEqual([c[1] for c in commands], ["close", "set_force", "hold", "move", "open"])
|
||||
self.assertFalse(any(s["release_armed"] for _, s in statuses))
|
||||
|
||||
def test_close_timeout_still_opens(self):
|
||||
config = Demo02Config(open_wait_stable_samples=2, close_timeout_seconds=0.3)
|
||||
commands, _, _ = self.run_replay(
|
||||
lambda t, _: feedback(normal=0 if t < 0.1 else 0.2, right_normal=0, shear=0, stamp=t),
|
||||
config=config, duration=0.8)
|
||||
self.assertTrue(any(c[1] == "open" and c[2][1] == "close-timeout-open" for c in commands))
|
||||
|
||||
def test_missing_feedback_cancels_pending_release(self):
|
||||
def curve(t, _):
|
||||
if 0.82 <= t < 1.0:
|
||||
return None
|
||||
return self.contact_curve(t, shear=0.35 if t >= 0.8 else 0.05)
|
||||
commands, statuses, _ = self.run_replay(curve)
|
||||
self.assertFalse(any(c[1] == "open" for c in commands))
|
||||
gaps = [s for _, s in statuses if s["action"] == "waiting-for-sensor-samples"]
|
||||
self.assertTrue(gaps)
|
||||
self.assertTrue(all(not s["release_armed"] for s in gaps))
|
||||
|
||||
def test_current_json_ramp_and_release(self):
|
||||
config = Demo02Config.from_args([])
|
||||
|
||||
def curve(t, actuator):
|
||||
if t < 0.4:
|
||||
return feedback(normal=0, shear=0, stamp=t)
|
||||
if t < 0.5:
|
||||
return feedback(normal=0.3, right_normal=0, shear=0, stamp=t)
|
||||
shear = 0.4 if t >= 4 else 0.05
|
||||
return feedback(normal=0.5 + actuator.force_steps * 0.1, shear=shear, stamp=t)
|
||||
|
||||
commands, _, _ = self.run_replay(curve, config=config, duration=4.4)
|
||||
releases = [c for c in commands if c[1] == "open"]
|
||||
self.assertEqual(len(releases), 1)
|
||||
self.assertGreaterEqual(releases[0][0] - 10, 4)
|
||||
self.assertLessEqual(releases[0][0] - 10 - 4, 0.2)
|
||||
|
||||
|
||||
class FeedbackFreshnessTests(unittest.TestCase):
|
||||
def test_only_fresh_side_advances_its_filter(self):
|
||||
converter = SimpleNamespace(convert=float)
|
||||
processor = TactileFeedbackProcessor(Demo02Config(filter_alpha=0.5), converter, converter)
|
||||
samples = [dict(fnormal=0, fshearx=0, fsheary=0, timestamp=1.0) for _ in range(2)]
|
||||
reader = SimpleNamespace(get_samples=lambda **_: samples)
|
||||
processor.read(reader)
|
||||
samples[0] = dict(fnormal=1, fshearx=0, fsheary=0, timestamp=2.0)
|
||||
first = processor.read(reader)
|
||||
again = processor.read(reader)
|
||||
self.assertEqual(first.left_normal, 0.5)
|
||||
self.assertEqual(again.left_normal, first.left_normal)
|
||||
samples[1] = dict(fnormal=1, fshearx=0, fsheary=0, timestamp=2.1)
|
||||
other = processor.read(reader)
|
||||
self.assertEqual(other.left_normal, 0.5)
|
||||
self.assertEqual(other.right_normal, 0.5)
|
||||
processor.reset_filters()
|
||||
self.assertEqual(processor.read(reader).left_normal, 1)
|
||||
|
||||
|
||||
class WorkerMotionTests(unittest.TestCase):
|
||||
def test_open_drops_queued_grip_force(self):
|
||||
worker = ActuatorWorker(None)
|
||||
worker.set_force(80)
|
||||
worker.hold(80)
|
||||
worker.set_force(90)
|
||||
worker.open(20)
|
||||
self.assertEqual([c.kind for c in worker._commands], ["open"])
|
||||
|
||||
def test_motion_snapshot_covers_queue_and_slow_actual_io(self):
|
||||
entered, finish = threading.Event(), threading.Event()
|
||||
|
||||
def set_force(_):
|
||||
entered.set()
|
||||
finish.wait(timeout=2)
|
||||
|
||||
gripper = SimpleNamespace(connect=lambda: None, read_position=lambda: 0, set_force=set_force)
|
||||
worker = ActuatorWorker(gripper)
|
||||
worker.set_force(30)
|
||||
self.assertEqual(worker.motion_status, (True, None))
|
||||
worker.start()
|
||||
try:
|
||||
self.assertTrue(entered.wait(timeout=1))
|
||||
self.assertEqual(worker.motion_status, (True, None))
|
||||
self.assertFalse(worker.wait_idle(timeout=0.02))
|
||||
before_completion = time.perf_counter()
|
||||
finish.set()
|
||||
self.assertTrue(worker.wait_idle(timeout=1))
|
||||
pending, completed = worker.motion_status
|
||||
self.assertFalse(pending)
|
||||
self.assertGreaterEqual(completed, before_completion)
|
||||
finally:
|
||||
finish.set()
|
||||
worker.stop()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,302 @@
|
||||
import math
|
||||
import random
|
||||
import unittest
|
||||
|
||||
from gripper.config.settings import Demo02Config
|
||||
from gripper.control.release_watch import ReleaseWatch
|
||||
from test_release_watch import feedback
|
||||
|
||||
|
||||
class SettlingCurveTests(unittest.TestCase):
|
||||
def watch(self):
|
||||
cfg = Demo02Config()
|
||||
watch = ReleaseWatch(cfg)
|
||||
watch.contact(0, feedback())
|
||||
for i in range(1, 16):
|
||||
watch.step(i / 30, feedback(stamp=i / 30))
|
||||
return watch
|
||||
|
||||
def test_smooth_rise_flatten_and_return_are_not_a_new_touch(self):
|
||||
for side_only in (False, True):
|
||||
for hz in (20, 30):
|
||||
with self.subTest(side_only=side_only, hz=hz):
|
||||
watch = self.watch()
|
||||
for i in range(1, 2 * hz + 1):
|
||||
t = i / hz
|
||||
value = 0.05 + 0.08 * (1 - math.cos(2 * math.pi * min(t, 1.6) / 1.6))
|
||||
sample = feedback(shear=value, right_shear=0.05 if side_only else value, stamp=0.5 + t)
|
||||
result = watch.step(0.5 + t, sample, motion_pending=side_only)
|
||||
self.assertFalse(result.release, f"false release at t={t:.3f}: {result}")
|
||||
|
||||
def test_new_touch_on_top_of_smooth_settling_still_releases(self):
|
||||
for touch_at in (0.3, 0.8, 1.1):
|
||||
for side_only in (False, True):
|
||||
with self.subTest(touch_at=touch_at, side_only=side_only):
|
||||
watch = self.watch()
|
||||
released_at = None
|
||||
for i in range(1, 48):
|
||||
t = i / 30
|
||||
value = 0.05 + 0.08 * (1 - math.cos(2 * math.pi * t / 1.6))
|
||||
if t + 1e-9 >= touch_at:
|
||||
value += 0.1
|
||||
sample = feedback(shear=value, right_shear=0.05 if side_only else value, stamp=0.5 + t)
|
||||
result = watch.step(0.5 + t, sample, motion_pending=side_only)
|
||||
if result.release:
|
||||
released_at = t
|
||||
break
|
||||
self.assertIsNotNone(released_at)
|
||||
self.assertGreaterEqual(released_at + 1e-9, touch_at)
|
||||
self.assertLessEqual(released_at - touch_at, 0.2)
|
||||
|
||||
def test_large_smooth_change_cannot_bypass_shape_gate_via_weak_path(self):
|
||||
watch = self.watch()
|
||||
watch.config.release_min_change = 0.5
|
||||
for i in range(1, 61):
|
||||
t = i / 30
|
||||
value = 0.05 + 0.6 * (1 - math.cos(math.pi * min(t, 1.5) / 1.5))
|
||||
result = watch.step(0.5 + t, feedback(shear=value, stamp=0.5 + t))
|
||||
self.assertFalse(result.release)
|
||||
|
||||
def test_contact_inherits_the_already_rising_curve(self):
|
||||
watch = ReleaseWatch(Demo02Config())
|
||||
history = [(i / 30, feedback(shear=value, right_shear=0.05, stamp=i / 30))
|
||||
for i, value in enumerate((0.05, 0.08, 0.11))]
|
||||
watch.contact(2 / 30, history[-1][1], history=history)
|
||||
results = []
|
||||
for i, value in enumerate((0.135, 0.153, 0.165, 0.17, 0.168, 0.16, 0.15, 0.14), 3):
|
||||
result = watch.step(i / 30, feedback(shear=value, right_shear=0.05, stamp=i / 30), motion_pending=True)
|
||||
self.assertFalse(result.release)
|
||||
results.append(result)
|
||||
self.assertTrue(any(r.curve_motion == "settling_turn" for r in results))
|
||||
|
||||
def test_high_rate_existing_curve_does_not_bypass_onset_gate(self):
|
||||
for side_only in (False, True):
|
||||
with self.subTest(side_only=side_only):
|
||||
watch = ReleaseWatch(Demo02Config())
|
||||
|
||||
def sample_at(t):
|
||||
value = 0.3 + 4.5 * t - 4.0 * t * t
|
||||
return feedback(shear=value, right_shear=0.05 if side_only else value, stamp=t)
|
||||
|
||||
history = [(i / 30, sample_at(i / 30)) for i in range(3)]
|
||||
watch.contact(2 / 30, history[-1][1], history=history)
|
||||
for i in range(3, 27):
|
||||
result = watch.step(i / 30, sample_at(i / 30))
|
||||
self.assertFalse(result.release)
|
||||
|
||||
def test_side_curve_uses_sensor_time_despite_control_tick_jitter(self):
|
||||
watch = self.watch()
|
||||
receipt = 10.0
|
||||
for i in range(1, 49):
|
||||
t = i / 30
|
||||
receipt += 0.018 if i % 2 else 0.0486666667
|
||||
value = 0.05 + 0.08 * (1 - math.cos(2 * math.pi * t / 1.6))
|
||||
sample = feedback(shear=value, right_shear=0.05, stamp=1000 + t)
|
||||
result = watch.step(receipt, sample, motion_pending=True)
|
||||
self.assertFalse(result.release)
|
||||
|
||||
def test_predictor_cannot_invent_displacement_from_bounded_noise(self):
|
||||
for protected in (False, True):
|
||||
with self.subTest(protected=protected):
|
||||
watch = self.watch()
|
||||
rng = random.Random(17)
|
||||
for i in range(1, 301):
|
||||
t = 0.5 + i / 30
|
||||
sample = feedback(normal=0.5 + rng.uniform(-0.012, 0.012),
|
||||
right_normal=0.5 + rng.uniform(-0.012, 0.012),
|
||||
shear=0.05 + rng.uniform(-0.012, 0.012),
|
||||
right_shear=0.05 + rng.uniform(-0.012, 0.012), stamp=t)
|
||||
result = watch.step(t, sample, motion_pending=protected)
|
||||
self.assertFalse(result.release)
|
||||
|
||||
@staticmethod
|
||||
def noisy_curve(noise, side_only, seed, touch_at=None):
|
||||
rng = random.Random(seed)
|
||||
for i in range(1, 61):
|
||||
t = i / 30
|
||||
value = 0.05 + 0.08 * (1 - math.cos(2 * math.pi * min(t, 1.6) / 1.6))
|
||||
offset = 0.1 if touch_at is not None and t + 1e-9 >= touch_at else 0.0
|
||||
left = value + rng.uniform(-noise, noise) + offset
|
||||
right = 0.05 if side_only else value + rng.uniform(-noise, noise) + offset
|
||||
yield t, feedback(shear=left, right_shear=right, stamp=0.5 + t)
|
||||
|
||||
def test_noise_on_a_smooth_curve_cannot_be_amplified_into_a_touch(self):
|
||||
# Both values move well beyond the amplitude floor over time, but the
|
||||
# only departures from the settling curve are tiny bounded noise.
|
||||
# Three-point extrapolation previously falsely released 56/60 cases.
|
||||
for noise in (0.002, 0.004, 0.006):
|
||||
for side_only in (False, True):
|
||||
for seed in range(10):
|
||||
with self.subTest(noise=noise, side_only=side_only, seed=seed):
|
||||
watch = self.watch()
|
||||
for t, sample in self.noisy_curve(noise, side_only, seed):
|
||||
result = watch.step(0.5 + t, sample, motion_pending=side_only)
|
||||
self.assertFalse(result.release, f"false release at {t:.3f}: {result}")
|
||||
|
||||
def test_noise_near_a_turn_does_not_cancel_a_real_touch(self):
|
||||
for touch_at in (0.3, 0.8, 1.1):
|
||||
for side_only in (False, True):
|
||||
for seed in range(10):
|
||||
with self.subTest(touch_at=touch_at, side_only=side_only, seed=seed):
|
||||
watch = self.watch()
|
||||
released_at = None
|
||||
for t, sample in self.noisy_curve(0.006, side_only, seed, touch_at):
|
||||
result = watch.step(0.5 + t, sample, motion_pending=side_only)
|
||||
if result.release:
|
||||
released_at = t
|
||||
break
|
||||
self.assertIsNotNone(released_at)
|
||||
self.assertGreaterEqual(released_at + 1e-9, touch_at)
|
||||
self.assertLessEqual(released_at - touch_at, 0.2)
|
||||
|
||||
|
||||
class LongHistorySlopeTests(unittest.TestCase):
|
||||
def test_force_curve_fit_slope_decays_as_ramp_flattens(self):
|
||||
from gripper.control.force_curve import ForceCurve
|
||||
|
||||
cfg = Demo02Config(
|
||||
release_curve_history=30,
|
||||
release_slope_window_seconds=0.8,
|
||||
release_min_change=0.04,
|
||||
release_side_touch_change=0.06,
|
||||
)
|
||||
curve = ForceCurve(cfg)
|
||||
rising = []
|
||||
for i in range(1, 36):
|
||||
t = i / 30
|
||||
# Fast early climb then flatten: slope should shrink.
|
||||
value = 0.5 + 0.4 * (1 - math.exp(-4.0 * t))
|
||||
curve.observe(t, value)
|
||||
rising.append(abs(curve.fit_slope))
|
||||
self.assertGreaterEqual(len(curve.history), 30)
|
||||
self.assertGreater(max(rising[:10]), rising[-1])
|
||||
self.assertLess(rising[-1], 0.25)
|
||||
|
||||
def test_stable_mode_blocks_shear_side_touch_before_plateau(self):
|
||||
cfg = Demo02Config()
|
||||
cfg.apply_grip_behavior_mode("stable")
|
||||
cfg.release_confirm_samples = 2
|
||||
watch = ReleaseWatch(cfg)
|
||||
watch.contact(0.0, feedback())
|
||||
for i in range(1, 25):
|
||||
t = i / 30
|
||||
normal = 0.5 + 0.8 * (1 - math.exp(-3.0 * t))
|
||||
sample = feedback(normal=normal, right_normal=normal * 0.95,
|
||||
shear=0.05 + 0.02 * i, right_shear=0.05, stamp=t)
|
||||
result = watch.step(t, sample, motion_pending=True)
|
||||
self.assertFalse(result.release)
|
||||
self.assertTrue(result.reason in {"force-settling", "grip-settle", "self-settle"})
|
||||
|
||||
def test_stable_mode_open_works_during_adaptive_force_after_plateau(self):
|
||||
cfg = Demo02Config()
|
||||
cfg.apply_grip_behavior_mode("stable")
|
||||
cfg.release_confirm_samples = 2
|
||||
cfg.release_min_hold_seconds = 0.05
|
||||
watch = ReleaseWatch(cfg)
|
||||
watch.contact(0.0, feedback())
|
||||
for i in range(1, 55):
|
||||
t = i / 30
|
||||
sample = feedback(normal=0.9, right_normal=0.9, shear=0.05, right_shear=0.05, stamp=t)
|
||||
self.assertFalse(watch.step(t, sample).release)
|
||||
self.assertTrue(watch._plateau_latched)
|
||||
# Adaptive 加力 is still pending, but intentional shear open must work.
|
||||
released = False
|
||||
for i in range(55, 70):
|
||||
t = i / 30
|
||||
sample = feedback(normal=0.95, right_normal=0.95,
|
||||
shear=0.35, right_shear=0.05, stamp=t)
|
||||
result = watch.step(t, sample, motion_pending=True)
|
||||
if result.release:
|
||||
released = True
|
||||
break
|
||||
self.assertTrue(released)
|
||||
|
||||
def test_stable_mode_ignores_normal_open_after_plateau(self):
|
||||
cfg = Demo02Config()
|
||||
cfg.apply_grip_behavior_mode("stable")
|
||||
cfg.release_confirm_samples = 2
|
||||
cfg.release_min_hold_seconds = 0.05
|
||||
watch = ReleaseWatch(cfg)
|
||||
watch.contact(0.0, feedback())
|
||||
for i in range(1, 55):
|
||||
t = i / 30
|
||||
sample = feedback(normal=0.9, right_normal=0.9, shear=0.05, right_shear=0.05, stamp=t)
|
||||
self.assertFalse(watch.step(t, sample).release)
|
||||
for i in range(55, 70):
|
||||
t = i / 30
|
||||
sample = feedback(normal=1.25, right_normal=0.9, shear=0.05, right_shear=0.05, stamp=t)
|
||||
result = watch.step(t, sample)
|
||||
self.assertFalse(result.release, f"normal false open: {result}")
|
||||
|
||||
def test_stable_mode_allows_shear_open_after_plateau(self):
|
||||
cfg = Demo02Config()
|
||||
cfg.apply_grip_behavior_mode("stable")
|
||||
cfg.release_confirm_samples = 2
|
||||
cfg.release_min_hold_seconds = 0.05
|
||||
watch = ReleaseWatch(cfg)
|
||||
watch.contact(0.0, feedback())
|
||||
for i in range(1, 55):
|
||||
t = i / 30
|
||||
sample = feedback(normal=0.9, right_normal=0.9, shear=0.05, right_shear=0.05, stamp=t)
|
||||
self.assertFalse(watch.step(t, sample).release)
|
||||
released = False
|
||||
for i in range(55, 70):
|
||||
t = i / 30
|
||||
sample = feedback(normal=0.9, right_normal=0.9,
|
||||
shear=0.35, right_shear=0.05, stamp=t)
|
||||
result = watch.step(t, sample)
|
||||
if result.release:
|
||||
released = True
|
||||
break
|
||||
self.assertTrue(released)
|
||||
|
||||
def test_intentional_open_after_plateau_not_revoked_by_rising_slope(self):
|
||||
cfg = Demo02Config()
|
||||
cfg.apply_grip_behavior_mode("stable")
|
||||
cfg.release_confirm_samples = 2
|
||||
cfg.release_min_hold_seconds = 0.05
|
||||
watch = ReleaseWatch(cfg)
|
||||
watch.contact(0.0, feedback())
|
||||
for i in range(1, 55):
|
||||
t = i / 30
|
||||
watch.step(t, feedback(normal=0.9, right_normal=0.9, stamp=t))
|
||||
self.assertTrue(watch._plateau_latched)
|
||||
released = False
|
||||
for i in range(55, 75):
|
||||
t = i / 30
|
||||
n = 0.9 + 0.05 * (i - 54)
|
||||
self.assertTrue(watch._force_settled(t))
|
||||
result = watch.step(
|
||||
t,
|
||||
feedback(normal=n, right_normal=0.9,
|
||||
shear=0.05 + 0.03 * (i - 54), right_shear=0.05, stamp=t),
|
||||
motion_pending=True,
|
||||
)
|
||||
if result.release:
|
||||
released = True
|
||||
break
|
||||
self.assertTrue(released)
|
||||
|
||||
def test_short_normal_spike_does_not_count_as_settled_plateau(self):
|
||||
cfg = Demo02Config()
|
||||
cfg.apply_grip_behavior_mode("stable")
|
||||
cfg.release_confirm_samples = 2
|
||||
cfg.release_min_hold_seconds = 0.0
|
||||
cfg.release_slope_hold_seconds = 0.4
|
||||
watch = ReleaseWatch(cfg)
|
||||
watch.contact(0.0, feedback())
|
||||
for i in range(1, 12):
|
||||
t = i / 30
|
||||
n = 0.9 if i >= 6 else 0.1 + 0.15 * i
|
||||
sample = feedback(normal=n, right_normal=n, stamp=t)
|
||||
watch.step(t, sample, motion_pending=True)
|
||||
for i in range(12, 20):
|
||||
t = i / 30
|
||||
sample = feedback(normal=1.2, right_normal=0.9, stamp=t)
|
||||
result = watch.step(t, sample, motion_pending=True)
|
||||
self.assertFalse(result.release)
|
||||
self.assertIn(result.reason, {"force-settling", "grip-settle", "self-settle"})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,347 @@
|
||||
import unittest
|
||||
|
||||
from gripper.common.filters import ExponentialFilter
|
||||
from gripper.config.settings import Demo02Config
|
||||
from gripper.control.feedback import ForceFeedback
|
||||
from gripper.control.release_watch import ReleasePhase, ReleaseWatch
|
||||
|
||||
|
||||
def feedback(normal=0.5, shear=0.05, *, right_normal=None, right_shear=None, stamp=None):
|
||||
rn = normal if right_normal is None else right_normal
|
||||
rs = shear if right_shear is None else right_shear
|
||||
return ForceFeedback(normal, rn, min(normal, rn), max(normal, rn), normal - rn,
|
||||
shear, rs, max(shear, rs), shear - rs,
|
||||
left_timestamp=stamp, right_timestamp=stamp)
|
||||
|
||||
|
||||
class ReleaseReplayTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.config = Demo02Config()
|
||||
self.watch = ReleaseWatch(self.config)
|
||||
self.now = 0.0
|
||||
self.watch.contact(self.now)
|
||||
|
||||
def tick(self, **kwargs):
|
||||
self.now += 1 / 30
|
||||
motion = {k: kwargs.pop(k) for k in tuple(kwargs) if k.startswith("motion_")}
|
||||
kwargs.setdefault("stamp", self.now)
|
||||
return self.watch.step(self.now, feedback(**kwargs), **motion)
|
||||
|
||||
def ready(self):
|
||||
for _ in range(15):
|
||||
result = self.tick()
|
||||
self.assertFalse(result.release)
|
||||
self.assertEqual(self.watch.phase, ReleasePhase.WATCHING)
|
||||
return result
|
||||
|
||||
def test_contact_spikes_and_ringing_do_not_release(self):
|
||||
for i in range(20):
|
||||
result = self.tick(normal=0.5 + (0.3 if i % 2 else 0))
|
||||
self.assertFalse(result.release)
|
||||
if self.now < self.config.release_min_hold_seconds:
|
||||
self.assertEqual(result.reason, "grip-settle")
|
||||
self.ready()
|
||||
|
||||
def test_contact_minimum_is_not_restarted_by_self_motion(self):
|
||||
for i in range(15):
|
||||
result = self.tick(motion_completed_at=0.06)
|
||||
if self.now < self.config.release_min_hold_seconds:
|
||||
self.assertFalse(self.watch.armed)
|
||||
self.assertTrue(self.watch.armed)
|
||||
|
||||
def test_step_then_plateau_confirms_without_second_rate_spike(self):
|
||||
self.ready()
|
||||
result = self.tick(shear=0.2)
|
||||
self.assertEqual(result.reason, "sudden")
|
||||
self.assertFalse(result.release)
|
||||
result = self.tick(shear=0.2)
|
||||
self.assertTrue(result.release)
|
||||
self.assertEqual(result.shear_rate, 0.0)
|
||||
self.assertFalse(self.tick(shear=0.2).release, "release effect must be emitted once")
|
||||
|
||||
def test_normal_unloading_and_either_side_shear(self):
|
||||
for changes in ({"normal": 0.3}, {"right_normal": 0.3},
|
||||
{"shear": 0.25}, {"right_shear": 0.25}):
|
||||
with self.subTest(changes=changes):
|
||||
self.setUp()
|
||||
self.ready()
|
||||
self.tick(**changes)
|
||||
self.assertTrue(self.tick(**changes).release)
|
||||
|
||||
def test_single_frame_pulse_and_filtered_falling_edge_do_not_release(self):
|
||||
for alpha in (1.0, 0.55, 0.65):
|
||||
with self.subTest(alpha=alpha):
|
||||
self.setUp()
|
||||
self.ready()
|
||||
filt = ExponentialFilter(alpha)
|
||||
filt.update(0.05)
|
||||
for raw in [0.85] + [0.05] * 20:
|
||||
result = self.tick(shear=filt.update(raw))
|
||||
self.assertFalse(result.release)
|
||||
self.assertTrue(self.watch.armed)
|
||||
|
||||
def test_reversal_or_channel_swap_cannot_combine_votes(self):
|
||||
self.ready()
|
||||
self.tick(shear=0.3, right_shear=0.05)
|
||||
result = self.tick(shear=0.05, right_shear=0.3)
|
||||
self.assertFalse(result.release)
|
||||
self.assertEqual(result.reason, "settling-turn")
|
||||
|
||||
def test_slow_internal_drift_never_accumulates_into_release(self):
|
||||
self.ready()
|
||||
for i in range(1, 601):
|
||||
result = self.tick(normal=0.5 + i * 0.005, shear=0.05 + i * 0.003)
|
||||
self.assertFalse(result.release)
|
||||
self.assertEqual(result.reason, "slow_internal")
|
||||
|
||||
def test_medium_rate_push_uses_separate_longer_confirmation(self):
|
||||
# Exercise the legacy weak path independently of the new touch path.
|
||||
self.config.release_min_change = 0.4
|
||||
self.ready()
|
||||
counts = []
|
||||
# A new, large but sub-sudden-rate onset followed by a plateau.
|
||||
# A large displacement accumulated along a smooth curve is no longer
|
||||
# sufficient evidence for this path.
|
||||
self.now += 0.19
|
||||
result = self.watch.step(self.now, feedback(shear=0.34, stamp=self.now))
|
||||
counts.append(result.confirm_count)
|
||||
for i in range(10):
|
||||
result = self.tick(shear=0.34)
|
||||
if result.confirm_count:
|
||||
counts.append(result.confirm_count)
|
||||
if result.release:
|
||||
break
|
||||
self.assertTrue(result.release)
|
||||
self.assertEqual(result.reason, "slow_external")
|
||||
self.assertEqual(counts, [1, 2, 3, 4, 5])
|
||||
|
||||
def test_noise_floor_rejects_tiny_high_rate_change(self):
|
||||
self.ready()
|
||||
for i in range(10):
|
||||
self.now += 0.001
|
||||
result = self.watch.step(self.now, feedback(shear=0.05 + (0.01 if i % 2 else 0), stamp=self.now))
|
||||
self.assertFalse(result.release)
|
||||
self.assertNotEqual(result.phase, ReleasePhase.CONFIRMING)
|
||||
|
||||
def test_queued_delayed_command_and_tail_are_protected(self):
|
||||
self.ready()
|
||||
for i in range(12):
|
||||
result = self.tick(normal=0.5 + i * 0.1, motion_pending=True)
|
||||
self.assertEqual(result.reason, "self-settle")
|
||||
self.assertFalse(result.release)
|
||||
completed = self.now
|
||||
for i in range(4):
|
||||
result = self.tick(normal=1.8, motion_completed_at=completed)
|
||||
self.assertFalse(result.release)
|
||||
self.assertFalse(result.allow_force_update)
|
||||
for _ in range(10):
|
||||
result = self.tick(normal=1.8, motion_completed_at=completed)
|
||||
self.assertFalse(result.release)
|
||||
self.assertTrue(result.allow_force_update)
|
||||
self.tick(normal=1.8, shear=0.3, motion_completed_at=completed)
|
||||
self.assertTrue(self.tick(normal=1.8, shear=0.3, motion_completed_at=completed).release)
|
||||
|
||||
def test_repeated_self_steps_leave_observation_window(self):
|
||||
self.ready()
|
||||
completed = None
|
||||
force = 0.5
|
||||
commands, command_times = 0, []
|
||||
for _ in range(100):
|
||||
result = self.tick(normal=force, motion_completed_at=completed)
|
||||
self.assertFalse(result.release)
|
||||
if result.allow_force_update:
|
||||
commands += 1
|
||||
command_times.append(self.now)
|
||||
force += 0.15
|
||||
completed = self.now
|
||||
self.assertGreater(commands, 3)
|
||||
self.assertTrue(all(b - a >= 0.2 for a, b in zip(command_times, command_times[1:])))
|
||||
|
||||
def test_cached_frames_and_opposite_side_updates_do_not_confirm(self):
|
||||
self.ready()
|
||||
result = self.tick(shear=0.3, right_shear=0.05)
|
||||
stamp = self.now
|
||||
for i in range(1, 5):
|
||||
self.now += 1 / 30
|
||||
sample = feedback(shear=0.3, right_shear=0.05, stamp=stamp)
|
||||
sample.right_timestamp = self.now
|
||||
result = self.watch.step(self.now, sample)
|
||||
self.assertFalse(result.release)
|
||||
self.assertEqual(result.phase, ReleasePhase.CONFIRMING)
|
||||
|
||||
def test_stale_gap_and_timestamp_restart_discard_candidate(self):
|
||||
for restarted in (False, True):
|
||||
with self.subTest(restarted=restarted):
|
||||
self.setUp()
|
||||
self.ready()
|
||||
self.tick(shear=0.3)
|
||||
self.now += 0.4
|
||||
result = self.watch.step(self.now, feedback(shear=0.3, stamp=0.0 if restarted else self.now))
|
||||
self.assertFalse(result.release)
|
||||
self.assertEqual(result.reason, "sample-gap")
|
||||
for _ in range(15):
|
||||
self.assertFalse(self.tick(shear=0.3).release)
|
||||
|
||||
def test_expired_candidate_cannot_release_on_late_second_frame(self):
|
||||
self.config.release_confirm_window_seconds = 0.05
|
||||
self.ready()
|
||||
self.tick(shear=0.3)
|
||||
self.now += 0.08
|
||||
result = self.watch.step(self.now, feedback(shear=0.3, stamp=self.now))
|
||||
self.assertFalse(result.release)
|
||||
self.assertEqual(result.reason, "confirm-expired")
|
||||
|
||||
def test_filtered_release_latency_at_30hz_and_60hz_control(self):
|
||||
for control_hz in (30, 60):
|
||||
for alpha in (0.55, 0.65):
|
||||
with self.subTest(control_hz=control_hz, alpha=alpha):
|
||||
self.setUp()
|
||||
self.config.control_hz = control_hz
|
||||
self.ready()
|
||||
filt = ExponentialFilter(alpha)
|
||||
filt.update(0.05)
|
||||
onset = self.now
|
||||
result = None
|
||||
for frame in range(6):
|
||||
stamp = self.now + 1 / 30
|
||||
filtered = filt.update(0.35)
|
||||
for _ in range(control_hz // 30):
|
||||
self.now += 1 / control_hz
|
||||
result = self.watch.step(self.now, feedback(shear=filtered, stamp=stamp))
|
||||
if result.release:
|
||||
break
|
||||
if result.release:
|
||||
break
|
||||
self.assertTrue(result.release)
|
||||
self.assertLessEqual(self.now - onset, 0.2)
|
||||
|
||||
def test_reset_discards_previous_grip(self):
|
||||
self.ready()
|
||||
self.tick(shear=0.3)
|
||||
self.watch.reset()
|
||||
self.assertFalse(self.tick(shear=0.3).release)
|
||||
self.assertEqual(self.watch.phase, ReleasePhase.DISARMED)
|
||||
self.watch.contact(self.now)
|
||||
self.assertEqual(self.tick(shear=0.3).reason, "grip-settle")
|
||||
|
||||
def test_suspend_discards_candidate_until_fresh_curve_settles(self):
|
||||
self.ready()
|
||||
self.tick(shear=0.3)
|
||||
self.watch.suspend()
|
||||
self.assertFalse(self.watch.armed)
|
||||
for _ in range(15):
|
||||
self.assertFalse(self.tick(shear=0.3).release)
|
||||
self.assertTrue(self.watch.armed)
|
||||
self.tick(shear=0.5)
|
||||
self.assertTrue(self.tick(shear=0.5).release)
|
||||
|
||||
def test_stopped_side_cannot_supply_quiet_curve_from_duplicate_frames(self):
|
||||
self.ready()
|
||||
self.tick(normal=0.8, motion_pending=True)
|
||||
stamp = self.now
|
||||
for i in range(1, 20):
|
||||
self.now += 1 / 30
|
||||
sample = feedback(normal=0.8, stamp=stamp)
|
||||
sample.right_timestamp = self.now
|
||||
result = self.watch.step(self.now, sample)
|
||||
self.assertFalse(result.release)
|
||||
self.assertFalse(self.watch.armed)
|
||||
|
||||
def test_small_continuous_noise_cannot_lock_contact_protection(self):
|
||||
for i in range(30):
|
||||
result = self.tick(normal=0.5 + (0.012 if i % 2 else -0.012))
|
||||
self.assertFalse(result.release)
|
||||
self.assertTrue(self.watch.armed)
|
||||
self.tick(normal=0.512, shear=0.13)
|
||||
self.assertTrue(self.tick(normal=0.488, shear=0.13).release)
|
||||
|
||||
def test_light_filtered_touch_below_sudden_rate_releases(self):
|
||||
self.ready()
|
||||
filt = ExponentialFilter(0.65)
|
||||
filt.update(0.05)
|
||||
onset = self.now
|
||||
for _ in range(6):
|
||||
result = self.tick(shear=filt.update(0.12))
|
||||
if result.release:
|
||||
break
|
||||
self.assertTrue(result.release)
|
||||
self.assertLessEqual(self.now - onset, 0.2 + 1e-9)
|
||||
|
||||
def test_gentle_push_is_not_absorbed_as_slow_reference_drift(self):
|
||||
self.ready()
|
||||
onset = self.now
|
||||
for i in range(1, 7):
|
||||
result = self.tick(shear=0.05 + 0.01 * i)
|
||||
if result.release:
|
||||
break
|
||||
self.assertTrue(result.release)
|
||||
self.assertLessEqual(self.now - onset, 0.2 + 1e-9)
|
||||
|
||||
def test_noise_after_force_command_does_not_extend_protection_forever(self):
|
||||
self.ready()
|
||||
completed = self.now
|
||||
for i in range(15):
|
||||
result = self.tick(normal=0.7 + (0.012 if i % 2 else -0.012),
|
||||
motion_completed_at=completed)
|
||||
self.assertFalse(result.release)
|
||||
self.assertTrue(self.watch.armed)
|
||||
|
||||
def test_side_touch_does_not_wait_for_contact_timer_or_serial(self):
|
||||
self.watch.contact(self.now, feedback())
|
||||
first = self.tick(shear=0.15, right_shear=0.05, motion_pending=True)
|
||||
self.assertEqual(first.reason, "side_touch")
|
||||
self.assertFalse(first.release)
|
||||
second = self.tick(shear=0.15, right_shear=0.05, motion_pending=True)
|
||||
self.assertTrue(second.release)
|
||||
self.assertLess(self.now, 0.1)
|
||||
self.assertLess(self.now, self.config.release_min_hold_seconds)
|
||||
|
||||
def test_bilateral_contact_shear_is_not_early_side_touch(self):
|
||||
self.watch.contact(self.now, feedback())
|
||||
for i in range(10):
|
||||
result = self.tick(normal=0.5 + i * 0.1, shear=0.05 + i * 0.06,
|
||||
motion_pending=True)
|
||||
self.assertFalse(result.release)
|
||||
|
||||
def test_normal_self_offset_is_not_replayed_after_cooldown(self):
|
||||
self.ready()
|
||||
completed = self.now
|
||||
for _ in range(15):
|
||||
result = self.tick(normal=0.8, right_normal=0.5, motion_completed_at=completed)
|
||||
self.assertFalse(result.release)
|
||||
|
||||
def test_early_side_touch_requires_fresh_both_sides(self):
|
||||
self.watch.contact(self.now, feedback())
|
||||
self.tick(shear=0.15, right_shear=0.05, motion_pending=True)
|
||||
stamp = self.now
|
||||
for _ in range(4):
|
||||
self.now += 1 / 30
|
||||
sample = feedback(shear=0.15, right_shear=0.05, stamp=stamp)
|
||||
sample.left_timestamp = self.now
|
||||
result = self.watch.step(self.now, sample, motion_pending=True)
|
||||
self.assertFalse(result.release)
|
||||
|
||||
def test_early_filtered_contact_pulse_does_not_open(self):
|
||||
for alpha in (0.55, 0.65):
|
||||
for peak in (0.85, 2.05, 8.05):
|
||||
with self.subTest(alpha=alpha, peak=peak):
|
||||
self.setUp()
|
||||
self.watch.contact(self.now, feedback())
|
||||
filt = ExponentialFilter(alpha)
|
||||
filt.update(0.05)
|
||||
for raw in [peak] + [0.05] * 20:
|
||||
result = self.tick(shear=filt.update(raw), right_shear=0.05, motion_pending=True)
|
||||
self.assertFalse(result.release)
|
||||
|
||||
def test_slow_creep_does_not_permanently_prevent_force_updates(self):
|
||||
self.ready()
|
||||
allowed = 0
|
||||
for i in range(1, 60):
|
||||
result = self.tick(normal=0.5 + 0.005 * i)
|
||||
self.assertFalse(result.release)
|
||||
allowed += result.allow_force_update
|
||||
self.assertGreater(allowed, 40)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user