Skip to content

Time Models

mosaicolabs.models.data.header

Header Definitions.

This module defines the standard Header class used to provide metadata to ontology data.

Header

Bases: Serializable

A heading, typically associated with a sensor measurement

It is composed of Optional fields depending on the type contained information in the sensor measurement.

Attributes:

Name Type Description
timestamp Optional[Time]

Time (seconds and nanoseconds) passed since the epoch (Unix time) or process start (clock time). It can be omitted if not available.
frame_id Optional[string]

Reference frame name used for the measurement. It can be omitted if not available.
sample_counter Optional[uint64]

Integer indicating the number of samples elapsed since process start. It can be omitted if not available.
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, Header, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter Header with time seconds-component AND time nanoseconds-component
    qresponse = client.query(
        QueryOntologyCatalog(Header.Q.timestamp.seconds.lt(20.0))
            .with_expression(Header.Q.timestamp.nanoseconds.gt(100000))
    )

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

timestamp class-attribute instance-attribute 

timestamp = MosaicoField(
    nullable=True,
    default=None,
    description="Timestamp representing when the data has been measured",
)

Time (seconds and nanoseconds) passed since the epoch (Unix time) or process start (clock time).

Querying with the .Q Proxy

Timestamp components are queryable through the timestamp field prefix.

Field Access Path Queryable Type Supported Operators
Header.Q.timestamp.seconds Numeric .eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Header.Q.timestamp.nanoseconds Numeric .eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Example 
from mosaicolabs import MosaicoClient, Header, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Find headers where the timestamp exceeds 5 seconds
    qresponse = client.query(
        QueryOntologyCatalog(Header.Q.timestamp.seconds.gt(5.0))
    )

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

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

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

frame_id class-attribute instance-attribute 

frame_id = MosaicoField(
    nullable=True,
    default=None,
    description="String representing the acquired data reference system name",
)

String representing the acquired data reference system name. It may be None if it is unknown or the measurement does not support one (like an audio stream).

Querying with the .Q Proxy

Frame id component is queryable through the frame_id field prefix.

