Python Examples

Complete Python examples demonstrating HORUS capabilities.

Available Examples

Example	Description	Key Features
Basic Node	Simple sensor-to-motor node	Node, `recv()`, `send()`
Typed Topics	Direct pub/sub with typed messages	Topic, CmdVel, LaserScan
Async Node	I/O-bound work with async/await	`Node(tick=async def)`, `asyncio`
ML Inference	ONNX inference over a Tensor topic	`horus.Tensor`, `onnxruntime`
Cross-Language	Python-Rust communication	Typed topics, shared memory
Multi-Node System	Scheduler with multiple nodes	Scheduler, run()
Sensor Processing	LaserScan obstacle avoidance	Topic, LaserScan, CmdVel
Camera Images	Send and receive frames	`horus.Image`, numpy/torch interop

Basic Node

A minimal node that reads sensor data and publishes motor commands:

from horus import Node, run

def controller(node):
    """Simple obstacle avoidance"""
    if node.has_msg("sensor.distance"):
        distance = node.recv("sensor.distance")
        if distance < 0.5:
            node.send("motor.cmd", {"linear": 0.0, "angular": 0.5})
        else:
            node.send("motor.cmd", {"linear": 1.0, "angular": 0.0})

node = Node(
    name="obstacle_avoider",
    subs=["sensor.distance"],
    pubs=["motor.cmd"],
    tick=controller,
    rate=10
)

run(node)

Typed Pub/Sub

Using typed Topic for direct pub/sub with message classes:

from horus import Topic, CmdVel, LaserScan

# Create typed topics
scan_topic = Topic(LaserScan)
cmd_topic = Topic(CmdVel)

# Send a velocity command
cmd_topic.send(CmdVel(linear=1.0, angular=0.0))

# Receive laser scan (returns None if no message)
scan = scan_topic.recv()
if scan:
    print(f"Got {len(scan)} valid range readings")
    # scan.ranges is a fixed 360-slot array, so min() over it would pick up
    # unfilled slots — min_range() returns the nearest *valid* reading.
    print(f"Closest obstacle: {scan.min_range():.2f}m")

Backend selection is automatic — there is no backend= argument. Every topic is shared-memory backed; HORUS reads the topic's publisher/subscriber counts and whether the message is a fixed-size POD, then picks the ring: SpscShm (~85ns) for one publisher and one subscriber, MpscShm (~65ns) for many publishers into one subscriber, PodShm (~50ns) to broadcast a POD type to many subscribers, and FanoutShm (~40ns) for non-POD broadcast and many-to-many. The only tuning knobs are capacity and endpoint:

from horus import Topic, CmdVel

# Deeper ring buffer
buffered_topic = Topic(CmdVel, capacity=2048)

# Explicit endpoint — overrides the shared-memory topic name.
# This is NOT a network address: Python topics are shared-memory only.
named_topic = Topic(CmdVel, endpoint="cmd_vel")

# Read back which backend was chosen
print(buffered_topic.backend_type)

Async Node

There is no AsyncNode class — pass an async def function as a node's tick and HORUS runs it on its async I/O executor. Pair a slow async producer with a fast sync consumer:

import asyncio
import horus

async def fetch_weather(node):
    """I/O-bound: await the network instead of blocking on it."""
    await asyncio.sleep(0.2)             # stands in for an aiohttp GET
    node.send("weather", {"temp": 21.5})

def log_weather(node):
    """Sync tick — keeps its own rate no matter how slow the fetch is."""
    if node.has_msg("weather"):
        weather = node.recv("weather")
        node.log_info(f"Temperature: {weather['temp']}C")

horus.run(
    # rate= sets the polling interval; there is no horus.sleep().
    horus.Node(name="fetcher", tick=fetch_weather, rate=2, order=0, pubs=["weather"]),
    horus.Node(name="logger", tick=log_weather, rate=10, order=1, subs=["weather"]),
    duration=2.0,
)

An async tick blocks until its coroutine finishes, so rate is an upper bound for that node. The sync logger is unaffected and keeps its full 10 Hz. See Async Nodes for the details.

ML Inference

HORUS ships no model-loading or inference helpers — horus.ml_utils does not exist. Inference is your own onnxruntime / torch code; HORUS carries the data. Use horus.Tensor so frames cross shared memory without a copy.

import numpy as np
import horus

session = None

def load_model(node):
    """Runs once, before the first tick."""
    global session
    import onnxruntime
    session = onnxruntime.InferenceSession("pose.onnx")

def estimate(node):
    frame = node.recv("camera.raw")
    if frame is None:
        return
    # frame is a horus.Tensor: view it as numpy without copying
    img = frame.numpy()
    outputs = session.run(None, {"input": img[None, ...]})
    node.send("poses", horus.Tensor.from_numpy(np.asarray(outputs[0])))

node = horus.Node(
    name="pose_estimation",
    tick=estimate,
    init=load_model,
    subs=[horus.Sub("camera.raw", horus.Tensor)],
    pubs=[horus.Pub("poses", horus.Tensor)],
    rate=30,
    compute=True,   # inference is CPU-bound
)

