roboto.domain.topics.operations#

Module Contents#

class roboto.domain.topics.operations.AddMessagePathRepresentationRequest(/, **data)#

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 (Any)

message_path_id: str#

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class roboto.domain.topics.operations.AddMessagePathRequest(/, **data)#

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 (Any)

message_path#: Dot-delimited path to the attribute (e.g., “pose.position.x”).

data_type#: Native data type as it appears in the original data source (e.g., “float32”, “geometry_msgs/Pose”). Used for display purposes.

canonical_data_type#: Normalized Roboto data type that enables specialized platform features for maps, images, timestamps, and other data.

metadata#: Initial key-value pairs to associate with the message path.

canonical_data_type: roboto.domain.topics.record.CanonicalDataType#

data_type: str#

message_path: str#

metadata: dict[str, Any] = None#

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class roboto.domain.topics.operations.BaseAddRepresentationRequest(/, **data)#

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 (Any)

association: roboto.association.Association#

storage_format: roboto.domain.topics.record.RepresentationStorageFormat#

version: int#

class roboto.domain.topics.operations.CreateTopicRequest(/, **data)#

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 (Any)

association: roboto.association.Association#

end_time: int | None = None#

message_count: int | None = None#

message_paths: collections.abc.Sequence[AddMessagePathRequest] | None = None#

metadata: collections.abc.Mapping[str, Any] | None = None#

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

schema_checksum: str | None = None#

schema_name: str | None = None#

start_time: int | None = None#

topic_name: str#

class roboto.domain.topics.operations.DeleteMessagePathRequest(/, **data)#

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 (Any)

message_path: str#: Message path name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class roboto.domain.topics.operations.MessagePathChangeset(/, **data)#

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 (Any)

check_replace_all_correctness()#

Return type:: MessagePathChangeset

classmethod from_replacement_message_paths(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.abc.Sequence[AddMessagePathRequest]) – Sequence of message path requests to replace existing paths.
Returns:: MessagePathChangeset configured to replace all existing message paths.
Return type:: MessagePathChangeset

Examples

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

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.
Return type:: bool

message_paths_to_add: collections.abc.Sequence[AddMessagePathRequest] | None = None#: Message paths to add to a topic.

message_paths_to_delete: collections.abc.Sequence[DeleteMessagePathRequest] | None = None#: Message paths to delete from a topic.

message_paths_to_update: collections.abc.Sequence[UpdateMessagePathRequest] | None = None#: Message paths to update on a topic.

replace_all: bool = False#

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.

class roboto.domain.topics.operations.MessagePathRepresentationMapping(/, **data)#

Bases: pydantic.BaseModel

Mapping between message paths and their data representation.

Associates a set of message paths with a specific representation that contains their data. This mapping is used to efficiently locate and access data for specific message paths within topic representations.

Parameters:: data (Any)

message_paths: collections.abc.MutableSequence[roboto.domain.topics.record.MessagePathRecord]#

representation: roboto.domain.topics.record.RepresentationRecord#

class roboto.domain.topics.operations.SetDefaultRepresentationRequest(/, **data)#

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 (Any)

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class roboto.domain.topics.operations.UpdateMessagePathRequest(/, **data)#

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 (Any)

canonical_data_type: roboto.domain.topics.record.CanonicalDataType | roboto.sentinels.NotSetType#

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.

data_type: str | roboto.sentinels.NotSetType#: Native data type for the data under this message path (optional).

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.
Return type:: bool

message_path: str#: Message path name (required).

metadata_changeset: roboto.updates.TaglessMetadataChangeset | roboto.sentinels.NotSetType#: A set of changes to the message path’s metadata (optional).

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class roboto.domain.topics.operations.UpdateTopicRequest(/, **data)#

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 (Any)

end_time: int | None | roboto.sentinels.NotSetType#

message_count: int | roboto.sentinels.NotSetType#

message_path_changeset: MessagePathChangeset | roboto.sentinels.NotSetType#

metadata_changeset: roboto.updates.MetadataChangeset | roboto.sentinels.NotSetType#

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

schema_checksum: str | None | roboto.sentinels.NotSetType#

schema_name: str | None | roboto.sentinels.NotSetType#

start_time: int | None | roboto.sentinels.NotSetType#