Custom Messages (horus.msggen)

The horus.msggen module lets you define custom typed messages directly in Python — no hand-written Rust required. You describe the fields in Python, run a one-time build step, and the messages become native zero-copy types (~3-5μs) importable from horus exactly like the built-in messages.

The workflow is always the same:

Define the message with register_message(name, topic, fields)
Build it once with build_messages() — this generates Rust code and runs maturin develop
Use it — from horus import YourMessage, then send it over a typed Topic

Defining a Message

Describe the message name, its topic, and a list of (field_name, type) tuples. Fixed-size scalar fields keep the message a zero-copy Pod.

# simplified
from horus.msggen import register_message

# Register a custom message type
register_message('RobotStatus', 'robot.status', [
    ('battery_level', 'f32'),
    ('error_code', 'i32'),
    ('is_active', 'bool'),
    ('timestamp', 'u64'),
])

register_message records the definition — it does not create a usable class by itself. Run build_messages() (shown in the next sections) to compile every registered message, then import the generated class from horus and construct instances:

# simplified — after build_messages() has run
from horus import RobotStatus

status = RobotStatus(battery_level=85.0, error_code=0, is_active=True, timestamp=0)
print(status.battery_level)  # 85.0
status.error_code = 5

Supported Types

Type String	Size	Description
`f32` / `float32`	4 bytes	32-bit float
`f64` / `float64`	8 bytes	64-bit float
`i8`	1 byte	Signed 8-bit int
`i16`	2 bytes	Signed 16-bit int
`i32`	4 bytes	Signed 32-bit int
`i64`	8 bytes	Signed 64-bit int
`u8`	1 byte	Unsigned 8-bit int
`u16`	2 bytes	Unsigned 16-bit int
`u32`	4 bytes	Unsigned 32-bit int
`u64`	8 bytes	Unsigned 64-bit int
`bool`	1 byte	Boolean

Sending and Receiving

Once built, a custom message is a typed message like any built-in type. Create a typed Topic for it and pass instances directly — no manual serialization:

# simplified
import horus
from horus.msggen import register_message, build_messages

register_message('RobotStatus', 'robot.status', [
    ('battery_level', 'f32'),
    ('error_code', 'i32'),
])
build_messages()  # one-time compile; a no-op if nothing changed

from horus import RobotStatus, Topic

# Publisher — a typed topic carries RobotStatus instances directly
pub_topic = Topic(RobotStatus)
pub_topic.send(RobotStatus(battery_level=85.0, error_code=0))

# Subscriber (different process) — recv() returns a typed RobotStatus (or None)
sub_topic = Topic(RobotStatus)
received = sub_topic.recv()
if received:
    print(received.battery_level)  # 85.0

Or use the Node convenience API:

# simplified
import horus
from horus.msggen import register_message, build_messages

register_message('RobotStatus', 'robot.status', [
    ('battery_level', 'f32'),
    ('error_code', 'i32'),
])
build_messages()  # one-time compile

from horus import RobotStatus

def publisher_tick(node):
    node.send("robot.status", RobotStatus(battery_level=85.0, error_code=0))

def subscriber_tick(node):
    if node.has_msg("robot.status"):
        status = node.recv("robot.status")
        print(status.battery_level)  # 85.0

# Declare the typed message on pubs/subs so the topic is typed end-to-end
pub = horus.Node("publisher", pubs={"robot.status": RobotStatus}, tick=publisher_tick)
sub = horus.Node("subscriber", subs={"robot.status": RobotStatus}, tick=subscriber_tick)
horus.run(pub, sub, duration=3)

Registering Multiple Messages

You can register several messages before building — build_messages() compiles them all in one pass. Here is the full register → build → use cycle end to end. Generated messages are PyO3 types with the same zero-copy performance (~3-5μs) as the built-in messages.

Step 1: Register Messages

# simplified
from horus.msggen import register_message

# Register one or more messages
register_message('RobotStatus', 'robot.status', [
    ('battery_level', 'f32'),
    ('error_code', 'i32'),
    ('is_active', 'bool'),
    ('timestamp', 'u64'),
])

register_message('SensorReading', 'sensor.reading', [
    ('x', 'f64'),
    ('y', 'f64'),
    ('z', 'f64'),
])

Step 2: Build

# simplified
from horus.msggen import build_messages

# Generate Rust code and rebuild
build_messages()  # Runs: maturin develop --release

This generates Rust code in horus_py/src/custom_messages/ and rebuilds the module.

Step 3: Use

After building, your messages are available directly from horus:

# simplified
from horus import RobotStatus, SensorReading, Topic

# Create typed topic
topic = Topic(RobotStatus)

