用 Rerun 可视化实时空中交通数据：GeoPandas + pyproj + `send_columns` 实战解析

用 Rerun 可视化实时空中交通数据：GeoPandas + pyproj +send_columns实战解析

【免费下载链接】rerunVisualize, query, and stream to train on multimodal robotics data.项目地址: https://gitcode.com/GitHub_Trending/re/rerun

本教程围绕 Rerun 官方示例 examples/python/air_traffic_data 展开，演示如何把 INVOLI 提供的真实空中交通数据（航班位置、高度、航速等）接入 Rerun viewer：在地图视图（Map View）与 3D 视图中同时呈现航班轨迹，用 pyproj 完成坐标系重投影，用 GeoPandas 加载真实国界地图数据，并对比行式rr.log与列式rr.send_columns两种日志写入策略。读完本文，你将掌握一套可复用的"地理空间数据 + Rerun 多视图"的可视化工程方案。

示例概览：一个多视图的空中交通数据看板

示例的核心脚本是 air_traffic_data.py，它展示了 Rerun viewer 的多个能力维度：

• 使用**地图视图（Map View）**显示地理空间原始图元（GeoPoints等）；
• 使用pyproj将地理数据从一个坐标系变换到另一个坐标系（EPSG:4326 经纬度 → 最适合当前区域的 UTM 投影坐标系）；
• 使用GeoPandas加载真实行政边界矢量数据（EU NUTS 边界），并投影到 3D 视图；
• 使用Polars对数据分批（batch），再通过rr.send_columns()一次调用批量下发（--batch模式）。

示例代码内部构建了一个固定布局的 blueprint：上方横向并排一个 3D 视图（Spatial3DView）与一个地图视图（MapView），下方是一个时间序列视图（TimeSeriesView），并设置行高比例为row_shares=[3, 1]：

blueprint = rrb.Vertical( rrb.Horizontal(rrb.Spatial3DView(origin="/"), rrb.MapView(origin="/")), rrb.TimeSeriesView(origin="/aircraft"), row_shares=[3, 1], ) rr.script_setup(args, "rerun_example_air_traffic_data", default_blueprint=blueprint)

关于地图视图的能力边界，可以参考仓库内文档 map_view.md：它是一个 2D 地图视图，用于展示地理空间图元，可配置zoom（缩放级别）与background（底图），当前可可视化的 archetype 为GeoLineStrings与GeoPoints；其中GeoPoints的位置以 EPSG:4326 经纬度（北/东为正的度数）表达，可选颜色与半径。

运行示例

环境准备

示例要求本机已安装 Rerun SDK 并克隆本仓库（Rerun 最新版本 SDK 对应的 Python 版本要求见仓库文档），随后按 README 完成安装：

pip install --upgrade rerun-sdk # 安装最新版 Rerun SDK git clone [e-mail]:rerun-io/rerun.git # 克隆仓库 cd rerun git checkout latest # 检出与最新 SDK 发布版匹配的提交

示例的 Python 依赖声明在 pyproject.toml 中，包括geopandas、numpy、polars、pyproj、requests、rerun-sdk、shapely、tqdm。以可编辑模式安装：

pip install -e examples/python/air_traffic_data

启动与参数

直接运行主脚本即可看到效果：

python -m air_traffic_data

如需定制行为或保存数据，用--help查看全部命令行参数：

python -m air_traffic_data --help

从源码看，示例实际支持的 CLI 参数包括：

参数	默认值	说明
--dataset	2h	选择自动下载的数据集，可选10min或2h（见下方"数据集"一节）
--raw	False（store_true）	若开启，直接记录含缺失值等问题的原始数据，用于在 viewer 中压测边界情况
--batch	True	若开启，使用批处理 logger（基于rr.send_columns，要求 Rerun 0.18+）
--batch-size	8192	批处理 logger 的批次大小
--dir	无	使用本地目录中的 JSON 数据，而不是下载数据集

此外，rr.script_add_args(args)还会注入 Rerun 脚本的标准参数（如连接/服务/保存等），使得该示例可以直接以rerun支持的各种运行方式（连接 viewer、起 server、保存.rrd文件）启动。

