企业官网建设流程全解析

Mosaico是一个面向 Physical AI 的数据平台（The data platform for Physical AI），其 Python SDK 内置了一套流畅的查询工作流，让你能够按传感器原始物理值（如加速度 > 15 m/s²、点云距离 < 50 米）对存储在平台中的 Lidar、IMU 等时序数据进行服务端深度过滤，而不必把数据拉回本地再逐帧扫描。本文用 10 个实用技巧，带你从.Q代理基础一步步掌握 IMU 冲击检测、Lidar 点云物理过滤、时间窗口聚类与跨传感器关联查询，快速定位最有价值的训练片段。

查询三要素：Builder、.Q代理与with_expression

在展开 10 招之前，先花 1 分钟理解 Mosaico 查询的三个核心构件（详见官方文档 The Query Workflow）：

多个 Builder 同时传给client.query()时，也会被逻辑AND连接，且整条请求在服务端原子执行——这是"按物理值过滤"高效的关键。

技巧 1：用.Q代理写出第一条 IMU 物理值过滤

IMU 本体模型定义了acceleration（线性加速度，m/s²）与angular_velocity（角速度，rad/s）等字段，全部可通过.Q代理逐级下钻。以"找出 X 轴加速度超过 15 m/s² 的冲击事件"为例：

from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog with MosaicoClient.connect("localhost", 6726) as client: qresponse = client.query( QueryOntologyCatalog(IMU.Q.acceleration.x.gt(15.0)) ) if qresponse is not None: for item in qresponse: print(f"Sequence: {item.sequence.name}") print(f"Topics: {[t.name for t in item.topics]}")

结果自动按 Sequence 分组返回（QueryResponse），item.topics中每一项都给出了归一化的 Topic 名称，可直接用于后续读取。IMU 的完整可查字段（含orientation四元数、header.timestamp等）见 imu.py。

数值字段支持的 8 个比较算子：.eq()、.lt()、.leq()、.gt()、.geq()、.between()、.in_()、.outside()。其中.between([min, max])是范围过滤的首选，后面会反复用到。

技巧 2：一次查询组合多个轴与多个条件（AND 链式）

.with_expression()可以连续链式调用，把"X 轴强减速 + Y 轴大幅偏移"这类复合物理条件合并为一次请求：

qresponse = client.query( QueryOntologyCatalog() .with_expression(IMU.Q.acceleration.x.lt(-4.0)) .with_expression(IMU.Q.acceleration.y.gt(5.0)) .with_expression(IMU.Q.angular_velocity.z.gt(1.0)) )

⚠️重要约束（Single Occurrence Constraint）：同一个字段路径在一个 Builder 实例中只能出现一次。想表达"加速度在 0.5~1.0 之间"，不要链两次.gt()/.lt()，而要用.between([0.5, 1.0])；查询同一模型的不同字段（acceleration.x与acceleration.y）则完全没问题。

技巧 3：Lidar 点云是列表字段，先量化再比较

Lidar 本体把每次扫描存成一组扁平列表：x/y/z坐标、intensity、range、beam_id、azimuth、elevation、confidence等，第i个元素对应第i个点（见 lidar.py）。列表字段不能直接比较，必须先用量词"收窄"到元素级别，再链上数值算子：

# 错误：Lidar.Q.range.gt(1) —— 列表不能直接比较 # 正确写法（.any() / .all() / 索引之后必须再链算子）： QueryOntologyCatalog(Lidar.Q.intensity.any().gt(200.0)) # 本帧存在强反射点 QueryOntologyCatalog(Lidar.Q.range.all().leq(50.0)) # 整帧点云都在 50 米内

记住这个"两步走"语法，是玩转所有点云查询的前提。

技巧 4：用range与.between()做距离/区域过滤

range字段是传感器原点到每个点的径向距离（米）。结合.between()即可把点云"裁剪"到感兴趣的物理区域：

# 只保留 X 坐标落在 ±1 米"近场"内的扫描帧 qresponse = client.query( QueryOntologyCatalog(Lidar.Q.x.all().between([-1.0, 1.0])) )

典型用途：近场障碍物检测片段、近距建图片段、剔除超远噪声点。between的语义是闭区间，边界值会被保留。

技巧 5：用beam_id按激光环过滤，再按方位角切扇区

多线激光雷达的beam_id记录每个点来自哪一圈发射器（等价于 ROSPointCloud2中的ring）。先锁定特定环，再叠加azimuth（方位角，弧度）扇区过滤：

