feat(lerobot_v3): add image-mode output to the v3 writer

CHANGELOG.md

@@ -9,6 +9,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Image-mode output for the LeRobot v3 writer.**
+  `LeRobotV3WriterConfig(video=False)` PNG-encodes each camera frame inline into
+  the data parquet (`dtype: "image"` features) instead of encoding MP4s — no
+  `videos/` directory and no per-camera `videos/…` pointer columns are written,
+  and `info.json`'s `video_path` / `video_files_size_in_mb` are `null` (the keys
+  stay present, as v3 loaders require). Set it from a conversion config with
+  `writer_config: {video: false}`. Needs Pillow (now part of the `[lerobot]`
+  extra). Video mode stays the default and is unchanged.
+  The LeRobot v3 **reader** decodes these inline PNG columns, so image-mode
+  datasets round-trip through Forge.
+
 - **Streaming reads for cloud datasets (LeRobot-v3, Zarr).** `forge inspect`
   **and `forge ingest`** on `s3://…` / `gs://…` now read over the network with
   **range requests** instead of downloading the whole dataset.
@@ -148,6 +159,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
     vars / profiles / IAM roles; GCP Application Default Credentials). Forge
     never handles credentials itself.
 
+### Fixed
+
+- **`writer_config` now reaches parallel conversion workers.** `ConversionConfig`
+  dropped `writer_config` when serialized, so with `--workers > 1` every
+  writer-specific option (including the new `video: false`) was silently ignored
+  by the worker processes.
+
 ### Changed
 
 - `fsspec` is now a core dependency.

docs/configuration.md

@@ -100,6 +100,18 @@ video:
   compress: true       # Enable compression
 ```
 
+### Writer Settings
+
+Options passed straight to the target format's writer config (unknown keys are
+ignored):
+
+```yaml
+writer_config:
+  video: false         # lerobot-v3: store camera frames as PNG bytes inline in
+                       # the data parquet (dtype "image") instead of MP4 videos.
+                       # Default true. Needs the [lerobot] extra (Pillow).
+```
+
 ### Behavior Settings
 
 ```yaml

forge/config/models.py

@@ -309,6 +309,9 @@ def to_dict(self) -> dict[str, Any]:
         if self.num_workers > 1:
             result["num_workers"] = self.num_workers
 
+        if self.writer_config:
+            result["writer_config"] = self.writer_config
+
         return result
 
     def to_yaml(self, path: Path | str) -> None:

forge/convert/converter.py

@@ -62,6 +62,51 @@ class _EpisodeResult:
     error: str | None = None
 
 
