Requirements	Compatibility	Tests Matrix
		Python: 3.10, 3.11, 3.12, 3.13, 3.14 NumPy: 1.26, 2.x

cydr is a fast, pythonic, focused XCDR1 serializer/deserializer for Python with a schema API and just-in-time compilation. After defining your message schema with a Python class, cydr generates a Cython codec and compiles it down to C unlocking sub-microsecond operations.

Priorities:

1. SPEED
2. Pythonic-style
3. JIT compilation

Runnable examples:

• examples/roundtrip_custom_message.py for the minimal low-level nested-dict interface
• examples/roundtrip_xcdrstruct_message.py for the higher-level XcdrStruct interface based on msgspec

Those benchmarks were performed on JointState messages containing 3 sequence of floats, 1 sequence of strings, one nested Header (containing a string, and another nested Time with 2 integers). The Count row of table idicates the length of the 4 sequences.

from typing import Any
import msgspec
import numpy as np
from nptyping import Bytes, Float64, NDArray

from cydr.idl import XcdrStruct
from cydr.types import int32, string, uint32

# Define adn use your message schema using a simple python class
# Type hints (`: int32` / `: uint32`) describe the cdr type of the field
class Time(XcdrStruct):
    sec: int32 = np.int32(0)
    nanosec: uint32 = np.uint32(0)

# Create more schemas using one you previously created
class Header(XcdrStruct):
    stamp: Time = msgspec.field(default_factory=Stamp)
    frame_id: string = b""

# And also define collections (arrays of set/varialbe length) using numpy and nptyping
class JointState(XcdrStruct):
    header: Header = msgspec.field(default_factory=Header)
    name: NDArray[Any, Bytes] = msgspec.field(
        default_factory=lambda: np.array([], dtype=np.bytes_)
    )
    position: NDArray[Any, Float64] = msgspec.field(
        default_factory=lambda: np.array([], dtype=np.float64)
    )
    velocity: NDArray[Any, Float64] = msgspec.field(
        default_factory=lambda: np.array([], dtype=np.float64)
    )
    effort: NDArray[Any, Float64] = msgspec.field(
        default_factory=lambda: np.array([], dtype=np.float64)
    )

# Optional: forces caching and compilation of the schema
JointState.brew()

# Instatiate a message with its nested components and arrays
message = JointState(
    header=Header(
        stamp=Stamp(sec=np.int32(10), nanosec=np.uint32(123)),
        frame_id=b"map",
    ),
    name=np.array([b"joint_a", b"joint_b"], dtype=np.bytes_),
    position=np.array([1.0, 2.0], dtype=np.float64),
)

# Encode / Decode
payload = message.serialize()
decoded = JointState.deserialize(payload)

• WIRE -> Python

• boolean -> np.bool_
• byte, uint8 -> np.uint8
• int8 -> np.int8
• int16 -> np.int16
• uint16 -> np.uint16
• int32 -> np.int32
• uint32 -> np.uint32
• int64 -> np.int64
• uint64 -> np.uint64
• float32 -> np.float32
• float64 -> np.float64
• string -> UTF-8 bytes

• Sequence[Primitive] -> NDArray[Any, dtype] defines a variable-length 1D collection of one primitive dtype.
• Array[n, Primitive] -> NDArray[Shape["n"], dtype] defines a fixed-length 1D collection of one primitive dtype.
• Sequence[string] -> NDArray[Any, Bytes] Represent a collection of strings with a numpy array of np.bytes_ by default.
  • When deserializing you can choose the representation using the Enum StringCollectionMode. .LIST mode (list[bytes]) is more performant from small arrays, or you can use the most performant .RAW mode, to manipulate a list like C wrapper directly.
  • When serializing, you can either pass list[bytes] or array[np.bytes_]

Generated codecs are cached by the hash of the flattened field-type sequence.

• Default cache dir: ./.cydr_cache
• Override globally with CYDR_CACHE_DIR=/path/to/cache