Futures Models

mosaicolabs.models.futures.laser._LaserScanBase

Bases: BaseModel

Internal generic base model shared by laser scan ontologies.

Encodes the scan geometry, timing metadata, and range and intensity arrays that are common to both single-return and multi-echo laser scanners.

This class is not intended to be instantiated directly. Use one of the concrete subclasses: LaserScan or MultiEchoLaserScan.

Attributes:

Name	Type	Description
angle_min	float32	Start angle of the scan in radians.
angle_max	float32	End angle of the scan in radians.
angle_increment	float32	Angular step between consecutive beams in radians.
time_increment	float32	Time elapsed between consecutive beam measurements in seconds.
scan_time	float32	Total duration of one full scan in seconds.
range_min	float32	Minimum valid range value in meters; measurements below this threshold should be discarded.
range_max	float32	Maximum valid range value in meters; measurements above this threshold should be discarded.
ranges	float32	Range measurements for each beam. Shape depends on T.
intensities	float32	Intensity measurements for each beam, co-indexed with ranges (optional). Shape depends on T.

Querying with the .Q Proxy

Scalar fields on this model are fully queryable via the .Q proxy. ranges and intensities are declared by each concrete subclass, and their queryability depends on the return shape: single-return arrays (as on LaserScan) are queryable via all(), any() or index access [i], while multi-echo arrays (as on MultiEchoLaserScan) are not queryable, since nested lists are not supported by the .Q proxy.

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.angle_min	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.angle_max	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.angle_increment	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.time_increment	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.scan_time	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.range_min	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.range_max	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        # Find scans with a wide field of view and a long maximum range
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.range_max.gt(30.0))
            .with_expression(LaserScan.Q.angle_max.geq(3.14)),
        )

        # Inspect the response
        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

angle_min class-attribute instance-attribute

angle_min = MosaicoField(
    description="start angle of the scan in rad."
)

Start angle of the scan in radians.

Defines the angular position of the first beam in the sweep.Together with angle_max and angle_increment, it fully characterises the angular coverage of the scan.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.angle_min	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.angle_min.geq(-3.14))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

angle_max class-attribute instance-attribute

angle_max = MosaicoField(
    description="end angle of the scan in rad."
)

End angle of the scan in radians.

Defines the angular position of the last beam in the sweep. The total field of view of the scanner is angle_max - angle_min.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.angle_max	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.angle_max.geq(3.14))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

angle_increment class-attribute instance-attribute

angle_increment = MosaicoField(
    description="angular distance between measurements in rad."
)

Angular step between consecutive beams in radians.

The number of beams in a sweep can be derived as round((angle_max - angle_min) / angle_increment) + 1. A negative value indicates a clockwise scan direction.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.angle_increment	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        # Find high-resolution scans (small angular step)
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.angle_increment.lt(0.01))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

time_increment class-attribute instance-attribute

time_increment = MosaicoField(
    description="time between measurements in seconds."
)

Time elapsed between consecutive beam measurements, in seconds.

If the scanner is moving, this will be used in interpoling position of 3D points.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.time_increment	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.time_increment.lt(0.0001))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

scan_time class-attribute instance-attribute

scan_time = MosaicoField(
    description="time between scans in seconds."
)

Time between scans in seconds.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.scan_time	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        # Find sequences recorded at 10 Hz (scan_time ≈ 0.1 s)
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.scan_time.between([0.09, 0.11]))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

range_min class-attribute instance-attribute

range_min = MosaicoField(
    description="minimum range value in meters."
)

Minimum valid range value, in meters.

Measurements strictly below this threshold are outside the sensor's reliable operating range and should be discarded or treated as invalid during downstream processing.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.range_min	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.range_min.leq(0.1))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

range_max class-attribute instance-attribute

range_max = MosaicoField(
    description="maximum range value in meters."
)

Maximum valid range value, in meters.

Measurements strictly above this threshold exceed the sensor's maximum detection distance and should be discarded or treated as invalid during downstream processing.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.range_max	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        # Find long-range scanner sessions
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.range_max.gt(30.0))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

mosaicolabs.models.futures.LaserScan

Bases: _LaserScanBase, Serializable, HeaderMixin

Single-return 2D laser scan data.

This model represents one sweep of a single-return laser range finder. Each beam yields exactly one range measurement, corresponding to the strongest or first detected echo.

ranges and intensities are flat List[float] whose i-th element corresponds to the beam at angular position angle_min + i * angle_increment.

Attributes:

Name	Type	Description
angle_min	float32	Start angle of the scan in radians.
angle_max	float32	End angle of the scan in radians.
angle_increment	float32	Angular step between consecutive beams in radians.
time_increment	float32	Time between consecutive beam measurements in seconds.
scan_time	float32	Total duration of one full scan in seconds.
range_min	float32	Minimum valid range threshold in meters.
range_max	float32	Maximum valid range threshold in meters.
ranges	SingleRange	Measured distance per beam in meters.
intensities	Optional[SingleRange]	Signal amplitude per beam (optional).

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. Scalar fields support the standard numeric operators directly. ranges and intensities are list-typed: use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - LaserScan.Q.ranges.all() -> invalid expression - LaserScan.Q.ranges.gt(1) -> invalid expression - LaserScan.Q.ranges.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.angle_min	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.angle_max	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.angle_increment	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.time_increment	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.scan_time	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.range_min	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.range_max	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.ranges.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.ranges.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.ranges.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.intensities.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.intensities.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.intensities.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.model.futures import LaserScan

with MosaicoClient.connect("localhost", 6726) as client:
    # Find long-range, wide-FOV scans
    qresponse = client.query(
        QueryOntologyCatalog(LaserScan.Q.range_max.gt(30.0))
            .with_expression(LaserScan.Q.angle_max.geq(3.14)),
    )

    if qresponse is not None:
        for item in qresponse:
            print(f"Sequence: {item.sequence.name}")
            print(f"Topics: {[topic.name for topic in item.topics]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

header class-attribute instance-attribute

header = MosaicoField(
    nullable=True,
    default=None,
    description="Contains measure metadata like timestamp, reference frame and samples counter.",
)

Measure header containing measurement timestamp and reference frame.

Querying with the .Q Proxy

Header components are queryable through the header field prefix.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.header.timestamp.seconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.timestamp.nanoseconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.frame_id	String	.eq(), .match(), .in_(), .lt(), .gt(), .leq(), .geq(), .between(), .outside()
<Model>.Q.header.sample_counter	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, ForceTorque, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Find where the measure lasts at least 10 seconds
    qresponse = client.query(QueryOntologyCatalog(ForceTorque.Q.header.timestamp.seconds.gt(10.0)))

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

angle_min class-attribute instance-attribute

angle_min = MosaicoField(
    description="start angle of the scan in rad."
)

Start angle of the scan in radians.

Defines the angular position of the first beam in the sweep.Together with angle_max and angle_increment, it fully characterises the angular coverage of the scan.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.angle_min	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.angle_min.geq(-3.14))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

angle_max class-attribute instance-attribute

angle_max = MosaicoField(
    description="end angle of the scan in rad."
)

End angle of the scan in radians.

Defines the angular position of the last beam in the sweep. The total field of view of the scanner is angle_max - angle_min.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.angle_max	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.angle_max.geq(3.14))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

angle_increment class-attribute instance-attribute

angle_increment = MosaicoField(
    description="angular distance between measurements in rad."
)

Angular step between consecutive beams in radians.

The number of beams in a sweep can be derived as round((angle_max - angle_min) / angle_increment) + 1. A negative value indicates a clockwise scan direction.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.angle_increment	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        # Find high-resolution scans (small angular step)
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.angle_increment.lt(0.01))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

time_increment class-attribute instance-attribute

time_increment = MosaicoField(
    description="time between measurements in seconds."
)

Time elapsed between consecutive beam measurements, in seconds.

If the scanner is moving, this will be used in interpoling position of 3D points.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.time_increment	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.time_increment.lt(0.0001))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

scan_time class-attribute instance-attribute

scan_time = MosaicoField(
    description="time between scans in seconds."
)

Time between scans in seconds.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.scan_time	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        # Find sequences recorded at 10 Hz (scan_time ≈ 0.1 s)
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.scan_time.between([0.09, 0.11]))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

range_min class-attribute instance-attribute

range_min = MosaicoField(
    description="minimum range value in meters."
)

Minimum valid range value, in meters.

Measurements strictly below this threshold are outside the sensor's reliable operating range and should be discarded or treated as invalid during downstream processing.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.range_min	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.range_min.leq(0.1))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

range_max class-attribute instance-attribute

range_max = MosaicoField(
    description="maximum range value in meters."
)

Maximum valid range value, in meters.

Measurements strictly above this threshold exceed the sensor's maximum detection distance and should be discarded or treated as invalid during downstream processing.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.range_max	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        # Find long-range scanner sessions
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.range_max.gt(30.0))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

ranges class-attribute instance-attribute

ranges = MosaicoField(
    description="range data in meters. Ranges need to be between range min and max otherwise discarded."
)

Range measurements for each beam.

A flat list of float values, one per beam, representing the measured distance in meters.

Values outside the [range_min, range_max] interval should be considered invalid.

Querying with the .Q Proxy

The range measurements are queryable via the ranges field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - LaserScan.Q.ranges.all() -> invalid expression - LaserScan.Q.ranges.gt(1) -> invalid expression - LaserScan.Q.ranges.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.ranges.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.ranges.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.ranges.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        # Find scans with at least one beam returning within 1 meter
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.ranges.any().leq(1.0))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

intensities class-attribute instance-attribute

intensities = MosaicoField(
    default=None, description="intensity data."
)

Intensity measurements for each beam (optional).

A flat list of float values, carries the signal amplitude of each beam.

