Platform Models
mosaicolabs.models.platform.Topic ¶
Bases: BaseModel, _QueryProxyMixin
Represents a read-only view of a server-side Topic platform resource.
The Topic class provides access to topic-specific system metadata, such as the ontology tag (e.g., 'imu', 'camera') and the serialization format.
It serves as a metadata-rich view of an individual data stream within the platform catalog.
Data Retrieval
This class provides a server-side metadata-only view of the topic.
To retrieve the actual time-series messages contained within the topic, you must
use the TopicHandler.get_data_streamer()
method from a TopicHandler
instance.
Querying with the .Q Proxy¶
Warning: Deprecated
Querying the topic user-custom metadata via the user_metadata field of this class is deprecated.
Use the QueryTopic.with_user_metadata() builder instead.
Example
from mosaicolabs import MosaicoClient, Topic, QueryTopic
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific data value (using constructor)
qresponse = client.query(
QueryTopic()
.with_user_metadata("update_rate_hz", gt=100)
.with_user_metadata("interface.type", eq="canbus")
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
user_metadata
instance-attribute
¶
Custom user-defined key-value pairs associated with the entity.
Querying with the .Q Proxy¶
Warning: Deprecated
Querying the topic user-custom metadata via the user_metadata field of this class is deprecated.
Use the QueryTopic.with_user_metadata() builder instead.
name
property
¶
The unique identifier or resource name of the entity.
Querying with Query Builders¶
The name property is queryable when constructing a QueryTopic
via the convenience methods:
Example
from mosaicolabs import MosaicoClient, QueryTopic
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific data value (using constructor)
qresponse = client.query(
QueryTopic().with_name("/front/imu"),
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
created_timestamp
property
¶
The UTC timestamp indicating when the entity was created on the server.
Querying with Query Builders¶
The created_timestamp property is queryable when constructing a QueryTopic
via the convenience method:
Example
from mosaicolabs import MosaicoClient, Topic, IMU, QueryTopic, Time
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific topic creation time
qresponse = client.query(
QueryTopic().with_created_timestamp(time_start=Time.from_float(1765432100)),
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
ontology_tag
property
¶
The ontology type identifier (e.g., 'imu', 'gnss').
This corresponds to the __ontology_tag__ defined in the
Serializable class registry.
Querying with Query Builders¶
The ontology_tag property is queryable when constructing a QueryTopic
via the convenience method QueryTopic.with_ontology_tag().
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().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]}")
sequence_name
property
¶
The name of the parent sequence containing this topic.
Querying with Query Builders¶
The sequence_name property is not queryable directly. Use QuerySequence to query for sequences.
Example
from mosaicolabs import MosaicoClient, Topic, QuerySequence
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific data value (using constructor)
qresponse = client.query(
QuerySequence().with_name("test_winter_20260129_103000")
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
chunks_number
property
¶
The number of physical data chunks stored for this topic.
May be None if the server did not provide detailed storage statistics.
Querying with Query Builders¶
The chunks_number property is not queryable.
serialization_format
property
¶
The format used to serialize the topic data (e.g., 'arrow', 'image').
This corresponds to the SerializationFormat enum.
Querying with Query Builders¶
The serialization_format property is not queryable.
mosaicolabs.models.platform.Session
dataclass
¶
Represents a read-only view of a server-side writing Session platform resource.
The Session class is designed to hold system-level metadata. It serves as the primary
metadata container for a logical grouping of topics written in the writing session.
Data Retrieval
This class provides a server-side metadata-only view of the session.
To retrieve the actual time-series data contained within the topics of the session, you must
use the TopicHandler.get_data_streamer()
method from a TopicHandler instance.
Querying with the .Q Proxy¶
The session fields are not queryable via the .Q proxy.
mosaicolabs.models.platform.Sequence ¶
Bases: BaseModel, _QueryProxyMixin
Represents a read-only view of a server-side Sequence platform resource.
The Sequence class is designed to hold system-level metadata and enable fluid querying of
user-defined properties. It serves as the primary metadata container
for a logical grouping of related topics.
Data Retrieval
This class provides a server-side metadata-only view of the sequence.
To retrieve the actual time-series data contained within the sequence, you must
use the SequenceHandler.get_data_streamer()
method from a SequenceHandler
instance.
Querying with the .Q Proxy¶
Warning: Deprecated
Querying the sequence user-custom metadata via the user_metadata field of this class is deprecated.
Use the QuerySequence.with_user_metadata() builder instead.
Example
from mosaicolabs import MosaicoClient, Sequence, QuerySequence
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific data value (using constructor)
qresponse = client.query(
QuerySequence()
.with_user_metadata("project", eq="Apollo")
.with_user_metadata("vehicle.software_stack.planning", eq="plan-4.1.7")
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
user_metadata
instance-attribute
¶
Custom user-defined key-value pairs associated with the entity.
Querying with the .Q Proxy¶
Warning: Deprecated
Querying the sequence user-custom metadata via the user_metadata field of this class is deprecated.
Use the QuerySequence.with_user_metadata() builder instead.
topics
property
¶
Returns the list of names for all topics contained within this sequence.
Accessing Topic Data
This property returns string identifiers. To interact
with topic data or metadata, use the
MosaicoClient.topic_handler()
factory.
Querying with Query Builders¶
The topics property is not queryable directly. Use QueryTopic to query for topics.
Example
from mosaicolabs import MosaicoClient, QueryTopic
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific data value (using constructor)
qresponse = client.query(
QueryTopic().with_name("/sensors/camera/front/image_raw")
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
name
property
¶
The unique identifier or resource name of the entity.
Querying with Query Builders¶
The name property is queryable when constructing a QuerySequence
via the convenience methods:
Example
from mosaicolabs import MosaicoClient, QuerySequence
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific data value (using constructor)
qresponse = client.query(
QuerySequence().with_name_match("test_winter_2025_01_"),
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
created_timestamp
property
¶
The UTC timestamp indicating when the entity was created on the server.
Querying with Query Builders¶
The created_timestamp property is queryable when constructing a QuerySequence
via the convenience method:
Example
from mosaicolabs import MosaicoClient, QuerySequence, Time
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific sequence creation time
qresponse = client.query(
QuerySequence().with_created_timestamp(time_start=Time.from_float(1765432100)),
)
# Inspect the response
if qresponse is not None:
# Results are automatically grouped by Sequence for easier data management
for item in qresponse:
print(f"Sequence: {item.sequence.name}")
print(f"Topics: {[topic.name for topic in item.topics]}")
updated_timestamps
property
¶
The UTC timestamps indicating when the entity was updated on the server.
Querying with Query Builders¶
The updated_timestamps property is not queryable.