Data Types and Mixins

mosaicolabs.models.data.base_types

This module provides specialized wrapper classes for standard Python primitive types, including Integers, Floating-point numbers, Booleans, and Strings.

In the Mosaico ecosystem, raw primitives cannot be transmitted directly because the platform requires structured metadata and explicit serialization schemas. These wrappers elevate basic data types to "first-class citizens" of the messaging system by inheriting from Serializable.

Key Features: * Explicit Serialization: Each class defines a precise pyarrow.StructType schema, ensuring consistent bit-width (e.g., 8-bit vs 64-bit) across the entire data platform. * Registry Integration: Wrapped types are automatically registered in the Mosaico ontology, allowing them to be used in platform-side Queries.

Integer8

Bases: Serializable

A wrapper for a signed 8-bit integer.

Attributes:

Name	Type	Description
data	int8	The underlying 8-bit integer value.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, Integer8, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Integer8.Q.data.gt(123)))

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

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

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

data class-attribute instance-attribute

data = MosaicoField(description='8-bit Integer data.')

The underlying integer value.

Querying with the .Q Proxy

This field is queryable when constructing a QueryOntologyCatalog via the .Q proxy.

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

Example

from mosaicolabs import MosaicoClient, Integer8, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Integer8.Q.data.gt(-10)))

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

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

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

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

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

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

Returns:

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

Raises:

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

Practical Application: Topic Filtering

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

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

Integer16

Bases: Serializable

A wrapper for a signed 16-bit integer.

Attributes:

Name	Type	Description
data	int16	The underlying 16-bit integer value.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, Integer16, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Integer16.Q.data.gt(123)))

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

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

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

data class-attribute instance-attribute

data = MosaicoField(description='16-bit Integer data.')

The underlying integer value.

Querying with the .Q Proxy

This field is queryable when constructing a QueryOntologyCatalog via the .Q proxy.

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

Example

from mosaicolabs import MosaicoClient, Integer16, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Integer16.Q.data.gt(-10)))

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

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

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

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

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

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

Returns:

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

Raises:

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

Practical Application: Topic Filtering

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

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

Integer32

Bases: Serializable

A wrapper for a signed 32-bit integer.

Attributes:

Name	Type	Description
data	int32	The underlying 32-bit integer value.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, Integer32, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Integer32.Q.data.gt(123)))

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

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

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

data class-attribute instance-attribute

data = MosaicoField(description='32-bit Integer data.')

The underlying integer value.

Querying with the .Q Proxy

This field is queryable when constructing a QueryOntologyCatalog via the .Q proxy.

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

Example

from mosaicolabs import MosaicoClient, Integer32, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Integer32.Q.data.gt(-10)))

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

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

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

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

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

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

Returns:

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

Raises:

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

Practical Application: Topic Filtering

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

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

Integer64

Bases: Serializable

A wrapper for a signed 64-bit integer.

Attributes:

Name	Type	Description
data	int64	The underlying 64-bit integer value.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, Integer64, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Integer64.Q.data.gt(123)))

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

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

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

data class-attribute instance-attribute

data = MosaicoField(description='64-bit Integer data.')

The underlying integer value.

Querying with the .Q Proxy

This field is queryable when constructing a QueryOntologyCatalog via the .Q proxy.

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

Example

from mosaicolabs import MosaicoClient, Integer64, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Integer64.Q.data.gt(123)))

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

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

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

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

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

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

Returns:

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

Raises:

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

Practical Application: Topic Filtering

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

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

Unsigned8

Bases: Serializable

A wrapper for an unsigned 8-bit integer.

Attributes:

Name	Type	Description
data	uint8	The underlying unsigned 8-bit integer value.

Raises:

Type	Description
ValueError	If data is initialized with a negative value.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, Unsigned8, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Unsigned8.Q.data.gt(123)))

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

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

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

data class-attribute instance-attribute

data = MosaicoField(description='8-bit Unsigned data.')

The underlying unsigned integer value.

Querying with the .Q Proxy

This field is queryable when constructing a QueryOntologyCatalog via the .Q proxy.

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

