Custom Messages (horus.msggen)

There are two ways to move a payload HORUS has no built-in type for. Only the first works on an installed HORUS:

Dict Topics (No Build Step)

Any bare-string topic is a generic topic: it serializes whatever you send with MessagePack, so a plain dict of primitives crosses processes with no schema, no code generation and no compilation. This is the path build_messages() itself points you at when you have no source checkout.

Basic Usage

from horus import Topic

# Publisher
pub_topic = Topic("robot.status")
pub_topic.send({
    'battery_level': 85.0,
    'error_code': 0,
    'is_active': True,
    'timestamp': 0,
})

# Subscriber (different process)
sub_topic = Topic("robot.status")
data = sub_topic.recv()          # -> dict, or None if nothing queued
if data:
    print(data['battery_level'])  # 85.0

Supported Values

A generic topic carries JSON-shaped values: dicts, lists, strings, numbers and booleans, nested freely. It does not accept bytes — send(b'...') raises TypeError: Failed to convert Python object: invalid type: byte array, expected any valid JSON value. complex is rejected the same way, which catches people sending np.fft output directly. To move raw bytes, carry them inside the dict as a list of ints or as a base64 string field.

There is also a size ceiling: the encoded payload must be at most 4096 bytes. Exceeding it raises

ValueError: Invalid input: 'data' out of range: expected [0..4096], got 14619

which is roughly a thousand floats. That is ample for control values, poses and feature vectors, and far too small for an image or a point cloud — those belong on typed topics carrying horus.Image / horus.PointCloud, which are pool-backed and never pass through this encoder.

Note that recv() returns a plain dict, not an object — read fields with data['battery_level'], not data.battery_level.

With Node

The same dict goes through the Node convenience API:

import horus

def publisher_tick(node):
    node.send("robot.status", {'battery_level': 85.0, 'error_code': 0})

def subscriber_tick(node):
    if node.has_msg("robot.status"):
        data = node.recv("robot.status")   # -> dict
        print(data['battery_level'])       # 85.0

pub = horus.Node("publisher", pubs="robot.status", tick=publisher_tick)
sub = horus.Node("subscriber", subs="robot.status", tick=subscriber_tick)
horus.run(pub, sub, duration=3)

Compiled Messages (Source Checkout Only)

horus.msggen turns Python message definitions into Rust PyO3 source and runs maturin over it. It is for adding message types to a horus_py source checkout; the generated files are not wired into the extension for you, and the resulting class does not get the zero-copy backend that built-in types use — see Step 3.

Step 1: Define Messages

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'),
])

Field types accepted by register_message

[T; N] array literals and raw Vec<...> strings also parse. Anything else raises ValueError: Unknown type: ....

Step 2: Build

from horus.msggen import build_messages

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

This writes one .rs file per message plus a mod.rs into horus_py/src/custom_messages/, then runs maturin.

Step 3: Import It

No wiring step is needed. horus_py/src/lib.rs declares mod custom_messages; unconditionally and calls custom_messages::register_custom_messages(m)? from its #[pymodule], and custom_messages/mod.rs is a committed stub that the generator rewrites with one mod declaration and one add_class call per message. build_messages() is therefore sufficient on its own.

error[E0428]: the name `custom_messages` is defined multiple times

The generated class is importable from the extension module and constructible:

from horus._horus import RobotStatus

status = RobotStatus(battery_level=85.0, error_code=0, is_active=True, timestamp=0)
print(status)   # RobotStatus(battery_level=85.000, error_code=0, is_active=true, timestamp=0)

ImportError: cannot import name 'RobotStatus' from 'horus'

It is not usable as a topic message type. Topic() selects the zero-copy POD backend only for the built-in types in the pod_topic_types! macro in horus_py/src/topic.rs; a generated name falls through to the GenericMessage MessagePack path, and the generated #[pyclass] derives no serde impls, so sending an instance raises TypeError: Failed to convert Python object. Send its fields as a dict instead.

YAML Schema (Code Generation Only)