数据集与数据模型

INVOLI 数据集

数据由 INVOLI 提供，脚本内置了两个可直接下载的数据集（air_traffic_data.py）：

INVOLI_DATASETS = { "10min": "https://storage.googleapis.com/rerun-example-datasets/involi/involi_demo_set_1_10min.zip", "2h": "https://storage.googleapis.com/rerun-example-datasets/involi/involi_demo_set_2_2h.zip", }

下载带 tqdm 进度条、按 chunk 流式写入内存（io.BytesIO），并校验content-length是否一致；随后解压到仓库内examples/python/air_traffic_data/dataset/目录（首次运行时自动创建）。如果你已经有同类 JSON 数据，也可以用--dir指向本地目录，脚本会递归收集该目录下所有*.json文件（get_paths_for_directory）。

Measurement 数据模型

每一条航班记录被建模为Measurementdataclass（air_traffic_data.py），字段与原始 JSON 的对应关系如下：

字段	类型	来源
icao_id	str	data["ids"]["icao"]（ICAO 24 位地址码，同时用作实体路径）
latitude/longitude	float \| None	data.get("latitude")等，允许缺失
barometric_altitude	float \| None	气压高度（用于 3D 点位与时间序列）
wg84_altitude	float \| None	WGS84 高度
course	float \| None	航向
ground_speed	float \| None	地速
vertical_speed	float \| None	垂直速度
ground_status	str \| None	地面状态（如起降阶段标记）
timestamp	float	data["timestamp"][0] + data["timestamp"][1] / 1e9，即"秒 + 纳秒"组合时间戳换算成浮点秒

注意原始数据中大量字段允许为空（None），这一特点直接决定了后面批处理 logger 需要按字段分别drop_nulls再下发的设计。

坐标系变换：从经纬度到最优 UTM 投影

这是本示例最有价值的技术点之一：把地理坐标（经度、纬度）投影为适合展示的平面坐标，同时保留一份经纬度原始坐标用于地图视图。

自动挑选最优 UTM 分带

find_best_utm_crs（air_traffic_data.py）根据所有测量点的经纬度跨度计算AreaOfInterest，再通过pyproj.database.query_utm_crs_info(datum_name="WGS 84", area_of_interest=...)查询覆盖该区域的最佳 UTM 带，最终CRS.from_epsg(...)返回对应 CRS：

area_of_interest = get_area_of_interest(measurements) utm_crs_list = query_utm_crs_info(datum_name="WGS 84", area_of_interest=area_of_interest) return CRS.from_epsg(utm_crs_list[0].code)

因为航班数据集中在欧洲（德国、瑞士上空），自动选出的 UTM 带天然覆盖该区域，无需硬编码投影参数。

Transformer 的使用

随后构造一个"永远 x=经度、y=纬度"顺序的变换器（always_xy=True避免经纬度顺序歧义）：

proj = Transformer.from_crs("EPSG:4326", utm_crs, always_xy=True)

在行式 logger 中，每个测量点这样投影为 3D 坐标：

rr.log( entity_path, rr.Points3D( [self._proj.transform(measurement.longitude, measurement.latitude, measurement.barometric_altitude)], colors=color, ), rr.GeoPoints(lat_lon=[measurement.latitude, measurement.longitude]), )

一次rr.log同时写入两类图元：Points3D（UTM 平面坐标 + 高度）进入 3D 视图，GeoPoints（保留经纬度）进入地图视图。航线所在实体的颜色由实体路径字符串确定性生成：rr.components.Color.from_string(entity_path)，同一架飞机在多个视图中颜色一致。

高度夸张与 Transform3D

由于航班高度（约万米级）相对横向距离尺度极小，脚本对整棵aircraft实体树施加了一个静态的 z 轴 10 倍缩放：

rr.log("aircraft", rr.Transform3D(scale=[1, 1, 10]), static=True)

这是 Rerun 典型的"静态数据 + 运行时变换"组合用法：static=True表示该变换不会随时间变化，一次性写入即可作用于其下所有子实体。