Querying with the .Q Proxy

The intensity measurements are queryable via the intensities field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - LaserScan.Q.intensities.all() -> invalid expression - LaserScan.Q.intensities.gt(1) -> invalid expression - LaserScan.Q.intensities.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.intensities.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.intensities.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
LaserScan.Q.intensities.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        # Find scans with at least one high-intensity beam return
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.intensities.any().gt(200.0))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

Retrieves the unique identifier (tag) for the current ontology class, automatically generated during class definition.

This method provides the string key used by the Mosaico platform to identify and route specific data types within the ontology registry. It abstracts away the internal naming conventions, ensuring that you always use the correct identifier for queries and serialization.

Returns:

Name	Type	Description
str	str	The registered string tag for this class (e.g., "imu", "gps").

Raises:

Type	Description
Exception	If the class was not properly initialized via __pydantic_init_subclass__.

Practical Application: Topic Filtering

This method is particularly useful when constructing QueryTopic requests. By using the convenience method QueryTopic.with_ontology_tag(), you can filter topics by data type without hardcoding strings that might change.

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value (using constructor)
    qresponse = client.query(
        QueryTopic(
            Topic.with_ontology_tag(IMU.ontology_tag()),
        )
    )

    # 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]}")

mosaicolabs.models.futures.MultiEchoLaserScan

Bases: _LaserScanBase, Serializable, HeaderMixin

Multi-echo 2D laser scan data.

This model represents one sweep of a multi-echo laser range finder. Multi-echo scanners record several range returns per beam, allowing the sensor to detect overlapping surfaces, semi-transparent objects such as vegetation or rain drops, and retroreflective targets simultaneously.

ranges and intensities are List[List[float]] arrays where the i-th inner list contains all echo returns for the beam at angular position angle_min + i * angle_increment, ordered from nearest to farthest. An empty inner list indicates no valid return for that beam.

Attributes:

Name	Type	Description
angle_min	float32	Start angle of the scan in radians.
angle_max	float32	End angle of the scan in radians.
angle_increment	float32	Angular step between consecutive beams in radians.
time_increment	float32	Time between consecutive beam measurements in seconds.
scan_time	float32	Total duration of one full scan in seconds.
range_min	float32	Minimum valid range threshold in meters.
range_max	float32	Maximum valid range threshold in meters.
ranges	MultiRange	List of echo distances per beam in meters; may contain multiple returns per beam.
intensities	Optional[MultiRange]	List of echo amplitudes per beam, co-indexed with ranges (optional).

Querying with the .Q Proxy

Scalar fields are fully queryable via the .Q proxy. ranges and intensities are not queryable: each is a nested list (a list of echoes per beam), and the .Q proxy does not support indexing or filtering into nested lists.

Field Access Path	Queryable Type	Supported Operators
MultiEchoLaserScan.Q.angle_min	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
MultiEchoLaserScan.Q.angle_max	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
MultiEchoLaserScan.Q.angle_increment	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
MultiEchoLaserScan.Q.time_increment	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
MultiEchoLaserScan.Q.scan_time	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
MultiEchoLaserScan.Q.range_min	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
MultiEchoLaserScan.Q.range_max	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import MultiEchoLaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        # Find long-range, wide-FOV multi-echo scans
        qresponse = client.query(
            QueryOntologyCatalog(MultiEchoLaserScan.Q.range_max.gt(30.0))
                .with_expression(MultiEchoLaserScan.Q.angle_max.geq(3.14)),
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

header class-attribute instance-attribute

header = MosaicoField(
    nullable=True,
    default=None,
    description="Contains measure metadata like timestamp, reference frame and samples counter.",
)

Measure header containing measurement timestamp and reference frame.

Querying with the .Q Proxy

Header components are queryable through the header field prefix.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.header.timestamp.seconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.timestamp.nanoseconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.frame_id	String	.eq(), .match(), .in_(), .lt(), .gt(), .leq(), .geq(), .between(), .outside()
<Model>.Q.header.sample_counter	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, ForceTorque, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Find where the measure lasts at least 10 seconds
    qresponse = client.query(QueryOntologyCatalog(ForceTorque.Q.header.timestamp.seconds.gt(10.0)))

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

angle_min class-attribute instance-attribute

angle_min = MosaicoField(
    description="start angle of the scan in rad."
)

Start angle of the scan in radians.

Defines the angular position of the first beam in the sweep.Together with angle_max and angle_increment, it fully characterises the angular coverage of the scan.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.angle_min	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.angle_min.geq(-3.14))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

angle_max class-attribute instance-attribute

angle_max = MosaicoField(
    description="end angle of the scan in rad."
)

End angle of the scan in radians.

Defines the angular position of the last beam in the sweep. The total field of view of the scanner is angle_max - angle_min.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.angle_max	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.angle_max.geq(3.14))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

angle_increment class-attribute instance-attribute

angle_increment = MosaicoField(
    description="angular distance between measurements in rad."
)

Angular step between consecutive beams in radians.

The number of beams in a sweep can be derived as round((angle_max - angle_min) / angle_increment) + 1. A negative value indicates a clockwise scan direction.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.angle_increment	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        # Find high-resolution scans (small angular step)
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.angle_increment.lt(0.01))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

time_increment class-attribute instance-attribute

time_increment = MosaicoField(
    description="time between measurements in seconds."
)

Time elapsed between consecutive beam measurements, in seconds.

If the scanner is moving, this will be used in interpoling position of 3D points.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.time_increment	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.time_increment.lt(0.0001))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

scan_time class-attribute instance-attribute

scan_time = MosaicoField(
    description="time between scans in seconds."
)

Time between scans in seconds.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.scan_time	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        # Find sequences recorded at 10 Hz (scan_time ≈ 0.1 s)
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.scan_time.between([0.09, 0.11]))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

range_min class-attribute instance-attribute

range_min = MosaicoField(
    description="minimum range value in meters."
)

Minimum valid range value, in meters.

Measurements strictly below this threshold are outside the sensor's reliable operating range and should be discarded or treated as invalid during downstream processing.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.range_min	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.range_min.leq(0.1))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

range_max class-attribute instance-attribute

range_max = MosaicoField(
    description="maximum range value in meters."
)

Maximum valid range value, in meters.

Measurements strictly above this threshold exceed the sensor's maximum detection distance and should be discarded or treated as invalid during downstream processing.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
LaserScan.Q.range_max	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

    from mosaicolabs import MosaicoClient, QueryOntologyCatalog
    from mosaicolabs.model.futures import LaserScan

    with MosaicoClient.connect("localhost", 6726) as client:
        # Find long-range scanner sessions
        qresponse = client.query(
            QueryOntologyCatalog(LaserScan.Q.range_max.gt(30.0))
        )

        if qresponse is not None:
            for item in qresponse:
                print(f"Sequence: {item.sequence.name}")
                print(f"Topics: {[topic.name for topic in item.topics]}")

                # Clusterize all topics within the sequence to extract the time intervals
                clusters_dict = item.clusterize_all()

                # Since clusterize_all() used default clustering_dt_ns, each topic will have
                # just one cluster representing the first and last moment the query was satisfied
                for t_name, clusters in clusters_dict.items():
                    print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

ranges class-attribute instance-attribute

ranges = MosaicoField(
    description="range data in meters. Ranges need to be between range min and max otherwise discarded."
)

Range measurements for each beam.

A list of lists, where the i-th inner list contains all echo distances returned by the i-th beam, ordered from nearest to farthest. An empty inner list indicates no valid return for that beam.

Values outside the [range_min, range_max] interval should be considered invalid.

Querying with the .Q Proxy

The ranges field is not queryable via the .Q proxy: it is a nested list (a list of echo distances per beam), and the .Q proxy does not support indexing or filtering into nested lists.

intensities class-attribute instance-attribute

intensities = MosaicoField(
    default=None, description="intensity data."
)

Intensity measurements for each beam. (optional).

A flat list of list of float value carries the signal amplitude of each returned echo.

Querying with the .Q Proxy

The intensities field is not queryable via the .Q proxy: it is a nested list (a list of echo amplitudes per beam), and the .Q proxy does not support indexing or filtering into nested lists.

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

Retrieves the unique identifier (tag) for the current ontology class, automatically generated during class definition.

This method provides the string key used by the Mosaico platform to identify and route specific data types within the ontology registry. It abstracts away the internal naming conventions, ensuring that you always use the correct identifier for queries and serialization.

Returns:

Name	Type	Description
str	str	The registered string tag for this class (e.g., "imu", "gps").

Raises:

Type	Description
Exception	If the class was not properly initialized via __pydantic_init_subclass__.

Practical Application: Topic Filtering

This method is particularly useful when constructing QueryTopic requests. By using the convenience method QueryTopic.with_ontology_tag(), you can filter topics by data type without hardcoding strings that might change.

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value (using constructor)
    qresponse = client.query(
        QueryTopic(
            Topic.with_ontology_tag(IMU.ontology_tag()),
        )
    )

    # 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]}")

mosaicolabs.models.futures.Radar

Bases: Serializable, HeaderMixin

Radar Ontology.

This model represents a set of detections acquired from a Radar sensor. Each detection corresponds to a target or a reflection point in the sensor's field of view, characterised by its position, optional velocity, and signal-quality metrics.

Each field is a flat list whose i-th element corresponds to the i-th detection in the scan.

Unlike a LiDAR, Radar detections are inherently sparse and carry additional electromagnetic attributes such as Radar Cross Section (RCS), Signal-to-Noise Ratio (SNR), and Doppler velocity, which are not available from purely optical sensors.

Attributes:

Name	Type	Description
x	list_(float32)	X coordinates of each detection in meters.
y	list_(float32)	Y coordinates of each detection in meters.
z	list_(float32)	Z coordinates of each detection in meters.
range	Optional[list_(float32)]	Radial distance from the sensor origin to each detection in meters (optional).
azimuth	Optional[list_(float32)]	Azimuth angle in radians for each detection (optional).
elevation	Optional[list_(float32)]	Elevation angle in radians for each detection (optional).
rcs	Optional[list_(float32)]	Radar Cross Section of each detection in dBm (optional).
snr	Optional[list_(float32)]	Signal-to-Noise Ratio of each detection in dB (optional).
doppler_velocity	Optional[list_(float32)]	Doppler radial velocity of each detection in m/s (optional).
vx	Optional[list_(float32)]	X component of the velocity of each detection in m/s (optional).
vy	Optional[list_(float32)]	Y component of the velocity of each detection in m/s (optional).
vx_comp	Optional[list_(float32)]	Ego-motion-compensated X velocity of each detection in m/s (optional).
vy_comp	Optional[list_(float32)]	Ego-motion-compensated Y velocity of each detection in m/s (optional).
ax	Optional[list_(float32)]	X component of the acceleration of each detection in m/s² (optional).
ay	Optional[list_(float32)]	Y component of the acceleration of each detection in m/s² (optional).
radial_speed	Optional[list_(float32)]	Radial speed of each detection in m/s (optional).

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter Radar data based on thresholds values within a QueryOntologyCatalog. Expressions entailing lists of values can be queried using any between all(), any() or index access [i] followed by the contained type supported operations.

Example

from mosaicolabs import MosaicoClient, QueryTopic
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Fetch all sequences that contain at least one Radar topic
    qresponse = client.query(QueryTopic().with_ontology_tag(Radar.ontology_tag()))

    if qresponse is not None:

        for item in qresponse.items:
            print(f"Sequence: {item.name}")
            print(f"Topics:   {[topic.name for topic in item.topics]}")

header class-attribute instance-attribute

header = MosaicoField(
    nullable=True,
    default=None,
    description="Contains measure metadata like timestamp, reference frame and samples counter.",
)

Measure header containing measurement timestamp and reference frame.

Querying with the .Q Proxy

Header components are queryable through the header field prefix.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.header.timestamp.seconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.timestamp.nanoseconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.frame_id	String	.eq(), .match(), .in_(), .lt(), .gt(), .leq(), .geq(), .between(), .outside()
<Model>.Q.header.sample_counter	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, ForceTorque, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Find where the measure lasts at least 10 seconds
    qresponse = client.query(QueryOntologyCatalog(ForceTorque.Q.header.timestamp.seconds.gt(10.0)))

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

x class-attribute instance-attribute

x = MosaicoField(description='x coordinates in meters.')

X coordinates of each detection, in meters.

Querying with the .Q Proxy