Example

from mosaicolabs import MosaicoClient, Unsigned8, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Unsigned8.Q.data.leq(253)))

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

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

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

model_post_init

model_post_init(context)

Validates that the input data is non-negative.

Parameters:

Name	Type	Description	Default
context	Any	The Pydantic validation context passed by the base class.	required

Raises:

Type	Description
ValueError	If data < 0.

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

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

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

Returns:

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

Raises:

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

Practical Application: Topic Filtering

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

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

Unsigned16

Bases: Serializable

A wrapper for an unsigned 16-bit integer.

Attributes:

Name	Type	Description
data	uint16	The underlying unsigned 16-bit integer value.

Raises:

Type	Description
ValueError	If data is initialized with a negative value.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, Unsigned16, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Unsigned16.Q.data.gt(123)))

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

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

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

data class-attribute instance-attribute

data = MosaicoField(description='16-bit Unsigned data.')

The underlying unsigned integer value.

Querying with the .Q Proxy

This field is queryable when constructing a QueryOntologyCatalog via the .Q proxy.

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

Example

from mosaicolabs import MosaicoClient, Unsigned16, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Unsigned16.Q.data.eq(2)))

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

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

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

model_post_init

model_post_init(context)

Validates that the input data is non-negative.

Parameters:

Name	Type	Description	Default
context	Any	The Pydantic validation context passed by the base class.	required

Raises:

Type	Description
ValueError	If data < 0.

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

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

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

Returns:

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

Raises:

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

Practical Application: Topic Filtering

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

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

Unsigned32

Bases: Serializable

A wrapper for an unsigned 32-bit integer.

Attributes:

Name	Type	Description
data	uint32	The underlying unsigned 32-bit integer value.

Raises:

Type	Description
ValueError	If data is initialized with a negative value.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, Unsigned32, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Unsigned32.Q.data.gt(123)))

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

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

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

data class-attribute instance-attribute

data = MosaicoField(description='32-bit Unsigned data.')

The underlying unsigned integer value.

Querying with the .Q Proxy

This field is queryable when constructing a QueryOntologyCatalog via the .Q proxy.

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

Example

from mosaicolabs import MosaicoClient, Unsigned32, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Unsigned32.Q.data.gt(123)))

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

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

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

model_post_init

model_post_init(context)

Validates that the input data is non-negative.

Parameters:

Name	Type	Description	Default
context	Any	The Pydantic validation context passed by the base class.	required

Raises:

Type	Description
ValueError	If data < 0.

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

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

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

Returns:

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

Raises:

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

Practical Application: Topic Filtering

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

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

Unsigned64

Bases: Serializable

A wrapper for an unsigned 64-bit integer.

Attributes:

Name	Type	Description
data	uint64	The underlying unsigned 64-bit integer value.

Raises:

Type	Description
ValueError	If data is initialized with a negative value.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, Unsigned64, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Unsigned64.Q.data.gt(123)))

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

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

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

data class-attribute instance-attribute

data = MosaicoField(description='64-bit Unsigned data.')

The underlying unsigned integer value.

Querying with the .Q Proxy

This field is queryable when constructing a QueryOntologyCatalog via the .Q proxy.

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

Example

from mosaicolabs import MosaicoClient, Unsigned64, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Unsigned64.Q.data.gt(123)))

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

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

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

model_post_init

model_post_init(context)

Validates that the input data is non-negative.

Parameters:

Name	Type	Description	Default
context	Any	The Pydantic validation context passed by the base class.	required

Raises:

Type	Description
ValueError	If data < 0.

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

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

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

Returns:

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

Raises:

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

Practical Application: Topic Filtering

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

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

Floating16

Bases: Serializable

A wrapper for a 16-bit half-precision floating-point number.

Attributes:

Name	Type	Description
data	float16	The underlying half-precision float.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, Floating16, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Floating16.Q.data.gt(123.45)))

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

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

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

data class-attribute instance-attribute

data = MosaicoField(
    description="16-bit Floating-point data."
)