Field Access Path Queryable Type Supported Operators
Header.Q.frame_id String .eq(), .match(), .in_(), .lt(), .gt(), .leq(), .geq(), .between(), .outside()
Example 
from mosaicolabs import MosaicoClient, Header, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Find headers where reference system is base_link
    qresponse = client.query(
        QueryOntologyCatalog(Header.Q.frame_id.eq("base_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]}")

        # 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))

sample_counter class-attribute instance-attribute 

sample_counter = MosaicoField(
    nullable=True,
    default=None,
    description="An optional counter used to track how many samples have been processed. It needs to be monotonically increasing",
)

Counter used to track how many samples have been processed by the sensor. It should always be monotonically increasing.

Querying with the .Q Proxy

Sample counters component is queryable through the sample_counter field prefix.

Field Access Path Queryable Type Supported Operators
Header.Q.sample_counter Numeric .eq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .outside()
Example 
from mosaicolabs import MosaicoClient, Header, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Find headers where the sample counters exceeds 300-th sample
    qresponse = client.query(
        QueryOntologyCatalog(Header.Q.sample_counter.gt(300.0))
    )

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

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

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

is_registered classmethod 

is_registered()

Checks if a class is registered.

Returns:

Name Type Description
bool bool

True if registered.

ontology_tag classmethod 

ontology_tag()

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

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

Returns:

Name Type Description
str str

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

Raises:

Type Description
Exception

If the class was not properly initialized via __pydantic_init_subclass__.
Practical Application: Topic Filtering

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

Example: 

from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic

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

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

mosaicolabs.models.data.time

Time Definitions.

This module defines the standard Time and Duration classes used to provide temporal context to ontology data. Both classes use a high-precision seconds/nanoseconds split based on the ROS convention.

_TemporalBase is a private base class that holds the common fields, validators, and conversion logic shared by Time and Duration.

Time

Bases: _TemporalBase, Serializable

A high-precision time representation.

The Time class splits a timestamp into a 32-bit integer for seconds and a 32-bit unsigned integer for nanoseconds.

Attributes:

Name Type Description
seconds int32

Seconds passed since the epoch (Unix time) or process start (clock time).
nanoseconds uint32

Nanoseconds component within the current second, ranging from 0 to 999,999,999.
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.

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

Example: 

from mosaicolabs import MosaicoClient, Time, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter Time with time seconds-component AND time nanoseconds-component
    qresponse = client.query(
        QueryOntologyCatalog(Time.Q.seconds.lt(20.0))
            .with_expression(Time.Q.nanoseconds.gt(100000))
    )

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

seconds class-attribute instance-attribute 

seconds = MosaicoField(description='Time in seconds.')

Seconds since the epoch (Unix time) or since process start (clock time).

nanoseconds class-attribute instance-attribute 

nanoseconds = MosaicoField(
    description="Time in nanoseconds."
)

Nanoseconds component within the current second, ranging from 0 to 999,999,999.

from_datetime classmethod 

from_datetime(dt)

Factory method to create a Time object from a Python datetime instance.

Parameters:

Name Type Description Default
dt datetime

A standard Python datetime object.

 required

Returns:

Name Type Description
Time Time

A Time instance reflecting the datetime's timestamp.

now classmethod 

now()

Factory method that returns the current system UTC time in high precision.

Returns:

Name Type Description
Time Time

A Time instance representing the current UTC time.

to_datetime 

to_datetime()

Converts the time to a Python UTC datetime object.

Microsecond Limitation

Python's datetime objects typically support microsecond precision; nanosecond data below that threshold will be truncated.

Returns:

Name Type Description
datetime datetime

A UTC datetime object representing the time.

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

validate_nanosec classmethod 

validate_nanosec(v)

Ensures nanoseconds are within the valid [0, 1e9) range.

from_float classmethod 

from_float(ftime)

Factory method to create an instance from a float (seconds since epoch).

This method carefully handles floating-point artifacts by using rounding for the fractional part to ensure stable nanosecond conversion.

Parameters:

Name Type Description Default
ftime float

Total seconds since epoch (e.g., from time.time()).

 required

Returns:

Name Type Description
Self Self

A normalized instance.

from_milliseconds classmethod 

from_milliseconds(total_milliseconds)

Factory method to create an instance from a total count of milliseconds.

Parameters:

Name Type Description Default
total_milliseconds int

Total time elapsed in milliseconds.

 required

Returns:

Name Type Description
Self Self

An instance with split sec/nanosec components.

from_nanoseconds classmethod 

from_nanoseconds(total_nanoseconds)

Factory method to create an instance from a total count of nanoseconds.

Parameters:

Name Type Description Default
total_nanoseconds int

Total time elapsed in nanoseconds.

 required

Returns:

Name Type Description
Self Self

An instance with split sec/nanosec components.

to_float 

to_float()

Converts the high-precision time to a float.

Precision Loss

Converting to a 64-bit float may result in the loss of nanosecond precision due to mantissa limitations.

Returns:

Name Type Description
float float

Total seconds as a float, potentially losing sub-microsecond precision.

to_nanoseconds 

to_nanoseconds()

Converts the time to a total integer of nanoseconds.

This conversion preserves full precision.

Returns:

Name Type Description
int int

Total time in nanoseconds.

to_milliseconds 

to_milliseconds()

Converts the time to a total integer of milliseconds.

This conversion preserves full precision.

Returns:

Name Type Description
int int

Total time in milliseconds.

Duration

Bases: _TemporalBase, Serializable

A high-precision duration representation.

The Duration class represents a span of time split into a 32-bit integer for seconds and a 32-bit unsigned integer for nanoseconds.

Attributes:

Name Type Description
seconds int32

seconds component of the duration.
nanoseconds uint32

Nanoseconds component within the current second, ranging from 0 to 999,999,999.
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.

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

Example: 

from mosaicolabs import MosaicoClient, Duration, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter Duration with seconds-component AND nanoseconds-component
    qresponse = client.query(
        QueryOntologyCatalog(Duration.Q.seconds.lt(20.0))
            .with_expression(Duration.Q.nanoseconds.gt(100000))
    )

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

seconds class-attribute instance-attribute 

seconds = MosaicoField(description='Time in seconds.')

Seconds since the epoch (Unix time) or since process start (clock time).

nanoseconds class-attribute instance-attribute 

nanoseconds = MosaicoField(
    description="Time in nanoseconds."
)

Nanoseconds component within the current second, ranging from 0 to 999,999,999.

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

validate_nanosec classmethod 

validate_nanosec(v)

Ensures nanoseconds are within the valid [0, 1e9) range.

from_float classmethod 

from_float(ftime)

Factory method to create an instance from a float (seconds since epoch).

This method carefully handles floating-point artifacts by using rounding for the fractional part to ensure stable nanosecond conversion.

Parameters:

Name Type Description Default
ftime float

Total seconds since epoch (e.g., from time.time()).

 required

Returns:

Name Type Description
Self Self

A normalized instance.

from_milliseconds classmethod 

from_milliseconds(total_milliseconds)

Factory method to create an instance from a total count of milliseconds.

Parameters:

Name Type Description Default
total_milliseconds int

Total time elapsed in milliseconds.

 required

Returns:

Name Type Description
Self Self

An instance with split sec/nanosec components.

from_nanoseconds classmethod 

from_nanoseconds(total_nanoseconds)

Factory method to create an instance from a total count of nanoseconds.

Parameters:

Name Type Description Default
total_nanoseconds int

Total time elapsed in nanoseconds.

 required

Returns:

Name Type Description
Self Self

An instance with split sec/nanosec components.

to_float 

to_float()

Converts the high-precision time to a float.

Precision Loss

Converting to a 64-bit float may result in the loss of nanosecond precision due to mantissa limitations.

Returns:

Name Type Description
float float

Total seconds as a float, potentially losing sub-microsecond precision.

to_nanoseconds 

to_nanoseconds()

Converts the time to a total integer of nanoseconds.

This conversion preserves full precision.

Returns:

Name Type Description
int int

Total time in nanoseconds.

to_milliseconds 

to_milliseconds()

Converts the time to a total integer of milliseconds.

This conversion preserves full precision.

Returns:

Name Type Description
int int

Total time in milliseconds.