The X coordinates value are queryable via the x field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.x.all() -> invalid expression - Radar.Q.x.gt(1) -> invalid expression - Radar.Q.x.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.x.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.x.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.x.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar values on X within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.x.all().between([-1.0, 1.0]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

y class-attribute instance-attribute

y = MosaicoField(description='y coordinates in meters.')

Y coordinates of each detection, in meters.

Querying with the .Q Proxy

The Y coordinates value are queryable via the y field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.y.all() -> invalid expression - Radar.Q.y.gt(1) -> invalid expression - Radar.Q.y.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.y.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.y.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.y.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar values on Y within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.y.all().between([-1.0, 1.0]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

z class-attribute instance-attribute

z = MosaicoField(description='z coordinates in meters.')

Z coordinates of each detection, in meters.

Querying with the .Q Proxy

The Z coordinates value are queryable via the z field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.z.all() -> invalid expression - Radar.Q.z.gt(1) -> invalid expression - Radar.Q.z.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.z.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.z.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.z.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar values on Z within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.z.all().between([-1.0, 1.0]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

range class-attribute instance-attribute

range = MosaicoField(
    default=None, description="radial distance in meters."
)

Radial distance from the sensor origin to each detection, in meters.

Represents the straight-line distance along the beam axis.

Querying with the .Q Proxy

The range value is queryable via the range field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.range.all() -> invalid expression - Radar.Q.range.gt(1) -> invalid expression - Radar.Q.range.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.range.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.range.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.range.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar detections within 100 meters of the sensor
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.range.all().leq(100.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

azimuth class-attribute instance-attribute

azimuth = MosaicoField(
    default=None, description="azimuth angle in radians."
)

Horizontal (azimuth) angle of each detection in radians.

Measured in the sensor's horizontal plane, typically from 0 to 2π, with 0 aligned to the sensor's forward axis.

Querying with the .Q Proxy

The azimuth value is queryable via the azimuth field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.azimuth.all() -> invalid expression - Radar.Q.azimuth.gt(1) -> invalid expression - Radar.Q.azimuth.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.azimuth.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.azimuth.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.azimuth.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar detections within a specific azimuth range
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.azimuth.all().between([-1.57, 1.57]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

elevation class-attribute instance-attribute

elevation = MosaicoField(
    default=None, description="elevation angle in radians."
)

Vertical (elevation) angle of each detection in radians.

Measured from the sensor's horizontal plane; positive values point upward.

Querying with the .Q Proxy

The elevation value is queryable via the elevation field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.elevation.all() -> invalid expression - Radar.Q.elevation.gt(1) -> invalid expression - Radar.Q.elevation.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.elevation.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.elevation.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.elevation.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar detections within a specific elevation range
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.elevation.all().between([-0.26, 0.26]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

rcs class-attribute instance-attribute

rcs = MosaicoField(
    default=None, description="radar cross section in dBm."
)

Radar Cross Section (RCS) of each detection, in dBm.

Quantifies the effective scattering area of the target as seen by the sensor. Higher values typically correspond to larger or more reflective objects. Useful for target classification and false-positive filtering.

Querying with the .Q Proxy

The RCS value is queryable via the rcs field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.rcs.all() -> invalid expression - Radar.Q.rcs.gt(1) -> invalid expression - Radar.Q.rcs.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.rcs.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.rcs.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.rcs.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar detections with a large radar cross section
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.rcs.any().gt(10.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

snr class-attribute instance-attribute

snr = MosaicoField(
    default=None, description="signal to noise ratio in dB."
)

Signal-to-Noise Ratio (SNR) of each detection, in dB.

Indicates the quality of the received echo relative to background noise. Low-SNR detections are generally less reliable and may be filtered out during object-level processing.

Querying with the .Q Proxy

The SNR value is queryable via the snr field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.snr.all() -> invalid expression - Radar.Q.snr.gt(1) -> invalid expression - Radar.Q.snr.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.snr.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.snr.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.snr.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar detections with a low signal-to-noise ratio
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.snr.all().lt(5.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

doppler_velocity class-attribute instance-attribute

doppler_velocity = MosaicoField(
    default=None, description="doppler velocity in m/s."
)

Doppler radial velocity of each detection, in m/s.

Represents the component of the target's velocity along the sensor's line of sight, derived directly from the frequency shift of the returned signal. Positive values conventionally indicate motion away from the sensor.

Querying with the .Q Proxy

The doppler velocity value is queryable via the doppler_velocity field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.doppler_velocity.all() -> invalid expression - Radar.Q.doppler_velocity.gt(1) -> invalid expression - Radar.Q.doppler_velocity.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.doppler_velocity.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.doppler_velocity.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.doppler_velocity.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar detections moving away from the sensor
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.doppler_velocity.any().gt(0.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

vx class-attribute instance-attribute

vx = MosaicoField(
    default=None, description="x velocity in m/s."
)

X component of the estimated velocity of each detection, in m/s.

Expressed in the sensor frame. This is a Cartesian decomposition of the target velocity, as opposed to the purely radial doppler_velocity.

Querying with the .Q Proxy

The X velocity value is queryable via the vx field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.vx.all() -> invalid expression - Radar.Q.vx.gt(1) -> invalid expression - Radar.Q.vx.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.vx.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.vx.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.vx.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar detections moving fast along X
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.vx.any().gt(5.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

vy class-attribute instance-attribute

vy = MosaicoField(
    default=None, description="y velocity in m/s."
)

Y component of the estimated velocity of each detection, in m/s.

Expressed in the sensor frame. See vx for further context.

Querying with the .Q Proxy

The Y velocity value is queryable via the vy field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.vy.all() -> invalid expression - Radar.Q.vy.gt(1) -> invalid expression - Radar.Q.vy.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.vy.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.vy.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.vy.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar detections moving fast along Y
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.vy.any().gt(5.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

vx_comp class-attribute instance-attribute

vx_comp = MosaicoField(
    default=None,
    description="x compensated velocity in m/s.",
)

Ego-motion-compensated X velocity of each detection, in m/s.

Obtained by subtracting the host vehicle's own velocity from vx, yielding the detection's absolute velocity in the world frame along the X axis.

Querying with the .Q Proxy

The compensated X velocity value is queryable via the vx_comp field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.vx_comp.all() -> invalid expression - Radar.Q.vx_comp.gt(1) -> invalid expression - Radar.Q.vx_comp.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.vx_comp.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.vx_comp.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.vx_comp.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar detections that are stationary in the world frame
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.vx_comp.all().between([-0.5, 0.5]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

vy_comp class-attribute instance-attribute

vy_comp = MosaicoField(
    default=None,
    description="y compensated velocity in m/s.",
)

Ego-motion-compensated Y velocity of each detection, in m/s.

Analogous to vx_comp along the Y axis. See vx_comp for further context.

Querying with the .Q Proxy

The compensated Y velocity value is queryable via the vy_comp field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.vy_comp.all() -> invalid expression - Radar.Q.vy_comp.gt(1) -> invalid expression - Radar.Q.vy_comp.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.vy_comp.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.vy_comp.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.vy_comp.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar detections that are stationary in the world frame
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.vy_comp.all().between([-0.5, 0.5]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

ax class-attribute instance-attribute

ax = MosaicoField(
    default=None, description="x acceleration in m/s^2."
)

X component of the estimated acceleration of each detection, in m/s².

Available only on sensors that track detections across multiple scans and report per-point kinematic state (e.g. high-level object-list outputs).

Querying with the .Q Proxy

The X acceleration value is queryable via the ax field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.ax.all() -> invalid expression - Radar.Q.ax.gt(1) -> invalid expression - Radar.Q.ax.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.ax.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.ax.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.ax.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar detections that are accelerating hard along X
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.ax.any().gt(3.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

ay class-attribute instance-attribute

ay = MosaicoField(
    default=None, description="y acceleration in m/s^2."
)

Y component of the estimated acceleration of each detection, in m/s².

Analogous to ax along the Y axis. See ax for further context.

Querying with the .Q Proxy

The Y acceleration value is queryable via the ay field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.ay.all() -> invalid expression - Radar.Q.ay.gt(1) -> invalid expression - Radar.Q.ay.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.ay.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.ay.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.ay.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar detections that are accelerating hard along Y
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.ay.any().gt(3.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

radial_speed class-attribute instance-attribute

radial_speed = MosaicoField(
    default=None, description="radial speed in m/s."
)

Radial speed of each detection, in m/s.

Represents the magnitude of the velocity component along the line of sight, without sign convention. Distinct from doppler_velocity, which may carry a directional sign depending on the sensor's convention.

Querying with the .Q Proxy

The radial speed value is queryable via the radial_speed field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Radar.Q.radial_speed.all() -> invalid expression - Radar.Q.radial_speed.gt(1) -> invalid expression - Radar.Q.radial_speed.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Radar.Q.radial_speed.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.radial_speed.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Radar.Q.radial_speed.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Radar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Radar detections with a high radial speed
    qresponse = client.query(
        QueryOntologyCatalog(Radar.Q.radial_speed.any().gt(10.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

Retrieves the unique identifier (tag) for the current ontology class, automatically generated during class definition.

This method provides the string key used by the Mosaico platform to identify and route specific data types within the ontology registry. It abstracts away the internal naming conventions, ensuring that you always use the correct identifier for queries and serialization.

Returns:

Name	Type	Description
str	str	The registered string tag for this class (e.g., "imu", "gps").

Raises:

Type	Description
Exception	If the class was not properly initialized via __pydantic_init_subclass__.

Practical Application: Topic Filtering

This method is particularly useful when constructing QueryTopic requests. By using the convenience method QueryTopic.with_ontology_tag(), you can filter topics by data type without hardcoding strings that might change.

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value (using constructor)
    qresponse = client.query(
        QueryTopic(
            Topic.with_ontology_tag(IMU.ontology_tag()),
        )
    )

    # 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]}")

mosaicolabs.models.futures.Lidar

Bases: Serializable, HeaderMixin

LiDAR Ontology.

This model represents a 3D point cloud acquired from a LiDAR sensor. Each field is a flat list whose i-th element corresponds to the i-th point in the scan. All lists within a single instance are therefore guaranteed to have the same length.

Attributes:

Name	Type	Description
x	list_(float32)	X coordinates of each point in meters.
y	list_(float32)	Y coordinates of each point in meters.
z	list_(float32)	Z coordinates of each point in meters.
intensity	Optional[list_(float32)]	Strength of the returned signal for each point (optional).
reflectivity	Optional[list_(uint16)]	Surface reflectivity per point (optional).
beam_id	Optional[list_(uint16)]	Laser beam index (ring / channel / line) that fired each point (optional).
range	Optional[list_(float32)]	Distance from the sensor origin to each point in meters (optional).
near_ir	Optional[list_(float32)]	Near-infrared ambient light reading per point, useful as a noise/ambient estimate (optional).
azimuth	Optional[list_(float32)]	Azimuth angle in radians for each point (optional).
elevation	Optional[list_(float32)]	Elevation angle in radians for each point (optional).
confidence	Optional[list_(uint8)]	Per-point validity or confidence flags as a manufacturer-specific bitmask (optional).
return_type	Optional[list_(uint8)]	Single/dual return classification, manufacturer-specific (optional).
point_timestamp	Optional[list_(float64)]	Per-point acquisition time offset from the scan start, in seconds (optional).

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter Lidar data based on thresholds values within a QueryOntologyCatalog. Expressions entailing lists of values can be queried using any between all(), any() or index access [i] followed by the contained type supported operations.

Example

from mosaicolabs import MosaicoClient, QueryTopic
from mosaicolabs.models.futures import Lidar

with MosaicoClient.connect("localhost", 6726) as client:
    # Fetch all sequences that contain at least one Lidar topic
    qresponse = client.query(QueryTopic().with_ontology_tag(Lidar.ontology_tag()))

    if qresponse is not None:

        for item in qresponse.items:
            print(f"Sequence: {item.name}")
            print(f"Topics:   {[topic.name for topic in item.topics]}")

header class-attribute instance-attribute

header = MosaicoField(
    nullable=True,
    default=None,
    description="Contains measure metadata like timestamp, reference frame and samples counter.",
)

Measure header containing measurement timestamp and reference frame.

Querying with the .Q Proxy

Header components are queryable through the header field prefix.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.header.timestamp.seconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.timestamp.nanoseconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.frame_id	String	.eq(), .match(), .in_(), .lt(), .gt(), .leq(), .geq(), .between(), .outside()
<Model>.Q.header.sample_counter	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, ForceTorque, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Find where the measure lasts at least 10 seconds
    qresponse = client.query(QueryOntologyCatalog(ForceTorque.Q.header.timestamp.seconds.gt(10.0)))

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

x class-attribute instance-attribute

x = MosaicoField(description='x coordinates in meters')

X coordinates of each point in the cloud, in meters.

Querying with the .Q Proxy

The X cordinates value are queryable via the x field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Lidar.Q.x.all() -> invalid expression - Lidar.Q.x.gt(1) -> invalid expression - Lidar.Q.x.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Lidar.Q.x.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.x.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.x.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Lidar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Lidar values on X within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(Lidar.Q.x.all().between([-1.0, 1.0]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

y class-attribute instance-attribute

y = MosaicoField(description='y coordinates in meters')

Y coordinates of each point in the cloud, in meters.

Querying with the .Q Proxy

The Y cordinates value are queryable via the y field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Lidar.Q.y.all() -> invalid expression - Lidar.Q.y.gt(1) -> invalid expression - Lidar.Q.y.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Lidar.Q.y.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.y.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.y.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Lidar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Lidar values on Y within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(Lidar.Q.y.all().between([-1.0, 1.0]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

z class-attribute instance-attribute

z = MosaicoField(description='z coordinates in meters')

Z coordinates of each point in the cloud, in meters.

Querying with the .Q Proxy

The Z cordinates value are queryable via the y field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Lidar.Q.z.all() -> invalid expression - Lidar.Q.z.gt(1) -> invalid expression - Lidar.Q.z.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Lidar.Q.z.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.z.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.z.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Lidar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Lidar values on Z within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(Lidar.Q.z.all().between([-1.0, 1.0]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

intensity class-attribute instance-attribute

intensity = MosaicoField(
    default=None,
    description="Surface reflectivity per point.",
)

Strength of the returned laser signal for each point.

Querying with the .Q Proxy

The intensity value is queryable via the intensity field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Lidar.Q.intensity.all() -> invalid expression - Lidar.Q.intensity.gt(1) -> invalid expression - Lidar.Q.intensity.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Lidar.Q.intensity.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.intensity.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.intensity.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Lidar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Lidar points with at least one strong return
    qresponse = client.query(
        QueryOntologyCatalog(Lidar.Q.intensity.any().gt(200.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

reflectivity class-attribute instance-attribute

reflectivity = MosaicoField(
    default=None,
    description="Surface reflectivity per point.",
)

Surface reflectivity per point.

Encodes the estimated reflectance of the surface that produced each return, independently of the distance. Manufacturer-specific scaling applies.

Querying with the .Q Proxy

The reflectivity value is queryable via the reflectivity field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Lidar.Q.reflectivity.all() -> invalid expression - Lidar.Q.reflectivity.gt(1) -> invalid expression - Lidar.Q.reflectivity.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Lidar.Q.reflectivity.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.reflectivity.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.reflectivity.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Lidar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Lidar points with high surface reflectivity
    qresponse = client.query(
        QueryOntologyCatalog(Lidar.Q.reflectivity.any().geq(200))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

beam_id class-attribute instance-attribute

beam_id = MosaicoField(
    default=None,
    description="Laser beam index (ring / channel / line) that fired each point.",
)

Laser beam index (ring / channel / line) that fired each point.

Identifies which physical emitter in the sensor array produced the return. Equivalent to the ring field commonly found in ROS PointCloud2 messages from multi-beam sensors such as Velodyne or Ouster.

Querying with the .Q Proxy

The beam id value is queryable via the beam_id field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Lidar.Q.beam_id.all() -> invalid expression - Lidar.Q.beam_id.gt(1) -> invalid expression - Lidar.Q.beam_id.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Lidar.Q.beam_id.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.beam_id.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.beam_id.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Lidar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Lidar points fired by a specific beam
    qresponse = client.query(
        QueryOntologyCatalog(Lidar.Q.beam_id.any().eq(0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

range class-attribute instance-attribute

range = MosaicoField(
    default=None,
    description="Distance from the sensor origin to each point, in meters.",
)

Distance from the sensor origin to each point, in meters.

Represents the raw radial distance along the beam axis, before projection onto Cartesian coordinates. Not always provided by all sensor drivers.

Querying with the .Q Proxy

The range value is queryable via the range field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Lidar.Q.range.all() -> invalid expression - Lidar.Q.range.gt(1) -> invalid expression - Lidar.Q.range.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Lidar.Q.range.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.range.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.range.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Lidar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Lidar points within 50 meters of the sensor
    qresponse = client.query(
        QueryOntologyCatalog(Lidar.Q.range.all().leq(50.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

near_ir class-attribute instance-attribute

near_ir = MosaicoField(
    default=None,
    description="Near-infrared ambient light reading per point.",
)

Near-infrared ambient light reading per point.

Captured passively by the sensor between laser pulses. Useful as a proxy for ambient illumination or for filtering sun-noise artefacts. Exposed as the ambient channel in Ouster drivers.

Querying with the .Q Proxy

The near-infrared value is queryable via the near_ir field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Lidar.Q.near_ir.all() -> invalid expression - Lidar.Q.near_ir.gt(1) -> invalid expression - Lidar.Q.near_ir.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Lidar.Q.near_ir.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.near_ir.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.near_ir.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Lidar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Lidar points captured under strong ambient IR light
    qresponse = client.query(
        QueryOntologyCatalog(Lidar.Q.near_ir.any().gt(500.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

azimuth class-attribute instance-attribute

azimuth = MosaicoField(
    default=None,
    description="Horizontal (azimuth) angle of each point in radians.",
)

Horizontal (azimuth) angle of each point in radians.

Querying with the .Q Proxy

The azimuth value is queryable via the azimuth field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Lidar.Q.azimuth.all() -> invalid expression - Lidar.Q.azimuth.gt(1) -> invalid expression - Lidar.Q.azimuth.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Lidar.Q.azimuth.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.azimuth.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.azimuth.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Lidar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Lidar points within a specific azimuth range
    qresponse = client.query(
        QueryOntologyCatalog(Lidar.Q.azimuth.all().between([-1.57, 1.57]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

elevation class-attribute instance-attribute

elevation = MosaicoField(
    default=None,
    description="Vertical (elevation) angle of each point in radians.",
)

Vertical (elevation) angle of each point in radians.

Querying with the .Q Proxy

The elevation value is queryable via the elevation field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Lidar.Q.elevation.all() -> invalid expression - Lidar.Q.elevation.gt(1) -> invalid expression - Lidar.Q.elevation.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Lidar.Q.elevation.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.elevation.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.elevation.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Lidar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Lidar points within a specific elevation range
    qresponse = client.query(
        QueryOntologyCatalog(Lidar.Q.elevation.all().between([-0.26, 0.26]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

confidence class-attribute instance-attribute

confidence = MosaicoField(
    default=None,
    description="Per-point validity or confidence flags.",
)

Per-point validity or confidence flags.

Stored as a manufacturer-specific bitmask (equivalent to the tag or flags fields in Ouster point clouds). Individual bits may signal saturated returns, calibration issues, or other quality indicators.

Querying with the .Q Proxy

The confidence value is queryable via the confidence field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Lidar.Q.confidence.all() -> invalid expression - Lidar.Q.confidence.gt(1) -> invalid expression - Lidar.Q.confidence.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Lidar.Q.confidence.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.confidence.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.confidence.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Lidar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Lidar points with at least one low-confidence flag
    qresponse = client.query(
        QueryOntologyCatalog(Lidar.Q.confidence.any().lt(10))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

return_type class-attribute instance-attribute

return_type = MosaicoField(
    default=None,
    description="Single/dual return classification per point.",
)

Single/dual return classification per point.

Indicates whether a point originates from the first return, last return, strongest return, etc. Encoding is manufacturer-specific.

Querying with the .Q Proxy

The return type value is queryable via the return_type field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Lidar.Q.return_type.all() -> invalid expression - Lidar.Q.return_type.gt(1) -> invalid expression - Lidar.Q.return_type.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Lidar.Q.return_type.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.return_type.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.return_type.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Lidar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Lidar points with a specific return classification
    qresponse = client.query(
        QueryOntologyCatalog(Lidar.Q.return_type.any().eq(1))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

point_timestamp class-attribute instance-attribute

point_timestamp = MosaicoField(
    default=None,
    description="Per-point acquisition time offset from the scan start, in seconds.",
)

Per-point acquisition time offset from the scan start, in seconds.

Allows precise temporal localisation of individual points within a single sweep, which is important for motion-distortion correction during point-cloud registration.

Querying with the .Q Proxy

The point timestamp value is queryable via the point_timestamp field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - Lidar.Q.point_timestamp.all() -> invalid expression - Lidar.Q.point_timestamp.gt(1) -> invalid expression - Lidar.Q.point_timestamp.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
Lidar.Q.point_timestamp.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.point_timestamp.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Lidar.Q.point_timestamp.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import Lidar

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Lidar points acquired within the first 10ms of the sweep
    qresponse = client.query(
        QueryOntologyCatalog(Lidar.Q.point_timestamp.all().leq(0.01))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

Retrieves the unique identifier (tag) for the current ontology class, automatically generated during class definition.

This method provides the string key used by the Mosaico platform to identify and route specific data types within the ontology registry. It abstracts away the internal naming conventions, ensuring that you always use the correct identifier for queries and serialization.

Returns:

Name	Type	Description
str	str	The registered string tag for this class (e.g., "imu", "gps").

Raises:

Type	Description
Exception	If the class was not properly initialized via __pydantic_init_subclass__.

Practical Application: Topic Filtering

This method is particularly useful when constructing QueryTopic requests. By using the convenience method QueryTopic.with_ontology_tag(), you can filter topics by data type without hardcoding strings that might change.

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value (using constructor)
    qresponse = client.query(
        QueryTopic(
            Topic.with_ontology_tag(IMU.ontology_tag()),
        )
    )

    # 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]}")

mosaicolabs.models.futures.depth_camera._DepthCameraBase

Bases: BaseModel

Internal base model shared by all depth camera ontologies.

Defines the spatial core and common optional channels that every depth camera variant exposes, regardless of the underlying acquisition technology.

This class is not intended to be instantiated directly. Use one of the concrete subclasses: RGBDCamera, ToFCamera, or StereoCamera.

Attributes:

Name	Type	Description
x	list_(float32)	Horizontal positions of each point, derived from depth, in meters.
y	list_(float32)	Vertical positions of each point, derived from depth, in meters.
z	list_(float32)	Depth values (distance along the optical axis) of each point, in meters.
rgb	Optional[list_(float32)]	Packed RGB colour value per point (optional).
intensity	Optional[list_(float32)]	Signal amplitude or intensity per point (optional).

x class-attribute instance-attribute

x = MosaicoField(
    description="Horizontal position derived from depth."
)

Horizontal position of each point derived from the depth map, in meters.

y class-attribute instance-attribute

y = MosaicoField(
    description="Vertical position derived from depth."
)

Vertical position of each point derived from the depth map, in meters.

z class-attribute instance-attribute

z = MosaicoField(
    description="Depth value directly (distance along optical axis)."
)

Depth value of each point, in meters.

Represents the distance along the camera's optical axis (Z-forward convention). This is the primary measurement from which x and y are projected using the sensor's intrinsic parameters.

rgb class-attribute instance-attribute

rgb = MosaicoField(
    default=None, description="Packed RGB color value."
)

Packed RGB colour value per point.

Each element encodes the red, green, and blue channels of the pixel co-registered with the corresponding depth sample. The rgb field uses the packing convention (bits 16-23 = R, 8-15 = G, 0-7 = B, stored as a float32 reinterpretation of a uint32). Use pack_rgb() and unpack_rgb() to convert to/from this format.

intensity class-attribute instance-attribute

intensity = MosaicoField(
    default=None, description="Signal amplitude/intensity."
)

Signal amplitude or intensity per point.

mosaicolabs.models.futures.depth_camera.pack_rgb

pack_rgb(r, g, b)

Packs three RGB channels (8-bit each) into a single float32 value.

This utility combines Red, Green, and Blue components into a 32-bit integer and then bit-casts the result into a float. This is a standard technique used in point cloud processing to store color data efficiently within a single field.

Parameters:

Name	Type	Description	Default
r	int	Red intensity (0-255).	required
g	int	Green intensity (0-255).	required
b	int	Blue intensity (0-255).	required

Returns:

Name	Type	Description
float	float	The RGB color encoded as a 32-bit float.

mosaicolabs.models.futures.depth_camera.unpack_rgb

unpack_rgb(packed_rgb)

Unpacks a float32 value back into its original RGB components.

This is the inverse operation of pack_rgb(). It reinterprets the float's bits as an unsigned 32-bit integer and extracts the individual color bytes.

Parameters:

Name	Type	Description	Default
packed_rgb	float	The encoded float value containing RGB data.	required

Returns:

Type	Description
Tuple[int, int, int]	Tuple[int, int, int]: A tuple of (red, green, blue), where each value is between 0 and 255.

mosaicolabs.models.futures.RGBDCamera

Bases: _DepthCameraBase, Serializable, HeaderMixin

RGB-D camera ontology.

This model represents a registered depth-and-colour point cloud produced by an RGB-D sensor (e.g. Intel RealSense D-series, Microsoft Azure Kinect). Each point carries a 3D position in the camera frame together with an optional packed RGB colour value and an optional intensity channel.

RGB-D sensors typically fuse a structured-light or active-infrared depth map with a co-located colour camera, yielding a dense, pixel-aligned point cloud at video frame rates.

Each field is a flat list whose i-th element corresponds to the i-th point in the frame. All lists within a single instance are therefore guaranteed to have the same length.

Attributes:

Name	Type	Description
x	list_(float32)	Horizontal positions of each point, derived from depth, in meters.
y	list_(float32)	Vertical positions of each point, derived from depth, in meters.
z	list_(float32)	Depth values (distance along the optical axis) of each point, in meters.
rgb	Optional[list_(float32)]	Packed RGB colour value per point (optional).
intensity	Optional[list_(float32)]	Signal amplitude or intensity per point (optional).

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter RGBDCamera data based on thresholds values within a QueryOntologyCatalog. Expressions entailing lists of values can be queried using any between all(), any() or index access [i] followed by the contained type supported operations.

Example

from mosaicolabs import MosaicoClient, QueryTopic
from mosaicolabs.models.futures import RGBDCamera

with MosaicoClient.connect("localhost", 6726) as client:
    # Fetch all sequences that contain at least one RGBDCamera topic
    qresponse = client.query(QueryTopic().with_ontology_tag(RGBDCamera.ontology_tag()))

    if qresponse is not None:

        for item in qresponse.items:
            print(f"Sequence: {item.name}")
            print(f"Topics:   {[topic.name for topic in item.topics]}")

header class-attribute instance-attribute

header = MosaicoField(
    nullable=True,
    default=None,
    description="Contains measure metadata like timestamp, reference frame and samples counter.",
)

Measure header containing measurement timestamp and reference frame.

Querying with the .Q Proxy

Header components are queryable through the header field prefix.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.header.timestamp.seconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.timestamp.nanoseconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.frame_id	String	.eq(), .match(), .in_(), .lt(), .gt(), .leq(), .geq(), .between(), .outside()
<Model>.Q.header.sample_counter	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, ForceTorque, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Find where the measure lasts at least 10 seconds
    qresponse = client.query(QueryOntologyCatalog(ForceTorque.Q.header.timestamp.seconds.gt(10.0)))

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

x class-attribute instance-attribute

x = MosaicoField(
    description="Horizontal position derived from depth."
)

Horizontal position of each point derived from the depth map, in meters.

y class-attribute instance-attribute

y = MosaicoField(
    description="Vertical position derived from depth."
)

Vertical position of each point derived from the depth map, in meters.

z class-attribute instance-attribute

z = MosaicoField(
    description="Depth value directly (distance along optical axis)."
)

Depth value of each point, in meters.

Represents the distance along the camera's optical axis (Z-forward convention). This is the primary measurement from which x and y are projected using the sensor's intrinsic parameters.

rgb class-attribute instance-attribute

rgb = MosaicoField(
    default=None, description="Packed RGB color value."
)

Packed RGB colour value per point.

Each element encodes the red, green, and blue channels of the pixel co-registered with the corresponding depth sample. The rgb field uses the packing convention (bits 16-23 = R, 8-15 = G, 0-7 = B, stored as a float32 reinterpretation of a uint32). Use pack_rgb() and unpack_rgb() to convert to/from this format.

intensity class-attribute instance-attribute

intensity = MosaicoField(
    default=None, description="Signal amplitude/intensity."
)

Signal amplitude or intensity per point.

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

Retrieves the unique identifier (tag) for the current ontology class, automatically generated during class definition.

This method provides the string key used by the Mosaico platform to identify and route specific data types within the ontology registry. It abstracts away the internal naming conventions, ensuring that you always use the correct identifier for queries and serialization.

Returns:

Name	Type	Description
str	str	The registered string tag for this class (e.g., "imu", "gps").

Raises:

Type	Description
Exception	If the class was not properly initialized via __pydantic_init_subclass__.

Practical Application: Topic Filtering

This method is particularly useful when constructing QueryTopic requests. By using the convenience method QueryTopic.with_ontology_tag(), you can filter topics by data type without hardcoding strings that might change.

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value (using constructor)
    qresponse = client.query(
        QueryTopic(
            Topic.with_ontology_tag(IMU.ontology_tag()),
        )
    )

    # 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]}")

mosaicolabs.models.futures.StereoCamera

Bases: _DepthCameraBase, Serializable, HeaderMixin

Stereo camera ontology.

This model represents a dense point cloud produced by a passive stereo camera system (e.g. Stereolabs ZED, Luxonis OAK-D, Carnegie Robotics MultiSense). Depth is estimated by computing the horizontal disparity between a rectified left/right image pair and projecting it into 3D space using the known baseline and intrinsic parameters.

In addition to the common spatial and colour channels inherited from _DepthCamera, this model exposes two stereo-specific fields: luma, which carries the luminance of the source rectified image pixel, and cost, which encodes the confidence of the disparity estimate at each point.

Each field is a flat list whose i-th element corresponds to the i-th pixel in the disparity map (in row-major order). All lists within a single instance are therefore guaranteed to have the same length.

Attributes:

Name	Type	Description
x	list_(float32)	Horizontal positions of each point in meters.
y	list_(float32)	Vertical positions of each point in meters.
z	list_(float32)	Depth values (distance along the optical axis) of each point, in meters.
rgb	Optional[list_(float32)]	Packed RGB colour value per point (optional).
intensity	Optional[list_(float32)]	Signal amplitude or intensity per point (optional).
luma	Optional[list_(uint8)]	Luminance of the corresponding pixel in the rectified image (optional).
cost	Optional[list_(uint8)]	Stereo matching cost per point; lower values indicate higher disparity confidence (optional).

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter StereoCamera data based on thresholds values within a QueryOntologyCatalog. Expressions entailing lists of values can be queried using any between all(), any() or index access [i] followed by the contained type supported operations.

Example

```python from mosaicolabs import MosaicoClient, QueryTopic from mosaicolabs.models.futures import StereoCamera

with MosaicoClient.connect("localhost", 6726) as client: # Fetch all sequences that contain at least one stereo camera topic qresponse = client.query(QueryTopic().with_ontology_tag(StereoCamera.ontology_tag()))

if qresponse is not None:

    for item in qresponse.items:
        print(f"Sequence: {item.name}")
        print(f"Topics:   {[topic.name for topic in item.topics]}")

```

header class-attribute instance-attribute

header = MosaicoField(
    nullable=True,
    default=None,
    description="Contains measure metadata like timestamp, reference frame and samples counter.",
)

Measure header containing measurement timestamp and reference frame.

Querying with the .Q Proxy

Header components are queryable through the header field prefix.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.header.timestamp.seconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.timestamp.nanoseconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.frame_id	String	.eq(), .match(), .in_(), .lt(), .gt(), .leq(), .geq(), .between(), .outside()
<Model>.Q.header.sample_counter	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, ForceTorque, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Find where the measure lasts at least 10 seconds
    qresponse = client.query(QueryOntologyCatalog(ForceTorque.Q.header.timestamp.seconds.gt(10.0)))

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

x class-attribute instance-attribute

x = MosaicoField(
    description="Horizontal position derived from depth."
)

Horizontal position of each point derived from the depth map, in meters.

y class-attribute instance-attribute

y = MosaicoField(
    description="Vertical position derived from depth."
)

Vertical position of each point derived from the depth map, in meters.

z class-attribute instance-attribute

z = MosaicoField(
    description="Depth value directly (distance along optical axis)."
)

Depth value of each point, in meters.

Represents the distance along the camera's optical axis (Z-forward convention). This is the primary measurement from which x and y are projected using the sensor's intrinsic parameters.

rgb class-attribute instance-attribute

rgb = MosaicoField(
    default=None, description="Packed RGB color value."
)

Packed RGB colour value per point.

Each element encodes the red, green, and blue channels of the pixel co-registered with the corresponding depth sample. The rgb field uses the packing convention (bits 16-23 = R, 8-15 = G, 0-7 = B, stored as a float32 reinterpretation of a uint32). Use pack_rgb() and unpack_rgb() to convert to/from this format.

intensity class-attribute instance-attribute

intensity = MosaicoField(
    default=None, description="Signal amplitude/intensity."
)

Signal amplitude or intensity per point.

luma class-attribute instance-attribute

luma = MosaicoField(
    default=None,
    description="Luminance of the corresponding pixel in the rectified image.",
)

Luminance of the corresponding pixel in the rectified image.

Querying with the .Q Proxy

The luma value is queryable via the luma field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - StereoCamera.Q.luma.all() -> invalid expression - StereoCamera.Q.luma.gt(1) -> invalid expression - StereoCamera.Q.luma.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
StereoCamera.Q.luma.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
StereoCamera.Q.luma.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
StereoCamera.Q.luma.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import StereoCamera

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Stereo frames with bright pixels
    qresponse = client.query(
        QueryOntologyCatalog(StereoCamera.Q.luma.any().gt(200))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

cost class-attribute instance-attribute

cost = MosaicoField(
    default=None,
    description="Stereo matching cost (disparity confidence measure, 0 = high confidence).",
)

Stereo matching cost per point; lower values indicate higher disparity confidence.

Querying with the .Q Proxy

The cost value is queryable via the cost field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - StereoCamera.Q.cost.all() -> invalid expression - StereoCamera.Q.cost.gt(1) -> invalid expression - StereoCamera.Q.cost.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
StereoCamera.Q.cost.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
StereoCamera.Q.cost.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
StereoCamera.Q.cost.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import StereoCamera

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for Stereo frames with high-confidence disparity estimates
    qresponse = client.query(
        QueryOntologyCatalog(StereoCamera.Q.cost.all().leq(10))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

Retrieves the unique identifier (tag) for the current ontology class, automatically generated during class definition.

This method provides the string key used by the Mosaico platform to identify and route specific data types within the ontology registry. It abstracts away the internal naming conventions, ensuring that you always use the correct identifier for queries and serialization.

Returns:

Name	Type	Description
str	str	The registered string tag for this class (e.g., "imu", "gps").

Raises:

Type	Description
Exception	If the class was not properly initialized via __pydantic_init_subclass__.

Practical Application: Topic Filtering

This method is particularly useful when constructing QueryTopic requests. By using the convenience method QueryTopic.with_ontology_tag(), you can filter topics by data type without hardcoding strings that might change.

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value (using constructor)
    qresponse = client.query(
        QueryTopic(
            Topic.with_ontology_tag(IMU.ontology_tag()),
        )
    )

    # 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]}")

mosaicolabs.models.futures.ToFCamera

Bases: _DepthCameraBase, Serializable, HeaderMixin

Time-of-Flight (ToF) camera ontology.

This model represents a point cloud produced by a Time-of-Flight sensor (e.g. PMD Flexx2, ifm O3R, Sony DepthSense). ToF sensors measure depth by emitting amplitude-modulated infrared light and computing the phase shift of the returning signal, yielding per-pixel depth, amplitude, and noise estimates in a single acquisition.

In addition to the common spatial and colour channels inherited from _DepthCamera, this model exposes two ToF-specific fields: noise, which quantifies the per-pixel measurement uncertainty, and grayscale, which carries the passive greyscale amplitude captured alongside the active depth measurement.

Each field is a flat list whose i-th element corresponds to the i-th pixel in the depth frame (in row-major order). All lists within a single instance are therefore guaranteed to have the same length.

Attributes:

Name	Type	Description
x	list_(float32)	Horizontal positions of each point, derived from depth, in meters.
y	list_(float32)	Vertical positions of each point, derived from depth, in meters.
z	list_(float32)	Depth values (distance along the optical axis) of each point, in meters.
rgb	Optional[list_(float32)]	Packed RGB colour value per point (optional).
intensity	Optional[list_(float32)]	Signal amplitude or intensity per point (optional).
noise	Optional[list_(float32)]	Per-pixel noise estimate of the depth measurement (optional).
grayscale	Optional[list_(float32)]	Passive greyscale amplitude per pixel (optional).

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter ToFCamera data based on thresholds values within a QueryOntologyCatalog. Expressions entailing lists of values can be queried using any between all(), any() or index access [i] followed by the contained type supported operations.

Example

from mosaicolabs import MosaicoClient, QueryTopic
from mosaicolabs.models.futures import ToFCamera

with MosaicoClient.connect("localhost", 6726) as client:
    # Fetch all sequences that contain at least one ToF camera topic
    qresponse = client.query(QueryTopic().with_ontology_tag(ToFCamera.ontology_tag()))

    if qresponse is not None:

        for item in qresponse.items:
            print(f"Sequence: {item.name}")
            print(f"Topics:   {[topic.name for topic in item.topics]}")

header class-attribute instance-attribute

header = MosaicoField(
    nullable=True,
    default=None,
    description="Contains measure metadata like timestamp, reference frame and samples counter.",
)

Measure header containing measurement timestamp and reference frame.

Querying with the .Q Proxy

Header components are queryable through the header field prefix.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.header.timestamp.seconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.timestamp.nanoseconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.frame_id	String	.eq(), .match(), .in_(), .lt(), .gt(), .leq(), .geq(), .between(), .outside()
<Model>.Q.header.sample_counter	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, ForceTorque, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Find where the measure lasts at least 10 seconds
    qresponse = client.query(QueryOntologyCatalog(ForceTorque.Q.header.timestamp.seconds.gt(10.0)))

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

x class-attribute instance-attribute

x = MosaicoField(
    description="Horizontal position derived from depth."
)

Horizontal position of each point derived from the depth map, in meters.

y class-attribute instance-attribute

y = MosaicoField(
    description="Vertical position derived from depth."
)

Vertical position of each point derived from the depth map, in meters.

z class-attribute instance-attribute

z = MosaicoField(
    description="Depth value directly (distance along optical axis)."
)

Depth value of each point, in meters.

Represents the distance along the camera's optical axis (Z-forward convention). This is the primary measurement from which x and y are projected using the sensor's intrinsic parameters.

rgb class-attribute instance-attribute

rgb = MosaicoField(
    default=None, description="Packed RGB color value."
)

Packed RGB colour value per point.

Each element encodes the red, green, and blue channels of the pixel co-registered with the corresponding depth sample. The rgb field uses the packing convention (bits 16-23 = R, 8-15 = G, 0-7 = B, stored as a float32 reinterpretation of a uint32). Use pack_rgb() and unpack_rgb() to convert to/from this format.

intensity class-attribute instance-attribute

intensity = MosaicoField(
    default=None, description="Signal amplitude/intensity."
)

Signal amplitude or intensity per point.

noise class-attribute instance-attribute

noise = MosaicoField(
    default=None, description="Noise value per pixel."
)

Per-pixel noise estimate of the depth measurement.

High noise values typically indicate low-confidence depth samples caused by low signal return, multi-path interference, or motion blur, and should be treated with caution during downstream processing.

Querying with the .Q Proxy

The noise value is queryable via the noise field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - ToFCamera.Q.noise.all() -> invalid expression - ToFCamera.Q.noise.gt(1) -> invalid expression - ToFCamera.Q.noise.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
ToFCamera.Q.noise.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
ToFCamera.Q.noise.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
ToFCamera.Q.noise.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import ToFCamera

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for ToF frames with at least one high-noise pixel
    qresponse = client.query(
        QueryOntologyCatalog(ToFCamera.Q.noise.any().gt(0.5))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

grayscale class-attribute instance-attribute

grayscale = MosaicoField(
    default=None, description="Grayscale amplitude."
)

Passive greyscale amplitude per pixel.

Captured by the sensor's infrared photodiodes independently of the active modulation cycle. Provides a texture channel that can be used for feature extraction or visual odometry without requiring a separate colour camera.

Querying with the .Q Proxy

The grayscale value is queryable via the grayscale field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - ToFCamera.Q.grayscale.all() -> invalid expression - ToFCamera.Q.grayscale.gt(1) -> invalid expression - ToFCamera.Q.grayscale.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
ToFCamera.Q.grayscale.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
ToFCamera.Q.grayscale.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
ToFCamera.Q.grayscale.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog
from mosaicolabs.models.futures import ToFCamera

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for ToF frames with bright greyscale pixels
    qresponse = client.query(
        QueryOntologyCatalog(ToFCamera.Q.grayscale.any().gt(200.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

Retrieves the unique identifier (tag) for the current ontology class, automatically generated during class definition.

This method provides the string key used by the Mosaico platform to identify and route specific data types within the ontology registry. It abstracts away the internal naming conventions, ensuring that you always use the correct identifier for queries and serialization.

Returns:

Name	Type	Description
str	str	The registered string tag for this class (e.g., "imu", "gps").

Raises:

Type	Description
Exception	If the class was not properly initialized via __pydantic_init_subclass__.

Practical Application: Topic Filtering

This method is particularly useful when constructing QueryTopic requests. By using the convenience method QueryTopic.with_ontology_tag(), you can filter topics by data type without hardcoding strings that might change.

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value (using constructor)
    qresponse = client.query(
        QueryTopic(
            Topic.with_ontology_tag(IMU.ontology_tag()),
        )
    )

    # 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]}")

mosaicolabs.models.futures.GridCells

Bases: Serializable, HeaderMixin

Grid Cells data.

This class represents the grid cells.

Attributes:

Name	Type	Description
cell_width	float32	A MosaicoType.float32 that represents the width of each cell.
cell_height	float32	A MosaicoType.float32 that represents the width of each cell.
cells	list_(Point3d)	A MosaicoType.list_(Point2d) that represents the center point of each cell.
header	optional[Header]	Optional heading containing measurement metadata

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter grid cells data based on cell_width, cell_height, or cells field values within a QueryOntologyCatalog.

Example

from mosaicolabs import MosaicoClient, GridCells, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for cell grid width field values within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(GridCells.Q.cell_width.between(100, 200))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

header class-attribute instance-attribute

header = MosaicoField(
    nullable=True,
    default=None,
    description="Contains measure metadata like timestamp, reference frame and samples counter.",
)

Measure header containing measurement timestamp and reference frame.

Querying with the .Q Proxy

Header components are queryable through the header field prefix.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.header.timestamp.seconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.timestamp.nanoseconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.frame_id	String	.eq(), .match(), .in_(), .lt(), .gt(), .leq(), .geq(), .between(), .outside()
<Model>.Q.header.sample_counter	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, ForceTorque, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Find where the measure lasts at least 10 seconds
    qresponse = client.query(QueryOntologyCatalog(ForceTorque.Q.header.timestamp.seconds.gt(10.0)))

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

cell_width class-attribute instance-attribute

cell_width = MosaicoField(description="Width of each cell.")

Width of each cell.

Querying with the .Q Proxy

The grid cells width is queryable via the cell_width field.

Field Access Path	Queryable Type	Supported Operators
GridCells.Q.cell_width	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, GridCells, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for cell width within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(GridCells.Q.cell_width.between([100, 200]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

cell_height class-attribute instance-attribute

cell_height = MosaicoField(
    description="Height of each cell."
)

Height of each cell.

Querying with the .Q Proxy

The grid cells height is queryable via the cell_height field.

Field Access Path	Queryable Type	Supported Operators
GridCells.Q.cell_height	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, GridCells, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for cell width within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(GridCells.Q.cell_height.between([100, 200]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

cells class-attribute instance-attribute

cells = MosaicoField(
    description="The cell represented by a point at it's center."
)

The cell represented by a point at it's center.

Querying with the .Q Proxy

The cells value is queryable via the cells field. Since it represents a list of Point3d values, use all(), any() or index access [i] to narrow down to the list element, then continue the expression with the contained Point3d field (x, y or z). - GridCells.Q.cells.all() -> invalid expression - GridCells.Q.cells.x.gt(1) -> invalid expression - GridCells.Q.cells.all().x.gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
GridCells.Q.cells.all().x	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
GridCells.Q.cells.any().x	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
GridCells.Q.cells.[i].x	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, GridCells, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for grids with at least one cell beyond a specific X-coordinate
    qresponse = client.query(
        QueryOntologyCatalog(GridCells.Q.cells.any().x.gt(500.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

Retrieves the unique identifier (tag) for the current ontology class, automatically generated during class definition.

This method provides the string key used by the Mosaico platform to identify and route specific data types within the ontology registry. It abstracts away the internal naming conventions, ensuring that you always use the correct identifier for queries and serialization.

Returns:

Name	Type	Description
str	str	The registered string tag for this class (e.g., "imu", "gps").

Raises:

Type	Description
Exception	If the class was not properly initialized via __pydantic_init_subclass__.

Practical Application: Topic Filtering

This method is particularly useful when constructing QueryTopic requests. By using the convenience method QueryTopic.with_ontology_tag(), you can filter topics by data type without hardcoding strings that might change.

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value (using constructor)
    qresponse = client.query(
        QueryTopic(
            Topic.with_ontology_tag(IMU.ontology_tag()),
        )
    )

    # 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]}")

mosaicolabs.models.futures.MapMetadata

Bases: Serializable

Represents metadata about the map, like it's width and height. Typically used in combination with OccupancyGrid

Attributes:

Name	Type	Description
map_load_time	Time	A Time representing the time at which the map has been loaded.
resolution	float32	A MosaicoType.float32 representing the resolution of the map.
width	uint32	A MosaicoType.uint32 representing the number of cells that represent the width of the map.
height	uint32	A MosaicoType.uint32 representing the number of cells that represent the height of the map.
origin	Pose	A Pose that represents where the map starts in the real world.

Querying with the .Q Proxy

This class fields are queryable when constructing a QueryOntologyCatalog via the .Q proxy. Check the fields documentation for detailed description.

Example

from mosaicolabs import MosaicoClient, MapMetadata, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter MapMetadatas with width AND height
    qresponse = client.query(
        QueryOntologyCatalog(MapMetadata.Q.width.gt(100))
        .with_expression(MapMetadata.Q.height.lt(200))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

map_load_time class-attribute instance-attribute

map_load_time = MosaicoField(
    description="Time (in nanoseconds) at which the map has been loaded."
)

Time (in nanoseconds) at which the map has been loaded.

Querying with the .Q Proxy

The map metadata time is queryable via the map_load_time field.

Field Access Path	Queryable Type	Supported Operators
MapMetadata.Q.map_load_time	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, MapMetadata, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for map_load_time in nanoseconds within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(MapMetadata.Q.map_load_time.between([100000, 200000]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

resolution class-attribute instance-attribute

resolution = MosaicoField(
    description="Resolution of the map [m/cell]."
)

Resolution of the map.

Querying with the .Q Proxy

The map metadata resolution is queryable via the resolution field.

Field Access Path	Queryable Type	Supported Operators
MapMetadata.Q.resolution	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, MapMetadata, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for resolution within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(MapMetadata.Q.resolution.between([100000, 200000]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

width class-attribute instance-attribute

width = MosaicoField(
    description="Number of cells representing the width of the map [cells]."
)

Number of cells representing the width of the map.

Querying with the .Q Proxy

The map metadata width is queryable via the width field.

Field Access Path	Queryable Type	Supported Operators
MapMetadata.Q.width	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, MapMetadata, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for width within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(MapMetadata.Q.width.between([10, 20]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

height class-attribute instance-attribute

height = MosaicoField(
    description="Number of cells representing the height of the map [cells]."
)

Number of cells representing the height of the map.

Querying with the .Q Proxy

The map metadata height is queryable via the height field.

Field Access Path	Queryable Type	Supported Operators
MapMetadata.Q.height	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, MapMetadata, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for height within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(MapMetadata.Q.height.between([10, 20]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

origin class-attribute instance-attribute

origin = MosaicoField(
    description="Where the map starts in the real world."
)

The origin of the map [m, m, rad]. This is the real-world pose of the bottom left corner of cell (0,0) in the map.

Querying with the .Q Proxy

The map metadata origin is queryable via the origin field.

Field Access Path	Queryable Type	Supported Operators
MapMetadata.Q.origin.position.x	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
MapMetadata.Q.origin.position.y	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
MapMetadata.Q.origin.position.z	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
MapMetadata.Q.origin.orientation.x	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
MapMetadata.Q.origin.orientation.y	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
MapMetadata.Q.origin.orientation.z	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
MapMetadata.Q.origin.orientation.w	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, MapMetadata, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter map metadata where the object is beyond a specific X-coordinate
    qresponse = client.query(
        QueryOntologyCatalog(MapMetadata.Q.origin.position.x.gt(500.0))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

Retrieves the unique identifier (tag) for the current ontology class, automatically generated during class definition.

This method provides the string key used by the Mosaico platform to identify and route specific data types within the ontology registry. It abstracts away the internal naming conventions, ensuring that you always use the correct identifier for queries and serialization.

Returns:

Name	Type	Description
str	str	The registered string tag for this class (e.g., "imu", "gps").

Raises:

Type	Description
Exception	If the class was not properly initialized via __pydantic_init_subclass__.

Practical Application: Topic Filtering

This method is particularly useful when constructing QueryTopic requests. By using the convenience method QueryTopic.with_ontology_tag(), you can filter topics by data type without hardcoding strings that might change.

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value (using constructor)
    qresponse = client.query(
        QueryTopic(
            Topic.with_ontology_tag(IMU.ontology_tag()),
        )
    )

    # 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]}")

mosaicolabs.models.futures.OccupancyGrid

Bases: Serializable, HeaderMixin

Occupancy Grid data.

This class represents the occupancy grid.

Attributes:

Name	Type	Description
info	MapMetadata	A MapMetadata describing the occupancy grid.
data	list_(int8)	A MosaicoType.list_(MosaicoType.int8) representing data contained in the occupancy grid.
header	optional[Header]	Optional heading containing measurement metadata

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter occupancy grid data based on info or data field values within a QueryOntologyCatalog.

Example

from mosaicolabs import MosaicoClient, OccupancyGrid, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for grid width field values within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(OccupancyGrid.Q.info.width.between(-100, 100))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

header class-attribute instance-attribute

header = MosaicoField(
    nullable=True,
    default=None,
    description="Contains measure metadata like timestamp, reference frame and samples counter.",
)

Measure header containing measurement timestamp and reference frame.

Querying with the .Q Proxy

Header components are queryable through the header field prefix.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.header.timestamp.seconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.timestamp.nanoseconds	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.header.frame_id	String	.eq(), .match(), .in_(), .lt(), .gt(), .leq(), .geq(), .between(), .outside()
<Model>.Q.header.sample_counter	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, ForceTorque, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Find where the measure lasts at least 10 seconds
    qresponse = client.query(QueryOntologyCatalog(ForceTorque.Q.header.timestamp.seconds.gt(10.0)))

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

info class-attribute instance-attribute

info = MosaicoField(
    description="Info about the map like it's width and height."
)

Info about the map like it's width and height.

Querying with the .Q Proxy

The occupancy grid info is queryable via the info field.

Field Access Path	Queryable Type	Supported Operators
OccupancyGrid.Q.info.map_load_time	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
OccupancyGrid.Q.info.resolution	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
OccupancyGrid.Q.info.width	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
OccupancyGrid.Q.info.height	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
OccupancyGrid.Q.info.origin.position.x	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
OccupancyGrid.Q.info.origin.position.y	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
OccupancyGrid.Q.info.origin.position.z	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
OccupancyGrid.Q.info.origin.orientation.x	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
OccupancyGrid.Q.info.origin.orientation.y	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
OccupancyGrid.Q.info.origin.orientation.z	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
OccupancyGrid.Q.info.origin.orientation.w	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, OccupancyGrid, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for time seconds within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(OccupancyGrid.Q.info.map_load_time.between([100000, 200000]))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

data class-attribute instance-attribute

data = MosaicoField(
    description="Occupancy probability: 1 means occupied, 0 means unoccupied and -1 means unkown."
)

The map data, in row-major order, starting with (0,0). Occupancy probabilities are in the range [0,100]. Unknown is -1.

Querying with the .Q Proxy

The data value is queryable via the data field. Since it represents a list of values, use all(), any() or index access [i] to narrow down to the list element and compose a correct expression. - OccupancyGrid.Q.data.all() -> invalid expression - OccupancyGrid.Q.data.gt(1) -> invalid expression - OccupancyGrid.Q.data.all().gt(1) -> valid expression

Field Access Path	Queryable Type	Supported Operators
OccupancyGrid.Q.data.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
OccupancyGrid.Q.data.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
OccupancyGrid.Q.data.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, OccupancyGrid, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for occupancy grids with at least one fully occupied cell
    qresponse = client.query(
        QueryOntologyCatalog(OccupancyGrid.Q.data.any().eq(100))
    )

    # 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]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

Retrieves the unique identifier (tag) for the current ontology class, automatically generated during class definition.

This method provides the string key used by the Mosaico platform to identify and route specific data types within the ontology registry. It abstracts away the internal naming conventions, ensuring that you always use the correct identifier for queries and serialization.

Returns:

Name	Type	Description
str	str	The registered string tag for this class (e.g., "imu", "gps").

Raises:

Type	Description
Exception	If the class was not properly initialized via __pydantic_init_subclass__.

Practical Application: Topic Filtering

This method is particularly useful when constructing QueryTopic requests. By using the convenience method QueryTopic.with_ontology_tag(), you can filter topics by data type without hardcoding strings that might change.

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value (using constructor)
    qresponse = client.query(
        QueryTopic(
            Topic.with_ontology_tag(IMU.ontology_tag()),
        )
    )

    # 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]}")