# Send
status = RobotStatus(battery_level=85.0, error_code=0, is_active=True, timestamp=0)
topic.send(status)

# Receive (typed!)
received = topic.recv()
print(received.battery_level)

YAML Schema (Recommended for Teams)

For larger projects, define messages in YAML:

# messages.yaml
messages:
  - name: RobotStatus
    topic: robot.status
    fields:
      - name: battery_level
        type: f32
      - name: error_code
        type: i32
      - name: is_active
        type: bool

  - name: SensorReading
    topic: sensor.reading
    fields:
      - name: x
        type: f64
      - name: y
        type: f64
      - name: z
        type: f64

# simplified
from horus.msggen import generate_messages_from_yaml, build_messages

generate_messages_from_yaml('messages.yaml')
build_messages()

Rebuild Detection

The builder tracks message definitions via hash. It won't rebuild unless messages change:

# simplified
from horus.msggen import check_needs_rebuild, build_messages

if check_needs_rebuild():
    build_messages()
else:
    print("Messages are up to date")

Force rebuild with:

# simplified
build_messages(force=True)

Performance Comparison

Approach	Latency	Throughput	Use Case
Built-in (Rust)	~3μs	300K msgs/sec	CmdVel, Pose2D, etc.
Compiled Custom	~3-5μs	200K msgs/sec	Custom typed messages

Recommendation: For a quick, schema-free prototype, send a plain dict over a generic topic. When you need a stable typed contract — especially one shared with Rust nodes — define a custom message and build it.

API Reference

register_message

# simplified
def register_message(
    name: str,
    topic: str,
    fields: List[Tuple[str, str]],
    doc: Optional[str] = None,
) -> None

Parameters:

name: Class name (e.g., "RobotStatus")
topic: Topic name (e.g., "robot.status")
fields: List of (field_name, type_string) tuples
doc: Optional docstring for the generated type

build_messages

# simplified
def build_messages(
    force: bool = False,
    verbose: bool = True
) -> bool

Build all registered messages.

Parameters:

force: Rebuild even if unchanged
verbose: Print progress

Returns: True if successful

check_needs_rebuild

# simplified
def check_needs_rebuild() -> bool

Check if registered messages differ from last build.

Complete Example

# simplified
#!/usr/bin/env python3
"""Custom message example."""

import horus
from horus.msggen import register_message, build_messages

# Define and build a custom sensor message (build is a no-op once compiled)
register_message('MySensor', 'my.sensor', [
    ('distance', 'f32'),
    ('angle', 'f32'),
    ('confidence', 'f32'),
    ('object_id', 'u32'),
])
build_messages()

from horus import MySensor

def sensor_tick(node):
    """Publish sensor readings."""
    reading = MySensor(
        distance=2.5,
        angle=0.785,
        confidence=0.95,
        object_id=42,
    )
    node.send("my.sensor", reading)

def processor_tick(node):
    """Process sensor readings."""
    if node.has_msg("my.sensor"):
        reading = node.recv("my.sensor")
        print(f"Object {reading.object_id}: {reading.distance}m at {reading.angle}rad")

# Create nodes — declare the typed message so the topic is typed end-to-end
sensor = horus.Node("sensor", pubs={"my.sensor": MySensor}, tick=sensor_tick, rate=10)
processor = horus.Node("processor", subs={"my.sensor": MySensor}, tick=processor_tick)

# Run
horus.run(sensor, processor, duration=3)

When to Use Each Approach

Use Custom Messages When:

You need a stable, typed contract for your own data
Cross-language compatibility with Rust nodes is required
Type safety and zero-copy performance (~3-5μs) matter
The schema is stable enough to compile once

Use a Generic dict When:

Prototyping and the schema changes frequently
The data stays inside Python (does not cross to Rust)
You want to skip the build step entirely

Use Built-in Messages When:

Standard robotics types (CmdVel, Pose2D, LaserScan)
Maximum performance needed
Compatibility with other HORUS systems

Design Decisions

Why require a build step at all? Zero-copy IPC needs a fixed-size Pod layout known at compile time. Generating and compiling native PyO3 types is what lets a custom message be sent with the same performance and cross-language compatibility as a built-in type. For schema-free data that never leaves Python, send a plain dict over a generic topic instead — no build required.

Why YAML schema support? Teams need a shared source of truth for message definitions that is language-agnostic and version-controllable. YAML schemas serve as the canonical definition that generates the message types used from both Python and Rust nodes, ensuring cross-language compatibility.

Why generate Rust code for compiled messages instead of a Python-only binary format? Generated Rust code produces the same Pod types used by the standard library, so compiled custom messages get identical zero-copy IPC performance and are binary-compatible with Rust nodes. A Python-only approach would sacrifice cross-language support.