Sensors Models

mosaicolabs.models.sensors.IMU

Bases: Serializable

Inertial Measurement Unit data.

This model aggregates raw or estimated motion data from accelerometers and gyroscopes, providing a high-frequency snapshot of an object's inertial state.

Attributes:

Name	Type	Description
acceleration	Vector3d	Linear acceleration vector [ax, ay, az] in $m/s^2$.
angular_velocity	Vector3d	Angular velocity vector [wx, wy, wz] in $rad/s$.
orientation	Optional[Quaternion]	Optional estimated orientation expressed as a quaternion.

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter IMU data based on physical thresholds or metadata within a QueryOntologyCatalog.

Example

from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Find high-acceleration events (e.g., impacts) on the X-axis
    qresponse = client.query(
        QueryOntologyCatalog(IMU.Q.acceleration.x.gt(15.0))
        .with_expression(IMU.Q.angular_velocity.z.gt(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]}")

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(IMU.Q.acceleration.x.gt(15.0), include_timestamp_range=True)
        .with_expression(IMU.Q.angular_velocity.z.gt(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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

acceleration instance-attribute

acceleration

Linear acceleration component.

Querying with the .Q Proxy

Acceleration components are queryable through the acceleration field prefix.

Field Access Path	Queryable Type	Supported Operators
IMU.Q.acceleration.x	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
IMU.Q.acceleration.y	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
IMU.Q.acceleration.z	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()

Example

from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for high-impact events
    qresponse = client.query(
        QueryOntologyCatalog(IMU.Q.acceleration.z.gt(19.6))
    )

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

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(IMU.Q.acceleration.z.gt(19.6), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

angular_velocity instance-attribute

angular_velocity

Angular velocity component.

Querying with the .Q Proxy

Angular velocities components are queryable through the angular_velocity field prefix.

Field Access Path	Queryable Type	Supported Operators
IMU.Q.angular_velocity.x	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
IMU.Q.angular_velocity.y	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
IMU.Q.angular_velocity.z	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()

Example

from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for high-turns events
    qresponse = client.query(
        QueryOntologyCatalog(IMU.Q.angular_velocity.z.gt(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]}")

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(IMU.Q.angular_velocity.z.gt(1.0), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

orientation class-attribute instance-attribute

orientation = None

Estimated orientation [qx, qy, qz, qw] (optional).

Querying with the .Q Proxy

Estimated orientation components are queryable through the orientation field prefix.

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

Example

from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for orientation component values
    qresponse = client.query(
        QueryOntologyCatalog(IMU.Q.orientation.z.gt(0.707))
    )

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

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(IMU.Q.orientation.z.gt(0.707), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

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:

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

Raises:

Type	Description
Exception	If the class was not properly initialized via __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.sensors.GPSStatus

Bases: Serializable

Status of the GNSS receiver and satellite fix.

This class encapsulates quality metrics and operational state of the GNSS receiver, including fix type, satellite usage, and precision dilution factors.

Attributes:

Name	Type	Description
status	int	Fix status indicator (e.g., No Fix, 2D, 3D).
service	int	Service used for the fix (e.g., GPS, GLONASS, Galileo).
satellites	Optional[int]	Number of satellites currently visible or used in the solution.
hdop	Optional[float]	Horizontal Dilution of Precision (lower is better).
vdop	Optional[float]	Vertical Dilution of Precision (lower is better).

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter status data based on fix quality or precision metrics within a QueryOntologyCatalog.

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, GPSStatus

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for high-quality fixes (low HDOP)
    qresponse = client.query(
        QueryOntologyCatalog(GPSStatus.Q.hdop.lt(2.0))
        .with_expression(GPSStatus.Q.satellites.geq(6)),
    )

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

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(GPSStatus.Q.hdop.lt(2.0), include_timestamp_range=True)
        .with_expression(GPSStatus.Q.satellites.geq(6))
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

status instance-attribute

status

Fix status.

Querying with the .Q Proxy

The fix status is queryable via the status field.

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

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, GPSStatus

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for valid fixes
    qresponse = client.query(
        QueryOntologyCatalog(GPSStatus.Q.status.gt(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]}")

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(GPSStatus.Q.status.gt(0), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

service instance-attribute

service

Service used (GPS, GLONASS, etc).

Querying with the .Q Proxy

The service identifier is queryable via the service field.

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

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, GPSStatus

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for specific service ID
    qresponse = client.query(
        QueryOntologyCatalog(GPSStatus.Q.service.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]}")

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(GPSStatus.Q.service.eq(1), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

satellites class-attribute instance-attribute

satellites = None

Satellites visible/used.

Querying with the .Q Proxy

Satellite count is queryable via the satellites field.

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

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, GPSStatus

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for fixes with at least 6 satellites
    qresponse = client.query(
        QueryOntologyCatalog(GPSStatus.Q.satellites.geq(6))
    )

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

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(GPSStatus.Q.satellites.geq(6), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

hdop class-attribute instance-attribute

hdop = None

Horizontal Dilution of Precision.

Querying with the .Q Proxy

HDOP values are queryable via the hdop field.

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

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, GPSStatus

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for excellent horizontal precision
    qresponse = client.query(
        QueryOntologyCatalog(GPSStatus.Q.hdop.lt(1.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]}")

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(GPSStatus.Q.hdop.lt(1.5), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

vdop class-attribute instance-attribute

vdop = None

Vertical Dilution of Precision.

Querying with the .Q Proxy

VDOP values are queryable via the vdop field.

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

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, GPSStatus

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for good vertical precision
    qresponse = client.query(
        QueryOntologyCatalog(GPSStatus.Q.vdop.lt(2.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]}")

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(GPSStatus.Q.vdop.lt(2.0), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

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:

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

Raises:

Type	Description
Exception	If the class was not properly initialized via __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.sensors.GPS

Bases: Serializable

Processed GNSS fix containing Position, Velocity, and Status.

This class serves as the primary container for geodetic location data (WGS 84) and receiver state information.

Attributes:

Name	Type	Description
position	Point3d	Lat/Lon/Alt (WGS 84) represented as a Point3d.
velocity	Optional[Vector3d]	Velocity vector [North, East, Alt] in $m/s$.
status	Optional[GPSStatus]	Receiver status info including fix type and satellite count.

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter GPS data based on geodetic coordinates or signal quality within a QueryOntologyCatalog.

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, GPS

with MosaicoClient.connect("localhost", 6726) as client:
    # Find data collected above 1000m altitude
    qresponse = client.query(
        QueryOntologyCatalog(GPS.Q.position.z.gt(1000.0))
        .with_expression(GPS.Q.status.satellites.geq(6)),
    )

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

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(GPS.Q.position.z.gt(1000.0), include_timestamp_range=True)
        .with_expression(GPS.Q.status.satellites.geq(6)),
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

position instance-attribute

position

Lat/Lon/Alt (WGS 84).

Querying with the .Q Proxy

Position components are queryable through the position field prefix.

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

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, GPS

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific latitude range
    qresponse = client.query(
        QueryOntologyCatalog(GPS.Q.position.x.between([45.0, 46.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]}")

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(GPS.Q.position.x.between([45.0, 46.0]), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

velocity class-attribute instance-attribute

velocity = None

Velocity vector [North, East, Alt] m/s.

Querying with the .Q Proxy

Velocity components are queryable through the velocity field prefix.

Field Access Path	Queryable Type	Supported Operators
GPS.Q.velocity.x	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
GPS.Q.velocity.y	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
GPS.Q.velocity.z	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, GPS

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for high vertical velocity
    qresponse = client.query(
        QueryOntologyCatalog(GPS.Q.velocity.z.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]}")

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(GPS.Q.velocity.z.gt(5.0), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

status class-attribute instance-attribute

status = None

Receiver status information.

Querying with the .Q Proxy

Status components are queryable through the status field prefix.

Field Access Path	Queryable Type	Supported Operators
GPS.Q.status.satellites	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
GPS.Q.status.hdop	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
GPS.Q.status.vdop	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
GPS.Q.status.status	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
GPS.Q.status.service	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, GPS

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for high-precision fixes with at least 8 satellites
    qresponse = client.query(
        QueryOntologyCatalog(GPS.Q.status.satellites.geq(8))
    )

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

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(GPS.Q.status.status.eq(1), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

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:

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

Raises:

Type	Description
Exception	If the class was not properly initialized via __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.sensors.NMEASentence

Bases: Serializable

Raw NMEA 0183 sentence string.

Attributes:

Name	Type	Description
sentence	str	The NMEA 0183 sentence string.

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter NMEA data based on the sentence content within a QueryOntologyCatalog.

Field Access Path	Queryable Type	Supported Operators
NMEASentence.Q.sentence	String	.eq(), .neq(), .match(), .in_()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, NMEASentence

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for NMEA sentences containing "GPGGA"
    qresponse = client.query(
        QueryOntologyCatalog(NMEASentence.Q.sentence.match("GPGGA"))
    )

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

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(NMEASentence.Q.sentence.match("GPGGA"), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

sentence instance-attribute

sentence

Raw ASCII sentence.

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:

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

Raises:

Type	Description
Exception	If the class was not properly initialized via __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.sensors.Magnetometer

Bases: Serializable

Magnetic field measurement data.

This class represents the magnetic field measurements from a magnetometer sensor.

Attributes:

Name	Type	Description
magnetic_field	Vector3d	Magnetic field vector [mx, my, mz] in microTesla.

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter magnetometer data based on magnetic field values within a QueryOntologyCatalog.

Example

from mosaicolabs import MosaicoClient, Magnetometer, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for magnetic field values within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(Magnetometer.Q.magnetic_field.x.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]}")

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(Magnetometer.Q.magnetic_field.x.between(-100, 100), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

magnetic_field instance-attribute

magnetic_field

Magnetic field vector [mx, my, mz] in microTesla.

Querying with the .Q Proxy

The magnetic field vector is queryable via the magnetic_field field.

Field Access Path	Queryable Type	Supported Operators
Magnetometer.Q.magnetic_field.x	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
Magnetometer.Q.magnetic_field.y	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
Magnetometer.Q.magnetic_field.z	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()

Example

from mosaicolabs import MosaicoClient, Magnetometer, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for magnetic field values within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(Magnetometer.Q.magnetic_field.x.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]}")

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(Magnetometer.Q.magnetic_field.x.between(-100, 100), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

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:

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

Raises:

Type	Description
Exception	If the class was not properly initialized via __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.sensors.Pressure

Bases: Serializable, VarianceMixin

Represents a physical pressure value. The internal representation is always stored in Pascals (Pa).

Users are encouraged to use the from_* factory methods when initializing pressure values expressed in units other than Pascals.

Attributes:

Name	Type	Description
value	float	Pressure value in Pascals (Pa). When using the constructor directly, the value must be provided in Pascals.
variance	Optional[float]	The variance of the data.
variance_type	Optional[int]	Enum integer representing the variance parameterization.

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter pressure data based on pressure values within a QueryOntologyCatalog.

Example

from mosaicolabs import MosaicoClient, Pressure, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for pressure values within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(Pressure.Q.value.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]}")

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(Pressure.Q.value.between([100000, 200000]), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

variance class-attribute instance-attribute

variance = None

Optional 64-bit float representing the variance of the data.

This field is injected into the model via composition, ensuring that sensor data is paired with the optional variance attribute.

Querying with the .Q Proxy

The variance field is queryable with the .Q Proxy.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.variance	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()

The <Model> placeholder adapts based on how the VarianceMixin is integrated into the data structure:

• Direct Inheritance: Used for classes like Pressure or Temperature that inherit directly from VarianceMixin to represent 1D uncertainty.
• Composition (Nested Access): If a complex model contains a field that is a subclass of VarianceMixin, the proxy allows you to traverse the hierarchy to that specific attribute.

Example

from mosaicolabs import MosaicoClient, Pressure, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter Pressure data by a specific acquisition second
    qresponse = client.query(
        QueryOntologyCatalog(Pressure.Q.variance.lt(0.76))
    )

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

variance_type class-attribute instance-attribute

variance_type = None

Optional 16-bit integer representing the variance parameterization.

This field is injected into the model via composition, ensuring that sensor data is paired with the optional covariance type attribute.

Querying with the .Q Proxy

The variance_type field is fully queryable via the .Q Proxy.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.variance_type	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()

The <Model> placeholder adapts based on how the VarianceMixin is integrated into the data structure:

• Direct Inheritance: Used for classes like Pressure or Temperature that inherit directly from VarianceMixin to represent 1D uncertainty.
• Composition (Nested Access): If a complex model contains a field that is a subclass of VarianceMixin, the proxy allows you to traverse the hierarchy to that specific attribute.

Filtering by Precision

The following examples demonstrate how to filter sensor data based on the magnitude of the variance or the specific procedure used to calculate it.

from mosaicolabs import MosaicoClient, Pressure, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter by variance threshold
    results = client.query(QueryOntologyCatalog(
        Pressure.Q.variance.lt(0.76)
    ))

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

value instance-attribute

value

The absolute pressure reading from the sensor in Pascals.

Querying with the .Q Proxy

The pressure value is queryable via the value field.

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

Example

from mosaicolabs import MosaicoClient, Pressure, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for pressure values within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(Pressure.Q.value.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]}")

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(Pressure.Q.value.between([100000, 200000]), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

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:

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

Raises:

Type	Description
Exception	If the class was not properly initialized via __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]}")

from_atm classmethod

from_atm(*, value, variance=None, variance_type=None)

Creates a Pressure instance using the value in Atm and converting it in Pascal using the formula Pascal = Atm * 101325.

Parameters:

Name	Type	Description	Default
value	float	The pressure value in Atm.	required
variance	Optional[float]	The variance of the data.	None
variance_type	Optional[int]	Enum integer representing the variance parameterization.	None

Returns:

Name	Type	Description
Pressure	Pressure	A Pressure instance with value in Pascal.

from_bar classmethod

from_bar(*, value, variance=None, variance_type=None)

Creates a Pressure instance using the value in Bar and converting it in Pascal using the formula Pascal = Bar * 100000.

Parameters:

Name	Type	Description	Default
value	float	The pressure value in Bar.	required
variance	Optional[float]	The variance of the data.	None
variance_type	Optional[int]	Enum integer representing the variance parameterization.	None

Returns:

Name	Type	Description
Pressure	Pressure	A Pressure instance with value in Pascal.

from_psi classmethod

from_psi(*, value, variance=None, variance_type=None)

Creates a Pressure instance using the value in Psi and converting it in Pascal using the formula Pascal = Psi * 6894.7572931783.

Parameters:

Name	Type	Description	Default
value	float	The pressure value in Psi.	required
variance	Optional[float]	The variance of the data.	None
variance_type	Optional[int]	Enum integer representing the variance parameterization.	None

Returns:

Name	Type	Description
Pressure	Pressure	A Pressure instance with value in Pascal.

to_atm

to_atm()

Converts and returns the Pressure value in Atm using the formula Atm = Pascal / 101325.

Returns:

Name	Type	Description
float	float	The Pressure value in Atm.

to_bar

to_bar()

Converts and returns the Pressure value in Bar using the formula Bar = Pascal / 100000.

Returns:

Name	Type	Description
float	float	The Pressure value in Bar.

to_psi

to_psi()

Converts and returns the Pressure value in Psi using the formula Psi = Pascal / 6894.7572931783.

Returns:

Name	Type	Description
float	float	The Pressure value in Psi.

mosaicolabs.models.sensors.Range

Bases: Serializable, VarianceMixin

Represents a range measurement that defines a valid distance interval between the minimum and the maximum value. This with also the field of view, the radiation type and the range value. The internal representation is always stored in meters (m).

Attributes:

Name	Type	Description
radiation_type	int	Which type of radiation the sensor used.
field_of_view	float	The arc angle, in Radians (rad), over which the distance reading is valid.
min_range	float	Minimum range value in Meters (m). Fixed distance means that the minimum range must be equal to the maximum range.
max_range	float	Maximum range value in Meters (m). Fixed distance means that the minimum range must be equal to the maximum range.
range	float	Range value in Meters (m).
variance	Optional[float]	The variance of the data.
variance_type	Optional[int]	Enum integer representing the variance parameterization.

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter range data based on range parameters within a QueryOntologyCatalog.

Example

from mosaicolabs import MosaicoClient, Range, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for range data based on range parameters
    qresponse = client.query(
        QueryOntologyCatalog(Range.Q.range.between(0.0, 10.0))
        .with_epression(Range.Q.radiation_type.eq(0))
        .with_epression(Range.Q.max_range.between(70.0, 90.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]}")

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(Range.Q.range.between(0.0, 10.0), include_timestamp_range=True)
        .with_epression(Range.Q.radiation_type.eq(0))
        .with_epression(Range.Q.max_range.between(70.0, 90.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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

variance class-attribute instance-attribute

variance = None

Optional 64-bit float representing the variance of the data.

This field is injected into the model via composition, ensuring that sensor data is paired with the optional variance attribute.

Querying with the .Q Proxy

The variance field is queryable with the .Q Proxy.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.variance	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()

The <Model> placeholder adapts based on how the VarianceMixin is integrated into the data structure:

• Direct Inheritance: Used for classes like Pressure or Temperature that inherit directly from VarianceMixin to represent 1D uncertainty.
• Composition (Nested Access): If a complex model contains a field that is a subclass of VarianceMixin, the proxy allows you to traverse the hierarchy to that specific attribute.

Example

from mosaicolabs import MosaicoClient, Pressure, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter Pressure data by a specific acquisition second
    qresponse = client.query(
        QueryOntologyCatalog(Pressure.Q.variance.lt(0.76))
    )

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

variance_type class-attribute instance-attribute

variance_type = None

Optional 16-bit integer representing the variance parameterization.

This field is injected into the model via composition, ensuring that sensor data is paired with the optional covariance type attribute.

Querying with the .Q Proxy

The variance_type field is fully queryable via the .Q Proxy.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.variance_type	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()

The <Model> placeholder adapts based on how the VarianceMixin is integrated into the data structure:

• Direct Inheritance: Used for classes like Pressure or Temperature that inherit directly from VarianceMixin to represent 1D uncertainty.
• Composition (Nested Access): If a complex model contains a field that is a subclass of VarianceMixin, the proxy allows you to traverse the hierarchy to that specific attribute.

Filtering by Precision

The following examples demonstrate how to filter sensor data based on the magnitude of the variance or the specific procedure used to calculate it.

from mosaicolabs import MosaicoClient, Pressure, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter by variance threshold
    results = client.query(QueryOntologyCatalog(
        Pressure.Q.variance.lt(0.76)
    ))

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

radiation_type instance-attribute

radiation_type

Which type of radiation the sensor used.

Querying with the .Q Proxy

The radiation_type is queryable via the radiation_type field.

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

Example

from mosaicolabs import MosaicoClient, Range, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for range data based on radiation type
    qresponse = client.query(
        QueryOntologyCatalog(Range.Q.radiation_type.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]}")

field_of_view instance-attribute

field_of_view

The arc angle, in radians, over which the distance reading is valid.

Querying with the .Q Proxy

The field_of_view is queryable via the field_of_view field.

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

Example

from mosaicolabs import MosaicoClient, Range, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for range data based on field of view
    qresponse = client.query(
        QueryOntologyCatalog(Range.Q.field_of_view.between(0.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]}")

min_range instance-attribute

min_range

Minimum range value in meters. Fixed distance means that the minimum range must be equal to the maximum range.

Querying with the .Q Proxy

The min_range is queryable via the min_range field.

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

Example

from mosaicolabs import MosaicoClient, Range, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for range data based on minimum range
    qresponse = client.query(
        QueryOntologyCatalog(Range.Q.min_range.between(0.0, 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]}")

max_range instance-attribute

max_range

Maximum range value in meters. Fixed distance means that the minimum range must be equal to the maximum range.

Querying with the .Q Proxy

The max_range is queryable via the max_range field.

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

Example

from mosaicolabs import MosaicoClient, Range, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for range data based on maximum range
    qresponse = client.query(
        QueryOntologyCatalog(Range.Q.max_range.between(0.0, 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]}")

range instance-attribute

range

Range value in meters.

Querying with the .Q Proxy

The range is queryable via the range field.

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

Example

from mosaicolabs import MosaicoClient, Range, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for range data based on range
    qresponse = client.query(
        QueryOntologyCatalog(Range.Q.range.between(0.0, 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]}")

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(Range.Q.range.between(0.0, 10.0), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

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:

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

Raises:

Type	Description
Exception	If the class was not properly initialized via __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]}")

validate_min_and_max_range

validate_min_and_max_range()

Ensures that min_range is smaller or equal to max_range.

validate_range

validate_range()

Ensures that range is between min_range and max_range.

mosaicolabs.models.sensors.Temperature

Bases: Serializable, VarianceMixin

Represents a thermodynamic temperature. The internal representation is always stored in Kelvin (K).

Users are encouraged to use the from_* factory methods when initializing temperature values expressed in units other than Kelvin.

Attributes:

Name	Type	Description
value	float	Temperature value in Kelvin (K). When using the constructor directly, the value must be provided in Kelvin.
variance	Optional[float]	The variance of the data.
variance_type	Optional[int]	Enum integer representing the variance parameterization.

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter temperature data based on temperature values within a QueryOntologyCatalog.

Example

from mosaicolabs import MosaicoClient, Temperature, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for temperature values within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(Temperature.Q.value.between([273.15, 373.15]))
        .with_expression(Temperature.Q.timestamp_ns.between(1700000000, 1800000000)),
    )

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

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(Temperature.Q.value.between([273.15, 373.15]), include_timestamp_range=True)
    )

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

variance class-attribute instance-attribute

variance = None

Optional 64-bit float representing the variance of the data.

This field is injected into the model via composition, ensuring that sensor data is paired with the optional variance attribute.

Querying with the .Q Proxy

The variance field is queryable with the .Q Proxy.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.variance	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()

The <Model> placeholder adapts based on how the VarianceMixin is integrated into the data structure:

• Direct Inheritance: Used for classes like Pressure or Temperature that inherit directly from VarianceMixin to represent 1D uncertainty.
• Composition (Nested Access): If a complex model contains a field that is a subclass of VarianceMixin, the proxy allows you to traverse the hierarchy to that specific attribute.

Example

from mosaicolabs import MosaicoClient, Pressure, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter Pressure data by a specific acquisition second
    qresponse = client.query(
        QueryOntologyCatalog(Pressure.Q.variance.lt(0.76))
    )

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

variance_type class-attribute instance-attribute

variance_type = None

Optional 16-bit integer representing the variance parameterization.

This field is injected into the model via composition, ensuring that sensor data is paired with the optional covariance type attribute.

Querying with the .Q Proxy

The variance_type field is fully queryable via the .Q Proxy.

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.variance_type	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()

The <Model> placeholder adapts based on how the VarianceMixin is integrated into the data structure:

• Direct Inheritance: Used for classes like Pressure or Temperature that inherit directly from VarianceMixin to represent 1D uncertainty.
• Composition (Nested Access): If a complex model contains a field that is a subclass of VarianceMixin, the proxy allows you to traverse the hierarchy to that specific attribute.

Filtering by Precision

The following examples demonstrate how to filter sensor data based on the magnitude of the variance or the specific procedure used to calculate it.

from mosaicolabs import MosaicoClient, Pressure, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter by variance threshold
    results = client.query(QueryOntologyCatalog(
        Pressure.Q.variance.lt(0.76)
    ))

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

value instance-attribute

value

Temperature value in Kelvin.

Querying with the .Q Proxy

The temperature value is queryable via the value field.

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

Example

from mosaicolabs import MosaicoClient, Temperature, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for temperature values within a specific range
    qresponse = client.query(
        QueryOntologyCatalog(Temperature.Q.value.between([273.15, 373.15]))
    )

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

    # Filter for a specific component value and extract the first and last occurrence times
    qresponse = client.query(
        QueryOntologyCatalog(Temperature.Q.value.between([273.15, 373.15]), include_timestamp_range=True)
    )

    # 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:
                        [topic.timestamp_range.start, topic.timestamp_range.end]
                        for topic in item.topics}}")

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:

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

Raises:

Type	Description
Exception	If the class was not properly initialized via __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]}")

from_celsius classmethod

from_celsius(*, value, variance=None, variance_type=None)

Creates a Temperature instance using the value in Celsius and converting it in Kelvin using the formula Kelvin = Celsius + 273.15.

Parameters:

Name	Type	Description	Default
value	float	The temperature value in Celsius.	required
variance	Optional[float]	The variance of the data.	None
variance_type	Optional[int]	Enum integer representing the variance parameterization.	None

Returns:

Name	Type	Description
Temperature	Temperature	A Temperature instance with value in Kelvin.

from_fahrenheit classmethod

from_fahrenheit(
    *, value, variance=None, variance_type=None
)

Creates a Temperature instance using the value in Fahrenheit and converting it in Kelvin using the formula Kelvin = (Fahrenheit - 32) * 5 / 9 + 273.15.

Parameters:

Name	Type	Description	Default
value	float	The temperature value in Celsius.	required
variance	Optional[float]	The variance of the data.	None
variance_type	Optional[int]	Enum integer representing the variance parameterization.	None

Returns:

Name	Type	Description
Temperature	Temperature	A Temperature instance with value in Kelvin.

to_celsius

to_celsius()

Converts and returns the Temperature value in Celsius using the formula Celsius = Kelvin - 273.15.

Returns:

Name	Type	Description
float	float	The Temperature value in Celsius.

to_fahrenheit

to_fahrenheit()

Converts and returns the Temperature value in Fahrenheit using the formula Fahrenheit = (Kelvin - 273.15) * 9 / 5 + 32.

Returns:

Name	Type	Description
float	float	The Temperature value in Fahrenheit.

mosaicolabs.models.sensors.RobotJoint

Bases: Serializable

Snapshot of robot joint states.

Arrays must be index-aligned (e.g., names[0] corresponds to positions[0]).

Attributes:

Name	Type	Description
names	List[str]	Names of the different robot joints
positions	List[float]	Positions ([rad] or [m]) of the different robot joints
velocities	List[float]	Velocities ([rad/s] or [m/s]) of the different robot joints
efforts	List[float]	Efforts ([N] or [N/m]) applied to the different robot joints

Querying with the .Q Proxy

The robot joint states cannot be queried via the .Q proxy.

names instance-attribute

names

Names of the different robot joints

Querying with the .Q Proxy

The names are not queryable via the .Q proxy (Lists are not supported yet).

positions instance-attribute

positions

Positions ([rad] or [m]) of the different robot joints

Querying with the .Q Proxy

The positions are not queryable via the .Q proxy (Lists are not supported yet).

velocities instance-attribute

velocities

Velocities ([rad/s] or [m/s]) of the different robot joints

Querying with the .Q Proxy

The velocities are not queryable via the .Q proxy (Lists are not supported yet).

efforts instance-attribute

efforts

Efforts ([N] or [N/m]) applied to the different robot joints

Querying with the .Q Proxy

The efforts are not queryable via the .Q proxy (Lists are not supported yet).

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:

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

Raises:

Type	Description
Exception	If the class was not properly initialized via __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.sensors.Image

Bases: Serializable

Represents raw, uncompressed image data.

This class provides a flattened, row-major binary representation of an image. It is designed to handle: 1. Arbitrary Data Types: From standard uint8 RGB to float32 Depth and uint16 IR. 2. Memory Layouts: Explicit control over stride (stride) and endianness (is_bigendian). 3. Transport: Can act as a container for RAW bytes or wrap them in lossless containers (PNG).

Attributes:

Name	Type	Description
data	bytes	The flattened image memory buffer.
format	ImageFormat	The format used for serialization ('png' or 'raw').
width	int	The width of the image in pixels.
height	int	The height of the image in pixels.
stride	int	Bytes per row. Essential for alignment.
encoding	str	Pixel format (e.g., 'bgr8', 'mono16').
is_bigendian	bool	True if data is Big-Endian. Defaults to system endianness if null.

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter image data based on image parameters within a QueryOntologyCatalog.

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, Image

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for image data based on image parameters
    qresponse = client.query(
        QueryOntologyCatalog(Image.Q.width.between(1500, 2000))
        .with_expression(Image.Q.height.between(1500, 2000))
        .with_expression(Image.Q.format.eq("png")),
    )

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

data instance-attribute

data

The flattened image memory buffer.

Querying with the .Q Proxy

The data is not queryable via the data field (bytes are not comparable).

format instance-attribute

format

The format used for serialization ('png' or 'raw').

Querying with the .Q Proxy

The format is queryable via the format field.

Field Access Path	Queryable Type	Supported Operators
Image.Q.format	String	.eq(), .neq(), .match(), .in_()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, Image

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for image data based on format
    qresponse = client.query(
        QueryOntologyCatalog(Image.Q.format.eq(ImageFormat.PNG))
    )

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

width instance-attribute

width

The width of the image in pixels.

Querying with the .Q Proxy

The width is queryable via the width field.

Field Access Path	Queryable Type	Supported Operators
Image.Q.width	Integer	.eq(), .neq(), .gt(), .gte(), .lt(), .lte(), .between()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, Image

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for image data based on width
    qresponse = client.query(
        QueryOntologyCatalog(Image.Q.width.between(0, 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]}")

height instance-attribute

height

The height of the image in pixels.

Querying with the .Q Proxy

The height is queryable via the height field.

Field Access Path	Queryable Type	Supported Operators
Image.Q.height	Integer	.eq(), .neq(), .gt(), .gte(), .lt(), .lte(), .between()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, Image

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for image data based on height
    qresponse = client.query(
        QueryOntologyCatalog(Image.Q.height.between(0, 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]}")

stride instance-attribute

stride

The number of bytes per row of the image.

Querying with the .Q Proxy

The stride is queryable via the stride field.

Field Access Path	Queryable Type	Supported Operators
Image.Q.stride	Integer	.eq(), .neq(), .gt(), .gte(), .lt(), .lte(), .between()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, Image

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for image data based on stride
    qresponse = client.query(
        QueryOntologyCatalog(Image.Q.stride.between(0, 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]}")

encoding instance-attribute

encoding

The pixel encoding (e.g., 'bgr8', 'mono16'). Optional field.

Querying with the .Q Proxy

The encoding is queryable via the encoding field.

Field Access Path	Queryable Type	Supported Operators
Image.Q.encoding	String	.eq(), .neq(), .match(), .in_()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, Image

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for image data based on encoding
    qresponse = client.query(
        QueryOntologyCatalog(Image.Q.encoding.eq("bgr8"))
    )

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

is_bigendian class-attribute instance-attribute

is_bigendian = None

Store if the original data is Big-Endian. Optional field.

Querying with the .Q Proxy

The is_bigendian is queryable via the is_bigendian field.

Field Access Path	Queryable Type	Supported Operators
Image.Q.is_bigendian	Boolean	.eq(), .neq()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, Image

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for image data based on is_bigendian
    qresponse = client.query(
        QueryOntologyCatalog(Image.Q.is_bigendian.eq(True))
    )

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

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:

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

Raises:

Type	Description
Exception	If the class was not properly initialized via __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]}")

from_linear_pixels classmethod

from_linear_pixels(
    data,
    stride,
    height,
    width,
    encoding,
    is_bigendian=None,
    format=_DEFAULT_IMG_FORMAT,
)

Encodes linear pixel uint8 data into the storage container.

The "Wide Grayscale" Trick: When saving complex types (like float32 depth or uint16 raw) into standard image containers like PNG, we cannot rely on standard RGB encoders as they might apply color corrections or bit-depth reductions.

Instead, this method treats the data as a raw byte stream. It reshapes the stream into a 2D "Grayscale" image where: - Image_Height = Original_Height - Image_Width = Stride (The full row stride in bytes)

This guarantees that every bit of the original memory (including padding) is preserved losslessly.

Parameters:

Name	Type	Description	Default
data	List[int]	Flattened list of bytes (uint8).	required
stride	int	Row stride in bytes.	required
height	int	Image height.	required
width	int	Image width.	required
encoding	str	Pixel format string.	required
format	ImageFormat	Target container ('raw' or 'png').	_DEFAULT_IMG_FORMAT

Returns:

Name	Type	Description
Image	Image	An instantiated object.

to_linear_pixels

to_linear_pixels()

Decodes the storage container back to a linear byte list.

Reverses the "Wide Grayscale" encoding to return the original, flattened memory buffer.

Returns:

Type	Description
List[int]	List[int]: A list of uint8 integers representing the raw memory.

to_pillow

to_pillow()

Converts the raw binary data into a standard PIL Image.

This method performs the heavy lifting of interpretation: 1. Decoding: Unpacks the transport container (e.g., PNG -> bytes). 2. Casting: Interprets bytes according to self.encoding (e.g., as float32). 3. Endianness: Swaps bytes if the source endianness differs from the local CPU. 4. Color Swap: Converts BGR (common in OpenCV/Robotics) to RGB (required by PIL).

Returns:

Type	Description
Image	PILImage.Image: A visualizable image object.

Raises:

Type	Description
NotImplementedError	If the encoding is unknown.
ValueError	If data size doesn't match dimensions.

from_pillow classmethod

from_pillow(
    pil_image, target_encoding=None, output_format=None
)

Factory method to create an Image from a PIL object.

Automatically handles:

• Data flattening (row-major).
• Stride calculation.
• RGB to BGR conversion (if target_encoding requires it).
• Type casting (e.g., float -> uint8).

Parameters:

Name	Type	Description	Default
pil_image	Image	Source image.	required
target_encoding	Optional[str]	Target pixel format (e.g., "bgr8").	None
output_format	Optional[ImageFormat]	('raw' or 'png').	None

Returns:

Name	Type	Description
Image	Image	Populated data object.

mosaicolabs.models.sensors.CompressedImage

Bases: Serializable

Represents image data stored as a compressed binary blob (e.g. JPEG, PNG, H264, ...).

This class acts as a data container for encoded streams. It distinguishes between: 1. Stateless Formats (JPEG, PNG, TIFF): Can be decoded directly via .to_image(). 2. Stateful Formats (H.264, HEVC): Require a StatefulDecodingSession to maintain reference frames across multiple messages.

Attributes:

Name	Type	Description
data	bytes	The compressed binary payload.
format	str	The format identifier string (e.g., 'jpeg', 'png').

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter image data based on image parameters within a QueryOntologyCatalog.

Example "Reading H264 CompressedImage":

from mosaicolabs import MosaicoClient, CompressedImage, StatefulDecodingSession

with MosaicoClient.connect("localhost", 6726) as client:
    seq_handler = client.sequence_handler("multi_camera_mission")

    # Initialize the session contextually with the data streamer
    decoding_session = StatefulDecodingSession()

    # Iterate through interleaved topics
    for topic, msg in seq_handler.get_data_streamer():
        img = msg.get_data(CompressedImage)

        if img is None:
            # It is not a CompressedImage
            continue

        if img.format in [ImageFormat.H264, ImageFormat.HEVC]:
            # Use the session for stateful video decoding
            # The 'context' parameter ensures we use the correct reference frames for this topic
            pil_img = decoding_session.decode(
                img_data=img.data,
                format=img.format,
                context=topic
            )
        else:
            # Use standard stateless decoding for JPEG/PNG
            pil_img = img.to_image()

        if pil_img:
            pil_img.show()

    decoding_session.close()

data instance-attribute

data

The serialized (compressed) image data as bytes.

Querying with the .Q Proxy

The data is not queryable via the data field (bytes are not comparable).

format instance-attribute

format

The compression format (e.g., 'jpeg', 'png').

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
CompressedImage.Q.format	String	.eq(), .neq(), .match(), .in_()

Example

from mosaicolabs import MosaicoClient, CompressedImage, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for image data based on image parameters
    qresponse = client.query(
        QueryOntologyCatalog(CompressedImage.Q.format.eq("jpeg"))
    )

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

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:

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

Raises:

Type	Description
Exception	If the class was not properly initialized via __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]}")

to_image

to_image(codec=None, **kwargs)

Decompresses the stored binary data into a usable PIL Image object.

This method serves as the standard interface for converting compressed binary blobs back into pixel data. It defaults to a stateless decoding approach suitable for independent image frames.

Parameters:

Name	Type	Description	Default
codec	Optional[Any]	An optional codec instance implementing the .decode(img_bytes, format, **kwargs) interface. If None, it defaults to _StatelessDefaultCodec. This allows for the injection of custom, optimized, or hardware-accelerated decoders.	None
**kwargs	Any	Arbitrary keyword arguments passed directly to the codec's decode method (e.g., quality hints or specific decoder flags).	{}

Warning

Stateless Default: The default codec cannot maintain temporal state. For stateful formats like H.264 or HEVC, calling this method without a specialized codec will return None.

For multi-topic video sequences, use StatefulDecodingSession.decode() instead to prevent memory corruption and visual artifacts caused by interleaved P-frames.

Returns:

Type	Description
Optional[Image]	Optional[PILImage.Image]: A ready-to-use Pillow image object. Returns None if the data is empty, the format is stateful (and no compatible codec was provided), or decoding fails.

Example "Reading PNG CompressedImage":

from mosaicolabs import MosaicoClient, CompressedImage

with MosaicoClient.connect("localhost", 6726) as client:
    seq_handler = client.sequence_handler("multi_camera_mission")

    # Iterate through interleaved topics
    for topic, msg in seq_handler.get_data_streamer():
        img = msg.get_data(CompressedImage)

        if img is None:
            # It is not a CompressedImage
            continue

        # Standard usage for JPEG/PNG
        pil_img = img.to_image()

Example "Reading H264 CompressedImage":

from mosaicolabs import MosaicoClient, CompressedImage, StatefulDecodingSession

with MosaicoClient.connect("localhost", 6726) as client:
    seq_handler = client.sequence_handler("multi_camera_mission")

    # Initialize the session contextually with the data streamer
    decoding_session = StatefulDecodingSession()

    # Iterate through interleaved topics
    for topic, msg in seq_handler.get_data_streamer():
        img = msg.get_data(CompressedImage)

        if img is None:
            # It is not a CompressedImage
            continue

        if img.format in [ImageFormat.H264, ImageFormat.HEVC]:
            # Use the session for stateful video decoding
            # The 'context' parameter ensures we use the correct reference frames for this topic
            pil_img = decoding_session.decode(
                img_data=img.data,
                format=img.format,
                context=topic
            )
        else:
            # Use standard stateless decoding for JPEG/PNG
            pil_img = img.to_image()

        if pil_img:
            pil_img.show()

    decoding_session.close()

from_image classmethod

from_image(image, format=PNG, **kwargs)

Factory method to create a CompressedImage from a PIL Image.

NOTE: The function use the _DefaultCodec which is valid for stateless formats only ('png', 'jpeg', ...). If dealing with a stateful compressed image, the conversion must be made via user defined encoding algorithms.

Parameters:

Name	Type	Description	Default
image	Image	The source Pillow image.	required
format	ImageFormat	The target compression format (default: 'jpeg').	PNG
**kwargs	Any	Additional arguments passed to the codec's encode method (e.g., quality=90).	{}

Returns:

Name	Type	Description
CompressedImage	CompressedImage	A new instance containing the compressed bytes.

Raises:

Type	Description
ValueError	If no codec is found or encoding fails.

mosaicolabs.models.sensors.StatefulDecodingSession

StatefulDecodingSession()

Manages the stateful decoding of video streams for a specific reading session.

Unlike standard image formats (JPEG/PNG), video encodings like H.264 and HEVC utilize temporal compression (P-frames and B-frames), which require a persistent decoding state (reference frames).

This class maintains unique av.CodecContext instances for each provided context string (typically the topic_name), ensuring that interleaved frames from multiple video topics do not interfere with each other.

Supported Formats

• ImageFormat.H264
• ImageFormat.HEVC

Example

from mosaicolabs import MosaicoClient, CompressedImage, StatefulDecodingSession

with MosaicoClient.connect("localhost", 6726) as client:
    seq_handler = client.sequence_handler("multi_camera_mission")

    # Initialize the session contextually with the data streamer
    decoding_session = StatefulDecodingSession()

    # Iterate through interleaved topics
    for topic, msg in seq_handler.get_data_streamer():
        img = msg.get_data(CompressedImage)

        if img.format in [ImageFormat.H264, ImageFormat.HEVC]:
            # Use the session for stateful video decoding
            # The 'context' parameter ensures we use the correct reference frames for this topic
            pil_img = decoding_session.decode(
                img_data=img.data,
                format=img.format,
                context=topic
            )
        else:
            # Use standard stateless decoding for JPEG/PNG
            pil_img = img.to_image()

        if pil_img:
            pil_img.show()

    decoding_session.close()

decode

decode(*, img_bytes, format, context)

Decodes stateful compressed image data (video frames) using a persistent temporal context.

It utilizes the context string to look up or initialize a persistent av.CodecContext. This ensures that P-frames and B-frames are correctly applied to the reference frames of their specific stream.

Parameters:

Name	Type	Description	Default
img_bytes	bytes	The raw compressed binary blob extracted from a CompressedImage message.	required
format	ImageFormat	The encoding format. Expected to be a stateful format supported by the session (e.g., ImageFormat.H264 or ImageFormat.HEVC).	required
context	str	A unique identifier for the data stream, typically the topic_name. This key isolates the decoding state to prevent memory corruption when frames from multiple sources are interleaved in the same processing loop.	required

Returns:

Type	Description
Optional[Image]	Optional[PILImage.Image]: - A PIL.Image.Image object containing the decoded frame. - None if the format is unsupported, the data is corrupted, or the frame is a non-visual packet (e.g., internal metadata).

Note

The first few calls to this method for a new context may return None if the stream starts with inter-frames (P/B) before hitting an IDR-frame (I-frame/Keyframe).

close

close()

Explicitly release resources (optional, GC usually handles this).

mosaicolabs.models.sensors.ImageFormat

Bases: str, Enum

Defines the supported encoding and container formats for image data.

The enum differentiates between Stateless formats (independent frames) and Stateful formats (video streams with temporal dependencies).

RAW class-attribute instance-attribute

RAW = 'raw'

Uncompressed pixel data. Represents a raw buffer of pixels (e.g., RGB, BGR, or Grayscale).

PNG class-attribute instance-attribute

PNG = 'png'

Portable Network Graphics. A lossless compression format suitable for masks, overlays, and synthetic data where pixel perfection is required.

JPEG class-attribute instance-attribute

JPEG = 'jpeg'

Joint Photographic Experts Group. The standard lossy compression format for natural images, balancing file size and visual quality.

TIFF class-attribute instance-attribute

TIFF = 'tiff'

Tagged Image File Format. Preferred for high-bit depth (16-bit) or scientific data where metadata preservation is critical.

H264 class-attribute instance-attribute

H264 = 'h264'

Advanced Video Coding (AVC). A stateful format using inter-frame compression. Requires a StatefulDecodingSession to maintain temporal context.

HEVC class-attribute instance-attribute

HEVC = 'hevc'

High Efficiency Video Coding (H.265). A high-performance stateful format. Requires a StatefulDecodingSession and provides superior compression to H.264.

mosaicolabs.models.sensors.CameraInfo

Bases: Serializable

Meta-information for interpreting images from a calibrated camera.

This structure mirrors standard robotics camera models (e.g., ROS sensor_msgs/CameraInfo). It enables pipelines to rectify distorted images or project 3D points onto the 2D image plane.

Attributes:

Name	Type	Description
height	int	Height in pixels of the image with which the camera was calibrated
width	int	Width in pixels of the image with which the camera was calibrated
distortion_model	str	The distortion model used
distortion_parameters	list[float]	The distortion coefficients (k1, k2, t1, t2, k3...). Size depends on the model.
intrinsic_parameters	list[float]	The 3x3 Intrinsic Matrix (K) flattened row-major.
rectification_parameters	list[float]	The 3x3 Rectification Matrix (R) flattened row-major.
projection_parameters	list[float]	The 3x4 Projection Matrix (P) flattened row-major.
binning	Optional[Vector2d]	Hardware binning factor (x, y). If null, assumes (0, 0) (no binning).
roi	Optional[ROI]	Region of Interest. Used if the image is a sub-crop of the full resolution.

Querying with the .Q Proxy

This class is fully queryable via the .Q proxy. You can filter camera data based on camera parameters within a QueryOntologyCatalog.

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, CameraInfo

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for camera data based on camera parameters
    qresponse = client.query(
        QueryOntologyCatalog(CameraInfo.Q.height.between(1080, 2160))
    )

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

height instance-attribute

height

Height in pixels of the image with which the camera was calibrated

Querying with the .Q Proxy

The height is queryable via the height field.

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

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, CameraInfo

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for camera data based on camera parameters
    qresponse = client.query(
        QueryOntologyCatalog(CameraInfo.Q.height.between(1080, 2160))
    )

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

width instance-attribute

width

Width in pixels of the image with which the camera was calibrated

Querying with the .Q Proxy

The width is queryable via the width field.

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

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, CameraInfo

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for camera data based on camera parameters
    qresponse = client.query(
        QueryOntologyCatalog(CameraInfo.Q.width.between(1920, 3840))
    )

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

distortion_model instance-attribute

distortion_model

The distortion model used

Querying with the .Q Proxy

The distortion model is queryable via the distortion_model field.

Field Access Path	Queryable Type	Supported Operators
CameraInfo.Q.distortion_model	Categorical	.eq(), .neq(), .in_()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, CameraInfo

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for camera data based on camera parameters
    qresponse = client.query(
        QueryOntologyCatalog(CameraInfo.Q.distortion_model.eq("plumb_bob"))
    )

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

distortion_parameters instance-attribute

distortion_parameters

The distortion coefficients (k1, k2, t1, t2, k3...). Size depends on the model.

Querying with the .Q Proxy

The distortion parameters are not queryable via the .Q proxy (Lists are not supported yet).

intrinsic_parameters instance-attribute

intrinsic_parameters

The 3x3 Intrinsic Matrix (K) flattened row-major.

Querying with the .Q Proxy

The intrinsic parameters are not queryable via the .Q proxy (Lists are not supported yet).

rectification_parameters instance-attribute

rectification_parameters

The 3x3 Rectification Matrix (R) flattened row-major.

Querying with the .Q Proxy

The rectification parameters cannot be queried via the .Q proxy (Lists are not supported yet).

projection_parameters instance-attribute

projection_parameters

The 3x4 Projection Matrix (P) flattened row-major.

Querying with the .Q Proxy

The projection parameters cannot be queried via the .Q proxy (Lists are not supported yet).

binning class-attribute instance-attribute

binning = None

Hardware binning factor (x, y). If null, assumes (0, 0) (no binning).

Querying with the .Q Proxy

The binning parameters are queryable via the binning field.

Field Access Path	Queryable Type	Supported Operators
CameraInfo.Q.binning.x	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
CameraInfo.Q.binning.y	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, CameraInfo

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for camera data based on camera parameters
    qresponse = client.query(
        QueryOntologyCatalog(CameraInfo.Q.binning.x.eq(2))
    )

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

roi class-attribute instance-attribute

roi = None

Region of Interest. Used if the image is a sub-crop of the full resolution.

Querying with the .Q Proxy

The roi parameters are queryable via the roi field.

Field Access Path	Queryable Type	Supported Operators
CameraInfo.Q.roi.offset.x	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
CameraInfo.Q.roi.offset.y	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
CameraInfo.Q.roi.width	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()
CameraInfo.Q.roi.height	Numeric	.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between()

Example

from mosaicolabs import MosaicoClient, QueryOntologyCatalog, CameraInfo

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for camera data based on camera parameters
    qresponse = client.query(
        QueryOntologyCatalog(CameraInfo.Q.roi.offset.x.eq(2))
    )

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

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:

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

Raises:

Type	Description
Exception	If the class was not properly initialized via __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]}")