Querying Catalogs

Demonstrates three levels of catalog search: finding topics by partial name, filtering by sensor type, and running a multi-domain query that locates specific physical events — such as IMU lateral acceleration spikes — across the entire dataset in a single server-side call.

Python
C++
Rust

Learn

Doc: The Query Workflow

API Reference: Query Builders

API Reference: Query Response

Run
mosaicolabs.examples query_catalogs

Run the command with --help to see all available options.

The daemon must be running and contain data ingested via the ROS Ingestion example. The full source is on GitHub.

The Query Builder Pattern

Mosaico's query API uses fluent builder objects. Passing multiple builders to client.query() joins them with a logical AND, evaluated entirely server-side in a single round trip. Nothing is joined or filtered on the client side:

Two builders joined server-side
client.query(
    QueryTopic().with_name_match("*imu*"),      # AND
    QueryOntologyCatalog(IMU.Q.acceleration.y.geq(1.0)),
)

The sections below build up from a single builder to combinations of them.

Finding Topics by Name

QueryTopic.with_name_match() matches topic paths against a glob-style pattern rather than an exact string. A plain string like "image_raw" requires an exact match; wrap it in wildcards (e.g. "*image_raw*") to search for topics whose path merely contains that substring. See The Query Workflow for the full set of supported wildcards (*, ?, [], #). The result is a QueryResponse grouped by sequence, so each item contains the parent session name alongside the matching topic list.

Wildcard topic name search
from mosaicolabs import MosaicoClient, QueryTopic

with MosaicoClient.connect(host="localhost", port=6726) as client:
    results = client.query(
        QueryTopic().with_name_match("*image_raw*")
    )
    if results:
        for item in results:
            print(f"Sequence: {item.sequence.name}")
            for topic in item.topics:
                print(f"  {topic.name}")

Filtering by Sensor Type

with_ontology_tag() queries by the semantic type of the data rather than by path string. The query stays valid if topics are renamed, as long as the sensor type is unchanged.

Filter by sensor type
from mosaicolabs import IMU, QueryTopic

results = client.query(
    QueryTopic().with_ontology_tag(IMU.ontology_tag())
)
# Returns every IMU topic across all sequences in the catalog.

Multi-Domain Queries

QueryOntologyCatalog combined with QueryTopic lets you filter by both sensor type and field value in one call. The .Q proxy provides type-safe dot-notation expressions: IMU.Q.acceleration.y.geq(1.0) means "find messages where IMU y-axis acceleration is >= 1.0 m/s²". Here QueryTopic uses with_name(), i.e. an exact topic path, instead of the with_ontology_tag() seen above, since the physical filter on IMU.Q... already narrows the search to IMU data; restricting further to one specific topic path pins down which IMU stream the acceleration threshold applies to.

Multi-domain query with physics filter
from mosaicolabs import IMU, QueryOntologyCatalog, QueryTopic

results = client.query(
    QueryOntologyCatalog(
        IMU.Q.acceleration.y.geq(1.0)
    ),
    QueryTopic().with_name("/front_stereo_imu/imu")
)

Replaying an Event

note

This pattern is not part of the query_catalogs example script itself; it illustrates a natural next step once you have a query response, combining clusterize() with get_data_streamer() (see the Data Inspection example).

The returned query response can be used directly to slice a data stream to the exact windows containing the event. The example adds one second of padding on each side to capture the run-up and recovery.

Slice and replay the event window
if results:
    for item in results:
        for topic in item.topics:

            clusters = topic.clusterize()

            for cluster in clusters:
                streamer = client.sequence_handler(item.sequence.name).get_data_streamer(
                    topics=[topic.name],
                    start_timestamp_ns=cluster.timerange.start - 1_000_000_000,
                    end_timestamp_ns=cluster.timerange.end   + 1_000_000_000,
                )
                for msg in streamer:
                    pass  # process the event window

Which Builder For Which Question

Question	Builder	Method
"Which topics match this name pattern?"	`QueryTopic`	`with_name_match()`
"Which topics carry this sensor type?"	`QueryTopic`	`with_ontology_tag()`
"Which topics satisfy this physical condition?"	`QueryOntologyCatalog`	`.Q` proxy expression, e.g. `IMU.Q.acceleration.y.geq(1.0)`

The Query Builder Pattern​

Finding Topics by Name​

Filtering by Sensor Type​

Multi-Domain Queries​

Replaying an Event​

Which Builder For Which Question​