Nodes & Topics in Python

Python nodes run over the same shared memory as Rust nodes (~1.5μs per message on the typed zero-copy path, ~5–50μs for untyped topics) and can communicate seamlessly with Rust nodes on the same topics.

Creating a Node

Functional Style

import horus
from horus import CmdVel

def drive_tick(node):
    node.send("cmd_vel", CmdVel(linear=0.5, angular=0.0))

def drive_shutdown(node):
    print("Drive node stopped")

drive = horus.Node(
    name="drive_node",
    tick=drive_tick,
    shutdown=drive_shutdown,
    pubs=["cmd_vel"],
    rate=50,
    order=0
)
horus.run(drive)

Node callbacks:

Callback	Called	Required
`init(node)`	Once at startup	No
`tick(node)`	Every cycle	Yes
`shutdown(node)`	Once at exit	No

Minimal Style

For simple nodes, skip the optional callbacks and pass just a tick:

import horus

def publish_temp(node):
    node.send("temperature", 25.5)

sensor = horus.Node(
    name="temp_sensor",
    pubs="temperature",
    tick=publish_temp,
    rate=10
)
horus.run(sensor, duration=30)

Creating Topics

from horus import Topic, CmdVel, LaserScan, Image

# Typed topic (validates message types)
cmd = Topic(CmdVel)

# Untyped topic (accepts any serializable data)
data = Topic("raw_data")

# Large data types use zero-copy automatically
camera = Topic(Image, endpoint="camera.rgb")

A typed topic takes its name from the type — CmdVel → "cmd_vel", LaserScan → "scan". Pass endpoint="my_name" to override it (as Image does above — Image carries no name of its own, so without the override it would fall back to the lowercased type name, "image").

The topic name is the connection key. Any node (Python or Rust) publishing to "cmd_vel" connects to any subscriber on "cmd_vel".

⚠️Matching names is not enough — the encoding must match too

A typed topic (Topic(CmdVel), or pubs={"cmd_vel": {"type": CmdVel}}) carries the message as a fixed-size struct. A generic topic (a bare name with no type, carrying a dict) carries MessagePack bytes. They are different wire formats on the same topic name.

Connect one to the other and nothing complains: the typed subscriber reinterprets the MessagePack bytes as the struct and returns plausible-looking nonsense. There is no exception and no log, because at the shared-memory level both sides did exactly what they were told.

Pick one per topic and keep both ends on it. Use typed topics for anything a Rust or C++ node also touches — a dict topic is Python-to-Python only.

Sending Messages

from horus import Node, Topic, CmdVel

cmd = Topic(CmdVel)

# Send a typed message
cmd.send(CmdVel(linear=1.0, angular=0.5))

# With functional nodes, use node.send().
# Declare the type so this stays a typed topic — a bare `pubs=["cmd_vel"]`
# would make it a generic MessagePack topic that typed subscribers misread.
def my_tick(node):
    node.send("cmd_vel", {"linear": 1.0, "angular": 0.0})

node = Node(name="driver", tick=my_tick,
            pubs={"cmd_vel": {"type": CmdVel}})

Receiving Messages

from horus import Topic, LaserScan

scan = Topic(LaserScan)

# Get next message (returns None if empty)
msg = scan.recv()
if msg is not None:
    # ranges is always 360 slots, zero-filled past the real readings, so
    # min(msg.ranges) reports 0.0. min_range() returns the nearest reading
    # inside [range_min, range_max], or None if there is none.
    closest = msg.min_range()
    if closest is not None:
        print(f"Min range: {closest}")

With functional nodes, use node.recv():

def process(node):
    if node.has_msg("scan"):
        scan = node.recv("scan")           # One message
        all_scans = node.recv_all("scan")  # All pending messages

Timestamps

Timestamps are managed by the Rust Topic backend. Messages that include a timestamp field (e.g., msg.timestamp_ns) can be used for latency and staleness checks at the application level:

def monitor(node):
    if node.has_msg("sensor"):
        msg = node.recv("sensor")
        # Use message-level timestamps for timing
        if hasattr(msg, 'timestamp_ns') and msg.timestamp_ns:
            import time
            age_s = (time.time_ns() - msg.timestamp_ns) / 1e9
            if age_s > 0.5:
                node.log_warning("Sensor data is stale!")

Type-Safe Messages

Use HORUS built-in types for runtime type checking and cross-language compatibility:

from horus import CmdVel, Imu, Pose2D, LaserScan, Image

# Built-in types match Rust types exactly
cmd = CmdVel(linear=1.0, angular=0.5)
pose = Pose2D(x=1.0, y=2.0, theta=0.5)

Custom Messages

On an installed HORUS, the way to carry your own structure is a generic topic: give it a bare name and send a dict of primitives.

from horus import Topic

topic = Topic("motor.status")
topic.send({"motor_id": 1, "velocity": 3.14, "temperature": 45.0})

Generic topics take JSON-shaped values — dicts, lists, strings, numbers, booleans — up to 4096 bytes encoded. They are not the zero-copy path, but they need no build step.

horus.msggen generates real Rust message types instead, and is a different proposition: it requires a horus_py source checkout and maturin on PATH. On a pip-installed HORUS, build_messages() raises RuntimeError and points you back at dict topics. See Custom Messages for the full procedure — and note that a generated type still travels over the generic MessagePack path, so it is not the zero-copy backend the built-in types use.

Mixed Rust + Python Communication

Rust and Python nodes share the same shared memory. A topic name connects them automatically — no path configuration needed.

Rust publisher:

use horus::prelude::*;

let topic = Topic::<CmdVel>::new("cmd_vel")?;
topic.send(CmdVel::new(1.0, 0.0));

Python subscriber:

from horus import Topic, CmdVel

cmd = Topic(CmdVel)  # name derived from the type: "cmd_vel"
msg = cmd.recv()
if msg is not None:
    print(f"linear={msg.linear}, angular={msg.angular}")

Run them together:

horus run motor_control.rs planner.py logger.py

All processes communicate through shared memory — sub-microsecond between Rust nodes (198–304ns p50 cross-process), a few microseconds once Python is in the path.

Complete Example: Sensor Pipeline

import horus
from horus import LaserScan, CmdVel, Topic

scan_topic = Topic(LaserScan)
cmd_topic = Topic(CmdVel)

def avoider_tick(node):
    scan = scan_topic.recv()
    if scan is not None:
        min_range = min(r for r in scan.ranges
                       if scan.range_min < r < scan.range_max)
        if min_range < 0.5:
            cmd_topic.send(CmdVel(linear=0.0, angular=0.5))
        else:
            cmd_topic.send(CmdVel(linear=1.0, angular=0.0))

avoider = horus.Node(name="obstacle_avoider", tick=avoider_tick,
                     rate=30, order=0,
                     subs=["scan"], pubs=["cmd_vel"])
horus.run(avoider)