feat: add gripper trajectory record and replay

README.md

@@ -197,6 +197,72 @@ the frame the slave followed — a caller can show the jaws without opening a se
 - Follow gains default to the calibration's `kp` / `kd` (`100.0` / `2.0` out of the box); override
   with `kp=` / `kd=`.
 
+## Trajectory record and replay
+
+A motion you teach by hand can be captured once and repeated later. Recording puts the motor into
+zero-gravity so you can push the jaws through the motion; replay streams the captured openings back
+as MIT command frames. What is stored is the normalized opening in `[0, 1]`, exactly as teleop
+sends it, so a trajectory taught on one gripper replays on another with a different mount or
+calibration.
+
+```python
+from litegrip import LiteGrip
+
+with LiteGrip("can0") as gripper:
+    gripper.load_calibration()
+    gripper.enable()
+
+    taught = gripper.record(5.0)   # 5 s of hand-teaching; the jaws are slack
+    taught.save("pick")            # ~/.litegrip/trajectories/pick.lgt
+    gripper.play(taught)           # repeat it
+```
+
+| Method | Behaviour |
+| --- | --- |
+| `record(duration_s, rate_hz=100.0, zero_gravity=True)` | Blocking hand-teach. Returns the `Trajectory`. |
+| `record_start(rate_hz=100.0, zero_gravity=True, max_samples=None)` | Background recording; returns the status snapshot. |
+| `record_stop(allow_empty=False)` | Stops and returns the captured `Trajectory`. |
+| `play(trajectory, speed=1.0, kp=None, kd=None, align=True)` | Blocking replay. `loop` must be `False`. |
+| `play_start(trajectory, speed=1.0, kp=None, kd=None, loop=False, align=True)` | Background replay. |
+| `play_stop(timeout=2.0)` | Stops a replay and leaves the gripper holding. |
+| `trajectory_status()` | One snapshot for both directions. `active`, `kind`, `samples` and `error` are always there; a recording adds `rate_hz`, `zero_gravity` and `loop_hz`, a replay adds `frames`, `speed`, `openness` and `completed`. |
+
+`examples/trajectory.py` runs the same thing from the command line:
+
+```bash
+python3 examples/trajectory.py --record 5 --save pick   # hand-teach, then save
+python3 examples/trajectory.py --list                   # no hardware needed
+python3 examples/trajectory.py --play pick --repeat 3
+```
+
+`Trajectory.save("pick")` writes `~/.litegrip/trajectories/pick.lgt`; a name with a path separator
+in it is used as written. Set `LITEGRIP_TRAJ_DIR` to move that directory. `Trajectory.load("pick")`
+reads it back, and `--list` prints one line per file. The format is compact binary with an 8-byte
+magic header, and a file whose length does not match the sample count in its header is rejected
+rather than parsed into half a trajectory.
+
+- **Replay commands position, not force.** The recorded torque is stored for diagnostics and never
+  fed forward, so a squeeze recorded against an object repeats as a position path that presses with
+  whatever `kp` yields. The grip force you taught is not preserved — follow the replay with
+  `grasp(force_n=...)` if it matters.
+- **`record()` is exclusive and the jaws are slack for its whole duration.** It streams zero-torque
+  frames itself, so do not drive the gripper from the caller while it runs, and keep a hand on it:
+  nothing is holding the jaws.
+- **Record without zero-gravity when something else drives.** `record_start(zero_gravity=False)`
+  only *reads* state, so the caller may run a `grasp()` or a move sequence from another thread and
+  capture it. That is the way to record a programmatic motion.
+- **A capture that did not fill raises.** `record()` reports how many samples it got instead of
+  returning a short recording as if it were whole, and a sampling loop that died is never reported
+  as a good capture.
+- **A blocking `play()` returns with only one hold frame sent.** The motor self-locks a
+  communication-loss fault about 100 ms after the frames stop, so call the next action promptly —
+  or use `play_start(loop=True)` with `play_stop()` for a hold that lasts. A trajectory of one
+  sample is a pose with nothing to repeat, so looping it holds that opening.
+- **Recording, replay and teleop are mutually exclusive.** All three own the CAN I/O, and starting
+  a second one raises `TeleopBusyError` or `TrajectoryBusyError`. `disconnect()` stops whichever is
+  running.
+- Record and play both require a loaded calibration: without one the normalized opening is a guess.
+
 ## The six actions
 
 These are the supported entry points for moving the gripper. Each one verifies its own

examples/trajectory.py