The underlying half-precision float.

Querying with the .Q Proxy

This field is queryable when constructing a QueryOntologyCatalog via the .Q proxy.

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

Example

from mosaicolabs import MosaicoClient, Floating16, QueryOntologyCatalog

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

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

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

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

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

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

Returns:

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

Raises:

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

Practical Application: Topic Filtering

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

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

Floating32

Bases: Serializable

A wrapper for a 32-bit single-precision floating-point number.

Attributes:

Name	Type	Description
data	float32	The underlying single-precision float.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, Floating32, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Floating32.Q.data.gt(123.45)))

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

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

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

data class-attribute instance-attribute

data = MosaicoField(
    description="32-bit Floating-point data."
)

The underlying single-precision float.

Querying with the .Q Proxy

This field is queryable when constructing a QueryOntologyCatalog via the .Q proxy.

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

Example

from mosaicolabs import MosaicoClient, Floating32, QueryOntologyCatalog

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

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

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

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

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

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

Returns:

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

Raises:

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

Practical Application: Topic Filtering

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

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

Floating64

Bases: Serializable

A wrapper for a 64-bit double-precision floating-point number.

Attributes:

Name	Type	Description
data	float64	The underlying double-precision float.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, Floating64, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Floating64.Q.data.gt(123.45)))

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

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

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

data class-attribute instance-attribute

data = MosaicoField(
    description="64-bit Floating-point data."
)

The underlying double-precision float.

Querying with the .Q Proxy

This field is queryable when constructing a QueryOntologyCatalog via the .Q proxy.

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

Example

from mosaicolabs import MosaicoClient, Floating64, QueryOntologyCatalog

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

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

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

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

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

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

Returns:

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

Raises:

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

Practical Application: Topic Filtering

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

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

Boolean

Bases: Serializable

A wrapper for a standard boolean value.

Attributes:

Name	Type	Description
data	bool	The underlying boolean value.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, Boolean, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Boolean.Q.data.eq(True)))

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

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

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

data class-attribute instance-attribute

data = MosaicoField(description='Boolean data.')

The underlying boolean value.

Querying with the .Q Proxy

This field is queryable when constructing a QueryOntologyCatalog via the .Q proxy.

Field Access Path	Queryable Type	Supported Operators
Boolean.Q.data	Bool	.eq()

Example

from mosaicolabs import MosaicoClient, Boolean, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(Boolean.Q.data.eq(True)))

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

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

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

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

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

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

Returns:

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

Raises:

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

Practical Application: Topic Filtering

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

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

String

Bases: Serializable

A wrapper for a standard UTF-8 encoded string.

Attributes:

Name	Type	Description
data	string	The underlying string data.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, String, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value
    qresponse = client.query(QueryOntologyCatalog(String.Q.data.eq("hello")))

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

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

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

data class-attribute instance-attribute

data = MosaicoField(description='String data.')

The underlying string data.

Querying with the .Q Proxy

This field is queryable when constructing a QueryOntologyCatalog via the .Q proxy.

