dalaran-chunk-processing

# Dalaran chunk processing The pipeline layer between raw data and an DLR: readers produce `Chunk`s, streams transform them, terminal calls execute. This skill is the generic mechanics only. Decide the data model first (`dalaran-data-model`), then pick the importer skill for each source: | Source | Reader | Skill | | ----------------------------------------- | --------------------------------------- | --------------- | | MCAP file (ROS2, protobuf, Foxglove) | `McapReader(path).stream()` | `dalaran-mcap` | | URDF robot model (+ joint states → FK) | `UrdfTree.from_file_path(...).stream()` | `dalaran-urdf` | | Parquet table (trajectories, sensor logs) | `ParquetReader(path).stream()` | `dalaran-parquet` | | LeRobot dataset directory | built-in importer, then `RrdReader` | `dalaran-lerobot` | | Existing DLR | `RrdReader(path)` | here, below | | Sidecar files (JSON calib, metadata) | `Chunk.from_columns` + `from_iter` | here, below | The API is `dalaran.experimental`; when behavior matters, check the installed surface: `python -c "from dalaran.experimental import LazyChunkStream; help(LazyChunkStream)"`. ## Decision rule: where does each component come from? Default: **a reader produces the chunks; lenses shape them.** Walk this before writing any conversion code — most "build it by hand" instincts are wrong here: 1. **Source a reader supports?** Use the reader's `.stream()`; never hand-parse and re-log. MCAP→`McapReader`, URDF→`UrdfTree`, parquet→`ParquetReader`, DLR→`RrdReader`, LeRobot dir→`log_file_from_path`. 2. **A decoder already emits the archetype?** Foxglove gives `Transform3D`, `Pinhole`, `VideoStream` (real sample bytes) ready-made — **pass it through**, do not re-derive. Only custom-protobuf topics arrive as `<Name>:message` and need a lens (see `dalaran-mcap`). 3. **Fix an existing component in place** (swapped resolution, recolor, unit convert)? `MutateLens`, `output_mode="forward_unmatched"`. 4. **Derive a new component/entity** (FK→`/tf`, scalars from a message)? `DeriveLens`. To scatter one row into N (a joint batch → per-joint `/tf`), use the **two-lens pair**: derive the batch with `output_mode="forward_all"` (keeps the originals, e.g. the joint states), then a second `DeriveLens` with `scatter=True` and `output_mode="drop_unmatched"` (emits only the scattered rows). See the `robot_data_preprocessing` example. 5. **Genuine sidecar** no reader or lens can produce (JSON calibration offsets, hand-measured extrinsics, external metadata)? `Chunk.from_columns` + `from_iter`. 6. Finish with `LazyChunkStream.merge(...)` → `.collect(optimize=OptimizationProfile.OBJECT_STORE)` → `write_dlr(application_id, recording_id)`. Why this order: the pipeline stays lazy, columnar, multithreaded, and `OBJECT_STORE`-optimizable. A hand-built row loop or an out-of-lens `pa.array` throws all of that away — that is the path we are deliberately avoiding. ## Anti-patterns (use a reader + lens instead) If you are writing the left, stop and use the right: - **`for`-loop building rows/components** → a lens with a `Selector(...).pipe(...)` PyArrow-compute callback. - **`rr.init` + `rr.log` per message for conversion** → that is _live_ logging; for ingestion, read with a reader and `write_dlr`. - **`chunk.to_record_batch()` + `pc.filter` then rebuilding via `Chunk.from_columns`** (row-thinning by hand) → `stream.drop(content=...)`, `.split(...)`, or a `MutateLens` returning a filtered `pa.array`. - **`pa.array` / `pa.RecordBatch` / `np.frombuffer` assembled OUTSIDE a lens** → move the transform inside a `MutateLens`/`DeriveLens` selector callback. - **`rr.send_columns` hand-assembled from a custom parser** → use the matching reader; it produces chunks directly. - **Parsing MCAP/URDF with a non-Dalaran library then re-logging** → `McapReader` / `UrdfTree`. - **`Chunk.from_columns` for data a reader already decodes** (`Pinhole` intrinsics, `VideoStream`, `Transform3D` from a transforms topic) → keep it in the reader stream; fix with a `MutateLens` if needed. A wall of `pyarrow.compute` "missing-attribute" type errors (`pc.filter`, `pc.list_element`) usually means `pc.*` calls sit in module-level helpers instead of inside `Selector.pipe` lens callbacks. Refactor into a lens before suppressing the checker — the errors are a smell that the hand-building should not exist. **Porting an existing converter?** Hand-built converters predate decoder improvements and are not ground truth. Re-verify the decoder output (step 2) and check every `Chunk.from_columns` / for-loop against this list before copying. ## Core model - `LazyChunkStream` is a lazy pipeline DAG, not a collection. Building filters, lenses, maps, splits, and merges reads no source data. - Execution starts at terminal calls: `write_dlr(...)`, `collect()`, `to_chunks()`, or iterating the stream. - Execution is streaming, multithreaded, and mostly GIL-free. Prefer stream/lens operations and PyArrow compute over Python row loops. - **Move semantics**: builder calls (`filter`, `drop`, `lenses`, `map`, `flat_map`) consume the input stream; reusing a consumed stream raises. Reassign after each step. Terminal calls do not consume, but each terminal call re-executes the whole pipeline; `collect()` once if that is too costly. - `ChunkStore` is materialized in memory (`stream.collect()`, `ChunkStore.from_chunks`). `LazyStore` is manifest-indexed, loads chunks on demand (`RrdReader(path).store()`, catalog segment stores). Both have `schema()`, `summary()`, `stream()`, and `write_dlr(...)`. ## Stream composition ```python from dalaran.experimental import Chunk, LazyChunkStream, OptimizationProfile ``` - `stream.filter(content=, has_timeline=, is_static=, components=)` keeps the matching portion of each chunk; `stream.drop(...)` is its complement, same keyword filters. `content` takes an entity-path glob or a list of them. - `stream.map(fn)` applies `Chunk -> Chunk`; `stream.flat_map(fn)` applies `Chunk -> Iterable[Chunk]`. Escape hatches for chunk-level Python logic; prefer lenses for columnar work. - `stream.split(content=, ...)` returns `(matching, non_matching)`; both branches share the same upstream. - `LazyChunkStream.merge(*streams)` fans in any number of sources. - `LazyChunkStream.from_iter(chunks)` wraps hand-built chunks. ```python stream = source_stream() # any importer skill stream = stream.drop(content="/video_raw/**") stream = stream.lenses(fix_lens, content="/cam/**", output_mode="forward_unmatched") merged = LazyChunkStream.merge(stream, sidecar_stream) merged.write_dlr(out_path, application_id="my_app", recording_id=recording_id) ``` ## Hand-built chunks — sidecar only Use `Chunk.from_columns` ONLY for data no reader or lens can emit — JSON/CSV calibration, frame offsets, external metadata. If a reader (`McapReader`/`UrdfTree`/`ParquetReader`) decodes the topic or a lens can derive it, that is the idiomatic path; do not hand-assemble it here. In the `robot_data_preprocessing` example the _only_ hand-built chunk is the JSON offsets sidecar; the camera fix, FK→`/tf`, meshes, and recolor are all readers + lenses. `Chunk.from_columns(entity_path, indexes, columns)` mirrors `rr.send_columns(...)` and accepts the same archetype `.columns(...)` helpers. Empty `indexes` means static. ```python chunk = Chunk.from_columns( "/tf_static/robot_offsets", indexes=[], # static columns=rr.Transform3D.columns( translation=translations, quaternion=quaternions_xyzw, parent_frame=parents, child_frame=children, ), ) sidecar_stream = LazyChunkStream.from_iter([chunk]) ``` `rr.AnyValues.columns(...)` covers non-standard metadata fields. For inspection, a `Chunk` exposes `entity_path`, `num_rows`, `is_static`, `timeline_names`, `to_record_batch()`, and `format()` (human-readable table). ## Lenses Lenses reshape, fix, or derive components without iterating rows. Apply with `stream.lenses(lenses, output_mode=..., content=...)`. - `MutateLens(component, selector, keep_row_ids=False)` modifies an existing component in place. - `DeriveLens(component, output_entity=None, scatter=False)` creates new columns, optionally at another entity. Chain `.to_component(descriptor, selector)` per output; `.to_timeline(name, "sequence" | "duration_ns" | "timestamp_ns", selector)` extracts a time column from the data itself. `scatter=True` explodes one input row into N output rows (one per list element). - Scope with `content=` whenever the same component name exists under multiple entities. Output modes, and **the default is `drop_unmatched`**: - `drop_unmatched` (default): only lens outputs survive. Right for derive-only intermediate streams; silently discards everything else if applied broadly. - `forward_unmatched`: lens outputs plus the original components no lens consumed. Right for targeted fixes that preserve the rest of the stream. - `forward_all`: lens outputs plus all originals, including consumed ones. Can duplicate data. In-place fix (keep Arrow type and length intact): ```python stream = stream.lenses( MutateLens( "Pinhole:resolution", Selector(".").pipe( lambda res: pa.array( [(h, w) for w, h in res.to_pylist()], type=res.type, ) ), ), content=["/external/cam_low", "/external/cam_high"], output_mode="forward_unmatched", ) ``` Derive with unit conversion (PyArrow compute, no Python loop): ```python DeriveLens("schemas.proto.JointState:message", output_entity="/joints_deg/waist").to_component( rr.Scalars.descriptor_scalars(), Selector(".joint_positions").pipe(lambda arr: pc.multiply(pc.list_element(arr, 0), 180.0 / math.pi)), ) ``` ## Selector grammar `Selector("<query>")` navigates nested Arrow data, jq-style: - `.` current value; `.field` struct field - `[]` iterate list elements; `[N]` index a list - `?` suppress errors / skip missing optionals; `!` assert non-null - `|` pipe one expression into another `.pipe(fn)` chains a Python/PyArrow transform (or another Selector). `.execute(array)` runs it eagerly; `.execute_per_row(array)` guarantees the output row count matches the input (use inside lens callbacks that must stay row-aligned). ## Writing RRDs - `stream.write_dlr(path, application_id=..., recording_id=...)` executes and writes in one streaming pass. - `stream.collect(optimize=OptimizationProfile.OBJECT_STORE).write_dlr(...)` materializes, optimizes chunk layout, then writes. Memory scales with the materialized chunks. - Profiles: `OBJECT_STORE` (large chunks, for storage/query/catalog) and `LIVE` (small chunks, low-latency viewer). - Multiple physical RRDs form one logical recording when they share a `recording_id`; use this to separate base data, model/URDF data, and layers. **Always use `OptimizationProfile.OBJECT_STORE`** when the DLR is headed for a Dalaran catalog or Hub, unless explicitly asked otherwise. ## Chunk API vs logging API - Logging (`rr.log`, `rr.send_columns`, `RecordingStream`) is for live logging from user code; chunk processing is for ingestion, conversion, and postprocessing existing recordings. - Logging → chunks: write an DLR, read it back with `RrdReader`. `RrdReader(path)` lists `recordings()` / `blueprints()` (each a `StoreEntry` with `kind`, `application_id`, `recording_id`); `.stream(store=entry)` for sequential passes, `.store(store=entry)` for indexed access. - Chunks → logging: `dalaran.experimental.send_chunks(chunks, recording=...)` accepts a `Chunk`, `LazyChunkStream`, `LazyStore`, `ChunkStore`, or any iterable of chunks. The source store's `application_id`/`recording_id` are **not** preserved; the active recording's identity wins. ## Common gotchas - The default lens `output_mode` is `drop_unmatched`; forgetting to set `forward_unmatched` on a targeted fix silently drops the rest of the stream. - Do not reuse a consumed `LazyChunkStream`; reassign or `split` deliberately. - Scope lenses with `content=`; the same component name often exists under many entities. - Preserve Arrow array type and length in `MutateLens` transforms. - For catalog layers, the layer `recording_id` must equal the segment id. - This is `dalaran.experimental`; pin-check signatures when upgrading. ## References - End-to-end example (MCAP + URDF + JSON sidecar, lenses, merge, optimize): `https://github.com/Flaminis/Dalaran/tree/main/examples/python/robot_data_preprocessing` - Docs: `https://dalaran.dev/docs/concepts/logging-and-ingestion/chunk-processing-api`,