Geometry Models
mosaicolabs.models.data.geometry ¶
This module defines the fundamental building blocks for spatial representation, including vectors, points, quaternions, and rigid-body transforms.
The module follows a Two-Tier Architecture to optimize both internal efficiency and public usability:
- Internal Structs (
_Struct): Pure data containers that define the physical memory layout and the PyArrow schema. These are intended for embedding within larger composite objects (like aPoseorTransform) to avoid attaching redundant metadata headers or timestamps to every inner field. - Public Classes: High-level models that combine spatial data with Mosaico's transport and serialization logic.
These inherit from the internal structs and inject support for auto-registration (
Serializable), and uncertainty tracking (CovarianceMixin).
Vector2d ¶
Bases: _Vector2dStruct, Serializable, CovarianceMixin
A public 2D Vector for platform-wide transmission.
This class combines the [x, y] coordinates with full Mosaico transport logic. It is used to represent quantities such as velocity, acceleration, or directional forces.
Attributes:
| Name | Type | Description |
|---|---|---|
x |
float64
|
Vector X component. |
y |
float64
|
Vector Y component. |
covariance |
Optional[list_(float64)]
|
Optional flattened 2x2 covariance matrix representing the uncertainty of the vector measurement. |
covariance_type |
Optional[int16]
|
Enum integer representing the parameterization of the covariance matrix. |
Querying with the .Q Proxy¶
This class fields are queryable when constructing a QueryOntologyCatalog
via the .Q proxy. Check the fields documentation for detailed description.
Example
from mosaicolabs import MosaicoClient, Vector2d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(
QueryOntologyCatalog(Vector2d.Q.y.leq(123.4))
.with_expression(Vector2d.Q.timestamp_ns.between([1770282868, 1770290127]))
)
# 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(Vector2d.Q.y.leq(123.4), 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}}")
covariance
class-attribute
instance-attribute
¶
covariance = MosaicoField(
default=None,
nullable=True,
description="The covariance matrix (flattened) of the data.",
)
Optional list of 64-bit floats representing the flattened matrix.
Querying with the .Q Proxy¶
Non-Queryable
The field is not queryable with the .Q Proxy.
covariance_type
class-attribute
instance-attribute
¶
covariance_type = MosaicoField(
default=None,
nullable=True,
description="Enum integer representing the covariance parameterization.",
)
Optional 16-bit integer representing the covariance enum.
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 covariance_type field is fully queryable via the .Q Proxy. The <Model> placeholder
in the path represents any Mosaico class that exposes covariance information, either directly or through its internal fields.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.covariance_type |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
The <Model> placeholder adapts based on how the CovarianceMixin is integrated into your data structure:
- Direct Inheritance: Represents any class (e.g.,
Vector3d,Quaternion) that inherits directly fromCovarianceMixin. - Composition (Nested Fields): When a complex model (like
IMU) contains fields that are themselves covariance-aware, the proxy allows you to "drill down" to that specific attribute. For example, sinceIMU.accelerationis aVector3d, you access its covariance type viaIMU.Q.acceleration.covariance_type.
Filtering by Calibration Type
This example demonstrates searching for data segments where the acceleration was derived from a specific calibrated procedure.
from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog
# Assume FROM_CALIBRATED_PROCEDURE is a user-defined integer constant
with MosaicoClient.connect("localhost", 6726) as client:
# Target the covariance_type nested within the acceleration field
qbuilder = QueryOntologyCatalog(
IMU.Q.acceleration.covariance_type.eq(FROM_CALIBRATED_PROCEDURE)
)
results = client.query(qbuilder)
if results:
for item in results:
print(f"Sequence: {item.sequence.name}")
print(f"Matching Topics: {[topic.name for topic in item.topics]}")
x
class-attribute
instance-attribute
¶
x = MosaicoField(
nullable=True, description="Vector x component."
)
The Vector X component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector2dStruct (i.e. Vector2d, Point2d)
or any custom user-defined class that is a subclass of Serializable and derives from _Vector2dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector2d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector2d.Q.x.leq(123.4)))
# 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(Vector2d.Q.x.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Point2d.Q.x.leq(123.4)))
# 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]}")
y
class-attribute
instance-attribute
¶
y = MosaicoField(
nullable=True, description="Vector y component."
)
The Vector Y component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector2dStruct (i.e. Vector2d, Point2d)
or any custom user-defined class that inherits from Serializable and derives from _Vector2dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector2d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector2d.Q.y.leq(123.4)))
# 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(Vector2d.Q.y.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Point2d.Q.y.leq(123.4)))
# 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
¶
Checks if a class is registered.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if registered. |
ontology_tag
classmethod
¶
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., |
Raises:
| Type | Description |
|---|---|
Exception
|
If the class was not properly initialized via |
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_list
classmethod
¶
Creates a struct instance from a raw list.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
list[float]
|
A list containing exactly 2 float values: [x, y]. |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the input list does not have a length of 2. |
Vector3d ¶
Bases: _Vector3dStruct, Serializable, CovarianceMixin
A public 3D Vector for platform-wide transmission.
This class combines the [x, y, z] coordinates with full Mosaico transport logic. It is used to represent quantities such as velocity, acceleration, or directional forces.
Attributes:
| Name | Type | Description |
|---|---|---|
x |
float64
|
Vector X component. |
y |
float64
|
Vector Y component. |
z |
float64
|
Vector Z component. |
covariance |
Optional[list_(float64)]
|
Optional flattened 3x3 covariance matrix representing the uncertainty of the vector measurement. |
covariance_type |
Optional[int16]
|
Enum integer representing the parameterization of the covariance matrix. |
Querying with the .Q Proxy¶
This class fields are queryable when constructing a QueryOntologyCatalog
via the .Q proxy. Check the fields documentation for detailed description.
Example
from mosaicolabs import MosaicoClient, Vector3d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(
QueryOntologyCatalog(Vector3d.Q.y.leq(123.4))
.with_expression(Vector3d.Q.timestamp_ns.between([1770282868, 1770290127]))
)
# 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(Vector3d.Q.z.leq(123.4), 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}}")
covariance
class-attribute
instance-attribute
¶
covariance = MosaicoField(
default=None,
nullable=True,
description="The covariance matrix (flattened) of the data.",
)
Optional list of 64-bit floats representing the flattened matrix.
Querying with the .Q Proxy¶
Non-Queryable
The field is not queryable with the .Q Proxy.
covariance_type
class-attribute
instance-attribute
¶
covariance_type = MosaicoField(
default=None,
nullable=True,
description="Enum integer representing the covariance parameterization.",
)
Optional 16-bit integer representing the covariance enum.
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 covariance_type field is fully queryable via the .Q Proxy. The <Model> placeholder
in the path represents any Mosaico class that exposes covariance information, either directly or through its internal fields.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.covariance_type |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
The <Model> placeholder adapts based on how the CovarianceMixin is integrated into your data structure:
- Direct Inheritance: Represents any class (e.g.,
Vector3d,Quaternion) that inherits directly fromCovarianceMixin. - Composition (Nested Fields): When a complex model (like
IMU) contains fields that are themselves covariance-aware, the proxy allows you to "drill down" to that specific attribute. For example, sinceIMU.accelerationis aVector3d, you access its covariance type viaIMU.Q.acceleration.covariance_type.
Filtering by Calibration Type
This example demonstrates searching for data segments where the acceleration was derived from a specific calibrated procedure.
from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog
# Assume FROM_CALIBRATED_PROCEDURE is a user-defined integer constant
with MosaicoClient.connect("localhost", 6726) as client:
# Target the covariance_type nested within the acceleration field
qbuilder = QueryOntologyCatalog(
IMU.Q.acceleration.covariance_type.eq(FROM_CALIBRATED_PROCEDURE)
)
results = client.query(qbuilder)
if results:
for item in results:
print(f"Sequence: {item.sequence.name}")
print(f"Matching Topics: {[topic.name for topic in item.topics]}")
x
class-attribute
instance-attribute
¶
x = MosaicoField(
nullable=True, description="Vector x component."
)
The Vector X component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector3dStruct (i.e. Vector3d, Point3d)
or any custom user-defined class that inherits from Serializable and derives from _Vector3dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector3d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector3d.Q.x.leq(123.4)))
# 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(Vector3d.Q.x.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Point3d.Q.x.leq(123.4)))
# 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]}")
y
class-attribute
instance-attribute
¶
y = MosaicoField(
nullable=True, description="Vector y component."
)
The Vector Y component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector3dStruct (i.e. Vector3d, Point3d)
or any custom user-defined class that inherits from Serializable and derives from _Vector3dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector3d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector3d.Q.y.leq(123.4)))
# 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(Vector3d.Q.y.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Point3d.Q.y.leq(123.4)))
# 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]}")
z
class-attribute
instance-attribute
¶
z = MosaicoField(
nullable=True, description="Vector z component."
)
The Vector Z component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector3dStruct (i.e. Vector3d, Point3d)
or any custom user-defined Serializable class.
Example
from mosaicolabs import MosaicoClient, Vector3d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector3d.Q.z.leq(123.4)))
# 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(Vector3d.Q.z.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Point3d.Q.z.leq(123.4)))
# 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
¶
Checks if a class is registered.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if registered. |
ontology_tag
classmethod
¶
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., |
Raises:
| Type | Description |
|---|---|
Exception
|
If the class was not properly initialized via |
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_list
classmethod
¶
Creates a struct instance from a raw list.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
list[float]
|
A list containing exactly 3 float values: [x, y, z]. |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the input list does not have a length of 3. |
Vector4d ¶
Bases: _Vector4dStruct, Serializable, CovarianceMixin
A public 4D Vector for platform-wide transmission.
This class combines the [x, y, z, w] coordinates with full Mosaico transport logic. It is used to represent quantities such as velocity, acceleration, or directional forces.
Attributes:
| Name | Type | Description |
|---|---|---|
x |
float64
|
Vector X component. |
y |
float64
|
Vector Y component. |
z |
float64
|
Vector Z component. |
w |
float64
|
Vector W component. |
covariance |
Optional[list_(float64)]
|
Optional flattened 4x4 covariance matrix representing the uncertainty of the vector measurement. |
covariance_type |
Optional[int16]
|
Enum integer representing the parameterization of the covariance matrix. |
Querying with the .Q Proxy¶
This class fields are queryable when constructing a QueryOntologyCatalog
via the .Q proxy. Check the fields documentation for detailed description.
Example
from mosaicolabs import MosaicoClient, Vector4d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(
QueryOntologyCatalog(Vector4d.Q.y.leq(123.4))
.with_expression(Vector4d.Q.timestamp_ns.between([1770282868, 1770290127]))
)
# 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(Vector4d.Q.y.leq(123.4), 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}}")
covariance
class-attribute
instance-attribute
¶
covariance = MosaicoField(
default=None,
nullable=True,
description="The covariance matrix (flattened) of the data.",
)
Optional list of 64-bit floats representing the flattened matrix.
Querying with the .Q Proxy¶
Non-Queryable
The field is not queryable with the .Q Proxy.
covariance_type
class-attribute
instance-attribute
¶
covariance_type = MosaicoField(
default=None,
nullable=True,
description="Enum integer representing the covariance parameterization.",
)
Optional 16-bit integer representing the covariance enum.
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 covariance_type field is fully queryable via the .Q Proxy. The <Model> placeholder
in the path represents any Mosaico class that exposes covariance information, either directly or through its internal fields.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.covariance_type |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
The <Model> placeholder adapts based on how the CovarianceMixin is integrated into your data structure:
- Direct Inheritance: Represents any class (e.g.,
Vector3d,Quaternion) that inherits directly fromCovarianceMixin. - Composition (Nested Fields): When a complex model (like
IMU) contains fields that are themselves covariance-aware, the proxy allows you to "drill down" to that specific attribute. For example, sinceIMU.accelerationis aVector3d, you access its covariance type viaIMU.Q.acceleration.covariance_type.
Filtering by Calibration Type
This example demonstrates searching for data segments where the acceleration was derived from a specific calibrated procedure.
from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog
# Assume FROM_CALIBRATED_PROCEDURE is a user-defined integer constant
with MosaicoClient.connect("localhost", 6726) as client:
# Target the covariance_type nested within the acceleration field
qbuilder = QueryOntologyCatalog(
IMU.Q.acceleration.covariance_type.eq(FROM_CALIBRATED_PROCEDURE)
)
results = client.query(qbuilder)
if results:
for item in results:
print(f"Sequence: {item.sequence.name}")
print(f"Matching Topics: {[topic.name for topic in item.topics]}")
x
class-attribute
instance-attribute
¶
x = MosaicoField(
nullable=True, description="Vector x component."
)
The Vector X component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector4dStruct (i.e. Vector4d, Quaternion)
or any custom user-defined class that inherits from Serializable and derives from _Vector4dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector4d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector4d.Q.x.leq(123.4)))
# 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(Vector4d.Q.x.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Quaternion.Q.x.leq(0.4)))
# 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]}")
y
class-attribute
instance-attribute
¶
y = MosaicoField(
nullable=True, description="Vector y component."
)
The Vector Y component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector4dStruct (i.e. Vector4d, Quaternion)
or any custom user-defined class that inherits from Serializable and derives from _Vector4dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector4d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector4d.Q.y.leq(123.4)))
# 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(Vector4d.Q.y.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Quaternion.Q.y.leq(0.4)))
# 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]}")
z
class-attribute
instance-attribute
¶
z = MosaicoField(
nullable=True, description="Vector z component."
)
The Vector Z component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector4dStruct (i.e. Vector4d, Quaternion)
or any custom user-defined class that inherits from Serializable and derives from _Vector4dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector4d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector4d.Q.z.leq(123.4)))
# 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(Vector4d.Q.z.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Quaternion.Q.z.leq(0.4)))
# 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]}")
w
class-attribute
instance-attribute
¶
w = MosaicoField(
nullable=True, description="Vector w component."
)
The Vector W component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.w |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector4dStruct (i.e. Vector4d, Quaternion)
or any custom user-defined class that inherits from Serializable and derives from _Vector4dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector4d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector4d.Q.w.leq(123.4)))
# 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(Vector4d.Q.w.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Quaternion.Q.w.leq(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]}")
is_registered
classmethod
¶
Checks if a class is registered.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if registered. |
ontology_tag
classmethod
¶
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., |
Raises:
| Type | Description |
|---|---|
Exception
|
If the class was not properly initialized via |
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_list
classmethod
¶
Creates a struct instance from a raw list.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
list[float]
|
A list containing exactly 4 float values: [x, y, z, w]. |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the input list does not have a length of 4. |
Point2d ¶
Bases: _Vector2dStruct, Serializable, CovarianceMixin
Semantically represents a specific location (Point) in 2D space.
Structurally identical to a 2D Vector, but distinguished within the Mosaico API to clarify intent in spatial operations. Use this class for 2D coordinate data that requires Mosaico transport logic.
Attributes:
| Name | Type | Description |
|---|---|---|
x |
float64
|
Point X coordinate. |
y |
float64
|
Point Y coordinate. |
covariance |
Optional[list_(float64)]
|
Optional flattened 2x2 covariance matrix representing the uncertainty of the point measurement. |
covariance_type |
Optional[int16]
|
Enum integer representing the parameterization of the covariance matrix. |
Querying with the .Q Proxy¶
This class fields are queryable when constructing a QueryOntologyCatalog
via the .Q proxy. Check the fields documentation for detailed description.
Example
from mosaicolabs import MosaicoClient, Point2d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(
QueryOntologyCatalog(Point2d.Q.y.leq(123.4))
.with_expression(Point2d.Q.timestamp_ns.between([1770282868, 1770290127]))
)
# 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(Point2d.Q.y.leq(123.4), 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}}")
covariance
class-attribute
instance-attribute
¶
covariance = MosaicoField(
default=None,
nullable=True,
description="The covariance matrix (flattened) of the data.",
)
Optional list of 64-bit floats representing the flattened matrix.
Querying with the .Q Proxy¶
Non-Queryable
The field is not queryable with the .Q Proxy.
covariance_type
class-attribute
instance-attribute
¶
covariance_type = MosaicoField(
default=None,
nullable=True,
description="Enum integer representing the covariance parameterization.",
)
Optional 16-bit integer representing the covariance enum.
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 covariance_type field is fully queryable via the .Q Proxy. The <Model> placeholder
in the path represents any Mosaico class that exposes covariance information, either directly or through its internal fields.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.covariance_type |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
The <Model> placeholder adapts based on how the CovarianceMixin is integrated into your data structure:
- Direct Inheritance: Represents any class (e.g.,
Vector3d,Quaternion) that inherits directly fromCovarianceMixin. - Composition (Nested Fields): When a complex model (like
IMU) contains fields that are themselves covariance-aware, the proxy allows you to "drill down" to that specific attribute. For example, sinceIMU.accelerationis aVector3d, you access its covariance type viaIMU.Q.acceleration.covariance_type.
Filtering by Calibration Type
This example demonstrates searching for data segments where the acceleration was derived from a specific calibrated procedure.
from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog
# Assume FROM_CALIBRATED_PROCEDURE is a user-defined integer constant
with MosaicoClient.connect("localhost", 6726) as client:
# Target the covariance_type nested within the acceleration field
qbuilder = QueryOntologyCatalog(
IMU.Q.acceleration.covariance_type.eq(FROM_CALIBRATED_PROCEDURE)
)
results = client.query(qbuilder)
if results:
for item in results:
print(f"Sequence: {item.sequence.name}")
print(f"Matching Topics: {[topic.name for topic in item.topics]}")
x
class-attribute
instance-attribute
¶
x = MosaicoField(
nullable=True, description="Vector x component."
)
The Vector X component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector2dStruct (i.e. Vector2d, Point2d)
or any custom user-defined class that is a subclass of Serializable and derives from _Vector2dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector2d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector2d.Q.x.leq(123.4)))
# 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(Vector2d.Q.x.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Point2d.Q.x.leq(123.4)))
# 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]}")
y
class-attribute
instance-attribute
¶
y = MosaicoField(
nullable=True, description="Vector y component."
)
The Vector Y component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector2dStruct (i.e. Vector2d, Point2d)
or any custom user-defined class that inherits from Serializable and derives from _Vector2dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector2d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector2d.Q.y.leq(123.4)))
# 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(Vector2d.Q.y.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Point2d.Q.y.leq(123.4)))
# 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
¶
Checks if a class is registered.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if registered. |
ontology_tag
classmethod
¶
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., |
Raises:
| Type | Description |
|---|---|
Exception
|
If the class was not properly initialized via |
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_list
classmethod
¶
Creates a struct instance from a raw list.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
list[float]
|
A list containing exactly 2 float values: [x, y]. |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the input list does not have a length of 2. |
Point3d ¶
Bases: _Vector3dStruct, Serializable, CovarianceMixin
Semantically represents a specific location (Point) in 3D space.
The Point3d class is used to instantiate a 3D coordinate message for
transmission over the platform. It is structurally identical
to a 3D Vector but is used to denote state rather than direction.
Attributes:
| Name | Type | Description |
|---|---|---|
x |
float64
|
Point X coordinate. |
y |
float64
|
Point Y coordinate. |
z |
float64
|
Point Z coordinate. |
covariance |
Optional[list_(float64)]
|
Optional flattened 3x3 covariance matrix representing the uncertainty of the point measurement. |
covariance_type |
Optional[int16]
|
Enum integer representing the parameterization of the covariance matrix. |
Querying with the .Q Proxy¶
This class fields are queryable when constructing a QueryOntologyCatalog
via the .Q proxy. Check the fields documentation for detailed description.
Example
from mosaicolabs import MosaicoClient, Point3d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(
QueryOntologyCatalog(Point3d.Q.y.leq(123.4))
.with_expression(Point3d.Q.timestamp_ns.between([1770282868, 1770290127]))
)
# 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(Point3d.Q.z.leq(123.4), 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}}")
covariance
class-attribute
instance-attribute
¶
covariance = MosaicoField(
default=None,
nullable=True,
description="The covariance matrix (flattened) of the data.",
)
Optional list of 64-bit floats representing the flattened matrix.
Querying with the .Q Proxy¶
Non-Queryable
The field is not queryable with the .Q Proxy.
covariance_type
class-attribute
instance-attribute
¶
covariance_type = MosaicoField(
default=None,
nullable=True,
description="Enum integer representing the covariance parameterization.",
)
Optional 16-bit integer representing the covariance enum.
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 covariance_type field is fully queryable via the .Q Proxy. The <Model> placeholder
in the path represents any Mosaico class that exposes covariance information, either directly or through its internal fields.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.covariance_type |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
The <Model> placeholder adapts based on how the CovarianceMixin is integrated into your data structure:
- Direct Inheritance: Represents any class (e.g.,
Vector3d,Quaternion) that inherits directly fromCovarianceMixin. - Composition (Nested Fields): When a complex model (like
IMU) contains fields that are themselves covariance-aware, the proxy allows you to "drill down" to that specific attribute. For example, sinceIMU.accelerationis aVector3d, you access its covariance type viaIMU.Q.acceleration.covariance_type.
Filtering by Calibration Type
This example demonstrates searching for data segments where the acceleration was derived from a specific calibrated procedure.
from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog
# Assume FROM_CALIBRATED_PROCEDURE is a user-defined integer constant
with MosaicoClient.connect("localhost", 6726) as client:
# Target the covariance_type nested within the acceleration field
qbuilder = QueryOntologyCatalog(
IMU.Q.acceleration.covariance_type.eq(FROM_CALIBRATED_PROCEDURE)
)
results = client.query(qbuilder)
if results:
for item in results:
print(f"Sequence: {item.sequence.name}")
print(f"Matching Topics: {[topic.name for topic in item.topics]}")
x
class-attribute
instance-attribute
¶
x = MosaicoField(
nullable=True, description="Vector x component."
)
The Vector X component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector3dStruct (i.e. Vector3d, Point3d)
or any custom user-defined class that inherits from Serializable and derives from _Vector3dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector3d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector3d.Q.x.leq(123.4)))
# 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(Vector3d.Q.x.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Point3d.Q.x.leq(123.4)))
# 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]}")
y
class-attribute
instance-attribute
¶
y = MosaicoField(
nullable=True, description="Vector y component."
)
The Vector Y component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector3dStruct (i.e. Vector3d, Point3d)
or any custom user-defined class that inherits from Serializable and derives from _Vector3dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector3d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector3d.Q.y.leq(123.4)))
# 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(Vector3d.Q.y.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Point3d.Q.y.leq(123.4)))
# 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]}")
z
class-attribute
instance-attribute
¶
z = MosaicoField(
nullable=True, description="Vector z component."
)
The Vector Z component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector3dStruct (i.e. Vector3d, Point3d)
or any custom user-defined Serializable class.
Example
from mosaicolabs import MosaicoClient, Vector3d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector3d.Q.z.leq(123.4)))
# 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(Vector3d.Q.z.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Point3d.Q.z.leq(123.4)))
# 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
¶
Checks if a class is registered.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if registered. |
ontology_tag
classmethod
¶
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., |
Raises:
| Type | Description |
|---|---|
Exception
|
If the class was not properly initialized via |
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_list
classmethod
¶
Creates a struct instance from a raw list.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
list[float]
|
A list containing exactly 3 float values: [x, y, z]. |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the input list does not have a length of 3. |
Quaternion ¶
Bases: _Vector4dStruct, Serializable, CovarianceMixin
Represents a rotation in 3D space using normalized quaternions.
Structurally identical to a 4D vector [x, y, z, w], but semantically denotes an orientation. This representation avoids the gimbal lock issues associated with Euler angles.
Attributes:
| Name | Type | Description |
|---|---|---|
x |
float64
|
Vector X component. |
y |
float64
|
Vector Y component. |
z |
float64
|
Vector Z component. |
w |
float64
|
Vector W component. |
covariance |
Optional[list_(float64)]
|
Optional flattened 4x4 covariance matrix representing the uncertainty of the quaternion measurement. |
covariance_type |
Optional[int16]
|
Enum integer representing the parameterization of the covariance matrix. |
Querying with the .Q Proxy¶
This class fields are queryable when constructing a QueryOntologyCatalog
via the .Q proxy. Check the fields documentation for detailed description.
Example
from mosaicolabs import MosaicoClient, Quaternion, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(
QueryOntologyCatalog(Quaternion.Q.w.leq(0.707))
.with_expression(Quaternion.Q.timestamp_ns.between([1770282868, 1770290127]))
)
# 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(Quaternion.Q.w.leq(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}}")
covariance
class-attribute
instance-attribute
¶
covariance = MosaicoField(
default=None,
nullable=True,
description="The covariance matrix (flattened) of the data.",
)
Optional list of 64-bit floats representing the flattened matrix.
Querying with the .Q Proxy¶
Non-Queryable
The field is not queryable with the .Q Proxy.
covariance_type
class-attribute
instance-attribute
¶
covariance_type = MosaicoField(
default=None,
nullable=True,
description="Enum integer representing the covariance parameterization.",
)
Optional 16-bit integer representing the covariance enum.
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 covariance_type field is fully queryable via the .Q Proxy. The <Model> placeholder
in the path represents any Mosaico class that exposes covariance information, either directly or through its internal fields.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.covariance_type |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
The <Model> placeholder adapts based on how the CovarianceMixin is integrated into your data structure:
- Direct Inheritance: Represents any class (e.g.,
Vector3d,Quaternion) that inherits directly fromCovarianceMixin. - Composition (Nested Fields): When a complex model (like
IMU) contains fields that are themselves covariance-aware, the proxy allows you to "drill down" to that specific attribute. For example, sinceIMU.accelerationis aVector3d, you access its covariance type viaIMU.Q.acceleration.covariance_type.
Filtering by Calibration Type
This example demonstrates searching for data segments where the acceleration was derived from a specific calibrated procedure.
from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog
# Assume FROM_CALIBRATED_PROCEDURE is a user-defined integer constant
with MosaicoClient.connect("localhost", 6726) as client:
# Target the covariance_type nested within the acceleration field
qbuilder = QueryOntologyCatalog(
IMU.Q.acceleration.covariance_type.eq(FROM_CALIBRATED_PROCEDURE)
)
results = client.query(qbuilder)
if results:
for item in results:
print(f"Sequence: {item.sequence.name}")
print(f"Matching Topics: {[topic.name for topic in item.topics]}")
x
class-attribute
instance-attribute
¶
x = MosaicoField(
nullable=True, description="Vector x component."
)
The Vector X component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector4dStruct (i.e. Vector4d, Quaternion)
or any custom user-defined class that inherits from Serializable and derives from _Vector4dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector4d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector4d.Q.x.leq(123.4)))
# 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(Vector4d.Q.x.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Quaternion.Q.x.leq(0.4)))
# 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]}")
y
class-attribute
instance-attribute
¶
y = MosaicoField(
nullable=True, description="Vector y component."
)
The Vector Y component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector4dStruct (i.e. Vector4d, Quaternion)
or any custom user-defined class that inherits from Serializable and derives from _Vector4dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector4d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector4d.Q.y.leq(123.4)))
# 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(Vector4d.Q.y.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Quaternion.Q.y.leq(0.4)))
# 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]}")
z
class-attribute
instance-attribute
¶
z = MosaicoField(
nullable=True, description="Vector z component."
)
The Vector Z component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector4dStruct (i.e. Vector4d, Quaternion)
or any custom user-defined class that inherits from Serializable and derives from _Vector4dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector4d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector4d.Q.z.leq(123.4)))
# 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(Vector4d.Q.z.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Quaternion.Q.z.leq(0.4)))
# 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]}")
w
class-attribute
instance-attribute
¶
w = MosaicoField(
nullable=True, description="Vector w component."
)
The Vector W component
Querying with the .Q Proxy¶
This field is queryable when constructing a QueryOntologyCatalog
via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.w |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Universal Compatibility
The <Model> placeholder represents any Mosaico class derived by _Vector4dStruct (i.e. Vector4d, Quaternion)
or any custom user-defined class that inherits from Serializable and derives from _Vector4dStruct or its child classes.
Example
from mosaicolabs import MosaicoClient, Vector4d, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Vector4d.Q.w.leq(123.4)))
# 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(Vector4d.Q.w.leq(123.4), 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}}")
# Filter for a specific component value.
qresponse = client.query(QueryOntologyCatalog(Quaternion.Q.w.leq(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]}")
is_registered
classmethod
¶
Checks if a class is registered.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if registered. |
ontology_tag
classmethod
¶
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., |
Raises:
| Type | Description |
|---|---|
Exception
|
If the class was not properly initialized via |
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_list
classmethod
¶
Creates a struct instance from a raw list.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
list[float]
|
A list containing exactly 4 float values: [x, y, z, w]. |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the input list does not have a length of 4. |
Transform ¶
Bases: Serializable, CovarianceMixin
Represents a rigid-body transformation between two coordinate frames.
A transform consists of a translation followed by a rotation. It is typically used to describe the kinematic relationship between components (e.g., "Camera to Robot Base").
Attributes:
| Name | Type | Description |
|---|---|---|
translation |
Vector3d
|
A |
rotation |
Quaternion
|
A |
target_frame_id |
Optional[string]
|
The identifier of the destination coordinate frame. |
covariance |
Optional[list_(float64)]
|
Optional flattened 7x7 composed covariance matrix representing the uncertainty of the Translation+Rotation. |
covariance_type |
Optional[int16]
|
Enum integer representing the parameterization of the covariance matrix. |
Querying with the .Q Proxy¶
This class fields are queryable when constructing a QueryOntologyCatalog
via the .Q proxy. Check the fields documentation for detailed description.
Example
from mosaicolabs import MosaicoClient, Transform, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific component value.
qresponse = client.query(
QueryOntologyCatalog(Transform.Q.translation.x.gt(5.0))
.with_expression(Transform.Q.rotation.w.lt(0.707))
.with_expression(Transform.Q.timestamp_ns.between([1770282868, 1770290127]))
)
# 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(Transform.Q.translation.x.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}}")
translation
class-attribute
instance-attribute
¶
translation = MosaicoField(
description="3D translation vector."
)
The 3D translation vector component.
Querying with the .Q Proxy¶
Translation components are queryable through the translation field prefix.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
Transform.Q.translation.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Transform.Q.translation.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Transform.Q.translation.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Example
from mosaicolabs import MosaicoClient, Transform, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Find transforms where the linear X-translation exceeds 5 meters
qresponse = client.query(
QueryOntologyCatalog(Transform.Q.translation.x.gt(5.0))
.with_expression(Transform.Q.translation.z.lt(150.3))
)
# 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(Transform.Q.translation.x.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}}")
rotation
class-attribute
instance-attribute
¶
rotation = MosaicoField(
description="Quaternion representing rotation."
)
The rotation quaternion component (x, y, z, w).
Querying with the .Q Proxy¶
Rotation components are queryable through the rotation field prefix.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
Transform.Q.rotation.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Transform.Q.rotation.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Transform.Q.rotation.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Transform.Q.rotation.w |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Example
from mosaicolabs import MosaicoClient, Transform, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for specific orientation states
qresponse = client.query(
QueryOntologyCatalog(Transform.Q.rotation.w.geq(0.707))
.with_expression(Transform.Q.rotation.z.lt(0.4))
)
# 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(Transform.Q.rotation.x.gt(5.0), include_timestamp_range=True)
.with_expression(Transform.Q.rotation.z.lt(0.4))
)
# 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}}")
target_frame_id
class-attribute
instance-attribute
¶
target_frame_id = MosaicoField(
default=None, description="Target frame identifier."
)
Target coordinate frame identifier.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
Transform.Q.target_frame_id |
String |
.eq(), .match(), .in_(), .lt(), .gt(), .leq(), .geq() |
Example
from mosaicolabs import MosaicoClient, Transform, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for specific target frame id
qresponse = client.query(
QueryOntologyCatalog(Transform.Q.target_frame_id.eq("camera_link"))
)
# 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(Transform.Q.target_frame_id.eq("camera_link"), 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}}")
covariance
class-attribute
instance-attribute
¶
covariance = MosaicoField(
default=None,
nullable=True,
description="The covariance matrix (flattened) of the data.",
)
Optional list of 64-bit floats representing the flattened matrix.
Querying with the .Q Proxy¶
Non-Queryable
The field is not queryable with the .Q Proxy.
covariance_type
class-attribute
instance-attribute
¶
covariance_type = MosaicoField(
default=None,
nullable=True,
description="Enum integer representing the covariance parameterization.",
)
Optional 16-bit integer representing the covariance enum.
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 covariance_type field is fully queryable via the .Q Proxy. The <Model> placeholder
in the path represents any Mosaico class that exposes covariance information, either directly or through its internal fields.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.covariance_type |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
The <Model> placeholder adapts based on how the CovarianceMixin is integrated into your data structure:
- Direct Inheritance: Represents any class (e.g.,
Vector3d,Quaternion) that inherits directly fromCovarianceMixin. - Composition (Nested Fields): When a complex model (like
IMU) contains fields that are themselves covariance-aware, the proxy allows you to "drill down" to that specific attribute. For example, sinceIMU.accelerationis aVector3d, you access its covariance type viaIMU.Q.acceleration.covariance_type.
Filtering by Calibration Type
This example demonstrates searching for data segments where the acceleration was derived from a specific calibrated procedure.
from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog
# Assume FROM_CALIBRATED_PROCEDURE is a user-defined integer constant
with MosaicoClient.connect("localhost", 6726) as client:
# Target the covariance_type nested within the acceleration field
qbuilder = QueryOntologyCatalog(
IMU.Q.acceleration.covariance_type.eq(FROM_CALIBRATED_PROCEDURE)
)
results = client.query(qbuilder)
if results:
for item in results:
print(f"Sequence: {item.sequence.name}")
print(f"Matching Topics: {[topic.name for topic in item.topics]}")
is_registered
classmethod
¶
Checks if a class is registered.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if registered. |
ontology_tag
classmethod
¶
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., |
Raises:
| Type | Description |
|---|---|
Exception
|
If the class was not properly initialized via |
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]}")
Pose ¶
Bases: Serializable, CovarianceMixin
Represents the position and orientation of an object in a global or local frame.
While similar to a Transform, a
Pose semantically denotes the state of an object (its current location
and heading) rather than the mathematical shift between two frames.
Attributes:
| Name | Type | Description |
|---|---|---|
position |
Point3d
|
A |
orientation |
Quaternion
|
A |
covariance |
Optional[list_(float64)]
|
Optional flattened 7x7 composed covariance matrix representing the uncertainty of the Translation+Rotation. |
covariance_type |
Optional[int16]
|
Enum integer representing the parameterization of the covariance matrix. |
Querying with the .Q Proxy¶
This class fields are queryable when constructing a QueryOntologyCatalog
via the .Q proxy. Check the fields documentation for detailed description.
Example
from mosaicolabs import MosaicoClient, Pose, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter Poses with position X-component AND orientation W-component
qresponse = client.query(
QueryOntologyCatalog(Pose.Q.position.x.gt(5.0))
.with_expression(Pose.Q.orientation.w.lt(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(Pose.Q.position.x.gt(5.0), include_timestamp_range=True)
.with_expression(Pose.Q.orientation.w.lt(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:
[topic.timestamp_range.start, topic.timestamp_range.end]
for topic in item.topics}}")
position
class-attribute
instance-attribute
¶
position = MosaicoField(
description="3D translation vector."
)
The 3D position vector component.
Querying with the .Q Proxy¶
Position components are queryable through the position field prefix.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
Pose.Q.position.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Pose.Q.position.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Pose.Q.position.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Example
from mosaicolabs import MosaicoClient, Pose, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Find poses where the linear X-position exceeds 5 meters
qresponse = client.query(
QueryOntologyCatalog(Pose.Q.position.x.gt(123450.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(Pose.Q.position.x.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}}")
orientation
class-attribute
instance-attribute
¶
orientation = MosaicoField(
description="Quaternion representing rotation."
)
The orientation quaternion component (x, y, z, w).
Querying with the .Q Proxy¶
Rotation components are queryable through the orientation field prefix.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
Pose.Q.orientation.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Pose.Q.orientation.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Pose.Q.orientation.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Pose.Q.orientation.w |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Example
from mosaicolabs import MosaicoClient, Pose, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for specific orientation states
qresponse = client.query(
QueryOntologyCatalog(Pose.Q.orientation.w.geq(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(include_timestamp_range=True)
.with_expression(Pose.Q.orientation.w.geq(0.707))
.with_expression(Pose.Q.orientation.x.lt(0.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:
[topic.timestamp_range.start, topic.timestamp_range.end]
for topic in item.topics}}")
covariance
class-attribute
instance-attribute
¶
covariance = MosaicoField(
default=None,
nullable=True,
description="The covariance matrix (flattened) of the data.",
)
Optional list of 64-bit floats representing the flattened matrix.
Querying with the .Q Proxy¶
Non-Queryable
The field is not queryable with the .Q Proxy.
covariance_type
class-attribute
instance-attribute
¶
covariance_type = MosaicoField(
default=None,
nullable=True,
description="Enum integer representing the covariance parameterization.",
)
Optional 16-bit integer representing the covariance enum.
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 covariance_type field is fully queryable via the .Q Proxy. The <Model> placeholder
in the path represents any Mosaico class that exposes covariance information, either directly or through its internal fields.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.covariance_type |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
The <Model> placeholder adapts based on how the CovarianceMixin is integrated into your data structure:
- Direct Inheritance: Represents any class (e.g.,
Vector3d,Quaternion) that inherits directly fromCovarianceMixin. - Composition (Nested Fields): When a complex model (like
IMU) contains fields that are themselves covariance-aware, the proxy allows you to "drill down" to that specific attribute. For example, sinceIMU.accelerationis aVector3d, you access its covariance type viaIMU.Q.acceleration.covariance_type.
Filtering by Calibration Type
This example demonstrates searching for data segments where the acceleration was derived from a specific calibrated procedure.
from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog
# Assume FROM_CALIBRATED_PROCEDURE is a user-defined integer constant
with MosaicoClient.connect("localhost", 6726) as client:
# Target the covariance_type nested within the acceleration field
qbuilder = QueryOntologyCatalog(
IMU.Q.acceleration.covariance_type.eq(FROM_CALIBRATED_PROCEDURE)
)
results = client.query(qbuilder)
if results:
for item in results:
print(f"Sequence: {item.sequence.name}")
print(f"Matching Topics: {[topic.name for topic in item.topics]}")
is_registered
classmethod
¶
Checks if a class is registered.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if registered. |
ontology_tag
classmethod
¶
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., |
Raises:
| Type | Description |
|---|---|
Exception
|
If the class was not properly initialized via |
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]}")
Polygon ¶
Bases: Serializable
Polygon geometry defined by a list of points.
Each point represents a vertex of the polygon in 3D space.
Attributes:
| Name | Type | Description |
|---|---|---|
points |
list_(Point3d)
|
List of polygon vertices. |
Querying with the .Q Proxy¶
The points field is not queryable via the .Q proxy (lists are not supported yet).
points
class-attribute
instance-attribute
¶
points = MosaicoField(
description="List of polygon vertices as 3D points."
)
List of polygon vertices as 3D points.
Querying with the .Q Proxy¶
The points field is not queryable via the .Q proxy (lists are not supported yet).
is_registered
classmethod
¶
Checks if a class is registered.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if registered. |
ontology_tag
classmethod
¶
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., |
Raises:
| Type | Description |
|---|---|
Exception
|
If the class was not properly initialized via |
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.data.kinematics ¶
Kinematics Data Structures.
This module defines structures for analyzing motion: 1. Velocity (Twist): Linear and angular speed. 2. Acceleration: Linear and angular acceleration. 3. MotionState: A complete snapshot of an object's kinematics (Pose + Velocity + Acceleration).
These can be assigned to Message.data field to send data to the platform.
Velocity ¶
Bases: Serializable, CovarianceMixin
Represents 6-Degree-of-Freedom Velocity, commonly referred to as a Twist.
The Velocity class describes the instantaneous motion of an object, split into
linear and angular components.
Attributes:
| Name | Type | Description |
|---|---|---|
linear |
Optional[Vector3d]
|
Optional |
angular |
Optional[Vector3d]
|
Optional |
covariance |
Optional[list_(float64)]
|
Optional flattened 3x3 covariance matrix representing the uncertainty of the point measurement. |
covariance_type |
Optional[int16]
|
Enum integer representing the parameterization of the covariance matrix. |
Input Validation
A valid Velocity object must contain at least a linear or an angular
component; providing neither will raise a ValueError.
Querying with the .Q Proxy¶
This class fields are queryable when constructing a QueryOntologyCatalog
via the .Q proxy. Check the fields documentation for detailed description.
Example
from mosaicolabs import MosaicoClient, Velocity, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter Velocities with linear X-component AND angular Z-component
qresponse = client.query(
QueryOntologyCatalog(Velocity.Q.linear.x.gt(5.0))
.with_expression(Velocity.Q.angular.z.lt(10))
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
# Filter for a specific component value and extract the first and last occurrence times
qresponse = client.query(
QueryOntologyCatalog(Velocity.Q.linear.x.gt(5.0), include_timestamp_range=True)
.with_expression(Velocity.Q.angular.z.lt(10))
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {{topic.name:
[topic.timestamp_range.start, topic.timestamp_range.end]
for topic in item.topics}}")
linear
class-attribute
instance-attribute
¶
linear = MosaicoField(
default=None, description="3D linear velocity vector."
)
3D linear velocity vector
Querying with the .Q Proxy¶
Linear components are queryable through the linear field prefix.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
Velocity.Q.linear.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Velocity.Q.linear.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Velocity.Q.linear.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Example
from mosaicolabs import MosaicoClient, Velocity, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Find velocities where the linear X component exceeds 25 m/s
qresponse = client.query(
QueryOntologyCatalog(Velocity.Q.linear.x.gt(25.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(Velocity.Q.linear.x.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}}")
angular
class-attribute
instance-attribute
¶
angular = MosaicoField(
default=None, description="3D angular velocity vector."
)
3D angular velocity vector
Querying with the .Q Proxy¶
Angular components are queryable through the angular field prefix.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
Velocity.Q.angular.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Velocity.Q.angular.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Velocity.Q.angular.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Example
from mosaicolabs import MosaicoClient, Velocity, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Find velocities where the angular X component exceeds 2 rad//s
qresponse = client.query(
QueryOntologyCatalog(Velocity.Q.angular.x.gt(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(Velocity.Q.angular.x.gt(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}}")
covariance
class-attribute
instance-attribute
¶
covariance = MosaicoField(
default=None,
nullable=True,
description="The covariance matrix (flattened) of the data.",
)
Optional list of 64-bit floats representing the flattened matrix.
Querying with the .Q Proxy¶
Non-Queryable
The field is not queryable with the .Q Proxy.
covariance_type
class-attribute
instance-attribute
¶
covariance_type = MosaicoField(
default=None,
nullable=True,
description="Enum integer representing the covariance parameterization.",
)
Optional 16-bit integer representing the covariance enum.
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 covariance_type field is fully queryable via the .Q Proxy. The <Model> placeholder
in the path represents any Mosaico class that exposes covariance information, either directly or through its internal fields.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.covariance_type |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
The <Model> placeholder adapts based on how the CovarianceMixin is integrated into your data structure:
- Direct Inheritance: Represents any class (e.g.,
Vector3d,Quaternion) that inherits directly fromCovarianceMixin. - Composition (Nested Fields): When a complex model (like
IMU) contains fields that are themselves covariance-aware, the proxy allows you to "drill down" to that specific attribute. For example, sinceIMU.accelerationis aVector3d, you access its covariance type viaIMU.Q.acceleration.covariance_type.
Filtering by Calibration Type
This example demonstrates searching for data segments where the acceleration was derived from a specific calibrated procedure.
from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog
# Assume FROM_CALIBRATED_PROCEDURE is a user-defined integer constant
with MosaicoClient.connect("localhost", 6726) as client:
# Target the covariance_type nested within the acceleration field
qbuilder = QueryOntologyCatalog(
IMU.Q.acceleration.covariance_type.eq(FROM_CALIBRATED_PROCEDURE)
)
results = client.query(qbuilder)
if results:
for item in results:
print(f"Sequence: {item.sequence.name}")
print(f"Matching Topics: {[topic.name for topic in item.topics]}")
check_at_least_one_exists ¶
Ensures the velocity object is not empty.
Raises:
| Type | Description |
|---|---|
ValueError
|
If both |
is_registered
classmethod
¶
Checks if a class is registered.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if registered. |
ontology_tag
classmethod
¶
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., |
Raises:
| Type | Description |
|---|---|
Exception
|
If the class was not properly initialized via |
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]}")
Acceleration ¶
Bases: Serializable, CovarianceMixin
Represents 6-Degree-of-Freedom Acceleration.
This class provides a standardized way to transmit linear and angular acceleration data to the platform.
Attributes:
| Name | Type | Description |
|---|---|---|
linear |
Optional[Vector3d]
|
Optional 3D linear acceleration vector ($a_x, a_y, a_z$). |
angular |
Optional[Vector3d]
|
Optional 3D angular acceleration vector ($lpha_x, lpha_y, lpha_z$). |
covariance |
Optional[list_(float64)]
|
Optional flattened 3x3 covariance matrix representing the uncertainty of the point measurement. |
covariance_type |
Optional[int16]
|
Enum integer representing the parameterization of the covariance matrix. |
Input Validation
Similar to the Velocity class, an Acceleration instance requires
at least one non-null component (linear or angular).
Querying with the .Q Proxy¶
This class fields are queryable when constructing a QueryOntologyCatalog
via the .Q proxy. Check the fields documentation for detailed description.
Example
from mosaicolabs import MosaicoClient, Acceleration, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter Accelerations with linear X-component AND angular Z-component
qresponse = client.query(
QueryOntologyCatalog(Acceleration.Q.linear.x.gt(5.0))
.with_expression(Acceleration.Q.angular.z.lt(10))
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
# Filter for a specific component value and extract the first and last occurrence times
qresponse = client.query(
QueryOntologyCatalog(Acceleration.Q.linear.x.gt(5.0), include_timestamp_range=True)
.with_expression(Acceleration.Q.angular.z.lt(10))
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {{topic.name:
[topic.timestamp_range.start, topic.timestamp_range.end]
for topic in item.topics}}")
linear
class-attribute
instance-attribute
¶
linear = MosaicoField(
default=None,
description="3D linear acceleration vector.",
)
3D linear acceleration vector
Querying with the .Q Proxy¶
Linear components are queryable through the linear field prefix.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
Acceleration.Q.linear.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Acceleration.Q.linear.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Acceleration.Q.linear.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Example
from mosaicolabs import MosaicoClient, Acceleration, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Find accelerations where the linear X component exceeds 5 m/s^2
qresponse = client.query(
QueryOntologyCatalog(Acceleration.Q.linear.x.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(Acceleration.Q.linear.x.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}}")
angular
class-attribute
instance-attribute
¶
angular = MosaicoField(
default=None,
description="3D angular acceleration vector.",
)
3D angular acceleration vector
Querying with the .Q Proxy¶
Angular components are queryable through the angular field prefix.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
Acceleration.Q.angular.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Acceleration.Q.angular.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Acceleration.Q.angular.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Example
from mosaicolabs import MosaicoClient, Acceleration, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Find accelerations where the angular X component exceeds 1 rad/s^2
qresponse = client.query(
QueryOntologyCatalog(Acceleration.Q.angular.x.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(Acceleration.Q.angular.x.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}}")
covariance
class-attribute
instance-attribute
¶
covariance = MosaicoField(
default=None,
nullable=True,
description="The covariance matrix (flattened) of the data.",
)
Optional list of 64-bit floats representing the flattened matrix.
Querying with the .Q Proxy¶
Non-Queryable
The field is not queryable with the .Q Proxy.
covariance_type
class-attribute
instance-attribute
¶
covariance_type = MosaicoField(
default=None,
nullable=True,
description="Enum integer representing the covariance parameterization.",
)
Optional 16-bit integer representing the covariance enum.
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 covariance_type field is fully queryable via the .Q Proxy. The <Model> placeholder
in the path represents any Mosaico class that exposes covariance information, either directly or through its internal fields.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.covariance_type |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
The <Model> placeholder adapts based on how the CovarianceMixin is integrated into your data structure:
- Direct Inheritance: Represents any class (e.g.,
Vector3d,Quaternion) that inherits directly fromCovarianceMixin. - Composition (Nested Fields): When a complex model (like
IMU) contains fields that are themselves covariance-aware, the proxy allows you to "drill down" to that specific attribute. For example, sinceIMU.accelerationis aVector3d, you access its covariance type viaIMU.Q.acceleration.covariance_type.
Filtering by Calibration Type
This example demonstrates searching for data segments where the acceleration was derived from a specific calibrated procedure.
from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog
# Assume FROM_CALIBRATED_PROCEDURE is a user-defined integer constant
with MosaicoClient.connect("localhost", 6726) as client:
# Target the covariance_type nested within the acceleration field
qbuilder = QueryOntologyCatalog(
IMU.Q.acceleration.covariance_type.eq(FROM_CALIBRATED_PROCEDURE)
)
results = client.query(qbuilder)
if results:
for item in results:
print(f"Sequence: {item.sequence.name}")
print(f"Matching Topics: {[topic.name for topic in item.topics]}")
check_at_least_one_exists ¶
Ensures the acceleration object is not empty.
Raises:
| Type | Description |
|---|---|
ValueError
|
If both |
is_registered
classmethod
¶
Checks if a class is registered.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if registered. |
ontology_tag
classmethod
¶
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., |
Raises:
| Type | Description |
|---|---|
Exception
|
If the class was not properly initialized via |
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]}")
MotionState ¶
Bases: Serializable, CovarianceMixin
Aggregated Kinematic State.
MotionState groups Pose,
Velocity, and optional
Acceleration into a
single atomic update.
This is the preferred structure for:
- Trajectory Tracking: Recording the high-fidelity path of a robot or vehicle.
- State Estimation: Logging the output of Kalman filters or SLAM algorithms.
- Ground Truth: Storing reference data from simulation environments.
Attributes:
| Name | Type | Description |
|---|---|---|
pose |
Pose
|
The 6D pose representing current position and orientation. |
velocity |
Velocity
|
The 6D velocity (Twist). |
target_frame_id |
string
|
A string identifier for the target coordinate frame. |
acceleration |
Optional[Acceleration]
|
Optional 6D acceleration. |
covariance |
Optional[list_(float64)]
|
Optional flattened NxN composed covariance matrix representing the uncertainty of the Pose+Velocity+[Acceleration] measurement. |
covariance_type |
Optional[int16]
|
Enum integer representing the parameterization of the covariance matrix. |
Querying with the .Q Proxy¶
This class fields are queryable when constructing a QueryOntologyCatalog
via the .Q proxy. Check the fields documentation for detailed description.
Example
from mosaicolabs import MosaicoClient, MotionState, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter MotionStates with position X-component AND angular velocity Z-component
qresponse = client.query(
QueryOntologyCatalog(MotionState.Q.pose.position.x.gt(123456.9))
.with_expression(MotionState.Q.velocity.angular.z.lt(10))
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
# Filter for a specific component value and extract the first and last occurrence times
qresponse = client.query(
QueryOntologyCatalog(MotionState.Q.pose.position.x.gt(123456.9), include_timestamp_range=True)
.with_expression(MotionState.Q.velocity.angular.z.lt(10))
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {{topic.name:
[topic.timestamp_range.start, topic.timestamp_range.end]
for topic in item.topics}}")
pose
class-attribute
instance-attribute
¶
pose = MosaicoField(
description="6D pose with optional time and covariance info."
)
The 6D pose representing current position and orientation.
Querying with the .Q Proxy¶
Pose components are queryable through the pose field prefix.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
MotionState.Q.pose.position.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.pose.position.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.pose.position.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.pose.orientation.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.pose.orientation.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.pose.orientation.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.pose.orientation.w |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Example
from mosaicolabs import MosaicoClient, MotionState, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter snapshots where the object is beyond a specific X-coordinate
qresponse = client.query(
QueryOntologyCatalog(MotionState.Q.pose.position.x.gt(500.0))
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
# Filter for a specific component value and extract the first and last occurrence times
qresponse = client.query(
QueryOntologyCatalog(MotionState.Q.pose.position.x.gt(500.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 = MosaicoField(
description="6D velocity with optional time and covariance info."
)
The 6D velocity (Twist) describing instantaneous motion.
Querying with the .Q Proxy¶
Velocity components are queryable through the velocity field prefix.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
MotionState.Q.velocity.linear.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.velocity.linear.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.velocity.linear.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.velocity.angular.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.velocity.angular.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.velocity.angular.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Example
from mosaicolabs import MosaicoClient, MotionState, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Find states where linear velocity in X exceeds 2.5 m/s
qresponse = client.query(
QueryOntologyCatalog(MotionState.Q.velocity.linear.x.gt(2.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(MotionState.Q.velocity.linear.x.gt(5.0), include_timestamp_range=True)
.with_expression(MotionState.Q.velocity.angular.z.lt(10))
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {{topic.name:
[topic.timestamp_range.start, topic.timestamp_range.end]
for topic in item.topics}}")
target_frame_id
class-attribute
instance-attribute
¶
target_frame_id = MosaicoField(
description="Target frame identifier."
)
Identifier for the destination coordinate frame.
Querying with the .Q Proxy¶
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
MotionState.Q.target_frame_id |
String |
.eq(), .match(), .in_(), .lt(), .gt(), .leq(), .geq() |
Example
from mosaicolabs import MosaicoClient, MotionState, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Find states where target_frame_id is some link
qresponse = client.query(
QueryOntologyCatalog(MotionState.Q.target_frame_id.eq("moving_base"))
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
acceleration
class-attribute
instance-attribute
¶
acceleration = MosaicoField(
default=None,
description="6D acceleration with optional time and covariance info.",
)
Optional 6D acceleration components.
Querying with the .Q Proxy¶
Acceleration components are queryable through the acceleration field prefix if present.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
MotionState.Q.acceleration.linear.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.acceleration.linear.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.acceleration.linear.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.acceleration.angular.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.acceleration.angular.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
MotionState.Q.acceleration.angular.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Example
from mosaicolabs import MosaicoClient, MotionState, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Find states where centripetal acceleration exceeds 5 m/s^2
qresponse = client.query(
QueryOntologyCatalog(MotionState.Q.acceleration.linear.y.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(MotionState.Q.acceleration.linear.y.gt(5.0), include_timestamp_range=True)
.with_expression(MotionState.Q.acceleration.angular.z.lt(10))
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {{topic.name:
[topic.timestamp_range.start, topic.timestamp_range.end]
for topic in item.topics}}")
covariance
class-attribute
instance-attribute
¶
covariance = MosaicoField(
default=None,
nullable=True,
description="The covariance matrix (flattened) of the data.",
)
Optional list of 64-bit floats representing the flattened matrix.
Querying with the .Q Proxy¶
Non-Queryable
The field is not queryable with the .Q Proxy.
covariance_type
class-attribute
instance-attribute
¶
covariance_type = MosaicoField(
default=None,
nullable=True,
description="Enum integer representing the covariance parameterization.",
)
Optional 16-bit integer representing the covariance enum.
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 covariance_type field is fully queryable via the .Q Proxy. The <Model> placeholder
in the path represents any Mosaico class that exposes covariance information, either directly or through its internal fields.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.covariance_type |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
The <Model> placeholder adapts based on how the CovarianceMixin is integrated into your data structure:
- Direct Inheritance: Represents any class (e.g.,
Vector3d,Quaternion) that inherits directly fromCovarianceMixin. - Composition (Nested Fields): When a complex model (like
IMU) contains fields that are themselves covariance-aware, the proxy allows you to "drill down" to that specific attribute. For example, sinceIMU.accelerationis aVector3d, you access its covariance type viaIMU.Q.acceleration.covariance_type.
Filtering by Calibration Type
This example demonstrates searching for data segments where the acceleration was derived from a specific calibrated procedure.
from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog
# Assume FROM_CALIBRATED_PROCEDURE is a user-defined integer constant
with MosaicoClient.connect("localhost", 6726) as client:
# Target the covariance_type nested within the acceleration field
qbuilder = QueryOntologyCatalog(
IMU.Q.acceleration.covariance_type.eq(FROM_CALIBRATED_PROCEDURE)
)
results = client.query(qbuilder)
if results:
for item in results:
print(f"Sequence: {item.sequence.name}")
print(f"Matching Topics: {[topic.name for topic in item.topics]}")
is_registered
classmethod
¶
Checks if a class is registered.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if registered. |
ontology_tag
classmethod
¶
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., |
Raises:
| Type | Description |
|---|---|
Exception
|
If the class was not properly initialized via |
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.data.dynamics ¶
This module defines specialized ontology structures for representing physical dynamics, specifically linear forces and rotational moments (torques).
The primary structure, ForceTorque, implements a standard "Wrench" representation.
These models are designed to be assigned to the data field of a Message for transmission to the platform.
Key Features:
* Wrench Representation: Combines 3D linear force and 3D rotational torque into a single, synchronized state.
* Uncertainty Quantification: Inherits from CovarianceMixin to support $6 imes 6$ covariance matrices, allowing for the transmission of sensor noise characteristics or estimation confidence.
ForceTorque ¶
Bases: Serializable, CovarianceMixin
Represents a Wrench (Force and Torque) applied to a rigid body.
The ForceTorque class is used to describe the total mechanical action (wrench)
acting on a body at a specific reference point. By combining
linear force and rotational torque, it provides a complete description of
dynamics for simulation and telemetry.
Attributes:
| Name | Type | Description |
|---|---|---|
force |
Vector3d
|
A |
torque |
Vector3d
|
A |
covariance |
Optional[list_(float64)]
|
Optional flattened 6x6 composed covariance matrix representing the uncertainty of the force-torque measurement. |
covariance_type |
Optional[int16]
|
Enum integer representing the parameterization of the covariance matrix. |
Unit Standards
To ensure platform-wide consistency, all force components should be specified in Newtons and torque in Newton-meters.
Querying with the .Q Proxy¶
This class fields are queryable when constructing a QueryOntologyCatalog
via the .Q proxy. Check the fields documentation for detailed description.
Example
from mosaicolabs import MosaicoClient, ForceTorque, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Filter ForceTorques with force X-component AND torque Z-component
qresponse = client.query(
QueryOntologyCatalog(ForceTorque.Q.force.x.gt(5.0))
.with_expression(ForceTorque.Q.torque.z.lt(10))
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
# Filter for a specific data value and extract the first and last occurrence times
qresponse = client.query(
QueryOntologyCatalog(ForceTorque.Q.force.x.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}}")
force
class-attribute
instance-attribute
¶
force = MosaicoField(description='3D linear force.')
3D linear force vector
Querying with the .Q Proxy¶
Force components are queryable through the force field prefix.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
ForceTorque.Q.force.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
ForceTorque.Q.force.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
ForceTorque.Q.force.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Example
from mosaicolabs import MosaicoClient, ForceTorque, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Find where the linear X-force exceeds 50N
qresponse = client.query(QueryOntologyCatalog(ForceTorque.Q.force.x.gt(50.0)))
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
# Filter for a specific data value and extract the first and last occurrence times
qresponse = client.query(
QueryOntologyCatalog(ForceTorque.Q.force.x.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}}")
torque
class-attribute
instance-attribute
¶
torque = MosaicoField(description='3D torque vector.')
3D torque vector
Querying with the .Q Proxy¶
Torque components are queryable through the torque field prefix.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
ForceTorque.Q.torque.x |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
ForceTorque.Q.torque.y |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
ForceTorque.Q.torque.z |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Example
from mosaicolabs import MosaicoClient, ForceTorque, QueryOntologyCatalog
with MosaicoClient.connect("localhost", 6726) as client:
# Find where the linear Y-torque is small
qresponse = client.query(QueryOntologyCatalog(ForceTorque.Q.torque.y.lt(0.02)))
# 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 data value and extract the first and last occurrence times
qresponse = client.query(
QueryOntologyCatalog(ForceTorque.Q.torque.y.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}}")
covariance
class-attribute
instance-attribute
¶
covariance = MosaicoField(
default=None,
nullable=True,
description="The covariance matrix (flattened) of the data.",
)
Optional list of 64-bit floats representing the flattened matrix.
Querying with the .Q Proxy¶
Non-Queryable
The field is not queryable with the .Q Proxy.
covariance_type
class-attribute
instance-attribute
¶
covariance_type = MosaicoField(
default=None,
nullable=True,
description="Enum integer representing the covariance parameterization.",
)
Optional 16-bit integer representing the covariance enum.
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 covariance_type field is fully queryable via the .Q Proxy. The <Model> placeholder
in the path represents any Mosaico class that exposes covariance information, either directly or through its internal fields.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<Model>.Q.covariance_type |
Numeric |
.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
The <Model> placeholder adapts based on how the CovarianceMixin is integrated into your data structure:
- Direct Inheritance: Represents any class (e.g.,
Vector3d,Quaternion) that inherits directly fromCovarianceMixin. - Composition (Nested Fields): When a complex model (like
IMU) contains fields that are themselves covariance-aware, the proxy allows you to "drill down" to that specific attribute. For example, sinceIMU.accelerationis aVector3d, you access its covariance type viaIMU.Q.acceleration.covariance_type.
Filtering by Calibration Type
This example demonstrates searching for data segments where the acceleration was derived from a specific calibrated procedure.
from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog
# Assume FROM_CALIBRATED_PROCEDURE is a user-defined integer constant
with MosaicoClient.connect("localhost", 6726) as client:
# Target the covariance_type nested within the acceleration field
qbuilder = QueryOntologyCatalog(
IMU.Q.acceleration.covariance_type.eq(FROM_CALIBRATED_PROCEDURE)
)
results = client.query(qbuilder)
if results:
for item in results:
print(f"Sequence: {item.sequence.name}")
print(f"Matching Topics: {[topic.name for topic in item.topics]}")
is_registered
classmethod
¶
Checks if a class is registered.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if registered. |
ontology_tag
classmethod
¶
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., |
Raises:
| Type | Description |
|---|---|
Exception
|
If the class was not properly initialized via |
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]}")
Inertia ¶
Bases: Serializable
Inertia properties of a rigid body.
Includes mass, center of mass, and inertia tensor.
Attributes:
| Name | Type | Description |
|---|---|---|
mass |
float64
|
Mass of the object. |
center_of_mass |
Vector3d
|
Center of mass position. |
inertia |
list_(float64, list_size=6)
|
Inertia tensor (flattened 3x3 matrix). |
Querying with the .Q Proxy¶
Only scalar fields are queryable.
mass
class-attribute
instance-attribute
¶
mass = MosaicoField(description='Mass of the object.')
Mass of the object.
Querying with the .Q Proxy¶
The mass field is queryable.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
Inertia.Q.mass |
Numeric | .eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
center_of_mass
class-attribute
instance-attribute
¶
center_of_mass = MosaicoField(
description="Center of mass of the object."
)
Center of mass of the object.
Querying with the .Q Proxy¶
The center of mass is queryable via its components.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
Inertia.Q.center_of_mass.x |
Numeric | .eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Inertia.Q.center_of_mass.y |
Numeric | .eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
Inertia.Q.center_of_mass.z |
Numeric | .eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between() |
inertia
class-attribute
instance-attribute
¶
inertia = MosaicoField(
description="Inertia tensor components [ixx, ixy, ixz, iyy, iyz, izz]."
)
Inertia tensor represented by its 6 unique components [ixx, ixy, ixz, iyy, iyz, izz].
Querying with the .Q Proxy¶
This field is not queryable via the .Q proxy (lists are not supported yet).
is_registered
classmethod
¶
Checks if a class is registered.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if registered. |
ontology_tag
classmethod
¶
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., |
Raises:
| Type | Description |
|---|---|
Exception
|
If the class was not properly initialized via |
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]}")