+def _configure_writer(
+    writer: Any, config: ConversionConfig, dataset_info: DatasetInfo
+) -> None:
+    """Apply a conversion config to a writer's format-specific config.
+
+    Shared by the sequential path and the parallel worker so writer options
+    cannot drift between them.
+
+    Args:
+        writer: Writer instance to configure.
+        config: Conversion config supplying the overrides.
+        dataset_info: Dataset info for default values.
+    """
+    # Check if writer has a config attribute (like LeRobotV3Writer)
+    if not hasattr(writer, "config"):
+        return
+
+    wconfig = writer.config
+
+    # Apply defaults from dataset_info
+    if hasattr(wconfig, "fps") and dataset_info.inferred_fps:
+        wconfig.fps = dataset_info.inferred_fps
+
+    if hasattr(wconfig, "robot_type") and dataset_info.inferred_robot_type:
+        wconfig.robot_type = dataset_info.inferred_robot_type
+
+    if hasattr(wconfig, "camera_name_mapping") and config.camera_mapping:
+        wconfig.camera_name_mapping = config.camera_mapping
+
+    # Apply field mappings if writer supports them
+    if hasattr(wconfig, "field_mapping") and config.field_mapping:
+        wconfig.field_mapping = config.field_mapping
+
+    if hasattr(wconfig, "action_field") and config.action_field:
+        wconfig.action_field = config.action_field
+
+    if hasattr(wconfig, "state_field") and config.state_field:
+        wconfig.state_field = config.state_field
+
+    # Apply format-specific config from conversion config
+    for key, value in config.writer_config.items():
+        if hasattr(wconfig, key):
+            setattr(wconfig, key, value)
+
+
 def _process_episode_worker(
     source_path: str,
     output_path: str,
@@ -131,20 +176,7 @@ def _process_episode_worker(
             dataset_info.inferred_robot_type = config.robot_type
 
         # Configure writer
-        if hasattr(writer, "config"):
-            wconfig = writer.config
-            if hasattr(wconfig, "fps") and dataset_info.inferred_fps:
-                wconfig.fps = dataset_info.inferred_fps
-            if hasattr(wconfig, "robot_type") and dataset_info.inferred_robot_type:
-                wconfig.robot_type = dataset_info.inferred_robot_type
-            if hasattr(wconfig, "camera_name_mapping") and config.camera_mapping:
-                wconfig.camera_name_mapping = config.camera_mapping
-            if hasattr(wconfig, "field_mapping") and config.field_mapping:
-                wconfig.field_mapping = config.field_mapping
-            if hasattr(wconfig, "action_field") and config.action_field:
-                wconfig.action_field = config.action_field
-            if hasattr(wconfig, "state_field") and config.state_field:
-                wconfig.state_field = config.state_field
+        _configure_writer(writer, config, dataset_info)
 
         # Read and write the specific episode
         episodes_iter = reader.read_episodes(source)
@@ -291,7 +323,7 @@ def convert(
         self._apply_config_overrides(dataset_info)
 
         # 5. Configure writer with format-specific options
-        self._configure_writer(writer, dataset_info)
+        _configure_writer(writer, self.config, dataset_info)
 
         # 6. Convert episodes (parallel or sequential)
         total_episodes = dataset_info.num_episodes or 0
@@ -368,42 +400,6 @@ def _apply_config_overrides(self, dataset_info: DatasetInfo) -> None:
         if self.config.robot_type is not None:
             dataset_info.inferred_robot_type = self.config.robot_type
 
-    def _configure_writer(self, writer: Any, dataset_info: DatasetInfo) -> None:
-        """Configure writer with format-specific options.
-
-        Args:
-            writer: Writer instance to configure.
-            dataset_info: Dataset info for default values.
-        """
-        # Check if writer has a config attribute (like LeRobotV3Writer)
-        if hasattr(writer, "config"):
-            config = writer.config
-
-            # Apply defaults from dataset_info
-            if hasattr(config, "fps") and dataset_info.inferred_fps:
-                config.fps = dataset_info.inferred_fps
-
-            if hasattr(config, "robot_type") and dataset_info.inferred_robot_type:
-                config.robot_type = dataset_info.inferred_robot_type
-
-            if hasattr(config, "camera_name_mapping") and self.config.camera_mapping:
-                config.camera_name_mapping = self.config.camera_mapping
-
-            # Apply field mappings if writer supports them
-            if hasattr(config, "field_mapping") and self.config.field_mapping:
-                config.field_mapping = self.config.field_mapping
-
-            if hasattr(config, "action_field") and self.config.action_field:
-                config.action_field = self.config.action_field
-
-            if hasattr(config, "state_field") and self.config.state_field:
-                config.state_field = self.config.state_field
-
-            # Apply format-specific config from conversion config
-            for key, value in self.config.writer_config.items():
-                if hasattr(config, key):
-                    setattr(config, key, value)
-
     def _convert_parallel(
         self,
         source: Path,

forge/formats/lerobot_v3/reader.py

@@ -21,10 +21,15 @@
             │   ├── episode_000000.mp4
             │   └── ...
             └── ...
+
+Image-mode datasets (features with ``dtype: "image"``) have no videos/
+directory: each camera column holds PNG bytes inline in the data parquet,
+decoded lazily per frame.
 """
 
 from __future__ import annotations
 
+import io
 import json
 from collections.abc import Iterator
 from pathlib import Path
@@ -1144,12 +1149,17 @@ def _build_episode(
         # Get camera info from info.json for dimensions
         info_path = dataset_path / "meta" / "info.json"
         camera_features: dict[str, dict] = {}
+        # Image-mode cameras (dtype == "image"): inline PNG bytes live directly
+        # in the data parquet under their full feature key.
+        image_camera_features: dict[str, dict] = {}
         if info_path.exists():
             with open(info_path) as f:
                 info = json.load(f)
                 for key, spec in info.get("features", {}).items():
                     if spec.get("dtype") == "video":
                         camera_features[key] = spec
+                    elif spec.get("dtype") == "image":
+                        image_camera_features[key] = spec
 
         # Get video path template from info.json
         video_path_template = None
@@ -1343,6 +1353,49 @@ def make_loader(vp: Path = video_path, fi: int = local_frame_idx, d: tuple = dim
                         channels=dims[2],
                     )
 
+                # Inline image-mode cameras: PNG bytes live in this row's own
+                # struct<bytes, path> column, decoded lazily on load().
+                for image_key, spec in image_camera_features.items():
+                    if image_key not in ep_df.columns:
+                        continue
+                    cam_name = image_key.split(".")[-1]
+                    cell = row[image_key]
+                    png_bytes: bytes | None
+                    png_path: str | None
+                    if isinstance(cell, bytes):
+                        png_bytes, png_path = cell, None
+                    elif isinstance(cell, dict):
+                        png_bytes, png_path = cell.get("bytes"), cell.get("path")
+                    else:
+                        continue
+                    if png_bytes is None:
+                        if not png_path:
+                            continue
+                        resolved = Path(png_path)
+                        if not resolved.is_absolute():
+                            resolved = dataset_path / resolved
+                        if not resolved.exists():
+                            continue
+                        png_path = str(resolved)
+                    shape = spec.get("shape") or [0, 0, 3]
+
+                    def make_image_loader(
+                        data: bytes | None = png_bytes, path: str | None = png_path
+                    ) -> NDArray[Any]:
+                        import numpy as np
+                        from PIL import Image
+
+                        source: Any = io.BytesIO(data) if data is not None else path
+                        with Image.open(source) as img:
+                            return np.array(img)
+
+                    images[cam_name] = LazyImage(
+                        loader=make_image_loader,
+                        height=shape[0],
+                        width=shape[1],
+                        channels=shape[2] if len(shape) > 2 else 3,
+                    )
+
                 # Extract state and action
                 state = None
                 action = None