Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 31 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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. |
Expand All @@ -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 |
Expand Down
28 changes: 27 additions & 1 deletion readme_zn.md
Original file line number Diff line number Diff line change
Expand Up @@ -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。 |
Expand All @@ -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,所以一次没人答过的读也可以看着很「新」。两个都问。

### 使能与故障

| 方法 | 做什么 |
Expand Down
2 changes: 2 additions & 0 deletions src/litegrip/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,7 @@ def _detect_version(dist_name: str = "litegrip") -> str:

# ── Data models ─────────────────────────────────────────────────────────
from .models import (
STALE_AFTER_S,
GripperState,
GripperConfig,
GripperInfo,
Expand Down Expand Up @@ -220,6 +221,7 @@ def __dir__():
"clamp_to_travel",
"force_approach_terms",
# Models
"STALE_AFTER_S",
"GripperState",
"GripperConfig",
"GripperInfo",
Expand Down
22 changes: 22 additions & 0 deletions src/litegrip/can/motor.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

from __future__ import annotations

import time
from dataclasses import dataclass, field
from enum import IntEnum

Expand Down Expand Up @@ -62,6 +63,7 @@ class MotorState:
"_t_coil",
"rx_count",
"_last_update",
"_last_frame_time",
)

def __init__(self, params: MotorParams):
Expand All @@ -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 ──────────────────────────────────────────────────────

Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand Down
53 changes: 46 additions & 7 deletions src/litegrip/gripper.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -1936,16 +1944,26 @@ 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()
torque_nm = self._can.get_torque()
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
Expand All @@ -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
Expand Down
44 changes: 38 additions & 6 deletions src/litegrip/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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)."""
Expand Down
57 changes: 57 additions & 0 deletions src/litegrip/protocols/can_bus.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
Loading
Loading