From 30af30239a68e9b22ff4a884ef9d24c0a3659351 Mon Sep 17 00:00:00 2001 From: Everett Kleven Date: Thu, 21 May 2026 15:44:44 -0700 Subject: [PATCH 1/2] notebooks: MCAP robotics DataFrame demo MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Walks daft.read_mcap against a 1.19 GB MCAP from the DapengFeng/MCAP dataset on HuggingFace (FAST-LIVO/hku1: Livox LiDAR + IMU + stereo compressed cameras, 29,702 messages over 127.7s). Covers schema inference, per-topic groupby, topic+time pushdown, topic_start_time_resolver (per-file per-topic keyframe alignment, new in v0.7.2), and a 50ms-bucket join between LiDAR sweeps and IMU samples. Direct HTTP reads (daft.read_mcap("https://...")) tracked in Eventual-Inc/Daft#6983 — notebook downloads via huggingface_hub for now. Co-Authored-By: Claude Opus 4.7 --- notebooks/mcap_robotics_dataframe.ipynb | 904 ++++++++++++++++++++++++ 1 file changed, 904 insertions(+) create mode 100644 notebooks/mcap_robotics_dataframe.ipynb diff --git a/notebooks/mcap_robotics_dataframe.ipynb b/notebooks/mcap_robotics_dataframe.ipynb new file mode 100644 index 0000000..b407fab --- /dev/null +++ b/notebooks/mcap_robotics_dataframe.ipynb @@ -0,0 +1,904 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Reading MCAP in a DataFrame with Daft\n", + "\n", + "[MCAP](https://mcap.dev/) is the Foxglove-backed open standard for multimodal robotics logs \u2014 synchronized IMU, LiDAR, camera, ROS, and other message streams in a single file, schema-aware and replayable. Daft reads MCAP natively: every message becomes a row in a typed DataFrame, topic filters and time windows push down into the scanner, and a `topic_start_time_resolver` callback lets you align per-topic start times to keyframes.\n", + "\n", + "This notebook walks the API against a real robotics recording \u2014 a 1.19 GB MCAP from the [DapengFeng/MCAP](https://huggingface.co/datasets/DapengFeng/MCAP) dataset on HuggingFace (CC-BY-NC-4.0). The FAST-LIVO `hku1` recording has four topics (Livox LiDAR, Livox IMU, stereo compressed cameras from HKU's campus) and 29,702 messages over 127.7 seconds.\n", + "\n", + "> The MCAP file is ~1.2 GB and will be cached under `~/.cache/huggingface/`. Set `HF_HOME` before running if you want it elsewhere. Direct HTTP reads (`daft.read_mcap(\"https://...\")`) are tracked in [Daft #6983](https://github.com/Eventual-Inc/Daft/issues/6983) \u2014 until that lands, download via `huggingface_hub` first." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%pip install -q 'daft[mcap]>=0.7.13' 'huggingface_hub>=0.23' ipywidgets" + ] + }, + { + "cell_type": "code", + "execution_count": 8, + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "/Users/everettkleven/.cache/huggingface/hub/datasets--DapengFeng--MCAP/snapshots/579b47b2fef6b478dbd59947826ed27761bcdb60/FAST-LIVO/hku1/hku1_0.mcap\n" + ] + } + ], + "source": [ + "from huggingface_hub import hf_hub_download\n", + "\n", + "mcap_path = hf_hub_download(\n", + " repo_id=\"DapengFeng/MCAP\",\n", + " repo_type=\"dataset\",\n", + " filename=\"FAST-LIVO/hku1/hku1_0.mcap\",\n", + ")\n", + "print(mcap_path)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## 1. Open as a DataFrame\n", + "\n", + "`daft.read_mcap` returns a lazy DataFrame. The schema is fixed across every MCAP file:\n", + "\n", + "| column | type | meaning |\n", + "|--------|------|---------|\n", + "| `topic` | Utf8 | MCAP channel topic name (e.g. `/livox/imu`) |\n", + "| `log_time` | Int64 | message log time in nanoseconds since epoch |\n", + "| `publish_time` | Int64 | message publish time in nanoseconds since epoch |\n", + "| `sequence` | Int32 | per-channel sequence number |\n", + "| `data` | Utf8 | raw message bytes as a Python-style byte string |\n", + "\n", + "Note: `data` keeps the row self-contained but is not the place to decode camera frames or point clouds \u2014 narrow with `topics=[...]` and run a UDF after." + ] + }, + { + "cell_type": "code", + "execution_count": 9, + "metadata": {}, + "outputs": [ + { + "data": { + "text/html": [ + "\n", + "\n", + "\n", + "\n", + "\n", + "\n", + "\n", + "\n", + "\n", + "
column_nametype
topicString
log_timeInt64
publish_timeInt64
sequenceInt32
dataString
" + ], + "text/plain": [ + "\u256d\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u256e\n", + "\u2502 Column Name \u2506 DType \u2502\n", + "\u255e\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u256a\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2561\n", + "\u2502 topic \u2506 String \u2502\n", + "\u251c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u253c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u2524\n", + "\u2502 log_time \u2506 Int64 \u2502\n", + "\u251c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u253c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u2524\n", + "\u2502 publish_time \u2506 Int64 \u2502\n", + "\u251c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u253c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u2524\n", + "\u2502 sequence \u2506 Int32 \u2502\n", + "\u251c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u253c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u2524\n", + "\u2502 data \u2506 String \u2502\n", + "\u2570\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2534\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u256f" + ] + }, + "execution_count": 9, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "import daft\n", + "\n", + "df = daft.read_mcap(mcap_path)\n", + "df.schema()" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## 2. What's in this recording?\n", + "\n", + "Group by topic to count messages and bound the time range per channel. We project away `data` first so we never materialize the camera payloads." + ] + }, + { + "cell_type": "code", + "execution_count": 10, + "metadata": {}, + "outputs": [ + { + "data": { + "text/html": [ + "\n", + " \n", + "
\n", + "
\n", + "
\n", + "\n", + "\n", + "\n", + "\n", + "\n", + "\n", + "\n", + "
topic
String		messages
UInt64		t_start_ns
Int64		t_end_ns
Int64
/livox/imu
25871
1577836978701916933
1577837106395623922
/left_camera/image/compressed
1277
1577836978699826955
1577837106299479007
/right_camera/image/compressed
1277
1577836978699826955
1577837106299479007
/livox/lidar
1277
1577836978699826956
1577837106299479008
\n", + "
\n", + "
\n", + "
\n", + "
Cell Details
\n", + " \n", + "
\n", + "
\n", + "

Click on a cell to view its full content

\n", + "
\n", + "
\n", + "
\n", + " \n", + " \n", + " \n", + "(Showing first 4 of 4 rows)" + ], + "text/plain": [ + "" + ] + }, + "metadata": {}, + "output_type": "display_data" + } + ], + "source": [ + "(\n", + " df.select(\"topic\", \"sequence\", \"log_time\")\n", + " .groupby(\"topic\")\n", + " .agg(\n", + " daft.col(\"sequence\").count().alias(\"messages\"),\n", + " daft.col(\"log_time\").min().alias(\"t_start_ns\"),\n", + " daft.col(\"log_time\").max().alias(\"t_end_ns\"),\n", + " )\n", + " .sort(\"messages\", desc=True)\n", + " .show()\n", + ")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Expected: `/livox/imu` \u2248 25,871 messages, the three other topics \u2248 1,277 each.\n", + "\n", + "## 3. Topic filter pushes down into the scanner\n", + "\n", + "Passing `topics=[...]` to `read_mcap` means the MCAP reader walks only those channels. No camera payloads get pulled into memory." + ] + }, + { + "cell_type": "code", + "execution_count": 11, + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "[00:00] \ud83d\udde1\ufe0f \ud83d\udc1f Read MCAPSource (Python): 25,871 rows out, 0 B read | \ud83d\udde1\ufe0f \ud83d\udc1f Count: 25,871 rows in, 1 rows out | \ud83d\udde1\ufe0f \ud83d\udc1f Rename & Reorder: 1 rows in, 1 rows out\n", + "\n", + "IMU messages: 25871\n" + ] + } + ], + "source": [ + "imu_df = daft.read_mcap(mcap_path, topics=[\"/livox/imu\"])\n", + "print(\"IMU messages:\", imu_df.count_rows())" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## 4. Time-window a stream\n", + "\n", + "`start_time` and `end_time` are in nanoseconds, same unit as `log_time`. Anchor a 10-second window to the IMU's first log time." + ] + }, + { + "cell_type": "code", + "execution_count": 12, + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "[00:00] \ud83d\udde1\ufe0f \ud83d\udc1f Read MCAPSource (Python): 25,871 rows out, 0 B read | \ud83d\udde1\ufe0f \ud83d\udc1f Min: 25,871 rows in, 1 rows out\n", + "\n", + "IMU t0 (ns): 1577836978701916933\n", + "[00:00] \ud83d\udde1\ufe0f \ud83d\udc1f Read MCAPSource (Python): 2,028 rows out, 0 B read | \ud83d\udde1\ufe0f \ud83d\udc1f Count: 2,028 rows in, 1 rows out | \ud83d\udde1\ufe0f \ud83d\udc1f Rename & Reorder: 1 rows in, 1 rows out\n", + "\n", + "IMU messages in first 10s: 2028\n" + ] + } + ], + "source": [ + "NS_PER_S = 1_000_000_000\n", + "\n", + "t0 = imu_df.agg(daft.col(\"log_time\").min().alias(\"t0\")).to_pydict()[\"t0\"][0]\n", + "print(\"IMU t0 (ns):\", t0)\n", + "\n", + "window = daft.read_mcap(\n", + " mcap_path,\n", + " topics=[\"/livox/imu\"],\n", + " start_time=t0,\n", + " end_time=t0 + 10 * NS_PER_S,\n", + ")\n", + "print(\"IMU messages in first 10s:\", window.count_rows())" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## 5. Per-topic keyframe alignment\n", + "\n", + "`topic_start_time_resolver` is a callable that takes a file path and returns `{topic: start_time_ns}`. The reader expands one scan task per `(file, topic)`, each starting at its own offset.\n", + "\n", + "For video topics, that's the natural place to plug in your keyframe table \u2014 start decoding from an I-frame, not from the file head. Here we just skip the first 5 seconds of each topic relative to the file start to show the mechanism." + ] + }, + { + "cell_type": "code", + "execution_count": 13, + "metadata": {}, + "outputs": [ + { + "data": { + "text/html": [ + "\n", + " \n", + "
\n", + "
\n", + "
\n", + "\n", + "\n", + "\n", + "\n", + "\n", + "
topic
String		first_log_time_ns
Int64		messages
UInt64
/livox/lidar
1577836983799504995
1226
/livox/imu
1577836983702841043
24856
\n", + "
\n", + "
\n", + "
\n", + "
Cell Details
\n", + " \n", + "
\n", + "
\n", + "

Click on a cell to view its full content

\n", + "
\n", + "
\n", + "
\n", + " \n", + " \n", + " \n", + "(Showing first 2 of 2 rows)" + ], + "text/plain": [ + "" + ] + }, + "metadata": {}, + "output_type": "display_data" + } + ], + "source": [ + "def keyframe_resolver(file_path):\n", + " \"\"\"Return per-topic start offsets, in nanoseconds since epoch.\n", + "\n", + " In production this is where you'd look up an I-frame index next to each\n", + " recording \u2014 keyframe-aligned decoding for video topics, sensor-startup\n", + " masking for IMU/LiDAR.\"\"\"\n", + " return {\n", + " \"/livox/imu\": t0 + 5 * NS_PER_S,\n", + " \"/livox/lidar\": t0 + 5 * NS_PER_S,\n", + " }\n", + "\n", + "aligned = daft.read_mcap(\n", + " mcap_path,\n", + " topics=[\"/livox/imu\", \"/livox/lidar\"],\n", + " topic_start_time_resolver=keyframe_resolver,\n", + ")\n", + "\n", + "(\n", + " aligned.select(\"topic\", \"log_time\")\n", + " .groupby(\"topic\")\n", + " .agg(\n", + " daft.col(\"log_time\").min().alias(\"first_log_time_ns\"),\n", + " daft.col(\"log_time\").count().alias(\"messages\"),\n", + " )\n", + " .show()\n", + ")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## 6. Time-aligned join: IMU samples near every LiDAR sweep\n", + "\n", + "The kind of query that's painful in raw MCAP \u2014 for every LiDAR frame, pull the IMU samples within \u00b150 ms. Bucket both streams by 50 ms, join on the bucket, project the timestamps and sequence numbers.\n", + "\n", + "This is the metadata-level join. Decoding the LiDAR point cloud or fusing IMU rotation is a UDF step over the result." + ] + }, + { + "cell_type": "code", + "execution_count": 14, + "metadata": {}, + "outputs": [ + { + "data": { + "text/html": [ + "\n", + " \n", + "
\n", + "
\n", + "
\n", + "\n", + "\n", + "\n", + "\n", + "\n", + "\n", + "\n", + "\n", + "
lidar_seq
Int32		lidar_ts
Int64		imu_seq
Int32		imu_ts
Int64
0
1577836983699663162
0
1577836983652827978
0
1577836983699663162
0
1577836983657896042
0
1577836983699663162
0
1577836983662841082
0
1577836983699663162
0
1577836983667887211
0
1577836983699663162
0
1577836983672842026
\n", + "
\n", + "
\n", + "
\n", + "
Cell Details
\n", + " \n", + "
\n", + "
\n", + "

Click on a cell to view its full content

\n", + "
\n", + "
\n", + "
\n", + " \n", + " \n", + " \n", + "(Showing first 5 rows)" + ], + "text/plain": [ + "" + ] + }, + "metadata": {}, + "output_type": "display_data" + } + ], + "source": [ + "BUCKET_NS = 50_000_000 # 50 ms\n", + "\n", + "imu = (\n", + " daft.read_mcap(mcap_path, topics=[\"/livox/imu\"])\n", + " .select(\n", + " daft.col(\"log_time\").alias(\"imu_ts\"),\n", + " daft.col(\"sequence\").alias(\"imu_seq\"),\n", + " )\n", + " .with_column(\"bucket\", daft.col(\"imu_ts\") // BUCKET_NS)\n", + ")\n", + "\n", + "lidar = (\n", + " daft.read_mcap(mcap_path, topics=[\"/livox/lidar\"])\n", + " .select(\n", + " daft.col(\"log_time\").alias(\"lidar_ts\"),\n", + " daft.col(\"sequence\").alias(\"lidar_seq\"),\n", + " )\n", + " .with_column(\"bucket\", daft.col(\"lidar_ts\") // BUCKET_NS)\n", + ")\n", + "\n", + "joined = lidar.join(imu, on=\"bucket\").select(\n", + " \"lidar_seq\", \"lidar_ts\", \"imu_seq\", \"imu_ts\"\n", + ")\n", + "\n", + "print(\"matched (lidar, imu) pairs:\", joined.count_rows())\n", + "joined.sort(\"lidar_seq\").show(5)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## What you'd build next\n", + "\n", + "- **Decode payloads in a UDF.** Register a per-topic decoder once (e.g. `cv2.imdecode` for compressed images, the Livox `CustomMsg` parser for LiDAR), call it on `data` only after filtering. Daft distributes the work across the cluster.\n", + "- **Write filtered windows to a lakehouse.** Push the IMU/LiDAR join into an Iceberg table for downstream labeling, or land per-trip event tables in Delta.\n", + "- **Same code, fleet scale.** Point `daft.read_mcap` at an S3 prefix instead of a single file \u2014 the non-seekable stream wrapper means S3, GCS, and Azure all behave identically.\n", + "\n", + "### References\n", + "\n", + "- [`daft.read_mcap` docs](https://docs.daft.ai/) \u2014 full API reference\n", + "- [mcap.dev](https://mcap.dev/) \u2014 format spec, CLI, decoder libraries\n", + "- [foxglove.dev](https://foxglove.dev/) \u2014 visualization for the recordings you build queries on top of\n", + "- [DapengFeng/MCAP](https://huggingface.co/datasets/DapengFeng/MCAP) \u2014 the SLAM dataset used in this notebook\n", + "- [Daft PR #5886](https://github.com/Eventual-Inc/Daft/pull/5886) \u2014 the `topic_start_time_resolver` and non-seekable reader landing in v0.7.2" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": ".venv", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.12.12" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} From 0122de181cc9e916787c972ba69bdeb27156c46f Mon Sep 17 00:00:00 2001 From: Everett Kleven Date: Thu, 21 May 2026 16:00:50 -0700 Subject: [PATCH 2/2] chore: ruff format mcap_robotics_dataframe notebook Co-Authored-By: Claude Opus 4.7 --- notebooks/mcap_robotics_dataframe.ipynb | 137 +++++++++++++----------- 1 file changed, 76 insertions(+), 61 deletions(-) diff --git a/notebooks/mcap_robotics_dataframe.ipynb b/notebooks/mcap_robotics_dataframe.ipynb index b407fab..e5c085b 100644 --- a/notebooks/mcap_robotics_dataframe.ipynb +++ b/notebooks/mcap_robotics_dataframe.ipynb @@ -2,20 +2,22 @@ "cells": [ { "cell_type": "markdown", + "id": "7fb27b941602401d91542211134fc71a", "metadata": {}, "source": [ "# Reading MCAP in a DataFrame with Daft\n", "\n", - "[MCAP](https://mcap.dev/) is the Foxglove-backed open standard for multimodal robotics logs \u2014 synchronized IMU, LiDAR, camera, ROS, and other message streams in a single file, schema-aware and replayable. Daft reads MCAP natively: every message becomes a row in a typed DataFrame, topic filters and time windows push down into the scanner, and a `topic_start_time_resolver` callback lets you align per-topic start times to keyframes.\n", + "[MCAP](https://mcap.dev/) is the Foxglove-backed open standard for multimodal robotics logs — synchronized IMU, LiDAR, camera, ROS, and other message streams in a single file, schema-aware and replayable. Daft reads MCAP natively: every message becomes a row in a typed DataFrame, topic filters and time windows push down into the scanner, and a `topic_start_time_resolver` callback lets you align per-topic start times to keyframes.\n", "\n", - "This notebook walks the API against a real robotics recording \u2014 a 1.19 GB MCAP from the [DapengFeng/MCAP](https://huggingface.co/datasets/DapengFeng/MCAP) dataset on HuggingFace (CC-BY-NC-4.0). The FAST-LIVO `hku1` recording has four topics (Livox LiDAR, Livox IMU, stereo compressed cameras from HKU's campus) and 29,702 messages over 127.7 seconds.\n", + "This notebook walks the API against a real robotics recording — a 1.19 GB MCAP from the [DapengFeng/MCAP](https://huggingface.co/datasets/DapengFeng/MCAP) dataset on HuggingFace (CC-BY-NC-4.0). The FAST-LIVO `hku1` recording has four topics (Livox LiDAR, Livox IMU, stereo compressed cameras from HKU's campus) and 29,702 messages over 127.7 seconds.\n", "\n", - "> The MCAP file is ~1.2 GB and will be cached under `~/.cache/huggingface/`. Set `HF_HOME` before running if you want it elsewhere. Direct HTTP reads (`daft.read_mcap(\"https://...\")`) are tracked in [Daft #6983](https://github.com/Eventual-Inc/Daft/issues/6983) \u2014 until that lands, download via `huggingface_hub` first." + "> The MCAP file is ~1.2 GB and will be cached under `~/.cache/huggingface/`. Set `HF_HOME` before running if you want it elsewhere. Direct HTTP reads (`daft.read_mcap(\"https://...\")`) are tracked in [Daft #6983](https://github.com/Eventual-Inc/Daft/issues/6983) — until that lands, download via `huggingface_hub` first." ] }, { "cell_type": "code", "execution_count": null, + "id": "acae54e37e7d407bbb7b55eff062a284", "metadata": {}, "outputs": [], "source": [ @@ -25,6 +27,7 @@ { "cell_type": "code", "execution_count": 8, + "id": "9a63283cbaf04dbcab1f6479b197f3a8", "metadata": {}, "outputs": [ { @@ -48,6 +51,7 @@ }, { "cell_type": "markdown", + "id": "8dd0d8092fe74a7c96281538738b07e2", "metadata": {}, "source": [ "## 1. Open as a DataFrame\n", @@ -62,12 +66,13 @@ "| `sequence` | Int32 | per-channel sequence number |\n", "| `data` | Utf8 | raw message bytes as a Python-style byte string |\n", "\n", - "Note: `data` keeps the row self-contained but is not the place to decode camera frames or point clouds \u2014 narrow with `topics=[...]` and run a UDF after." + "Note: `data` keeps the row self-contained but is not the place to decode camera frames or point clouds — narrow with `topics=[...]` and run a UDF after." ] }, { "cell_type": "code", "execution_count": 9, + "id": "72eea5119410473aa328ad9291626812", "metadata": {}, "outputs": [ { @@ -85,19 +90,19 @@ "" ], "text/plain": [ - "\u256d\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u256e\n", - "\u2502 Column Name \u2506 DType \u2502\n", - "\u255e\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u256a\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2561\n", - "\u2502 topic \u2506 String \u2502\n", - "\u251c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u253c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u2524\n", - "\u2502 log_time \u2506 Int64 \u2502\n", - "\u251c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u253c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u2524\n", - "\u2502 publish_time \u2506 Int64 \u2502\n", - "\u251c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u253c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u2524\n", - "\u2502 sequence \u2506 Int32 \u2502\n", - "\u251c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u253c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u254c\u2524\n", - "\u2502 data \u2506 String \u2502\n", - "\u2570\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2534\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u256f" + "╭──────────────┬────────╮\n", + "│ Column Name ┆ DType │\n", + "╞══════════════╪════════╡\n", + "│ topic ┆ String │\n", + "├╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤\n", + "│ log_time ┆ Int64 │\n", + "├╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤\n", + "│ publish_time ┆ Int64 │\n", + "├╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤\n", + "│ sequence ┆ Int32 │\n", + "├╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤\n", + "│ data ┆ String │\n", + "╰──────────────┴────────╯" ] }, "execution_count": 9, @@ -114,6 +119,7 @@ }, { "cell_type": "markdown", + "id": "8edb47106e1a46a883d545849b8ab81b", "metadata": {}, "source": [ "## 2. What's in this recording?\n", @@ -124,6 +130,7 @@ { "cell_type": "code", "execution_count": 10, + "id": "10185d26023b46108eb7d9f57d49d2b3", "metadata": {}, "outputs": [ { @@ -198,7 +205,7 @@ "
\n", "
\n", "
Cell Details
\n", - " \n", + " \n", "
\n", "
\n", "

Click on a cell to view its full content

\n", @@ -315,22 +322,23 @@ "source": [ "(\n", " df.select(\"topic\", \"sequence\", \"log_time\")\n", - " .groupby(\"topic\")\n", - " .agg(\n", - " daft.col(\"sequence\").count().alias(\"messages\"),\n", - " daft.col(\"log_time\").min().alias(\"t_start_ns\"),\n", - " daft.col(\"log_time\").max().alias(\"t_end_ns\"),\n", - " )\n", - " .sort(\"messages\", desc=True)\n", - " .show()\n", + " .groupby(\"topic\")\n", + " .agg(\n", + " daft.col(\"sequence\").count().alias(\"messages\"),\n", + " daft.col(\"log_time\").min().alias(\"t_start_ns\"),\n", + " daft.col(\"log_time\").max().alias(\"t_end_ns\"),\n", + " )\n", + " .sort(\"messages\", desc=True)\n", + " .show()\n", ")" ] }, { "cell_type": "markdown", + "id": "8763a12b2bbd4a93a75aff182afb95dc", "metadata": {}, "source": [ - "Expected: `/livox/imu` \u2248 25,871 messages, the three other topics \u2248 1,277 each.\n", + "Expected: `/livox/imu` ≈ 25,871 messages, the three other topics ≈ 1,277 each.\n", "\n", "## 3. Topic filter pushes down into the scanner\n", "\n", @@ -340,13 +348,14 @@ { "cell_type": "code", "execution_count": 11, + "id": "7623eae2785240b9bd12b16a66d81610", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ - "[00:00] \ud83d\udde1\ufe0f \ud83d\udc1f Read MCAPSource (Python): 25,871 rows out, 0 B read | \ud83d\udde1\ufe0f \ud83d\udc1f Count: 25,871 rows in, 1 rows out | \ud83d\udde1\ufe0f \ud83d\udc1f Rename & Reorder: 1 rows in, 1 rows out\n", + "[00:00] 🗡️ 🐟 Read MCAPSource (Python): 25,871 rows out, 0 B read | 🗡️ 🐟 Count: 25,871 rows in, 1 rows out | 🗡️ 🐟 Rename & Reorder: 1 rows in, 1 rows out\n", "\n", "IMU messages: 25871\n" ] @@ -359,6 +368,7 @@ }, { "cell_type": "markdown", + "id": "7cdc8c89c7104fffa095e18ddfef8986", "metadata": {}, "source": [ "## 4. Time-window a stream\n", @@ -369,16 +379,17 @@ { "cell_type": "code", "execution_count": 12, + "id": "b118ea5561624da68c537baed56e602f", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ - "[00:00] \ud83d\udde1\ufe0f \ud83d\udc1f Read MCAPSource (Python): 25,871 rows out, 0 B read | \ud83d\udde1\ufe0f \ud83d\udc1f Min: 25,871 rows in, 1 rows out\n", + "[00:00] 🗡️ 🐟 Read MCAPSource (Python): 25,871 rows out, 0 B read | 🗡️ 🐟 Min: 25,871 rows in, 1 rows out\n", "\n", "IMU t0 (ns): 1577836978701916933\n", - "[00:00] \ud83d\udde1\ufe0f \ud83d\udc1f Read MCAPSource (Python): 2,028 rows out, 0 B read | \ud83d\udde1\ufe0f \ud83d\udc1f Count: 2,028 rows in, 1 rows out | \ud83d\udde1\ufe0f \ud83d\udc1f Rename & Reorder: 1 rows in, 1 rows out\n", + "[00:00] 🗡️ 🐟 Read MCAPSource (Python): 2,028 rows out, 0 B read | 🗡️ 🐟 Count: 2,028 rows in, 1 rows out | 🗡️ 🐟 Rename & Reorder: 1 rows in, 1 rows out\n", "\n", "IMU messages in first 10s: 2028\n" ] @@ -401,18 +412,20 @@ }, { "cell_type": "markdown", + "id": "938c804e27f84196a10c8828c723f798", "metadata": {}, "source": [ "## 5. Per-topic keyframe alignment\n", "\n", "`topic_start_time_resolver` is a callable that takes a file path and returns `{topic: start_time_ns}`. The reader expands one scan task per `(file, topic)`, each starting at its own offset.\n", "\n", - "For video topics, that's the natural place to plug in your keyframe table \u2014 start decoding from an I-frame, not from the file head. Here we just skip the first 5 seconds of each topic relative to the file start to show the mechanism." + "For video topics, that's the natural place to plug in your keyframe table — start decoding from an I-frame, not from the file head. Here we just skip the first 5 seconds of each topic relative to the file start to show the mechanism." ] }, { "cell_type": "code", "execution_count": 13, + "id": "504fb2a444614c0babb325280ed9130a", "metadata": {}, "outputs": [ { @@ -485,7 +498,7 @@ "
\n", "
\n", "
Cell Details
\n", - " \n", + " \n", "
\n", "
\n", "

Click on a cell to view its full content

\n", @@ -604,13 +617,14 @@ " \"\"\"Return per-topic start offsets, in nanoseconds since epoch.\n", "\n", " In production this is where you'd look up an I-frame index next to each\n", - " recording \u2014 keyframe-aligned decoding for video topics, sensor-startup\n", + " recording — keyframe-aligned decoding for video topics, sensor-startup\n", " masking for IMU/LiDAR.\"\"\"\n", " return {\n", - " \"/livox/imu\": t0 + 5 * NS_PER_S,\n", + " \"/livox/imu\": t0 + 5 * NS_PER_S,\n", " \"/livox/lidar\": t0 + 5 * NS_PER_S,\n", " }\n", "\n", + "\n", "aligned = daft.read_mcap(\n", " mcap_path,\n", " topics=[\"/livox/imu\", \"/livox/lidar\"],\n", @@ -619,22 +633,23 @@ "\n", "(\n", " aligned.select(\"topic\", \"log_time\")\n", - " .groupby(\"topic\")\n", - " .agg(\n", - " daft.col(\"log_time\").min().alias(\"first_log_time_ns\"),\n", - " daft.col(\"log_time\").count().alias(\"messages\"),\n", - " )\n", - " .show()\n", + " .groupby(\"topic\")\n", + " .agg(\n", + " daft.col(\"log_time\").min().alias(\"first_log_time_ns\"),\n", + " daft.col(\"log_time\").count().alias(\"messages\"),\n", + " )\n", + " .show()\n", ")" ] }, { "cell_type": "markdown", + "id": "59bbdb311c014d738909a11f9e486628", "metadata": {}, "source": [ "## 6. Time-aligned join: IMU samples near every LiDAR sweep\n", "\n", - "The kind of query that's painful in raw MCAP \u2014 for every LiDAR frame, pull the IMU samples within \u00b150 ms. Bucket both streams by 50 ms, join on the bucket, project the timestamps and sequence numbers.\n", + "The kind of query that's painful in raw MCAP — for every LiDAR frame, pull the IMU samples within ±50 ms. Bucket both streams by 50 ms, join on the bucket, project the timestamps and sequence numbers.\n", "\n", "This is the metadata-level join. Decoding the LiDAR point cloud or fusing IMU rotation is a UDF step over the result." ] @@ -642,6 +657,7 @@ { "cell_type": "code", "execution_count": 14, + "id": "b43b363d81ae4b689946ece5c682cd59", "metadata": {}, "outputs": [ { @@ -717,7 +733,7 @@ "
\n", "
\n", "
Cell Details
\n", - " \n", + " \n", "
\n", "
\n", "

Click on a cell to view its full content

\n", @@ -836,25 +852,23 @@ "\n", "imu = (\n", " daft.read_mcap(mcap_path, topics=[\"/livox/imu\"])\n", - " .select(\n", - " daft.col(\"log_time\").alias(\"imu_ts\"),\n", - " daft.col(\"sequence\").alias(\"imu_seq\"),\n", - " )\n", - " .with_column(\"bucket\", daft.col(\"imu_ts\") // BUCKET_NS)\n", + " .select(\n", + " daft.col(\"log_time\").alias(\"imu_ts\"),\n", + " daft.col(\"sequence\").alias(\"imu_seq\"),\n", + " )\n", + " .with_column(\"bucket\", daft.col(\"imu_ts\") // BUCKET_NS)\n", ")\n", "\n", "lidar = (\n", " daft.read_mcap(mcap_path, topics=[\"/livox/lidar\"])\n", - " .select(\n", - " daft.col(\"log_time\").alias(\"lidar_ts\"),\n", - " daft.col(\"sequence\").alias(\"lidar_seq\"),\n", - " )\n", - " .with_column(\"bucket\", daft.col(\"lidar_ts\") // BUCKET_NS)\n", + " .select(\n", + " daft.col(\"log_time\").alias(\"lidar_ts\"),\n", + " daft.col(\"sequence\").alias(\"lidar_seq\"),\n", + " )\n", + " .with_column(\"bucket\", daft.col(\"lidar_ts\") // BUCKET_NS)\n", ")\n", "\n", - "joined = lidar.join(imu, on=\"bucket\").select(\n", - " \"lidar_seq\", \"lidar_ts\", \"imu_seq\", \"imu_ts\"\n", - ")\n", + "joined = lidar.join(imu, on=\"bucket\").select(\"lidar_seq\", \"lidar_ts\", \"imu_seq\", \"imu_ts\")\n", "\n", "print(\"matched (lidar, imu) pairs:\", joined.count_rows())\n", "joined.sort(\"lidar_seq\").show(5)" @@ -862,21 +876,22 @@ }, { "cell_type": "markdown", + "id": "8a65eabff63a45729fe45fb5ade58bdc", "metadata": {}, "source": [ "## What you'd build next\n", "\n", "- **Decode payloads in a UDF.** Register a per-topic decoder once (e.g. `cv2.imdecode` for compressed images, the Livox `CustomMsg` parser for LiDAR), call it on `data` only after filtering. Daft distributes the work across the cluster.\n", "- **Write filtered windows to a lakehouse.** Push the IMU/LiDAR join into an Iceberg table for downstream labeling, or land per-trip event tables in Delta.\n", - "- **Same code, fleet scale.** Point `daft.read_mcap` at an S3 prefix instead of a single file \u2014 the non-seekable stream wrapper means S3, GCS, and Azure all behave identically.\n", + "- **Same code, fleet scale.** Point `daft.read_mcap` at an S3 prefix instead of a single file — the non-seekable stream wrapper means S3, GCS, and Azure all behave identically.\n", "\n", "### References\n", "\n", - "- [`daft.read_mcap` docs](https://docs.daft.ai/) \u2014 full API reference\n", - "- [mcap.dev](https://mcap.dev/) \u2014 format spec, CLI, decoder libraries\n", - "- [foxglove.dev](https://foxglove.dev/) \u2014 visualization for the recordings you build queries on top of\n", - "- [DapengFeng/MCAP](https://huggingface.co/datasets/DapengFeng/MCAP) \u2014 the SLAM dataset used in this notebook\n", - "- [Daft PR #5886](https://github.com/Eventual-Inc/Daft/pull/5886) \u2014 the `topic_start_time_resolver` and non-seekable reader landing in v0.7.2" + "- [`daft.read_mcap` docs](https://docs.daft.ai/) — full API reference\n", + "- [mcap.dev](https://mcap.dev/) — format spec, CLI, decoder libraries\n", + "- [foxglove.dev](https://foxglove.dev/) — visualization for the recordings you build queries on top of\n", + "- [DapengFeng/MCAP](https://huggingface.co/datasets/DapengFeng/MCAP) — the SLAM dataset used in this notebook\n", + "- [Daft PR #5886](https://github.com/Eventual-Inc/Daft/pull/5886) — the `topic_start_time_resolver` and non-seekable reader landing in v0.7.2" ] } ],