Run the subscriber at the publisher's rate. A Tensor topic drops almost everything for a subscriber ticking slower than the publisher — at 30 Hz publish / 10 Hz subscribe, recv() returns None on every tick. See Python Tensors for ML.

Cross-Language Communication

Python and Rust nodes communicate via typed topics through shared memory (same machine):

from horus import Topic, CmdVel
import time

# Create typed topic — shared memory, readable by Rust
topic = Topic(CmdVel)

# Publish velocity commands
while True:
    t = time.time()
    linear = 1.0 + 0.5 * (t % 10) / 10.0
    angular = 0.2 * ((t % 20) - 10) / 10.0

    topic.send(CmdVel(linear=linear, angular=angular))
    time.sleep(0.1)

Rust subscriber (in another process on the same machine):

use horus::prelude::*;

let topic: Topic<CmdVel> = Topic::new("cmd_vel")?;
loop {
    if let Some(cmd) = topic.recv() {
        println!("linear={}, angular={}", cmd.linear, cmd.angular);
    }
}

Multi-Node System

Running multiple nodes with the Scheduler:

from horus import Node, Scheduler

# Sensor node - reads and publishes data
def sensor_tick(node):
    # Simulated sensor reading
    node.send("sensor.distance", 1.5)

# Controller node - processes sensor data, outputs commands
def controller_tick(node):
    if node.has_msg("sensor.distance"):
        dist = node.recv("sensor.distance")
        if dist < 0.5:
            node.send("cmd_vel", {"linear": 0.0, "angular": 0.5})
        else:
            node.send("cmd_vel", {"linear": 1.0, "angular": 0.0})

# Logger node - records data
def logger_tick(node):
    if node.has_msg("cmd_vel"):
        cmd = node.recv("cmd_vel")
        print(f"Command: {cmd}")

# Execution order is set on the Node, not on scheduler.add()
sensor = Node(name="sensor", pubs=["sensor.distance"], tick=sensor_tick, rate=30, order=0)
controller = Node(name="controller", subs=["sensor.distance"], pubs=["cmd_vel"], tick=controller_tick, rate=30, order=1)
logger = Node(name="logger", subs=["cmd_vel"], tick=logger_tick, rate=10, order=2)

scheduler = Scheduler()
scheduler.add(sensor)
scheduler.add(controller)
scheduler.add(logger)
scheduler.run()

Or use the run() helper for quick prototyping:

from horus import Node, run

node = Node(
    name="echo",
    subs=["input"],
    pubs=["output"],
    tick=lambda n: n.send("output", n.recv("input")) if n.has_msg("input") else None,
    rate=30
)

run(node, duration=10)  # Run for 10 seconds

Sensor Processing Pipeline

Processing typed sensor data and publishing commands:

from horus import Node, Scheduler, Topic, LaserScan, CmdVel

scan_topic = None
cmd_topic = None

def init(node):
    global scan_topic, cmd_topic
    scan_topic = Topic(LaserScan)
    cmd_topic = Topic(CmdVel)

def obstacle_avoidance(node):
    scan = scan_topic.recv()
    if scan and scan.ranges:
        min_dist = min(r for r in scan.ranges if r > scan.range_min)
        if min_dist < 0.5:
            # Too close — turn away
            cmd_topic.send(CmdVel(linear=0.0, angular=0.5))
        else:
            # Clear ahead — drive forward
            cmd_topic.send(CmdVel(linear=1.0, angular=0.0))

node = Node(
    name="obstacle_avoider",
    tick=obstacle_avoidance,
    init=init,
    rate=10,
    order=0
)

scheduler = Scheduler()
scheduler.add(node)
scheduler.run()

Camera Image Pipeline

Send and receive camera images using the Image domain type with zero-copy shared memory.

Camera Sender

import horus
import numpy as np

# Create a 480x640 RGB8 image backed by shared memory
img = horus.Image(480, 640, "rgb8")

# Or build one from a NumPy array (copies the pixels into shared memory)
pixels = np.zeros((480, 640, 3), dtype=np.uint8)
pixels[:, :, 2] = 255  # Blue channel
img = horus.Image.from_numpy(pixels)

# Send over a topic
topic = horus.Topic(horus.Image, endpoint="camera/rgb")
topic.send(img)

Camera Receiver

import horus

topic = horus.Topic(horus.Image, endpoint="camera/rgb")

img = topic.recv()
if img is not None:
    # Convert to NumPy for processing (zero-copy)
    arr = img.to_numpy()
    print(f"Received {arr.shape[1]}x{arr.shape[0]} image")

    # Convert to PyTorch tensor (zero-copy via the array interface)
    tensor = img.as_tensor().torch()
    print(f"Torch tensor: {tensor.shape}, {tensor.dtype}")

Key Concepts:

horus.Image(height, width, encoding) — allocates from shared memory
Image.from_numpy(arr) — copy a NumPy array into a new shared-memory Image
img.to_numpy() / img.as_tensor().torch() — zero-copy conversion to frameworks
Same Topic API as Rust — Python and Rust processes share topics automatically