[python] Add direct LeRobot capture writer

docs/docs/pypaimon/multimodal-api.mdx

@@ -694,6 +694,66 @@ together, and keep writers paused until the call returns.
 Scalars map to scalar types, vectors to `VECTOR`, higher-rank tensors to nested
 `ARRAY`, and images to `BLOB`. Images keep their compressed bytes.
 
+## Capture LeRobot frames directly into Paimon
+
+`PaimonLeRobotWriter` implements the write-side surface used by LeRobot's
+recording loop without first creating a LeRobot Parquet/image dataset. Pass the
+same user feature mapping that would be passed to `LeRobotDataset.create`.
+The writer adds the standard `timestamp`, `frame_index`, `episode_index`,
+`index`, and `task_index` features itself.
+
+```python
+from pypaimon.multimodal.lerobot import PaimonLeRobotWriter
+
+writer = PaimonLeRobotWriter(
+    conn,
+    "robot_data",
+    fps=30,
+    features=dataset_features,
+)
+
+# LeRobot's record_loop only needs writer.fps, writer.features, and
+# writer.add_frame(frame), so the writer can be passed as its dataset argument.
+record_loop(..., fps=30, dataset=writer)
+writer.save_episode()
+
+# Optional durability/visibility boundary before finalize.
+writer.flush()
+writer.finalize()
+```
+
+Like native LeRobot, `add_frame` requires every declared user feature plus a
+string `task`, and rejects caller-provided generated fields. Numeric features
+must be NumPy arrays (Torch tensors are converted) with the declared dtype and
+shape. Image metadata uses `(channels, height, width)`; image values may be CHW,
+HWC, or PIL. Images are encoded as PNG bytes and stored in Paimon `BLOB`
+columns; no LeRobot data directory or MP4 is created. `video` features remain
+unsupported.
+
+`save_episode` accepts the current episode and writes it to a long-lived Paimon
+batch writer. The default `episodes_per_commit=-1` keeps all completed episodes
+in that batch until `finalize()`. Set a positive threshold for periodic commits,
+or call `flush()` at an operational boundary. `save_episode`, `flush`, and
+`finalize` always return `None`. Calling `save_episode()` without any buffered
+frames raises `ValueError`, matching native LeRobot.
+
+Call `clear_episode_buffer()` before `save_episode()` to discard a re-recorded
+episode without advancing frame or episode indices. Once `save_episode()`
+accepts an episode, `clear_episode_buffer()` no longer affects it, even when the
+batch has not yet been committed. `finalize()` rejects an unfinished episode
+instead of silently dropping its frames.
+
+The writer creates a missing table and appends to an existing compatible table.
+Before writing, it requires the table columns, order, Arrow types, nullability,
+and LeRobot feature metadata to match. On resume, new `index` and
+`episode_index` values continue after the existing maxima, existing task
+mappings are retained, and the episode-local `frame_index` starts again at zero.
+Each commit records the next global indices and task mapping in Snapshot
+properties. Resume normally reads only that metadata; a non-empty table created
+before these properties existed is scanned once and upgraded by its next
+commit. A commit exception has an unknown result and is not automatically
+retried.
+
 ## Overwrite
 
 `overwrite` accepts the same input formats as `add` and replaces existing data

paimon-python/pypaimon/multimodal/arrow_utils.py

@@ -26,6 +26,11 @@ def strict_arrow_table(
         source_path,
         batch_index,
         format_name):
+    """Validate one Arrow batch against the exact target table schema.
+
+    Reject missing, extra, or reordered columns, validate nested nullability,
+    and apply only Arrow safe casts before returning a ``pyarrow.Table``.
+    """
     if isinstance(data, pa.RecordBatch):
         table = pa.Table.from_batches([data])
     elif isinstance(data, pa.Table):

paimon-python/pypaimon/multimodal/lerobot/__init__.py

@@ -14,11 +14,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""One-time LeRobot Dataset v3 import into a multimodal Paimon table."""
+"""LeRobot Dataset v3 import and direct Paimon capture."""
 
 from pypaimon.multimodal.lerobot.api import load_from_lerobot
+from pypaimon.multimodal.lerobot.writer import PaimonLeRobotWriter
 
 
 __all__ = [
+    "PaimonLeRobotWriter",
     "load_from_lerobot",
 ]

paimon-python/pypaimon/multimodal/lerobot/loader.py

@@ -450,7 +450,7 @@ def _image_bytes(value, root):
     return _encode_media_frame(value)
 
 
-def _encode_media_frame(value):
+def _encode_media_frame(value, channel_first=None):
     try:
         import numpy as np
         from PIL import Image
@@ -466,7 +466,10 @@ def _encode_media_frame(value):
         if callable(detach):
             value = detach().cpu().numpy()
         array = np.asarray(value)
-        if array.ndim == 3 and array.shape[0] in (1, 3, 4):
+        if channel_first is True or (
+                channel_first is None
+                and array.ndim == 3
+                and array.shape[0] in (1, 3, 4)):
             array = np.transpose(array, (1, 2, 0))
         if np.issubdtype(array.dtype, np.floating):
             array = np.rint(np.clip(array, 0.0, 1.0) * 255.0).astype(np.uint8)