Field Access Path	Queryable Type	Supported Operators
String.Q.data	String	.eq(), .match(), .in_(), .lt(), .gt(), .leq(), .geq(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, String, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for strings containing a specific log pattern
    qresponse = client.query(QueryOntologyCatalog(String.Q.data.match("*[ERR]*")))

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

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

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

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

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

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

Returns:

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

Raises:

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

Practical Application: Topic Filtering

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

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

mosaicolabs.models.data.ROI

Bases: Serializable

Represents a rectangular Region of Interest (ROI) within a 2D coordinate system.

This class is primarily used in imaging and computer vision pipelines to define sub-windows for processing or rectification.

Attributes:

Name	Type	Description
offset	Vector2d	A Vector2d representing the top-left (leftmost, topmost) pixel coordinates of the ROI.
height	uint32	The vertical extent of the ROI in pixels.
width	uint32	The horizontal extent of the ROI in pixels.
do_rectify	Optional[bool]	Optional flag; True if a sub-window is captured and requires rectification.

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, ROI, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter ROIs with offset X-component AND offset Y-component
    qresponse = client.query(
        QueryOntologyCatalog(ROI.Q.offset.x.gt(5.0))
            .with_expression(ROI.Q.offset.y.lt(10))
    )

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

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

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

offset class-attribute instance-attribute

offset = MosaicoField(
    description="(Leftmost, Topmost) pixels of the ROI."
)

The top-left pixel coordinates of the ROI.

Querying with the .Q Proxy

Offset components are queryable through the offset field prefix.

Field Access Path	Queryable Type	Supported Operators
ROI.Q.offset.x	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
ROI.Q.offset.y	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Example

from mosaicolabs import MosaicoClient, ROI, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for ROIs starting between the 10th and 350th pixel vertically
    qresponse = client.query(
        QueryOntologyCatalog(ROI.Q.offset.x.gt(100))
            .with_expression(ROI.Q.offset.y.between(10, 350))
    )

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

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

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

height class-attribute instance-attribute

height = MosaicoField(
    description="Height pixel of the ROI."
)

Height of the ROI in pixels.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, ROI, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for ROIs with height beyond 100 pixels
    qresponse = client.query(
        QueryOntologyCatalog(ROI.Q.height.gt(100))
    )

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

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

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

width class-attribute instance-attribute

width = MosaicoField(description='Width pixel of the ROI.')

Width of the ROI in pixels.

Querying with the .Q Proxy

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

Example

from mosaicolabs import MosaicoClient, ROI, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for ROIs with width below (or equal to) 250 pixels
    qresponse = client.query(
        QueryOntologyCatalog(ROI.Q.width.leq(250))
    )

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

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

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

do_rectify class-attribute instance-attribute

do_rectify = MosaicoField(
    default=None,
    description="False if the full image is captured (ROI not used) and True if a subwindow is captured (ROI used) (optional). False if Null",
)

Flag indicating if the ROI requires rectification.

Querying with the .Q Proxy

Field Access Path	Queryable Type	Supported Operators
ROI.Q.do_rectify	Boolean	.eq()

Example

from mosaicolabs import MosaicoClient, ROI, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for explicitly non-rectified ROIs (not None)
    qresponse = client.query(
        QueryOntologyCatalog(ROI.Q.do_rectify.eq(False))
    )

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

is_registered classmethod

is_registered()

Checks if a class is registered.

Returns:

Name	Type	Description
bool	bool	True if registered.

ontology_tag classmethod

ontology_tag()

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

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

Returns:

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

Raises:

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

Practical Application: Topic Filtering

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

Example:

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

mosaicolabs.models.data.HeaderMixin

Bases: BaseModel

A mixin that adds header fields.

Recommended for sensors that provide a timestamp, a reference system or the number of produced samples.

Dynamic Schema Injection

This mixin uses the __init_subclass__ hook to perform a Schema Append operation:

1. It inspects the child class's existing __msco_pyarrow_struct__.
2. It appends a header field.
3. It reconstructs the final pa.struct for the class.

Collision Safety

The mixin performs a collision check during class definition. If the child class already defines a header field in its PyArrow struct, a ValueError will be raised to prevent schema corruption.

Querying with the .Q Proxy

When constructing a QueryOntologyCatalog, the class fields are queryable across any model inheriting from this mixin, according to the following table:

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

Universal Compatibility

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

Example

from mosaicolabs import MosaicoClient, ForceTorque, QueryOntologyCatalog

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

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

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

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

header class-attribute instance-attribute

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

Measure header containing measurement timestamp and reference frame.

Querying with the .Q Proxy

Header components are queryable through the header field prefix.

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

Example

from mosaicolabs import MosaicoClient, ForceTorque, QueryOntologyCatalog

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

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

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

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

mosaicolabs.models.data.CovarianceMixin

Bases: BaseModel

A mixin that adds uncertainty fields (covariance and covariance_type) to data models.

This is particularly useful for complex sensors like IMUs, Odometry, or GNSS receivers that provide multidimensional uncertainty matrices along with their primary measurements.

Dynamic Schema Injection

This mixin uses the __init_subclass__ hook to perform a Schema Append operation:

1. It inspects the child class's existing __msco_pyarrow_struct__.
2. It appends a covariance and covariance_type fields.
3. It reconstructs the final pa.struct for the class.

Collision Safety

The mixin performs a collision check during class definition. If the child class already defines a covariance or covariance_type field in its PyArrow struct, a ValueError will be raised to prevent schema corruption.

Attributes:

Name	Type	Description
covariance	Optional[list_(float64)]	Optional list of 64-bit floats representing the flattened matrix.
covariance_type	Optional[int16]	Optional 16-bit integer representing the covariance enum.

Querying with the .Q Proxy

When constructing a QueryOntologyCatalog, the class fields are queryable across any model inheriting from this mixin, according to the following table:

Field Access Path	Queryable Type	Supported Operators
<Model>.Q.covariance_type	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.covariance.all()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.covariance.any()	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
<Model>.Q.covariance.[i]	Numeric	.eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()

Universal Compatibility

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 from CovarianceMixin.
• 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, since IMU.acceleration is a Vector3d, you access its covariance type via IMU.Q.acceleration.covariance_type.

Example

from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter IMU data by a specific acquisition second
    # `FROM_CALIBRATED_PROCEDURE` is some enum value defined by the user
    qresponse = client.query(
        QueryOntologyCatalog(IMU.Q.acceleration.covariance_type.eq(FROM_CALIBRATED_PROCEDURE))
    )

    # Inspect the response
    if qresponse is not None:
        # Results 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]}")

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

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

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

Example

from mosaicolabs import MosaicoClient, Vector3d, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for measurements with at least one high-uncertainty covariance term
    qresponse = client.query(
        QueryOntologyCatalog(Vector3d.Q.covariance.any().gt(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 for topic in item.topics]}")

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

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

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(), .outside()

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 from CovarianceMixin.
• 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, since IMU.acceleration is a Vector3d, you access its covariance type via IMU.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]}")

