pamoja.ros2

Idiomatic ROS 2 naming and encoding facade.

What makes a topic name legal, what it becomes on the DDS wire, the RIHS type hash that identifies a message definition, and the CDR encoding the payload itself is written in.

None of it needs a ROS 2 installation. A gateway written in Python can validate a name, derive the DDS topic and the Zenoh key an rmw_zenoh peer subscribes on, and encode a geometry_msgs/msg/Twist with no ROS distribution anywhere near it. Driving a live graph does need one, so that stays in the Rust crate.

 1"""Idiomatic ROS 2 naming and encoding facade.
 2
 3What makes a topic name legal, what it becomes on the DDS wire, the RIHS type
 4hash that identifies a message definition, and the CDR encoding the payload
 5itself is written in.
 6
 7None of it needs a ROS 2 installation. A gateway written in Python can validate
 8a name, derive the DDS topic and the Zenoh key an ``rmw_zenoh`` peer subscribes
 9on, and encode a ``geometry_msgs/msg/Twist`` with no ROS distribution anywhere
10near it. Driving a live graph does need one, so that stays in the Rust crate.
11"""
12
13from __future__ import annotations
14
15import enum
16
17from pamoja._native import (
18    CdrReader,
19    CdrWriter,
20    ros2_dds_topic as dds_topic,
21    ros2_dds_type_name as dds_type_name,
22    ros2_entity_key as entity_key,
23    ros2_entity_kind_prefix as prefix_for,
24    ros2_is_fully_qualified as is_fully_qualified,
25    ros2_is_valid_name as is_valid_name,
26    ros2_percent_mangle as percent_mangle,
27    ros2_twist_from_cdr as twist_from_cdr,
28    ros2_twist_to_cdr as twist_to_cdr,
29    ros2_type_hash_digest as type_hash_digest,
30)
31
32__all__ = [
33    "CdrReader",
34    "CdrWriter",
35    "EntityKind",
36    "dds_topic",
37    "dds_type_name",
38    "entity_key",
39    "is_fully_qualified",
40    "is_valid_name",
41    "percent_mangle",
42    "prefix_for",
43    "twist_from_cdr",
44    "twist_to_cdr",
45    "type_hash_digest",
46]
47
48
49class EntityKind(str, enum.Enum):
50    """The ROS 2 subsystem a name belongs to, which fixes its DDS prefix."""
51
52    #: A topic, which takes the ``rt`` prefix.
53    TOPIC = "Topic"
54    #: The request side of a service, which takes the ``rq`` prefix.
55    SERVICE_REQUEST = "ServiceRequest"
56    #: The reply side of a service, which takes the ``rr`` prefix.
57    SERVICE_RESPONSE = "ServiceResponse"
class CdrReader:

A CDR decoder, which reads primitives back in the order they were written.

Reading past the end returns None rather than raising, because a short buffer is a wire condition rather than a programming error.

def read_i32(self, /):

Reads the next 32-bit signed integer, or None once exhausted.

def read_u32(self, /):

Reads the next 32-bit unsigned integer, or None once exhausted.

def read_f32(self, /):

Reads the next 32-bit float, or None once exhausted.

def read_f64(self, /):

Reads the next 64-bit float, or None once exhausted.

class CdrWriter:

A CDR encoder, which writes primitives with the alignment the wire format requires.

def write_i32(self, /, value):

Appends a 32-bit signed integer.

def write_u32(self, /, value):

Appends a 32-bit unsigned integer.

def write_f32(self, /, value):

Appends a 32-bit float.

def write_f64(self, /, value):

Appends a 64-bit float.

bytes

The bytes written so far.

class EntityKind(builtins.str, enum.Enum): 
50class EntityKind(str, enum.Enum):
51    """The ROS 2 subsystem a name belongs to, which fixes its DDS prefix."""
52
53    #: A topic, which takes the ``rt`` prefix.
54    TOPIC = "Topic"
55    #: The request side of a service, which takes the ``rq`` prefix.
56    SERVICE_REQUEST = "ServiceRequest"
57    #: The reply side of a service, which takes the ``rr`` prefix.
58    SERVICE_RESPONSE = "ServiceResponse"

The ROS 2 subsystem a name belongs to, which fixes its DDS prefix.

TOPIC = <EntityKind.TOPIC: 'Topic'>
SERVICE_REQUEST = <EntityKind.SERVICE_REQUEST: 'ServiceRequest'>
SERVICE_RESPONSE = <EntityKind.SERVICE_RESPONSE: 'ServiceResponse'>
def dds_topic(fqn, kind):

Returns the DDS topic a fully qualified name maps onto, or None if the name is not fully qualified.

def dds_type_name(ros_type):

Returns the DDS type name an interface type maps onto, or None if the type is not a valid package/namespace/Type.

def entity_key(domain_id, fqn, ros_type, type_hash):

Builds the Zenoh key an rmw_zenoh peer publishes an entity on, or None if the name, type, or hash is not usable.

def is_fully_qualified(name):

Reports whether a name is fully qualified, so it resolves with no namespace.

def is_valid_name(name):

Reports whether a string is a valid ROS 2 topic or service name.

def percent_mangle(name):

Percent-mangles a name the way a DDS partition requires.

def prefix_for(kind):

Returns the DDS topic prefix a subsystem uses.

def twist_from_cdr(data):

Decodes a twist from its CDR representation, or None if the bytes are not a well-formed twist.

Returns the linear and angular velocities as two (x, y, z) triples.

def twist_to_cdr(linear, angular):

Encodes a twist into its CDR representation.

The linear and angular velocities each cross as an (x, y, z) triple.

def type_hash_digest(text):

Returns the 32-byte digest a RIHS01 hash string carries, or None if the string is malformed.