Streaming reads for cloud LeRobot-v3 & Zarr (inspect + ingest)

CHANGELOG.md

@@ -9,6 +9,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **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.
+  - **Inspect** fetches metadata only — a 464 MB cloud LeRobot-v3 dataset yields
+    a full inspect from ~3 KB (its `info.json`), a ~150,000× reduction, in a
+    fraction of a second.
+  - **Ingest** (register + quality-score) reads only the **proprio parquet
+    columns** (state/action/timestamp) via column/range reads — **never the
+    videos**. Ingesting the same 464 MB dataset streams ~3 MB and produces
+    identical episodes and quality scores. Content hashes are identical whether
+    an episode is streamed or downloaded, so cross-path re-ingest is idempotent.
+  - LeRobot-v3 gets a proprio-only streaming `read_episodes`; Zarr streams via
+    its native fsspec store. `forge inspect --deep`, embeddings, and video
+    quality still fetch frames; other formats/commands are unchanged
+    (download-to-temp). New `forge.io.DataSource` helper; readers keep their
+    local paths untouched.
+
 - **Search- and selection-driven curation.** `forge curate` now selects
   episodes three ways — a SQL `--where` predicate, an explicit `--ids` list, or
   a `--from <selection.json>` file — so semantic search and Forge Studio can

README.md

@@ -286,6 +286,8 @@ forge inspect s3://my-bucket/datasets/run_0413
 forge convert gs://lab-data/rosbags ./out --format lerobot-v3
 ```
 
+**Streaming reads.** For **LeRobot-v3** and **Zarr**, `forge inspect` and `forge ingest` read over the network with **range requests** instead of downloading. Inspecting a 464 MB cloud LeRobot dataset fetches ~3 KB (its `info.json`); **ingesting it — register + quality-score every episode — reads only the proprio parquet columns (~3 MB), never the videos.** `forge inspect --deep` forces a full download when you need per-episode stats; embeddings/video quality still fetch frames.
+
 Auth uses each provider's standard credential chain (AWS env vars / profiles / IAM roles; GCP Application Default Credentials) — Forge never handles credentials itself. Catalogs can be read *and written* in the cloud; per-format conversion **outputs** are local-only for now. Full guide + connectivity troubleshooting: [forge/io/README.md](forge/io/README.md).
 
 ## Dataset registry
@@ -334,7 +336,7 @@ Full reference: [docs/cli.md](docs/cli.md). Run `forge <command> --help` for any
 - [x] Cloud storage — `s3://` / `gs://` on every command
 - [x] The catalog — ingest, SQL query, semantic search, dedup, curation, Studio
 - [ ] **Snapshots + export** — freeze a curated selection → LeRobot/RLDS for training
-- [ ] **Streaming reads** — process cloud datasets without a full download
+- [~] **Streaming reads** — `forge inspect` + `forge ingest` stream LeRobot-v3 / Zarr (metadata + proprio via range reads, no video download); frame/embedding streaming next
 - [ ] **Dataset merging & splitting** — combine datasets; stratified train/val/test
 - [ ] **Depth / point-cloud support** · **GR00T writer** · **distributed conversion**
 

forge/catalog/ingest.py

@@ -60,6 +60,26 @@ def _resolve_source(uri: str) -> Path:
     return Path(uri)
 
 
+def _resolve_for_ingest(uri: str):
+    """Resolve for ingest, streaming remote streamable formats instead of downloading.
+
+    Returns a source usable by the readers — the URI itself for a remote
+    ``lerobot-v3``/``zarr`` dataset (metadata + proprio streamed via range
+    reads; no video download), or a local path otherwise.
+    """
+    from forge.formats.registry import FormatRegistry
+    from forge.io import is_remote_uri
+
+    if is_remote_uri(uri):
+        try:
+            fmt = FormatRegistry.detect_format(uri)
+        except Exception:
+            fmt = None
+        if fmt in FormatRegistry.STREAMABLE_FORMATS:
+            return uri
+    return _resolve_source(uri)
+
+
 def _episode_content_hash(episode: Episode, frames: list) -> str:
     """xxh3 hash of an episode's content: proprio stream + shape signature.
 
@@ -245,7 +265,7 @@ def ingest(
             total = 0
             for src in sources:
                 try:
-                    path = _resolve_source(src)
+                    path = _resolve_for_ingest(src)
                     fmt = FormatRegistry.detect_format(path)
                     total += FormatRegistry.get_reader(fmt).inspect(path).num_episodes or 0
                 except Exception:
@@ -284,7 +304,7 @@ def _flush() -> None:
     try:
         for src in sources:
             try:
-                path = _resolve_source(src)
+                path = _resolve_for_ingest(src)
                 fmt = FormatRegistry.detect_format(path)
                 reader = FormatRegistry.get_reader(fmt)
                 dataset_info = None

forge/cli.py

@@ -417,24 +417,47 @@ def inspect_cmd(
 
     from forge.inspect import InspectionOptions, Inspector
 
-    # Resolve HuggingFace URLs to local paths
-    try:
-        resolved_path = _resolve_dataset_path(path)
-    except Exception as e:
-        console.print(f"[red]Error downloading dataset:[/red] {e}")
-        raise typer.Exit(1)
+    from forge.io import is_remote_uri
 
-    options = InspectionOptions(
-        sample_episodes=samples,
-        deep_scan=deep,
-    )
-    inspector = Inspector(options)
+    info = None
 
-    try:
-        info = inspector.inspect(resolved_path, format)
-    except ForgeError as e:
-        console.print(f"[red]Error:[/red] {e}")
-        raise typer.Exit(1)
+    # Streaming inspect for cloud lerobot-v3 / zarr: read metadata over the
+    # network (range reads) instead of downloading the whole dataset. Falls back
+    # to download for --deep, --generate-config, or non-streamable formats.
+    if is_remote_uri(path) and not deep and generate_config is None:
+        from forge.formats.registry import FormatRegistry
+
+        try:
+            detected = format or FormatRegistry.detect_format(path)
+        except Exception:
+            detected = None
+        if detected in FormatRegistry.STREAMABLE_FORMATS:
+            try:
+                info = FormatRegistry.get_reader(detected).inspect(path)
+                console.print("[dim]streamed metadata — no download[/dim]")
+            except ForgeError as e:
+                console.print(f"[red]Error:[/red] {e}")
+                raise typer.Exit(1)
+
+    if info is None:
+        # Resolve HuggingFace URLs / download cloud datasets to local paths
+        try:
+            resolved_path = _resolve_dataset_path(path)
+        except Exception as e:
+            console.print(f"[red]Error downloading dataset:[/red] {e}")
+            raise typer.Exit(1)
+
+        options = InspectionOptions(
+            sample_episodes=samples,
+            deep_scan=deep,
+        )
+        inspector = Inspector(options)
+
+        try:
+            info = inspector.inspect(resolved_path, format)
+        except ForgeError as e:
+            console.print(f"[red]Error:[/red] {e}")
+            raise typer.Exit(1)
 
     if output == "json":
         # Convert to JSON-serializable dict