mosaicolabs.models.data.VarianceMixin

Bases: BaseModel

A mixin that adds 1-dimensional uncertainty fields (variance and variance_type).

Recommended for sensors with scalar uncertain outputs, such as ultrasonic rangefinders, temperature sensors, or individual encoders.

Dynamic Schema Injection

This mixin uses the __init_subclass__ hook to perform a Schema Append operation:

1. It inspects the child class's existing __msco_pyarrow_struct__.
2. It appends a variance and variance_type field.
3. It reconstructs the final pa.struct for the class.

Collision Safety

The mixin performs a collision check during class definition. If the child class already defines a variance or variance_type field in its PyArrow struct, a ValueError will be raised to prevent schema corruption.

Querying with the .Q Proxy

When constructing a QueryOntologyCatalog, the class fields are queryable across any model inheriting from this mixin, according to the following table:

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

Universal Compatibility

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

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

Example

from mosaicolabs import MosaicoClient, Pressure, QueryOntologyCatalog

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

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

    # Filter Pressure data by a specific variance type
    # `FROM_CALIBRATED_PROCEDURE` is some enum value defined by the user
    qresponse = client.query(
        QueryOntologyCatalog(Pressure.Q.variance_type.eq(FROM_CALIBRATED_PROCEDURE))
    )

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

variance class-attribute instance-attribute

variance = MosaicoField(
    default=None,
    nullable=True,
    description="The variacne of the data.",
)

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

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

Querying with the .Q Proxy

The variance field is queryable with the .Q Proxy.

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

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

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

Example

from mosaicolabs import MosaicoClient, Pressure, QueryOntologyCatalog

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

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

variance_type class-attribute instance-attribute

variance_type = MosaicoField(
    default=None,
    nullable=True,
    description="Enum integer representing the variance parameterization.",
)

Optional 16-bit integer representing the variance parameterization.

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

Querying with the .Q Proxy

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

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

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

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

Filtering by Precision

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

from mosaicolabs import MosaicoClient, Pressure, QueryOntologyCatalog

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

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