去掉仿真运动序列的单步超时

偏航转过 90° 时轴仍在运动,20 秒超时会把全部轴停掉。pose move 测试通过。
This commit is contained in:
ykcheungab
2026-10-07 16:21:25 +08:00
parent cc848a273d
commit ff8288bcf9
2 changed files with 26 additions and 14 deletions
@@ -27,7 +27,7 @@ from __future__ import annotations
import time import time
from collections.abc import Mapping, Sequence from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field from dataclasses import dataclass
from PySide6.QtCore import QObject, Signal from PySide6.QtCore import QObject, Signal
@@ -43,10 +43,6 @@ from .units import user_units_to_deg
SETTLE_MM = 0.01 SETTLE_MM = 0.01
SETTLE_DEG = 0.01 SETTLE_DEG = 0.01
#: Per-step patience. A step that is neither arriving nor moving this long is
#: treated as stuck: reported and abandoned, with no further axes commanded.
STEP_TIMEOUT_S = 20.0
#: How long a step may report itself idle while still short of its target before it #: How long a step may report itself idle while still short of its target before it
#: counts as failed — a drive that has stopped without arriving will never arrive. #: counts as failed — a drive that has stopped without arriving will never arrive.
STALL_GRACE_S = 2.0 STALL_GRACE_S = 2.0
@@ -103,7 +99,6 @@ class MoveState:
finished: bool = False finished: bool = False
failure: str = "" failure: str = ""
cancelled: bool = False cancelled: bool = False
_step_started_at: float = field(default=0.0)
_idle_since: float | None = None _idle_since: float | None = None
@property @property
@@ -205,7 +200,7 @@ class PoseMover(QObject):
return MotionResult.refused("没有要移动的轴 / nothing to move") return MotionResult.refused("没有要移动的轴 / nothing to move")
stamp = time.monotonic() if now is None else now stamp = time.monotonic() if now is None else now
self.state = MoveState(plan=plan, started_at=stamp, _step_started_at=stamp) self.state = MoveState(plan=plan, started_at=stamp)
self.started.emit(plan) self.started.emit(plan)
self._issue_current() self._issue_current()
return MotionResult.success(plan.describe()) return MotionResult.success(plan.describe())
@@ -229,7 +224,6 @@ class PoseMover(QObject):
if step is None: if step is None:
self._complete() self._complete()
return return
self.state._step_started_at = time.monotonic()
self.state._idle_since = None self.state._idle_since = None
if not step.units == step.units: # NaN: no calibration for this role if not step.units == step.units: # NaN: no calibration for this role
self._fail(f"{step.role} 无法换算成用户单位 / cannot convert {step.role}") self._fail(f"{step.role} 无法换算成用户单位 / cannot convert {step.role}")
@@ -258,12 +252,6 @@ class PoseMover(QObject):
if step is None: if step is None:
self._complete() self._complete()
return return
now = time.monotonic()
if now - self.state._step_started_at > STEP_TIMEOUT_S:
self._fail(
f"{step.role} 等待超时 / timed out waiting for {step.role}"
)
return
state = snapshot.axis(step.axis) state = snapshot.axis(step.axis)
if state is None: if state is None:
@@ -292,6 +280,7 @@ class PoseMover(QObject):
return return
# Idle, but not there. A drive that has stopped short will not resume, so # Idle, but not there. A drive that has stopped short will not resume, so
# this is a failure rather than something to keep waiting for. # this is a failure rather than something to keep waiting for.
now = time.monotonic()
if self.state._idle_since is None: if self.state._idle_since is None:
self.state._idle_since = now self.state._idle_since = now
elif now - self.state._idle_since > STALL_GRACE_S: elif now - self.state._idle_since > STALL_GRACE_S:
+23
View File
@@ -245,6 +245,29 @@ def test_the_mover_waits_while_an_axis_is_still_moving():
assert mover.state.index == 0 assert mover.state.index == 0
def test_a_long_yaw_is_not_cut_off_while_it_is_still_moving(monkeypatch):
"""A yaw past 90° can take longer than a minute. Keep it while the axis moves.
The old step cap abandoned the plan after 20s and the window then stopped
every axis, so a 180° yaw never arrived.
"""
clock = {"t": 1_000.0}
monkeypatch.setattr(
"orisys_autocal.motion.pose_move.time.monotonic", lambda: clock["t"]
)
service = make_service(
limits={**LIMITS, "max_yaw_deg": 180.0, "max_pitch_deg": 360.0}
)
poller = FakePoller(service)
mover = PoseMover(service, poller.submit)
assert mover.start(plan_pose_move(service, {"yaw": 180.0})).ok
clock["t"] += 120.0
mover.on_snapshot(snapshot_of(service, {3: 1_000.0}, idle=False))
assert mover.running
assert mover.state.failure == ""
assert len(poller.commands) == 1
def test_the_mover_fails_when_an_axis_stops_short(): def test_the_mover_fails_when_an_axis_stops_short():
"""A drive that has stopped is not going to arrive: say so, do not wait.""" """A drive that has stopped is not going to arrive: say so, do not wait."""
service = make_service() service = make_service()