Platform Models¶
mosaicolabs.models.platform.Topic ¶
Bases: PlatformBase
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 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¶
The user_metadata field of this class is queryable when constructing a QueryTopic
via the .Q proxy.
Check the documentation of the PlatformBase to construct a
a valid expression for the builders involving the user_metadata component.
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(
Topic.Q.user_metadata["update_rate_hz"].eq(100), # Access the keys using the [] operator
Topic.Q.user_metadata["interface.type"].eq("canbus"), # Navigate the nested dicts using the dot notation
)
)
# # The same query using `with_expression`
# qresponse = client.query(
# QueryTopic()
# .with_expression(Topic.Q.user_metadata["update_rate_hz"].eq(100))
# .with_expression(
# Topic.Q.user_metadata["interface.type"].match("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¶
The user_metadata field is queryable when constructing a QueryTopic
or QuerySequence using the .Q proxy
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
Sequence.Q.user_metadata["key"] |
String, Numeric, Boolean |
.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .match() |
Sequence.Q.user_metadata["key.subkey.subsubkey..."] |
String, Numeric, Boolean |
.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .match() |
Topic.Q.user_metadata["key"] |
String, Numeric, Boolean |
.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .match() |
Topic.Q.user_metadata["key.subkey.subsubkey..."] |
String, Numeric, Boolean |
.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .match() |
Example
from mosaicolabs import MosaicoClient, Topic, Sequence, QueryTopic, QuerySequence
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific keys in sequence AND topic metadata.
qresponse = client.query(
QueryTopic(Topic.Q.user_metadata["update_rate_hz"].geq(100)),
QuerySequence(Sequence.Q.user_metadata["project.version"].match("v1.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]}")
name
property
¶
The unique identifier or resource name of the entity.
Querying with Query Builders¶
The name property is queryable when constructing a QueryTopic
or a QuerySequence
via the convenience methods:
QueryTopic.with_name()QueryTopic.with_name_match()QuerySequence.with_name()QuerySequence.with_name_match()
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_name("/front/imu"),
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_datetime
property
¶
The UTC timestamp indicating when the entity was created on the server.
Querying with Query Builders¶
The created_datetime property is queryable when constructing a QueryTopic
or a QuerySequence
via the convenience methods:
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]}")
# 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]}")
is_locked
property
¶
Indicates if the resource is currently locked.
A locked state typically occurs during active writing or maintenance operations, preventing deletion or structural modifications.
Querying with Query Builders¶
The is_locked property is not queryable.
total_size_bytes
property
¶
The total physical storage footprint of the entity on the server in bytes.
Querying with Query Builders¶
The total_size_bytes property is not queryable.
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.Sequence ¶
Bases: PlatformBase
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 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¶
The user_metadata field of this class is queryable when constructing a QuerySequence
via the .Q proxy.
Check the documentation of the PlatformBase to construct a
a valid expression for the builders involving the user_metadata component.
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(
Sequence.Q.user_metadata["project"].eq("Apollo"), # Access the keys using the [] operator
Sequence.Q.user_metadata["vehicle.software_stack.planning"].match("plan-4."), # Navigate the nested dicts using the dot notation
)
)
# # The same query using `with_expression`
# qresponse = client.query(
# QuerySequence()
# .with_expression(Sequence.Q.user_metadata["project"].eq("Apollo"))
# .with_expression(
# Sequence.Q.user_metadata["vehicle.software_stack.planning"].match("plan-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]}")
user_metadata
instance-attribute
¶
Custom user-defined key-value pairs associated with the entity.
Querying with the .Q Proxy¶
The user_metadata field is queryable when constructing a QueryTopic
or QuerySequence using the .Q proxy
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
Sequence.Q.user_metadata["key"] |
String, Numeric, Boolean |
.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .match() |
Sequence.Q.user_metadata["key.subkey.subsubkey..."] |
String, Numeric, Boolean |
.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .match() |
Topic.Q.user_metadata["key"] |
String, Numeric, Boolean |
.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .match() |
Topic.Q.user_metadata["key.subkey.subsubkey..."] |
String, Numeric, Boolean |
.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .match() |
Example
from mosaicolabs import MosaicoClient, Topic, Sequence, QueryTopic, QuerySequence
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific keys in sequence AND topic metadata.
qresponse = client.query(
QueryTopic(Topic.Q.user_metadata["update_rate_hz"].geq(100)),
QuerySequence(Sequence.Q.user_metadata["project.version"].match("v1.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]}")
name
property
¶
The unique identifier or resource name of the entity.
Querying with Query Builders¶
The name property is queryable when constructing a QueryTopic
or a QuerySequence
via the convenience methods:
QueryTopic.with_name()QueryTopic.with_name_match()QuerySequence.with_name()QuerySequence.with_name_match()
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_name("/front/imu"),
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_datetime
property
¶
The UTC timestamp indicating when the entity was created on the server.
Querying with Query Builders¶
The created_datetime property is queryable when constructing a QueryTopic
or a QuerySequence
via the convenience methods:
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]}")
# 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]}")
is_locked
property
¶
Indicates if the resource is currently locked.
A locked state typically occurs during active writing or maintenance operations, preventing deletion or structural modifications.
Querying with Query Builders¶
The is_locked property is not queryable.
total_size_bytes
property
¶
The total physical storage footprint of the entity on the server in bytes.
Querying with Query Builders¶
The total_size_bytes property 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 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]}")
mosaicolabs.models.platform.platform_base.PlatformBase ¶
Bases: BaseModel, _QueryProxyMixin
Base class for Mosaico Sequence and Topic entities.
The PlatformBase serves as a read-only view of a server-side resource.
It is designed to hold system-level metadata and enable fluid querying of
user-defined properties.
Core Functionality¶
- System Metadata: Consolidates attributes like storage size and locking status that are common across the catalog.
- Query Interface: Inherits from internal
_QueryableModelto support expressive syntax for filtering resources (e.g.,Sequence.Q.user_metadata["env"] == "prod").
Read-Only Entities
Instances of this class are factory-generated from server responses. Users should not instantiate this class directly.
Attributes:
| Name | Type | Description |
|---|---|---|
user_metadata |
Dict[str, Any]
|
A dictionary of custom key-value pairs assigned by the user. |
Querying with the .Q Proxy¶
The user_metadata field is queryable when constructing a QueryTopic
or QuerySequence via the .Q proxy.
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
<PlatformModel>.Q.user_metadata["key"] |
String, Numeric, Boolean |
.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .match() |
Universal Compatibility
The <PlatformModel> placeholder represents any Mosaico class derived by
PlatformBase
(i.e. Topic, Sequence)
Example
from mosaicolabs import MosaicoClient, Topic, Sequence, QueryTopic, QuerySequence
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific metadata key value.
qresponse = client.query(
QueryTopic(Topic.Q.user_metadata["update_rate_hz"].geq(100))
)
# Filter for a specific nested metadata key value.
qresponse = client.query(
QuerySequence(Sequence.Q.user_metadata["project.version"].match("v1.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]}")
user_metadata
instance-attribute
¶
Custom user-defined key-value pairs associated with the entity.
Querying with the .Q Proxy¶
The user_metadata field is queryable when constructing a QueryTopic
or QuerySequence using the .Q proxy
| Field Access Path | Queryable Type | Supported Operators |
|---|---|---|
Sequence.Q.user_metadata["key"] |
String, Numeric, Boolean |
.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .match() |
Sequence.Q.user_metadata["key.subkey.subsubkey..."] |
String, Numeric, Boolean |
.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .match() |
Topic.Q.user_metadata["key"] |
String, Numeric, Boolean |
.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .match() |
Topic.Q.user_metadata["key.subkey.subsubkey..."] |
String, Numeric, Boolean |
.eq(), .neq(), .lt(), .gt(), .leq(), .geq(), .in_(), .between(), .match() |
Example
from mosaicolabs import MosaicoClient, Topic, Sequence, QueryTopic, QuerySequence
with MosaicoClient.connect("localhost", 6726) as client:
# Filter for a specific keys in sequence AND topic metadata.
qresponse = client.query(
QueryTopic(Topic.Q.user_metadata["update_rate_hz"].geq(100)),
QuerySequence(Sequence.Q.user_metadata["project.version"].match("v1.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]}")
name
property
¶
The unique identifier or resource name of the entity.
Querying with Query Builders¶
The name property is queryable when constructing a QueryTopic
or a QuerySequence
via the convenience methods:
QueryTopic.with_name()QueryTopic.with_name_match()QuerySequence.with_name()QuerySequence.with_name_match()
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_name("/front/imu"),
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_datetime
property
¶
The UTC timestamp indicating when the entity was created on the server.
Querying with Query Builders¶
The created_datetime property is queryable when constructing a QueryTopic
or a QuerySequence
via the convenience methods:
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]}")
# 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]}")