@@ -0,0 +1,193 @@
+#!/usr/bin/env python3
+"""Teach a gripper a motion by hand and play it back.
+
+This is a runnable companion to the trajectory section of the README.
+
+    # Hand-teach 5 seconds and save it as "pick":
+    python3 examples/trajectory.py --record 5 --save pick
+
+    # List what has been saved, with no hardware attached:
+    python3 examples/trajectory.py --list
+
+    # Play it back three times:
+    python3 examples/trajectory.py --play pick --repeat 3
+
+During ``--record`` the motor goes into zero-gravity and the jaws are yours to
+push: take the part, move it through the approach, the squeeze and the release,
+and the samples are taken as you go.  Keep a hand on the gripper — nothing is
+holding the jaws while it is slack, and whatever is between them will drop.
+
+A saved trajectory stores the opening normalised by *this* unit's travel, so it
+replays on a gripper with a different mount or calibration.  What it does not
+store is force: replay commands position, with the gains you give it.  A squeeze
+recorded against an object repeats as a position path, not as the same grip
+force — use ``--kp`` and follow it with ``gripper.grasp(force_n=...)`` if the
+force matters.
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+import time
+
+from litegrip import LiteGrip, LiteGripError, Trajectory, trajectory_dir
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Record a gripper motion by hand, or replay a saved one.")
+    action = parser.add_mutually_exclusive_group(required=True)
+    action.add_argument(
+        "--record", type=float, metavar="SECONDS",
+        help="hand-teach for this many seconds (the jaws go slack)")
+    action.add_argument(
+        "--play", metavar="NAME",
+        help="replay a saved trajectory (a bare name, or a path to a .lgt file)")
+    action.add_argument(
+        "--list", action="store_true",
+        help="list the saved trajectories and exit; needs no hardware")
+
+    parser.add_argument(
+        "--channel", default="can0", help="CAN interface (default: can0)")
+    parser.add_argument(
+        "--can-id", type=lambda s: int(s, 0), default=0x08,
+        help="motor CAN ID (default: 0x08)")
+    parser.add_argument(
+        "--mount", choices=("normal", "reverse"), default=None,
+        help="load a mount template instead of this channel's calibration")
+    parser.add_argument(
+        "--save", metavar="NAME",
+        help="save the recording under this name (default: show it, save nothing)")
+    parser.add_argument(
+        "--rate", type=float, default=100.0,
+        help="samples per second while recording (default: 100)")
+    parser.add_argument(
+        "--speed", type=float, default=1.0,
+        help="playback speed multiplier; 0.5 is half speed (default: 1.0)")
+    parser.add_argument(
+        "--kp", type=float, default=None,
+        help="replay stiffness (default: the gripper's configured kp)")
+    parser.add_argument(
+        "--kd", type=float, default=None,
+        help="replay damping (default: the gripper's configured kd)")
+    parser.add_argument(
+        "--no-align", action="store_true",
+        help="replay: do not move to the first sample before following")
+    parser.add_argument(
+        "--repeat", type=int, default=1,
+        help="replay this many times (default: 1)")
+    parser.add_argument(
+        "--dry-run", action="store_true",
+        help="print the resolved plan and exit without touching hardware")
+    return parser
+
+
+def _describe(traj: Trajectory) -> str:
+    return (f"{len(traj)} samples, {traj.duration:.2f}s at {traj.sample_hz:.0f}Hz, "
+            f"mount={traj.mount}, travel="
+            f"{abs(traj.pos_open_rad - traj.pos_closed_rad) * traj.rad_to_mm:.1f}mm")
+
+
+def _list_saved() -> int:
+    """Print the saved trajectories.  Reads the directory, not the bus."""
+    import glob
+    import os
+
+    root = trajectory_dir()
+    paths = sorted(glob.glob(os.path.join(root, "*.lgt")))
+    if not paths:
+        print(f"no trajectories in {root}")
+        return 0
+    for path in paths:
+        try:
+            traj = Trajectory.load(path)
+        except LiteGripError as error:
+            # One unreadable file must not hide the rest of the list.
+            print(f"{os.path.basename(path)}: unreadable ({error})")
+            continue
+        print(f"{os.path.basename(path):<24} {_describe(traj)}")
+    return 0
+
+
+def _record(gripper: LiteGrip, args: argparse.Namespace) -> int:
+    print(f"recording {args.record}s at {args.rate}Hz — the jaws are slack now, "
+          f"push them through the motion")
+    traj = gripper.record(args.record, rate_hz=args.rate)
+    print(f"recorded: {_describe(traj)}")
+    if args.save:
+        written = traj.save(args.save)
+        print(f"saved to {written}")
+    else:
+        # Say so rather than letting the caller assume a file exists.
+        print("not saved (pass --save NAME to keep it)")
+    return 0
+
+
+def _play(gripper: LiteGrip, traj: Trajectory, args: argparse.Namespace) -> int:
+    for run in range(1, max(1, args.repeat) + 1):
+        status = gripper.play(
+            traj, speed=args.speed, kp=args.kp, kd=args.kd,
+            align=not args.no_align)
+        print(f"replay {run}/{args.repeat}: {status['frames']} frames, "
+              f"ended at openness {status['openness']:.3f}")
+        if run < args.repeat:
+            time.sleep(0.2)
+    return 0
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+
+    if args.list:
+        return _list_saved()
+
+    # Read the file before anything touches the bus: a mistyped name is worth
+    # finding out about without connecting, and the dry run should say which
+    # trajectory it means, not just which name was typed.
+    traj = None
+    if args.play is not None:
+        traj = Trajectory.load(args.play)
+        print(f"loaded {args.play}: {_describe(traj)}")
+
+    gripper = LiteGrip(channel=args.channel, can_id=args.can_id)
+    if args.mount is not None:
+        gripper.load_calibration(template=args.mount)
+    else:
+        gripper.load_calibration()
+    print(f"mount={gripper.mount} closed={gripper.config.pos_closed_rad:+.4f} "
+          f"open={gripper.config.pos_open_rad:+.4f} "
+          f"rad_to_mm={gripper.config.rad_to_mm}")
+
+    if args.dry_run:
+        what = (f"record {args.record}s at {args.rate}Hz"
+                if args.record is not None
+                else f"replay {args.play} x{args.repeat} at speed {args.speed}")
+        print(f"dry run: would {what} on {args.channel} "
+              f"(can_id=0x{args.can_id:02X}); nothing sent")
+        return 0
+
+    gripper.connect()
+    gripper.enable()
+    try:
+        if args.record is not None:
+            return _record(gripper, args)
+        return _play(gripper, traj, args)
+    finally:
+        # The blocking calls already hold the last position, but a Ctrl+C
+        # mid-call would otherwise leave the session claimed and the jaws slack.
+        gripper.play_stop()
+        gripper.disconnect()
+
+
+if __name__ == "__main__":
+    try:
+        sys.exit(main())
+    except LiteGripError as error:
+        print(f"error: {error}", file=sys.stderr)
+        sys.exit(1)
+    except FileNotFoundError as error:
+        # A mistyped --play name or --save directory; say which file, not a
+        # traceback the user has to read to find out.
+        print(f"error: no such file: {error.filename}", file=sys.stderr)
+        sys.exit(1)

readme_zn.md

@@ -174,6 +174,63 @@ python3 examples/teleop.py --mode slave  --channel can0 --host 192.168.1.20
   驱动之前就拒掉。
 - 跟随增益默认取标定里的 `kp` / `kd`（出厂是 `100.0` / `2.0`），用 `kp=` / `kd=` 覆盖。
 
+## 轨迹录制与回放
+
+你用手教一遍的动作可以录下来，之后反复重放。录制时电机进零重力，你直接掰爪子走完整个动作；
+回放把录到的张开度按 MIT 指令帧发回去。存的是归一化到 `[0, 1]` 的张开度，和遥操线上传的是同一个
+量，所以在一台夹爪上教出来的轨迹，换一台装法不同、标定不同的夹爪也能重放。
+
+```python
+from litegrip import LiteGrip
+
+with LiteGrip("can0") as gripper:
+    gripper.load_calibration()
+    gripper.enable()
+
+    taught = gripper.record(5.0)   # 手把手教 5 秒，期间爪子是卸力的
+    taught.save("pick")            # ~/.litegrip/trajectories/pick.lgt
+    gripper.play(taught)           # 重放
+```
+
+| 方法 | 行为 |
+| --- | --- |
+| `record(duration_s, rate_hz=100.0, zero_gravity=True)` | 阻塞式手把手录制，返回 `Trajectory`。 |
+| `record_start(rate_hz=100.0, zero_gravity=True, max_samples=None)` | 后台录制，返回状态快照。 |
+| `record_stop(allow_empty=False)` | 停止并返回录到的 `Trajectory`。 |
+| `play(trajectory, speed=1.0, kp=None, kd=None, align=True)` | 阻塞式回放。`loop` 必须是 `False`。 |
+| `play_start(trajectory, speed=1.0, kp=None, kd=None, loop=False, align=True)` | 后台回放。 |
+| `play_stop(timeout=2.0)` | 停止回放，并让夹爪持位。 |
+| `trajectory_status()` | 两个方向共用一个快照。`active`、`kind`、`samples`、`error` 一直都在；录制时另有 `rate_hz`、`zero_gravity`、`loop_hz`，回放时另有 `frames`、`speed`、`openness`、`completed`。 |
+
+`examples/trajectory.py` 在命令行做同样的事：
+
+```bash
+python3 examples/trajectory.py --record 5 --save pick   # 手把手录一段再存盘
+python3 examples/trajectory.py --list                   # 不需要接硬件
+python3 examples/trajectory.py --play pick --repeat 3
+```
+
+`Trajectory.save("pick")` 写到 `~/.litegrip/trajectories/pick.lgt`；带路径分隔符的名字按原样
+使用。目录可以用 `LITEGRIP_TRAJ_DIR` 改。`Trajectory.load("pick")` 读回来，`--list` 每个文件
+打一行。格式是紧凑二进制，开头 8 字节魔数；文件长度和头部声明的采样数对不上的会被拒绝，而不是
+解析出半截轨迹。
+
+- **回放的是位置，不是力。** 录到的力矩只是诊断信息，不会前馈下发，所以对着物体挤出来的那段，
+  重放时是一条位置轨迹，按 `kp` 顶上去 —— 你教的那个夹持力不会复现。力重要的话，回放完再调
+  `grasp(force_n=...)`。
+- **`record()` 期间独占，而且爪子全程卸力。** 它自己持续发零力矩帧，所以录制期间不要再从调用方
+  驱动夹爪，也要用手扶着：这期间没有任何东西托着爪子。
+- **有别的东西在驱动时，用非零重力模式录。** `record_start(zero_gravity=False)` 只读状态，调用方
+  可以在另一个线程里跑 `grasp()` 或一串运动，把它录下来。要录程序化的动作就走这条路。
+- **没录满会报错。** `record()` 会说清只录到几拍，而不是把一段短录制当成完整结果返回；采样循环
+  死掉也不会被报成一次好录制。
+- **阻塞式 `play()` 返回时只发了一帧持位。** 电机在停帧约 100 ms 后会因通信丢失自锁，所以要接着
+  调下一个动作 —— 想要持续持位就用 `play_start(loop=True)` 配 `play_stop()`。只有一拍采样的轨迹
+  是一个姿势、没有可循环的行程，循环它就等于一直保持那个张开度。
+- **录制、回放、遥操三者互斥。** 它们都独占 CAN 读写，起第二个会抛 `TeleopBusyError` 或
+  `TrajectoryBusyError`。`disconnect()` 会把正在跑的那个停掉。
+- 录制和回放都要求已加载标定：没有标定，归一化的张开度算不出来。
+
 ## 六个动作接口
 
 要让夹爪动起来就用这六个。每一个都会自己校验结果再报成功，所以调用方不必再重写斜坡和

src/litegrip/__init__.py

@@ -131,6 +131,23 @@ def _detect_version(dist_name: str = "litegrip") -> str:
     teleop_topic,
 )
 
+# ── Trajectory record and replay ────────────────────────────────────────
+from .trajectory import (
+    Trajectory,
+    TrajectorySample,
+    TrajectoryRecorder,
+    TrajectoryPlayer,
+    trajectory_dir,
+    resolve_path,
+    DEFAULT_RATE_HZ,
+    TrajectoryError,
+    TrajectoryBusyError,
+    TrajectoryNotActiveError,
+    TrajectoryEmptyError,
+    TrajectoryRecordingError,
+    TrajectoryFormatError,
+)
+
 # ── CAN subpackage (expert) ─────────────────────────────────────────────
 from . import can
 
@@ -222,6 +239,20 @@ def __dir__():
     "Listener",
     "Connector",
     "LatestSlot",
+    # Trajectory record and replay
+    "Trajectory",
+    "TrajectorySample",
+    "TrajectoryRecorder",
+    "TrajectoryPlayer",
+    "trajectory_dir",
+    "resolve_path",
+    "DEFAULT_RATE_HZ",
+    "TrajectoryError",
+    "TrajectoryBusyError",
+    "TrajectoryNotActiveError",
+    "TrajectoryEmptyError",
+    "TrajectoryRecordingError",
+    "TrajectoryFormatError",
     # Subpackages
     "can",
 ]