roboto.domain.topics.operations
Module Contents
AddMessagePathRepresentationRequest
Bases: BaseAddRepresentationRequest
Request to associate a message path with a representation.
Creates a link between a specific message path and a data representation, enabling efficient access to individual fields within topic data.
Parameters
data AnyAddMessagePathRequest
Bases: pydantic.BaseModel
Request to add a new message path to a topic.
Defines a new message path within a topic’s schema, specifying its data type, canonical type, and initial metadata. Used during topic creation or when extending an existing topic’s schema.
Parameters
data AnyAttributes
AddMessagePathRequest.canonical_data_type
Normalized Roboto data type that enables specialized platform features for maps, images, timestamps, and other data.
AddMessagePathRequest.data_type
Native data type as it appears in the original data source (e.g., “float32”, “geometry_msgs/Pose”). Used for display purposes.
AddMessagePathRequest.message_path
Dot-delimited path to the attribute (e.g., “pose.position.x”).
AddMessagePathRequest.metadata
Initial key-value pairs to associate with the message path.
AddMessagePathRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
AddMessagePathRequest.path_in_schema
List of path components representing the field’s location in the original data schema. Unlike message_path, which assumes dots separate path parts implying nested data, this preserves the exact path from the source data for accurate attribute access.
BaseAddRepresentationRequest
Bases: pydantic.BaseModel
Base request for adding a representation to a topic.
Defines the common fields required when creating any type of representation for topic data, including the storage location, format, and version information.
Parameters
data AnyAttributes
BaseAddRepresentationRequest.association
BaseAddRepresentationRequest.format
BaseAddRepresentationRequest.storage_format
BaseAddRepresentationRequest.transformations
BaseAddRepresentationRequest.version
CreateTopicRequest
Bases: pydantic.BaseModel
Request to create a new topic in the Roboto platform.
Contains all the information needed to register a topic found within a source recording file, including its schema, temporal boundaries, and initial message paths.
Parameters
data AnyAttributes
CreateTopicRequest.association
CreateTopicRequest.end_time
CreateTopicRequest.message_count
CreateTopicRequest.message_paths
CreateTopicRequest.metadata
CreateTopicRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
CreateTopicRequest.schema_checksum
CreateTopicRequest.schema_name
CreateTopicRequest.start_time
CreateTopicRequest.topic_name
DeleteMessagePathRequest
Bases: pydantic.BaseModel
Request to delete a message path from a topic.
Removes a message path from a topic’s schema. This operation cannot be undone and will remove all associated data and metadata for the specified path.
Parameters
data AnyMessagePathChangeset
Bases: pydantic.BaseModel
Changeset for batch operations on topic message paths.
Defines a collection of add, delete, and update operations to be applied to a topic’s message paths in a single atomic operation. Useful for making multiple schema changes efficiently.
Parameters
data AnyMessagePathChangeset.check_replace_all_correctness()
Return type
MessagePathChangeset.from_replacement_message_paths()
Create a changeset that replaces all existing message paths.
Creates a changeset that will replace all existing message paths on a topic with the provided set of message paths. This is useful for completely redefining a topic’s schema.
Parameters
message_paths collections.Sequence of message path requests to replace existing paths.
Returns
MessagePathChangeset configured to replace all existing message paths.
Usage
from roboto.domain.topics import AddMessagePathRequest, CanonicalDataType
new_paths = [
AddMessagePathRequest(
message_path="velocity.x", data_type="float32", canonical_data_type=CanonicalDataType.Number
)
]
changeset = MessagePathChangeset.from_replacement_message_paths(new_paths)MessagePathChangeset.has_changes()
Check whether the changeset contains any actual changes.
Returns
True if the changeset contains operations that would modify the topic’s message paths.
Attributes
MessagePathChangeset.message_paths_to_add
Message paths to add to a topic.
MessagePathChangeset.message_paths_to_delete
Message paths to delete from a topic.
MessagePathChangeset.message_paths_to_update
Message paths to update on a topic.
MessagePathChangeset.replace_all
Flag indicating whether this changeset should replace all message paths on a topic.
It assumes that the replacement message paths will be provided via message_paths_to_add. Rather than setting this flag directly, use appropriate class methods such as from_replacement_message_paths.
SetDefaultRepresentationRequest
Bases: BaseAddRepresentationRequest
Request to set the default representation for a topic.
Designates a specific representation as the default for accessing topic data. The default representation is used when no specific representation is requested for data access operations.
Parameters
data AnyAttributes
SetDefaultRepresentationRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
SetTimelineOffsetsRequest
Bases: pydantic.BaseModel
Request body for POST /v1/files/id/<id>/timeline-offsets.
Atomic request: the server applies all entries together or fails.
Parameters
data AnyAttributes
SetTimelineOffsetsRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
SetTimelineOffsetsRequest.offsets
Offsets to apply, at least one. An entry whose selectors reach no extent on this file is skipped as long as another entry reaches one; a request whose entries together reach none is refused. When two entries reach the same extent, the later entry’s offset applies.
TimelineOffsetEntry
Bases: pydantic.BaseModel
One offset for set_timeline_offsets().
Optional selectors narrow which of the file’s timelines the offset applies to.
Topic selector:
topic_name: the topic’s name (e.g."/imu/raw"); topic names are unique within a single file. Scopes the offset to a single topic on the file.
Timeline-source selector (give at most one):
timeline_source_name: the source’s name (e.g."header.stamp"). Source names are not unique, so the offset applies to every source with that name within the topic selector’s scope.timeline_source_id: explicit id, for callers holding aTimelineSourceRecord.
Omitting every selector applies the offset to every timeline on the file.
Parameters
data AnyAttributes
TimelineOffsetEntry.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
TimelineOffsetEntry.timeline_source_id
TimelineOffsetEntry.timeline_source_name
TimelineOffsetEntry.topic_name
TimelineOffsetEntry.unix_epoch_offset_ns
Nanoseconds added to the file’s stored timestamps so they read as Unix-epoch time (session_time_ns = stored_time_ns + unix_epoch_offset_ns). Must not be negative, and must fit in the signed 64-bit integer the platform stores it in. Also accepts any roboto.time.Time at runtime, read as roboto.time.to_epoch_nanoseconds() reads it, so an instant such as a datetime becomes the nanoseconds since the Unix epoch at which stored time 0 occurred; convert with that function first to satisfy a type checker.
TimelineSourceUpdate
Bases: pydantic.BaseModel
Partial update for a timeline source.
Fields left at NotSet are not modified.
Parameters
data AnyAttributes
TimelineSourceUpdate.is_default
TimelineSourceUpdate.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
TimelineSourceUpdate.name
UpdateMessagePathRequest
Bases: pydantic.BaseModel
Request to update an existing message path within a topic.
Allows modification of message path attributes including metadata, data type, and canonical data type. Used to correct or enhance message path definitions after initial creation.
Parameters
data AnyAttributes
UpdateMessagePathRequest.canonical_data_type
Canonical Roboto data type for the data under this message path (optional).
Note: updating this attribute should be done with care, as it affects Roboto’s ability to interpret and visualize the data.
UpdateMessagePathRequest.data_type
Native data type for the data under this message path (optional).
UpdateMessagePathRequest.has_updates()
Check whether this request would result in any message path modifications.
Returns
True if the request contains changes that would modify the message path.
Attributes
UpdateMessagePathRequest.metadata_changeset
A set of changes to the message path’s metadata (optional).
UpdateMessagePathRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
UpdateMessagePathRequest.path_in_schema
List of path components representing the field’s location in the source data schema (optional).
For nested fields like ‘position.x’, this would be [‘position’, ‘x’].
UpdateTopicRequest
Bases: pydantic.BaseModel
Request to update an existing topic’s properties.
Allows modification of topic attributes including temporal boundaries, message count, schema information, metadata, and message paths.
Parameters
data AnyAttributes
UpdateTopicRequest.end_time
UpdateTopicRequest.message_count
UpdateTopicRequest.message_path_changeset
UpdateTopicRequest.metadata_changeset
UpdateTopicRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].