diff --git a/README.md b/README.md index 2799d11..ceaa810 100644 --- a/README.md +++ b/README.md @@ -114,7 +114,8 @@ LiteGrip(channel="can0", can_id=0x08, mst_id=None, canfd_mode=None, | `connect()` → `bool` | Opens the CAN session. The `with` block does this for you. | | `disconnect()` → `None` | Closes it, disabling the motor unless `disable_on_disconnect=False`. | | `poll(timeout_s=0.0)` → `bool` | Reads one frame; `True` if one arrived. | -| `get_state(wait=True)` → `GripperState` | Position, velocity, torque, temperature, error bits. `wait=True` blocks for a fresh frame. | +| `get_state(wait=True)` → `GripperState` | Position, velocity, torque, temperature, error bits, plus how fresh they are. `wait=True` waits up to 50 ms for the frame that updates them — it does not guarantee one. | +| `refresh_status()` → `bool` | Asks the motor for a status frame (0xCC) without moving the jaws. Use it before `get_state()` when the gripper has been idle. | | `get_position()` / `get_position_rad()` → `float` | Measured opening, in mm / rad. | | `get_force()` → `float` | `get_torque()` × 10, in N. | | `get_torque()` → `float` | Measured torque, Nm. | @@ -124,6 +125,35 @@ LiteGrip(channel="can0", can_id=0x08, mst_id=None, canfd_mode=None, | `is_moving()` / `is_grasped()` → `bool` | Motion and grip predicates. | | `wait_for_ready(timeout=5.0)` → `bool` | Blocks until the motor accepts commands. | +The readings `get_state()` returns are a cache, and only a status frame updates it. A DM motor +answers the control frames it receives, so a gripper that is being driven reports on its own; one +holding a pose, or disabled, sends nothing, and then `wait=True` waits out its 50 ms and hands back +the previous frame. The snapshot says which happened: + +```python +state = gripper.get_state() +if not state.fresh: + gripper.refresh_status() # 0xCC: ask for a frame; the jaws do not move + state = gripper.get_state() + +if state.is_stale(): # older than 0.5 s, or never had a frame + raise RuntimeError(f"no status frame for {state.data_age_s} s") +print(state.position_mm, state.force_n, state.timestamp) +``` + +- `fresh` — a frame answered *this* read. False means the numbers are the previous frame's, however + old that is. +- `data_age_s` — seconds since the frame behind the snapshot arrived, or `None` when none ever has. +- `timestamp` — when that frame arrived, not when you asked. `0.0` means there is no frame at all and + every reading is a constructor default. **Do not** time a loop with it; it is for a human reading a + log, and the age arithmetic is what `data_age_s` is for. +- `is_stale(stale_after_s=STALE_AFTER_S)` — age past the threshold (`STALE_AFTER_S` is 0.5 s). It is + about the snapshot, not the read: a snapshot 20 ms old can still be `fresh=False`. + +`fresh` and `is_stale()` answer different questions, and neither substitutes for the other — the wait +window is 50 ms while the staleness threshold is 0.5 s, so a read no frame answered can still look +young. Ask both when the difference matters. + ### Enable and faults | Method | What it does | diff --git a/readme_zn.md b/readme_zn.md index 1841075..154ec6c 100644 --- a/readme_zn.md +++ b/readme_zn.md @@ -108,7 +108,8 @@ LiteGrip(channel="can0", can_id=0x08, mst_id=None, canfd_mode=None, | `connect()` → `bool` | 打开 CAN 会话。`with` 块已经替你做了。 | | `disconnect()` → `None` | 关闭会话；除非 `disable_on_disconnect=False`，否则会失能电机。 | | `poll(timeout_s=0.0)` → `bool` | 读一帧；读到返回 `True`。 | -| `get_state(wait=True)` → `GripperState` | 位置、速度、力矩、温度、错误位。`wait=True` 阻塞等一帧新数据。 | +| `get_state(wait=True)` → `GripperState` | 位置、速度、力矩、温度、错误位，以及这些读数有多新。`wait=True` 最多等 50 ms 等那一帧，但不等于是等到了。 | +| `refresh_status()` → `bool` | 主动要一帧状态（0xCC），夹爪不动。夹爪空闲过一阵就先调它再 `get_state()`。 | | `get_position()` / `get_position_rad()` → `float` | 实测开口，单位 mm / rad。 | | `get_force()` → `float` | `get_torque()` × 10，单位 N。 | | `get_torque()` → `float` | 实测力矩，Nm。 | @@ -118,6 +119,31 @@ LiteGrip(channel="can0", can_id=0x08, mst_id=None, canfd_mode=None, | `is_moving()` / `is_grasped()` → `bool` | 运动与夹持判据。 | | `wait_for_ready(timeout=5.0)` → `bool` | 阻塞到电机可以接受指令为止。 | +`get_state()` 返回的读数是缓存，只有状态帧能更新它。DM 电机是「收到控制帧就回一帧状态」，所以 +正在被驱动的夹爪会自己报数；而保持姿势的、失能的夹爪什么都不发，这时 `wait=True` 等满 50 ms +之后把上一帧交回来。快照会说明是哪一种情况： + +```python +state = gripper.get_state() +if not state.fresh: + gripper.refresh_status() # 0xCC：要一帧，夹爪不动 + state = gripper.get_state() + +if state.is_stale(): # 超过 0.5 s，或者从来就没有过帧 + raise RuntimeError(f"no status frame for {state.data_age_s} s") +print(state.position_mm, state.force_n, state.timestamp) +``` + +- `fresh` —— 这一次读有没有帧答过。False 表示这些数字是上一帧的，不管它多旧。 +- `data_age_s` —— 快照背后那一帧到现在多少秒；从来就没有过帧时是 `None`。 +- `timestamp` —— 那一帧到达的时刻，不是你调用的时刻。`0.0` 表示根本没有帧，读数全是构造默 + 认值。**别**拿它给循环计时 —— 它是给人看日志的，算 age 用 `data_age_s`。 +- `is_stale(stale_after_s=STALE_AFTER_S)` —— 年龄超过阈值（`STALE_AFTER_S` 是 0.5 s）。 + 它说的是快照，不是这次读：一个 20 ms 前的快照也可以是 `fresh=False`。 + +`fresh` 和 `is_stale()` 回答的是两个问题，谁也替代不了谁 —— 等帧窗口是 50 ms，而陈旧阈值是 +0.5 s，所以一次没人答过的读也可以看着很「新」。两个都问。 + ### 使能与故障 | 方法 | 做什么 | diff --git a/src/litegrip/__init__.py b/src/litegrip/__init__.py index d9fee15..ea4e963 100644 --- a/src/litegrip/__init__.py +++ b/src/litegrip/__init__.py @@ -89,6 +89,7 @@ def _detect_version(dist_name: str = "litegrip") -> str: # ── Data models ───────────────────────────────────────────────────────── from .models import ( + STALE_AFTER_S, GripperState, GripperConfig, GripperInfo, @@ -220,6 +221,7 @@ def __dir__(): "clamp_to_travel", "force_approach_terms", # Models + "STALE_AFTER_S", "GripperState", "GripperConfig", "GripperInfo", diff --git a/src/litegrip/can/motor.py b/src/litegrip/can/motor.py index 5089565..1f310f6 100644 --- a/src/litegrip/can/motor.py +++ b/src/litegrip/can/motor.py @@ -7,6 +7,7 @@ from __future__ import annotations +import time from dataclasses import dataclass, field from enum import IntEnum @@ -62,6 +63,7 @@ class MotorState: "_t_coil", "rx_count", "_last_update", + "_last_frame_time", ) def __init__(self, params: MotorParams): @@ -74,6 +76,7 @@ def __init__(self, params: MotorParams): self._t_coil: int = 0 self.rx_count: int = 0 self._last_update: float = 0.0 + self._last_frame_time: float = 0.0 # ── properties ────────────────────────────────────────────────────── @@ -131,8 +134,22 @@ def is_fault(self) -> bool: @property def last_update(self) -> float: + """Monotonic time of the newest status frame; ``0.0`` if none ever came. + + Monotonic because that is the clock the receive path stamps frames with + (``CanFrame.timestamp``) — subtract from ``time.monotonic()`` for an age. + """ return self._last_update + @property + def last_frame_time(self) -> float: + """Wall-clock time of the newest status frame; ``0.0`` if none ever came. + + The human-readable counterpart of :attr:`last_update`. The two clocks + are not interchangeable: this one cannot be subtracted from. + """ + return self._last_frame_time + # ── control mode management ───────────────────────────────────────── def set_mode(self, mode: ControlMode) -> None: @@ -157,6 +174,11 @@ def update_from_status(self, position: float, velocity: float, self._t_mos = t_mos self._t_coil = t_coil self._last_update = timestamp + # Same instant, other clock: ``timestamp`` is the monotonic stamp the + # transport put on the frame, and a snapshot that reports "when" to a + # human needs the wall clock. Sampling it here keeps the two within a + # few microseconds of the frame that produced them. + self._last_frame_time = time.time() self.rx_count += 1 def __repr__(self) -> str: diff --git a/src/litegrip/gripper.py b/src/litegrip/gripper.py index 403d708..0bb913d 100644 --- a/src/litegrip/gripper.py +++ b/src/litegrip/gripper.py @@ -1920,12 +1920,20 @@ def _bounded_move(target: float, label: str) -> None: # ═══════════════════════════════════════════════════════════════════ def get_state(self, wait: bool = True) -> GripperState: - """Return the current gripper state. + """Return the current gripper state, saying how fresh it is. Args: - wait: If True (default), waits up to 50 ms for a fresh status - frame. If False, returns immediately with the last cached - state (suitable for high-frequency control loops). + wait: If True (default), waits up to 50 ms for a status frame. + If False, returns immediately with the last cached state + (suitable for high-frequency control loops). + + ``wait=True`` asks for a frame; it does not guarantee one. A motor + answers a *control* frame, so an idle bus — one holding a pose, or + disabled — says nothing, and the wait ends on the cached readings. The + returned snapshot says which happened: ``fresh`` is False when no frame + arrived, ``data_age_s`` is the age of the frame the readings did come + from, and ``timestamp`` is that frame's arrival time, not this call's. + Call :meth:`refresh_status` first when the bus has been idle. Returns: GripperState snapshot. @@ -1936,9 +1944,9 @@ def get_state(self, wait: bool = True) -> GripperState: with self._io_lock: if wait: - self._can.update_state(timeout_s=0.05) + answered = self._can.update_state(timeout_s=0.05) else: - self._can.poll(timeout_s=0.0) + answered = self._can.poll(timeout_s=0.0) position_rad = self._can.get_position() velocity_rad_s = self._can.get_velocity() @@ -1946,6 +1954,16 @@ def get_state(self, wait: bool = True) -> GripperState: error_code = self._can.get_error() t_mos, t_coil = self._can.get_temperature() + # Both stamps come from the frame, not from this call. A duck-typed + # transport need not carry them, and then there is nothing to say: + # None age means "no frame to date the readings by". + last_update = getattr(self._can, "last_update", None) + frame_time = getattr(self._can, "last_frame_time", None) + # last_update is the monotonic stamp the receive path put on the + # frame, so the age has to be measured on that clock. + data_age_s = (None if last_update is None + else time.monotonic() - last_update) + # pos_closed_rad = closed (0 mm), pos_open_rad = open (max mm). # Which way the rad count runs depends on the mount, so close_sign # sets the sign; the result is 0 mm at the closed limit and +stroke @@ -1963,11 +1981,32 @@ def get_state(self, wait: bool = True) -> GripperState: temperature_mos=t_mos, temperature_coil=t_coil, error_code=error_code, - timestamp=time.time(), + # 0.0 = no frame ever arrived (see GripperState.timestamp). + timestamp=0.0 if frame_time is None else frame_time, position_mm=position_mm, force_n=force_n, + fresh=bool(answered), + data_age_s=data_age_s, ) + def refresh_status(self) -> bool: + """Ask the motor for a status frame (0xCC). Does not move the jaws. + + Use it before :meth:`get_state` on an idle bus. A motor answers the + control frames it receives, so a gripper holding a pose sends nothing on + its own and the status link goes quiet; this request makes it speak. + + Returns: + True if the request went out. False when there is no connection or + the write failed — the same reporting as ``enable()``, since this + asks rather than commands. + """ + self._check_connected() + if self._can is None: + return False + with self._io_lock: + return self._can.refresh_status() + def get_position(self) -> float: """Current position in mm.""" return self.get_state().position_mm diff --git a/src/litegrip/models.py b/src/litegrip/models.py index 7c687df..a72293c 100644 --- a/src/litegrip/models.py +++ b/src/litegrip/models.py @@ -2,9 +2,8 @@ from __future__ import annotations -from dataclasses import dataclass, field +from dataclasses import dataclass from enum import IntEnum, IntFlag -import time from typing import Optional from .constants import GripperGeometry, UnitConversion @@ -28,11 +27,21 @@ class GripperStatus(IntFlag): ERROR = 0x10 +#: Age past which a :class:`GripperState` counts as stale, in seconds. The same +#: threshold the C++ SDK uses, so a guard reads the same on either side. +STALE_AFTER_S = 0.5 + + @dataclass class GripperState: - """Live gripper state snapshot. - - All fields are updated on each :meth:`LiteGrip.get_state` call. + """Gripper state snapshot, and how fresh it is. + + The readings are a cache, updated only by a status frame. Whether such a + frame arrived is what :attr:`fresh`, :attr:`data_age_s` and + :meth:`is_stale` report — a snapshot from a bus that said nothing is not a + measurement, and it must not look like one. Check them before trusting the + numbers; :meth:`LiteGrip.refresh_status` elicits a frame when the bus is + idle. """ position_rad: float = 0.0 @@ -41,12 +50,35 @@ class GripperState: temperature_mos: int = 0 temperature_coil: int = 0 error_code: int = 0 - timestamp: float = field(default_factory=time.time) + #: Wall-clock time the frame behind this snapshot arrived — not the time of + #: the call that returned it. ``0.0`` means no frame has ever arrived, so + #: every reading above is still a constructor default. + timestamp: float = 0.0 # Convenience — computed from raw values with unit conversion position_mm: float = 0.0 force_n: float = 0.0 + #: Whether a status frame answered *this* read. False means the readings + #: are the ones from the previous frame, however old that is. + fresh: bool = False + #: Seconds since the frame behind this snapshot arrived, or ``None`` when + #: there is no such frame at all. + data_age_s: Optional[float] = None + + def is_stale(self, stale_after_s: float = STALE_AFTER_S) -> bool: + """True when the newest frame is older than *stale_after_s*, or absent. + + This is about the snapshot's age, not about the read that produced it: + a read no frame answered has ``fresh=False`` while the snapshot may + still be younger than the threshold. Ask both questions when the + difference matters — "how old is this" and "did anyone answer just now". + + ``stale_after_s`` defaults to :data:`STALE_AFTER_S`, the same threshold + the C++ SDK uses. + """ + return self.data_age_s is None or self.data_age_s > stale_after_s + @property def is_enabled(self) -> bool: """True if the motor is enabled (error_code == 1).""" diff --git a/src/litegrip/protocols/can_bus.py b/src/litegrip/protocols/can_bus.py index 1a6ec53..e9b3286 100644 --- a/src/litegrip/protocols/can_bus.py +++ b/src/litegrip/protocols/can_bus.py @@ -507,12 +507,69 @@ def last_send_error(self) -> Optional[Exception]: """ return self._last_send_error + @property + def last_update(self) -> Optional[float]: + """Monotonic time of the newest status frame, or ``None`` if none ever + arrived since the motor was registered. + + The readings :meth:`get_position` and friends return are a cache, and + the cache is only as good as this stamp. Subtract from + ``time.monotonic()`` for an age; do not treat the value as a wall-clock + time (see :attr:`last_frame_time` for that one). + """ + return self._motor_stamp("last_update") + + @property + def last_frame_time(self) -> Optional[float]: + """Wall-clock time of the newest status frame, or ``None``. + + The stamp to put on a snapshot a human reads. + """ + return self._motor_stamp("last_frame_time") + + def _motor_stamp(self, name: str) -> Optional[float]: + """Read one of the motor's frame stamps, mapping "never" to ``None``. + + ``0.0`` is what an un-updated :class:`MotorState` holds, and it is not a + time — a caller comparing it against a clock would get a nonsense age + rather than "no frame yet". + """ + motor = self._motor + if motor is None: + return None + stamp = getattr(motor, name, 0.0) + return stamp if stamp else None + def update_state(self, timeout_s: float = 0.05) -> bool: """Poll until at least one new status frame arrives.""" if self._controller is None or self._motor is None: return False return self._controller.poll_until(self._motor, timeout_s=timeout_s) + def refresh_status(self) -> bool: + """Ask the motor for a status frame (0xCC). + + A DM motor answers a control frame with a status frame, so a gripper + that is being driven reports on its own. One that is idle — holding a + pose, sitting disabled — sends nothing, and then + :meth:`update_state` has nothing to hear and the readings stay cached + however long the caller waits. This is the request that gives it + something to answer, and it does not change the motor's output. + + Returns: + True if the request went out, False when nothing is registered or + the write failed (the same ``False``-with-a-warning shape as + :meth:`enable` — this is a request, not a motion command). + """ + if self._controller is None or self._motor is None: + return False + try: + self._controller.refresh_status(self._motor) + return True + except Exception as e: + log.warning("状态帧请求下发失败（%s）：%r", self._channel, e) + return False + def get_position(self) -> float: """Current position in rad.""" if self._motor is None: diff --git a/tests/fake_can.py b/tests/fake_can.py index 3f1733b..7d71c18 100644 --- a/tests/fake_can.py +++ b/tests/fake_can.py @@ -20,6 +20,7 @@ from __future__ import annotations from dataclasses import dataclass +import time from typing import List, Optional import _sdkpath # noqa: F401 (把 src/ 插进 sys.path) @@ -222,6 +223,20 @@ def __init__(self, pos: float = 0.0, block_rad: Optional[float] = None, self.send_error: Optional[Exception] = None #: ``LiteGripCAN.last_send_error`` 的对应物：上一次下发失败的原因。 self.last_send_error: Optional[Exception] = None + #: 总线上还有没有状态帧 —— False = 空闲：没被驱动，电机不主动开口， + #: poll / update_state 都回 False，读数就是上一次缓存的那份。 + self.delivers = True + #: ``refresh_status()`` 的 0xCC 发出去了没有（False = 写失败）。 + self.refresh_ok = True + #: ``refresh_status()`` 被调了几次。 + self.refresh_calls = 0 + #: 要到的那一帧还没被读走 —— 下一次 poll / update_state 会读到它。 + self.refresh_pending = False + #: 最近一帧的两个时间戳。真层也是两条钟：``last_update`` 是 monotonic + #: （量 age 用），``last_frame_time`` 是 wall clock（人读的 timestamp + #: 用）。None = 还没有过帧，和真层 ``MotorState`` 的 0.0 一个意思。 + self.last_update: Optional[float] = None + self.last_frame_time: Optional[float] = None # ── 运动 ─────────────────────────────────────────────────────────── def control_mit(self, q_target, kp, kd, dq_target=0.0, @@ -256,10 +271,30 @@ def control_mit_stream(self, q_target, kp, kd, duration_s=0.5, # ── 状态 ─────────────────────────────────────────────────────────── def poll(self, timeout_s: float = 0.0) -> bool: - return True + return self._frame_available() def update_state(self, timeout_s: float = 0.05) -> bool: - return True + return self._frame_available() + + def _frame_available(self) -> bool: + """读到一帧了吗 —— 被驱动时总有，空闲时只有 ``refresh_status()`` 要到过。 + + 读到就把那对时间戳更新成「刚刚」；没读到就原样留着，于是读数仍然是 + 上一帧的，age 会如实涨上去。 + """ + if self.delivers or self.refresh_pending: + self.refresh_pending = False + self.last_update = time.monotonic() + self.last_frame_time = time.time() + return True + return False + + def refresh_status(self) -> bool: + """0xCC：电机收到就回一帧。空闲总线上这是唯一让它开口的办法。""" + self.refresh_calls += 1 + if self.refresh_ok: + self.refresh_pending = True + return self.refresh_ok def get_position(self) -> float: return self.motor.reported_pos() diff --git a/tests/test_state_staleness.py b/tests/test_state_staleness.py new file mode 100644 index 0000000..0a97983 --- /dev/null +++ b/tests/test_state_staleness.py @@ -0,0 +1,171 @@ +"""``get_state()`` 要说清那份读数是不是刚有帧答过。 + +``get_state(wait=True)`` 等 50 ms，然后**无论有没有等到帧**都把缓存读出来，还盖上 +``timestamp=time.time()``。于是空闲总线上的读数——没被驱动，电机根本不开口——看起来 +和一次测量一模一样：没有异常、没有告警、时间戳是「现在」。真机上这就是 +``g.stop()`` 之后连着十次 ``get_state()`` 都报出一个没人产生的「当前」值。 + +这个文件钉住新契约：``fresh`` 说这次读到帧没有，``data_age_s`` 说读数是哪一帧的， +``timestamp`` 是那一帧到达的时刻（不是这次调用的时刻），``is_stale()`` 把「多久了」 +和阈值比；以及给调用方一条出路 —— ``refresh_status()`` 发 0xCC 主动要一帧， +空闲总线上这是唯一让夹爪开口的办法（C++ 侧有，这边之前没有）。 + +CAN 层用真 ``LiteGripCAN`` + 假 controller，夹爪层用 ``fake_can.make_gripper`` +（``delivers=False`` 就是「总线安静」）。 +""" + +from __future__ import annotations + +import time +import unittest + +import _sdkpath # noqa: F401 +from fake_can import make_gripper +from litegrip import NotInitializedError, STALE_AFTER_S, GripperState +from litegrip.protocols.can_bus import LiteGripCAN + + +class FakeMotor: + """鸭子类型即可 —— 缺席 ``limits`` 时帧量程退回 DM4310（±10 Nm）。""" + + +class RecordingController: + """记下 0xCC 请求，别的什么都不做。""" + + def __init__(self, fail: bool = False): + self.refreshed = [] + self.fail = fail + + def refresh_status(self, motor): + if self.fail: + raise OSError(100, "Network is down") + self.refreshed.append(motor) + + +def make_can(controller=None): + can = LiteGripCAN(channel="vcan0") + if controller is not None: + can._motor = FakeMotor() + can._controller = controller + can._connected = True + return can + + +def age_the_last_frame(fake, seconds: float) -> None: + """把替身「最近一帧」的两个时间戳一起往后退 *seconds* 秒。""" + fake.last_update = time.monotonic() - seconds + fake.last_frame_time = time.time() - seconds + + +class TestGetStateSaysWhetherAFrameAnswered(unittest.TestCase): + + def test_a_frame_marks_the_snapshot_fresh(self): + g, _ = make_gripper() + + s = g.get_state() + + self.assertTrue(s.fresh) + self.assertIsNotNone(s.data_age_s) + self.assertLess(s.data_age_s, STALE_AFTER_S) + self.assertFalse(s.is_stale()) + + def test_an_idle_bus_returns_the_cache_and_says_so(self): + g, fake = make_gripper() + g.get_state() # 上一帧 + fake.delivers = False # 从这儿起没有帧了 + age_the_last_frame(fake, 2.0) + + before = time.time() + s = g.get_state(wait=True) + + self.assertFalse(s.fresh, "没等到帧，却报成刚读到的") + self.assertGreater(s.data_age_s, 1.0) + self.assertTrue(s.is_stale()) + self.assertLess(s.timestamp, before - 1.0, + "时间戳盖成了这次调用的时刻 —— 一个没人答过的读数因此" + "看起来是「现在」的") + + def test_a_bus_that_never_spoke_has_no_frame_to_date_it_by(self): + g, fake = make_gripper() + fake.delivers = False # 一帧都没有过 + + s = g.get_state(wait=True) + + self.assertFalse(s.fresh) + self.assertIsNone(s.data_age_s, "没有帧就没有 age —— 不是「很老」") + self.assertTrue(s.is_stale()) + self.assertEqual(s.timestamp, 0.0, + "0.0 才表示「没有帧」，盖个当前时刻就是谎报") + + def test_wait_false_reports_whether_the_poll_found_a_frame(self): + g, fake = make_gripper() + + fake.delivers = False + self.assertFalse(g.get_state(wait=False).fresh) + + fake.delivers = True + self.assertTrue(g.get_state(wait=False).fresh) + + def test_the_threshold_is_the_callers_to_pick(self): + s = GripperState(data_age_s=0.3, fresh=False) + + self.assertFalse(s.is_stale()) + self.assertTrue(s.is_stale(stale_after_s=0.1)) + self.assertEqual(STALE_AFTER_S, 0.5, "和 C++ 侧的阈值对齐") + + +class TestRefreshStatusElicitsAFrame(unittest.TestCase): + + def test_the_request_reaches_the_motor(self): + ctrl = RecordingController() + can = make_can(ctrl) + + self.assertTrue(can.refresh_status()) + self.assertEqual(len(ctrl.refreshed), 1) + + def test_an_unregistered_can_reports_false(self): + self.assertFalse(make_can().refresh_status()) + + def test_a_failed_write_reports_false(self): + can = make_can(RecordingController(fail=True)) + + with self.assertLogs("litegrip.can_bus", level="WARNING"): + self.assertFalse(can.refresh_status()) + + def test_a_disconnected_gripper_raises(self): + g, _ = make_gripper() + g._connected = False + + with self.assertRaises(NotInitializedError): + g.refresh_status() + + def test_the_gripper_delegates(self): + g, fake = make_gripper() + + self.assertTrue(g.refresh_status()) + self.assertEqual(fake.refresh_calls, 1) + + def test_a_refresh_gives_an_idle_bus_something_to_answer(self): + """请求 → 再读：空闲总线上那唯一的出路要真的走得通。""" + g, fake = make_gripper() + g.get_state() + fake.delivers = False + age_the_last_frame(fake, 3.0) + self.assertTrue(g.get_state(wait=True).is_stale()) + + self.assertTrue(g.refresh_status()) + + s = g.get_state(wait=True) + self.assertTrue(s.fresh) + self.assertFalse(s.is_stale()) + + def test_the_request_sends_no_control_frame(self): + g, fake = make_gripper() + + g.refresh_status() + + self.assertEqual(fake.frames, [], "要一帧状态不等于发一帧控制") + + +if __name__ == "__main__": + unittest.main()