For larger projects, message definitions can live 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

from horus.msggen import generate_messages_from_yaml

# Writes one .rs file per message into output_dir, and returns the generated source.
generate_messages_from_yaml('messages.yaml', output_dir='horus_py/src/custom_messages')

generate_messages_from_yaml is a code emitter only. It calls generate_message for each entry and never calls register_message, so nothing lands in the build registry — a following build_messages() prints No custom messages registered. Nothing to build. and returns True without building anything. Omit output_dir and it writes no files at all. It also requires pyyaml.

To go through the build path, read the YAML yourself and register each entry:

import yaml
from horus.msggen import register_message, build_messages

with open('messages.yaml') as f:
    schema = yaml.safe_load(f)

for msg in schema['messages']:
    register_message(
        msg['name'],
        msg['topic'],
        [(f['name'], f['type']) for f in msg['fields']],
        doc=msg.get('doc'),
    )

build_messages()

Rebuild Detection

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

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:

build_messages(force=True)

Performance

Topic() reaches the zero-copy POD backend only for the built-in message types listed in the pod_topic_types! macro in horus_py/src/topic.rs (CmdVel, Pose2D, Pose3D, Imu, Odometry, LaserScan, JointState, …). Every other type name — including generated custom messages — falls through to the GenericMessage path, which serializes with MessagePack. A custom message therefore cannot match built-in performance, no matter how it is defined.

For real numbers, run the benchmarks in the source tree rather than relying on figures quoted in docs:

cargo run --release -p horus_benchmarks --bin all_paths_latency

API Reference

horus.msggen exports exactly five names: register_message, build_messages, check_needs_rebuild, generate_message and generate_messages_from_yaml.

register_message

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

Register a message for compiled generation.

Parameters:

• name: Class name (e.g., "RobotStatus")
• topic: Topic name (e.g., "robot.status"). Only embedded in the generated Rust as the __topic_name__ class attribute, so any string will do — there is no separate naming convention for compiled messages
• fields: List of (field_name, type_string) tuples
• doc: Optional docstring, emitted as the generated Rust type's doc comment

build_messages

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

def check_needs_rebuild() -> bool: ...

Check if registered messages differ from last build.

generate_message

def generate_message(
    name: str,
    topic: str,
    fields: List[Tuple[str, str]],
    doc: Optional[str] = None,
    output_dir: Optional[Path] = None
) -> str: ...

Generate the Rust source for one message and return it. This is a code emitter: it does not add the message to the build registry, so a following build_messages() will not see it. Use register_message for that.

Parameters:

• name, topic, fields, doc: as for register_message
• output_dir: if given, the source is also written to <output_dir>/<name>.rs with the name lowercased (RobotStatus → robotstatus.rs), creating the directory if needed. Omit it and nothing is written to disk.

Returns: the generated Rust source

generate_messages_from_yaml

def generate_messages_from_yaml(
    yaml_path: str,
    output_dir: Optional[Path] = None
) -> List[str]: ...

Call generate_message once per entry under the messages: key of a YAML file. Same caveat: nothing is registered. Requires pyyaml.

Returns: one Rust source string per message, in file order

Complete Example

#!/usr/bin/env python3
"""Custom message example with dict topics."""

import horus

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

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

# Create nodes
sensor = horus.Node("sensor", pubs="my.sensor", tick=sensor_tick, rate=10)
processor = horus.Node("processor", subs="my.sensor", tick=processor_tick)

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

When to Use Each Approach

Use Dict Topics When:

• Prototyping new message types
• Message schema changes frequently
• You don't want to wait for compilation
• You are on a pip-installed HORUS (no source checkout)

Use Compiled Messages When:

• You are developing against a horus_py source checkout
• You want a real Rust type with named fields, and accept that its fields still travel over topics as a dict (see Step 3)

Use Built-in Messages When:

• Standard robotics types (CmdVel, Pose2D, LaserScan)
• Maximum performance needed — only built-in types reach the zero-copy POD backend
• Compatibility with other HORUS systems