Platform Models
mosaicolabs.models.platform.Topic
dataclass
¶
Topic(
user_metadata,
name,
sequence_name,
created_timestamp,
ontology_tag,
chunks_number,
serialization_format,
locked,
total_size_bytes,
)
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 Query Builders¶
Querying Topic specific attributes (like user_metadata or name) can be made using the
QueryTopic() query builder.
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_match("/sensors/imu") # (1)!
.with_user_metadata("update_rate_hz", gt=100) # (2)!
.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]}")
- Find all the topics which name matches the pattern
- Query the (key, value) in the
user_metadataJSON
user_metadata
instance-attribute
¶
Custom user-defined key-value pairs associated with the entity.
Querying with Query Builders¶
Querying the user_metadata attribute can be made using the
QueryTopic.with_user_metadata() query builder.
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_user_metadata("update_rate_hz", gt=100) # (1)!
.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]}")
- Query the (key, value) in the
user_metadataJSON
name
instance-attribute
¶
The unique identifier or resource name of the entity.
Querying with Query Builders¶
The name attribute 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]}")
sequence_name
instance-attribute
¶
The name of the parent sequence containing this topic.
Querying with Query Builders¶
The sequence_name attribute is queryable queryable when constructing a QuerySequence
via the convenience methods:
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_match("test_winter_2026")
)
# Inspect the response
if qresponse is not None:
# Results 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
instance-attribute
¶
The UTC timestamp indicating when the entity was created on the server.
Querying with Query Builders¶
The created_timestamp attribute 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
instance-attribute
¶
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 attribute is queryable when constructing a QueryTopic
via the convenience method:
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]}")
chunks_number
instance-attribute
¶
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 attribute is not queryable.
serialization_format
instance-attribute
¶
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 attribute 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.
locator
instance-attribute
¶
The session locator.
The locator is in the form 'sequence_name:session_identifier',
e.g.: 'test-sequence-datastream:01KQ9XQ0HJ3V39F87CBP6PYA1T'
created_timestamp
instance-attribute
¶
The UTC timestamp [ns] when the writing session started
completed_timestamp
instance-attribute
¶
The UTC timestamp [ns] of the session finalization.
mosaicolabs.models.platform.Sequence
dataclass
¶
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 Query Builders¶
Querying Sequence specific attributes (like user_metadata or name) can be made using the
QuerySequence() query builder.
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_") # (1)!
.with_user_metadata("project", eq="Apollo") # (2)!
.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]}")
- Find all the sequences which name matches the pattern
- Query the (key, value) in the
user_metadataJSON
user_metadata
instance-attribute
¶
Custom user-defined key-value pairs associated with the entity.
Querying with Query Builders¶
The user_metadata attribute is queryable when constructing a QuerySequence
via the convenience method:
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") # (1)!
.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]}")
- Query the (key, value) in the
user_metadataJSON
created_timestamp
instance-attribute
¶
The UTC timestamp when the sequence was created.
Querying with Query Builders¶
The `created_timestamp` attribute is queryable when constructing a [`QuerySequence`][mosaicolabs.models.query.QuerySequence]
via the convenience method:
* [`QuerySequence.with_created_timestamp()`][mosaicolabs.models.query.builders.QuerySequence.with_created_timestamp]
Example:
```python
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]}")
```
name
instance-attribute
¶
The name of the sequence.
Querying with Query Builders¶
The `name` attribute is queryable when constructing a [`QuerySequence`][mosaicolabs.models.query.QuerySequence]
via the convenience methods:
* [`QuerySequence.with_name()`][mosaicolabs.models.query.builders.QuerySequence.with_name]
* [`QuerySequence.with_name_match()`][mosaicolabs.models.query.builders.QuerySequence.with_name_match]
Example:
```python
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]}")
```
total_size_bytes
instance-attribute
¶
The aggregated total size of the sequence in bytes
Querying with Query Builders¶
The `total_size_bytes` attribute is not queryable.
sessions
instance-attribute
¶
The list of sessions in the sequence
Querying with Query Builders¶
The `sessions` attribute is not queryable.
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 queryable via the QueryTopic builder,
through 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("/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]}")