用 GeoPandas 加载真实地图边界

为了在 3D/2D 视图中显示国界、区域边界，脚本会从欧盟 GISCO 服务下载 NUTS 2021 边界数据（1:1000 万比例尺，EPSG:4326），解压到examples/python/air_traffic_data/dataset/map_data/，然后用 GeoPandas 读取并重投影到之前选定的 UTM CRS：

map_data = ( gpd.read_file(MAP_DATA_DIR / f"NUTS_RG_01M_2021_4326_LEVL_{level}.json") .set_crs("epsg:4326") .to_crs(crs) )

对每个属于目标国家（示例硬编码为德国DE、瑞士CH，level 0）的多边形：

1. 用shapely_geom_to_numpy把Polygon/MultiPolygon（含内环interiors）转成 numpy 坐标数组；
2. 同一实体路径下同时记录 2D 与 3D 两组LineStrips：2D 直接用平面坐标，3D 用np.hstack([line, np.zeros((len(line), 1))])补齐 z=0；
3. 以rr.AnyValues(**metadata)把该边界多边形的属性（如 NUTS_ID 等，剔除 geometry 字段）静态挂到实体上。

rr.log(entity_path + "/2D", rr.LineStrips2D(lines, colors=color), static=True) rr.log( entity_path + "/3D", rr.LineStrips3D([np.hstack([line, np.zeros((len(line), 1))]) for line in lines], colors=color), static=True, )

边界、地面状态等不随时间变化的数据都使用static=True，这正符合 Rerun 数据模型中"静态数据不占时间索引"的设计，详见 static.md。

两种日志写入策略：行式rr.log与列式rr.send_columns

示例最有教学价值之处在于，同一份数据分别实现了两种 logger（air_traffic_data.py），源码中用LoggerProtocol 统一接口（process_measurement+flush），log_everything根据--batch选择具体实现。

策略一：MeasurementLogger（逐行rr.log）

面向"数据在运行中逐步产生"的场景，逐条处理每条测量记录：

rr.set_time("unix_time", timestamp=measurement.timestamp) # ... 投影后的 Points3D + GeoPoints ... rr.log(entity_path, rr.AnyValues(**metadata)) # 其他属性 rr.log(entity_path + "/barometric_altitude", rr.Scalars(...), rr.SeriesLines(colors=color)) # 高度时间序列

要点：

• 时间通过rr.set_time("unix_time", ...)设置，实体路径按aircraft/{icao_id}组织，icao_id天然成为实体标识（_ignored_fields中排除icao_id与timestamp，避免重复记录）；
• 在非--raw模式下，dict_factory会过滤掉值为None的字段，让记录更干净；
• 高度被单独记录为Scalars+SeriesLines，供下方的TimeSeriesView展示。

策略二：MeasurementBatchLogger（批量rr.send_columns）

当数据已经是列式存储（例如从文件、数据库批量读出）时，逐行log会带来大量小调用。send_columns允许一次调用更新实体在多个时间点的状态，是更高效的方案。相关概念可参考仓库文档 send-columns.md：与面向行的logAPI 不同，send_columns会绕过时间上下文与微批处理（micro-batching），且不会自动添加log_time/log_tick等内置时间线，只包含调用中显式传入的时间列。

批处理 logger 的核心流程（air_traffic_data.py）：

df = polars.DataFrame(self._measurements).sort("timestamp") # 原始数据未按时间排序，先排序 for (icao_id,), group in df.group_by("icao_id"): self.log_position_and_altitude(group, icao_id) self.log_ground_status(group, icao_id) self.log_metadata(group, icao_id)

由于原始数据各字段缺失模式不同，位置/高度、地面状态、航向等元数据被拆成三次独立的send_columns调用，各自drop_nulls后再下发。位置与高度部分（air_traffic_data.py）最为典型：

