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_entity_kind_suffix as suffix_for,
25    ros2_is_fully_qualified as is_fully_qualified,
26    ros2_is_valid_name as is_valid_name,
27    ros2_percent_mangle as percent_mangle,
28    ros2_twist_from_cdr as twist_from_cdr,
29    ros2_twist_to_cdr as twist_to_cdr,
30    ros2_type_hash_digest as type_hash_digest,
31)
32
33__all__ = [
34    "CdrReader",
35    "CdrWriter",
36    "EntityKind",
37    "dds_topic",
38    "dds_type_name",
39    "entity_key",
40    "is_fully_qualified",
41    "is_valid_name",
42    "percent_mangle",
43    "prefix_for",
44    "suffix_for",
45    "twist_from_cdr",
46    "twist_to_cdr",
47    "type_hash_digest",
48]
49
50
51class EntityKind(str, enum.Enum):
52    """The ROS 2 subsystem a name belongs to, which fixes its DDS prefix and suffix."""
53
54    #: A topic, which takes the ``rt`` prefix.
55    TOPIC = "Topic"
56    #: The request side of a service, which takes the ``rq`` prefix and the ``Request`` suffix.
57    SERVICE_REQUEST = "ServiceRequest"
58    #: The reply side of a service, which takes the ``rr`` prefix and the ``Reply`` suffix.
59    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): 
52class EntityKind(str, enum.Enum):
53    """The ROS 2 subsystem a name belongs to, which fixes its DDS prefix and suffix."""
54
55    #: A topic, which takes the ``rt`` prefix.
56    TOPIC = "Topic"
57    #: The request side of a service, which takes the ``rq`` prefix and the ``Request`` suffix.
58    SERVICE_REQUEST = "ServiceRequest"
59    #: The reply side of a service, which takes the ``rr`` prefix and the ``Reply`` suffix.
60    SERVICE_RESPONSE = "ServiceResponse"

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

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 as rmw_zenoh writes it in a liveliness token, each / as %.

def prefix_for(kind):

Returns the DDS topic prefix a subsystem uses.

def suffix_for(kind):

Returns what the middleware appends to a name for a subsystem: nothing for a topic, Request for a service request, and Reply for a service response.

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.