qresponse = client.query( QueryOntologyCatalog() .with_expression(Lidar.Q.beam_id.any().eq(0)) # 关注第 0 环 .with_expression(Lidar.Q.azimuth.all().between([-1.57, 1.57])) # 只保留正前 ±90° 扇区 )

elevation（俯仰角）同理，可筛选地面/天空/地平线等不同高度带。这种"环 + 扇区"组合是激光雷达数据切片最常用的手法。

技巧 6：IMU 高值事件检测——冲击、急转、姿态异常

IMU 过滤是找出"事故/事件片段"的最高效手段。三个高频场景：

• 冲击/急刹：IMU.Q.acceleration.z.gt(19.6)（超过 2 倍重力，单位 m/s²）
• 急转弯：IMU.Q.angular_velocity.z.gt(1.0)（绕 Z 轴角速度超过 1 rad/s）
• 姿态异常：IMU.Q.orientation.z.gt(0.707)（四元数 Z 分量逼近 90° 翻滚）

from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog with MosaicoClient.connect("localhost", 6726) as client: qresponse = client.query( QueryOntologyCatalog( IMU.Q.acceleration.z.gt(15.0), ) )

这些条件的物理含义都是直观的：阈值用真实单位（m/s²、rad/s）书写，所见即所得。

技巧 7：用clusterize()把"匹配集合"切成时间窗口

查询返回的是一份"该 Topic 满足条件的完整时间范围"，通常还需要把它切成一个个离散事件。topic.clusterize(clustering_dt_ns=...)负责这件事：

• 间隔<clustering_dt_ns的匹配点会合并为同一个簇；
• 间隔≥该值的拆成不同簇，每个簇即一次独立事件；
• 默认0返回覆盖全程的单一区间。

from mosaicolabs import IMU, MosaicoClient, Query, QueryOntologyCatalog, QuerySequence with MosaicoClient.connect("localhost", 6726) as client: qresponse = client.query( query=Query( QuerySequence().with_name_match("robot-1"), QueryOntologyCatalog().with_expression(IMU.Q.acceleration.x.gt(5.0)), ) ) imu_topic = next(t for item in qresponse for t in item.topics if t.name == "front_imu") distinct_events = imu_topic.clusterize(clustering_dt_ns=int(1e8)) # 100ms 间隔 → 离散事件 active_window = imu_topic.clusterize(clustering_dt_ns=int(2e9)) # 2s 间隔 → 合并为活跃窗口

调参经验：IMU 是高频传感器，阈值给小（几十到几百毫秒）；GPS 等低频传感器要给大一些，否则采样间隔本身就会把一次事件切碎。时间窗口的完整机制见 temporal_windows.mdx。

技巧 8：intersect()关联 IMU 与 Lidar/GPS 的多传感器事件

单次查询里，同一 Builder 的多个条件只能作用在同一 Topic上。要回答"IMU 急减速且同时GPS 显示近乎静止"这类跨传感器问题，用item.intersect()：它会合并所有匹配 Topic 的表达式，返回所有条件同时成立的时间窗口。

# "急停"事件：IMU 强减速 + GPS 速度骤降，两者须发生在同一时刻 query = Query( QuerySequence().with_name_match("test_drive"), QueryOntologyCatalog() .with_expression(IMU.Q.acceleration.x.lt(-5.0)) .with_expression(GPS.Q.velocity.x.lt(1.0)), ) qresponse = client.query(query=query) for item in qresponse: hard_stop_windows = item.intersect( clustering_map={IMU.ontology_tag(): int(1e8), GPS.ontology_tag(): int(5e8)}, intersect_dt_ns=int(2e8), # 允许传感器间最多 200ms 时间漂移 )

intersect_dt_ns是关键旋钮：传感器采样时刻天然错开，给一点容差（百毫秒级）就能把"相机发现障碍物 → IMU 记录甩尾"这类因果链关联起来。Lidar 点云帧也可以参与同样的交集计算。

技巧 9：元数据 + 本体过滤的联合检索

真正好用的检索往往是"三层叠加"：会话名模式 + Topic 本体标签 + 物理值条件，一次client.query()全部完成：

from mosaicolabs import MosaicoClient, QuerySequence, QueryTopic, QueryOntologyCatalog, IMU with MosaicoClient.connect("localhost", 6726) as client: qresponse = client.query( QuerySequence().with_name_match("test_drive*"), # 第 1 层：会话名通配 QueryTopic().with_name("/sensors/imu"), # 第 2 层：指定 Topic QueryOntologyCatalog() .with_expression(IMU.Q.acceleration.x.gt(5.0)) # 第 3 层：物理值 .with_expression(IMU.Q.timestamp_ns.gt(1700134567)), # 附带时间下界 )

通配符*、?、[]、#可用于名称与元数据匹配（with_name_match()、with_user_metadata(..., match=...)），规则见 query.md 的 Wildcards 一节。

技巧 10：无类查询QueryableNumeric——查"未建模"的传感器数据

.Q代理需要一个 Python 类在手。但如果你入库的是未适配的自定义消息（经 ROS/MCAP 桥即时翻译、以本体标签如"GyroRaw"存储），类可能早已不存在。此时直接用字段路径字符串构造表达式即可，效果与.Q完全等价：

from mosaicolabs.query.queryable_fields import QueryableNumeric, QueryableString # 数值：找角速度 X 轴超过 0.5 rad/s 的序列（无需 GyroRaw 类） client.query(QueryOntologyCatalog().with_expression( QueryableNumeric("GyroRaw.gyro.x").gt(0.5))) # 字符串：找日志级别为 ERROR 的记录 client.query(QueryOntologyCatalog().with_expression( QueryableString("DiagnosticLog.level").eq("ERROR")))

注意：无类查询没有客户端 schema 校验——字段路径写错不会报错，只会匹配不到任何数据，所以路径要仔细核对。列表字段也可用字段[0]、字段[!]（全部）、字段[?]（任一）选择器表达，规则见 queryable_fields_queries.mdx。

快速对照表：10 招速查

延伸阅读

• 查询工作流完整机制（Builder、.Q代理、响应模型）：docs/py/docs/SDK/query.md
• 无类查询（Queryable Fields）详解：docs/main/docs/learn/queryable_fields_queries.mdx
• 时间窗口（clusterize / intersect）：docs/main/docs/learn/temporal_windows.mdx
• IMU 本体与可查字段：mosaico-sdk-py/src/mosaicolabs/models/sensors/imu.py
• Lidar 本体与可查字段：mosaico-sdk-py/src/mosaicolabs/models/futures/lidar.py

掌握这 10 招后，你就具备了在 Mosaico 平台上按物理语义"像写 SQL WHERE 子句一样"检索 Lidar 与 IMU 数据的能力——把海量录制直接浓缩成高质量训练片段，正是 Physical AI 数据闭环的核心一步。