Response

mosaicolabs.models.query.response

TimestampRange dataclass

TimestampRange(start, end)

Represents a temporal window defined by a start and end timestamp.

This utility class is used to define the bounds of sensor data or sequences within the Mosaico archive.

Attributes:

Name	Type	Description
start	int	The beginning of the range (inclusive), typically in nanoseconds.
end	int	The end of the range (inclusive), typically in nanoseconds.

QueryResponseItemSequence dataclass

QueryResponseItemSequence(name)

Metadata container for a single sequence discovered during a query.

Attributes:

Name	Type	Description
name	str	The unique identifier of the sequence in the Mosaico database.

QueryResponseItemTopic dataclass

QueryResponseItemTopic(name, timestamp_range)

Metadata for a specific topic (sensor stream) within a sequence.

Contains information about the topic's identity and its available time range in the archive.

Attributes:

Name	Type	Description
name	str	The name of the topic (e.g., 'front_camera/image_raw').
timestamp_range	Optional[TimestampRange]	The availability window of the data for this specific topic.

QueryResponseItem dataclass

QueryResponseItem(sequence, topics)

A unified result item representing a sequence and its associated topics.

This serves as the primary unit of data returned when querying the Mosaico metadata catalog.

Attributes:

Name	Type	Description
sequence	QueryResponseItemSequence	The parent sequence metadata.
topics	List[QueryResponseItemTopic]	The list of topics available within this sequence that matched the query criteria.

QueryResponse dataclass

QueryResponse(items=list())

An iterable collection of results returned by a Mosaico metadata query.

This class provides convenience methods to transform search results back into query builders, enabling a fluid, multi-stage filtering workflow.

Example

from mosaicolabs import MosaicoClient, IMU, Floating64, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter IMU data by a specific acquisition second
    qresponse = client.query(
        QueryOntologyCatalog(IMU.Q.header.stamp.sec.lt(1770282868))
    )

    # Inspect the response
    if qresponse is not None:
        # Results are automatically grouped by Sequence for easier data management
        for item in qresponse:
            print(f"Sequence: {item.sequence.name}")
            print(f"Topics: {[topic.name for topic in item.topics]}")

    # Filter primitive Floating64 telemetry by frame identifier
    qresponse = client.query(
        QueryOntologyCatalog(Floating64.Q.header.frame_id.eq("robot_base"))
    )

    # Inspect the response
    if qresponse is not None:
        # Results are automatically grouped by Sequence for easier data management
        for item in qresponse:
            print(f"Sequence: {item.sequence.name}")
            print(f"Topics: {[topic.name for topic in item.topics]}")

Attributes:

Name	Type	Description
items	List[QueryResponseItem]	The list of items matching the query.

to_query_sequence

to_query_sequence()

Converts the current response into a QuerySequence builder.

This allows for further filtering or operations on the specific set of sequences returned in this response.

Example

This demonstrates query chaining to narrow your search to specific sequences and topics. This is necessary when criteria span different data channels; otherwise, the resulting filters chained in AND in a single query would produce an empty result.

from mosaicolabs import MosaicoClient, QuerySequence

with MosaicoClient.connect("localhost", 6726) as client:
    # Broad Search: Find sequences with high-precision GPS
    initial_response = client.query(QueryOntologyCatalog(GPS.Q.status.status.eq(2)))

    # Chaining: Use results to "lock" the domain and find specific data in those sequences
    # on different data channels
    if not initial_response.is_empty():
        final_response = client.query(
            initial_response.to_query_sequence(),              # The "locked" sequence domain
            QueryTopic().with_name("/localization/log_string"), # Target a specific log topic
            QueryOntologyCatalog(String.Q.data.match("[ERR]"))  # Filter by content
        )

Returns:

Name	Type	Description
QuerySequence	QuerySequence	A builder initialized with an '$in' filter on the sequence names.

Raises:

Type	Description
ValueError	If the response is empty.

to_query_topic

to_query_topic()

Converts the current response into a QueryTopic builder.

Useful for narrowing down a search to specific topics found within the retrieved sequences.

Example

from mosaicolabs import MosaicoClient, QueryTopic

with MosaicoClient.connect("localhost", 6726) as client:
    # Broad Search: Find sequences with high-precision GPS
    initial_response = client.query(
            QueryTopic().with_name("/localization/log_string"), # Target a specific log topic
            QuerySequence().with_name_match("test_winter_2025_")  # Filter by content
        )

    # Chaining: Use results to "lock" the domain and find specific log-patterns in those sequences
    if not initial_response.is_empty():
        final_response = client.query(
            initial_response.to_query_topic(),              # The "locked" topic domain
            QueryOntologyCatalog(String.Q.data.match("[ERR]"))  # Filter by content
        )

Returns:

Name	Type	Description
QueryTopic	QueryTopic	A builder initialized with an '$in' filter on the topic names.

Raises:

Type	Description
ValueError	If the response is empty.

is_empty

is_empty()

Returns True if the response contains no results.