timestamps = rr.TimeColumn("unix_time", timestamp=df["timestamp"].to_numpy()) pos = self._proj.transform(df["longitude"], df["latitude"], df["barometric_altitude"]) rr.send_columns( entity_path, [timestamps], [ *rr.Points3D.columns(positions=np.vstack(pos).T), *rr.GeoPoints.columns(positions=np.vstack((df["latitude"], df["longitude"])).T), *rr.AnyValues.columns(latitude=..., longitude=..., barometric_altitude=...), ], ) rr.send_columns( entity_path + "/barometric_altitude", [timestamps], rr.Scalars.columns(scalars=df["barometric_altitude"].to_numpy()), )

几个值得注意的实现细节：

• 先注册静态指示器（indicator）：每个飞机实体第一次出现时，用rr.Points3D.from_fields(colors=color)、rr.SeriesLines.from_fields(colors=color)配合static=True写入一次，用于声明颜色等静态属性；此后send_columns不再重复传颜色。源码中对此有注释说明（_position_indicators集合去重），且留有一处已知限制的 TODO：GeoPoints.from_fields(colors=...)在 untagged index 下暂时是未定义行为，因此批处理模式下地图视图的点位颜色由 3D 指示器间接决定。
• 批大小默认 8192：process_measurement累积满self._batch_size即触发flush；最终在log_everything末尾还会再flush()一次，确保残留数据被写入。
• 时间列与组件列分离：rr.TimeColumn("unix_time", ...)作为索引列，*.columns(...)返回组件列；send_columns把"若干索引列 + 若干组件列"一次性提交，等价于多条时间戳的批量log。对应概念详见 chunks.md 与 chunk-processing-api.md。
• 排序的收益：get_paths_for_directory用自然排序（human sort，re.split(r"(\d+)")按数字段排序）整理 JSON 文件路径。虽然 Rerun 本身接受乱序数据，但乱序会带来（较小的）性能开销，日志侧先做廉价排序总是更优——源码注释明确说明了这一取舍。

实体层级与数据组织

从代码可以归纳出完整的实体树设计（供读者参考其组织方式）：

aircraft/ # Transform3D(scale=[1,1,10]) 静态变换 ├── {icao_id}/ # Points3D + GeoPoints + AnyValues(航向/地速/垂直速度/地面状态) │ └── barometric_altitude/ # Scalars + SeriesLines（高度时间序列） region_boundaries/ ├── DE/{level}/{NUTS_ID}/ # LineStrips2D + LineStrips3D + AnyValues(属性) └── CH/{level}/{NUTS_ID}/

• 时间线统一使用unix_time（浮点秒），由rr.set_time（行式）或rr.TimeColumn（列式）写入；
• 静态数据（边界、变换、指示器）用static=True；
• 动态数据（每架飞机的轨迹、高度序列）按时间推进。

延伸：把方案迁移到自己的地理数据

综合上述实现，若你要在自己的项目里复刻这套可视化管线，核心步骤可归纳为：

1. 确定 CRS：用query_utm_crs_info+AreaOfInterest自动挑选覆盖数据范围的最优 UTM 带，或按需固定使用某个投影；
2. 构造 Transformer：Transformer.from_crs("EPSG:4326", target_crs, always_xy=True)，同时保留原始经纬度；
3. 数据分层：位置/轨迹写入Points3D（投影坐标）与GeoPoints（经纬度）双份；静态背景（边界、底图要素）以LineStrips*+static=True写入；时序属性（高度、速度）单独写成Scalars供时间序列视图消费；
4. 选写入策略：实时流式数据用行式rr.log；批量/列式数据（Polars/DataFrame/文件）用rr.send_columns+rr.TimeColumn，并按字段缺失模式分组下发；
5. 组织布局：用rrb.Vertical/rrb.Horizontal/rrb.Spatial3DView/rrb.MapView/rrb.TimeSeriesView构建多视图 dashboard，把 3D、地图、时序三个视角并置。

本示例的完整代码（air_traffic_data.py）、依赖声明（pyproject.toml）与官方说明（README.md）都位于本仓库内，可随时对照阅读；地图视图与列式 API 的完整参考分别见 map_view.md 与 send-columns.md。

【免费下载链接】rerunVisualize, query, and stream to train on multimodal robotics data.项目地址: https://gitcode.com/GitHub_Trending/re/rerun