HORUS Python Bindings

# Install maturin (Python/Rust build tool)
# Option A: Via Cargo (recommended for Ubuntu 24.04+)
cargo install maturin

# Option B: Via pip (if not blocked by PEP 668)
# pip install maturin

# Build and install from source
cd horus_py
maturin develop --release

import horus

def process(node):
    node.send("output", "Hello HORUS!")

node = horus.Node(pubs="output", tick=process, rate=1)
horus.run(node, duration=3)

node = horus.Node(
    name="my_node",            # Optional: auto-generated if not provided
    pubs=["topic1", "topic2"], # Topics to publish to
    subs=["input1", "input2"], # Topics to subscribe to
    tick=my_function,          # Function called repeatedly
    rate=30,                   # Hz (default: 30)
    init=setup_fn,             # Optional: called once at start
    shutdown=cleanup_fn,       # Optional: called once at end
    on_error=error_fn,         # Optional: called on tick errors
    default_capacity=1024      # Optional: topic buffer capacity
)

import horus

class SensorNode(horus.Node):
    def __init__(self):
        super().__init__(
            name="sensor",
            pubs=["temperature"],
            rate=10
        )

    def tick(self, info=None):
        # Override tick method
        self.send("temperature", 25.0)

    def init(self, info=None):
        # Optional: override init
        print("Sensor initialized!")

    def shutdown(self, info=None):
        # Optional: override shutdown
        print("Sensor shutting down!")

# Use it
sensor = SensorNode()
horus.run(sensor)

def my_tick(node):
    # Check for messages
    if node.has_msg("input"):
        data = node.recv("input")  # Get one message

    # Get all messages
    all_msgs = node.recv_all("input")

    # Send messages
    node.send("output", {"value": 42})

# Single node
horus.run(node)

# Multiple nodes
horus.run(node1, node2, node3, duration=10)

import horus

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

sensor = horus.Node(
    name="temp_sensor",
    pubs="temperature",
    tick=publish_temperature,
    rate=1  # 1 Hz
)

horus.run(sensor, duration=10)

import horus

def display_temperature(node):
    if node.has_msg("temperature"):
        temp = node.recv("temperature")
        print(f"Temperature: {temp}°C")

display = horus.Node(
    name="display",
    subs="temperature",
    tick=display_temperature
)

horus.run(display)

import horus

def publish(node):
    node.send("raw", 42.0)

def process(node):
    if node.has_msg("raw"):
        data = node.recv("raw")
        result = data * 2.0
        node.send("processed", result)

def display(node):
    if node.has_msg("processed"):
        value = node.recv("processed")
        print(f"Result: {value}")

# Create pipeline
publisher = horus.Node("publisher", pubs="raw", tick=publish, rate=1)
processor = horus.Node("processor", subs="raw", pubs="processed", tick=process)
displayer = horus.Node("display", subs="processed", tick=display)

# Run all together
horus.run(publisher, processor, displayer, duration=5)

import horus

# Producer (inline)
producer = horus.Node(
    pubs="numbers",
    tick=lambda n: n.send("numbers", 42),
    rate=1
)

# Transformer (inline)
doubler = horus.Node(
    subs="numbers",
    pubs="doubled",
    tick=lambda n: n.send("doubled", n.recv("numbers") * 2) if n.has_msg("numbers") else None
)

horus.run(producer, doubler, duration=5)

import horus

def robot_controller(node):
    # Read from multiple sensors
    lidar_data = None
    camera_data = None

    if node.has_msg("lidar"):
        lidar_data = node.recv("lidar")

    if node.has_msg("camera"):
        camera_data = node.recv("camera")

    # Compute commands
    if lidar_data and camera_data:
        cmd = compute_navigation(lidar_data, camera_data)
        node.send("motors", cmd)
        node.send("status", "navigating")

robot = horus.Node(
    name="robot_controller",
    subs=["lidar", "camera"],
    pubs=["motors", "status"],
    tick=robot_controller,
    rate=50  # 50Hz control loop
)

import horus

class Context:
    def __init__(self):
        self.count = 0
        self.file = None

ctx = Context()

def init_handler(node):
    print("Starting up!")
    ctx.file = open("data.txt", "w")

def tick_handler(node):
    ctx.count += 1
    data = f"Tick {ctx.count}"
    node.send("data", data)
    ctx.file.write(data + "\n")

def shutdown_handler(node):
    print(f"Processed {ctx.count} messages")
    ctx.file.close()

node = horus.Node(
    pubs="data",
    init=init_handler,
    tick=tick_handler,
    shutdown=shutdown_handler,
    rate=10
)

horus.run(node, duration=5)

# Real-time mode is a scheduler-wide setting
scheduler = horus.Scheduler(rt=True)

from horus import us

motor_ctrl = horus.Node(name="motor", tick=drive, order=0, rate=1000, budget=300 * us)
planner    = horus.Node(name="planner", tick=plan, order=5, compute=True)
telemetry  = horus.Node(name="telem", tick=report, order=10, rate=1)

scheduler = horus.Scheduler(rt=True)
scheduler.add(motor_ctrl).add(planner).add(telemetry)

import horus

class MotorNode(horus.Node):
    def __init__(self):
        super().__init__(name="motor", rate=1000,
                         budget=300e-6, deadline=900e-6,
                         on_miss="safe_mode")
        self.pwm = 0.0

    def tick(self, info=None):
        self.pwm = self.compute_pwm()

    def compute_pwm(self):
        return 0.0

    def enter_safe_state(self):
        self.pwm = 0.0  # motors off

# Quick helper — creates a Scheduler, adds each node, and runs it
horus.run(sensor, controller, logger, duration=10)
horus.run(node, verbose=False)  # Quieter output

import horus

def sensor_tick(node):
    # The publisher stamps the message
    node.send("sensor_data", horus.CmdVel(1.0, 0.0, timestamp_ns=horus.timestamp_ns()))

def control_tick(node):
    msg = node.recv("sensor_data")
    if msg is None:
        return

    # Check message age
    age = (horus.timestamp_ns() - msg.timestamp_ns) / 1e9
    if age > 0.1:  # More than 100ms old
        node.log_warning(f"Stale data: {age*1000:.1f}ms old")
        return

    # Process fresh data
    process(msg)

# timestamp_ns lives on the typed message, so declare the topic with its type
sensor = horus.Node(name="sensor", pubs={"sensor_data": horus.CmdVel},
                    tick=sensor_tick, rate=20, order=0)
control = horus.Node(name="control", subs={"sensor_data": horus.CmdVel},
                     tick=control_tick, rate=20, order=1)
horus.run(sensor, control, duration=5)

# Run multiple Python files as separate processes
horus run node1.py node2.py node3.py

# Mix Python and Rust nodes
horus run sensor.rs controller.py visualizer.py

# Mix Rust and Python
horus run lidar_driver.rs planner.py motor_control.rs

# sensor_node.py
import horus

def sensor_tick(node):
    data = read_lidar()  # Your sensor code
    node.send("lidar_data", data)

sensor = horus.Node(name="lidar", pubs="lidar_data", tick=sensor_tick)
horus.run(sensor)

# controller_node.py
import horus

def control_tick(node):
    if node.has_msg("lidar_data"):
        data = node.recv("lidar_data")
        cmd = compute_control(data)
        node.send("motor_cmd", cmd)

controller = horus.Node(
    name="controller",
    subs="lidar_data",
    pubs="motor_cmd",
    tick=control_tick
)
horus.run(controller)

# Run both in separate processes
horus run sensor_node.py controller_node.py

import horus

def sensor_tick(node):
    """High-frequency sensor (100Hz)"""
    # The publisher stamps the message — nothing stamps it for you
    imu = {"accel_x": 1.0, "accel_y": 0.0, "accel_z": 9.8,
           "timestamp_ns": horus.timestamp_ns()}
    node.send("imu_data", imu)
    node.log_info("Published IMU")

def control_tick(node):
    """Medium-frequency control (50Hz)"""
    if node.has_msg("imu_data"):
        imu = node.recv("imu_data")

        # Check for stale data
        age = (horus.timestamp_ns() - imu["timestamp_ns"]) / 1e9
        if age > 0.05:
            node.log_warning("Stale IMU data!")
            return

        cmd = {"linear": 1.0, "angular": 0.0, "timestamp_ns": horus.timestamp_ns()}
        node.send("cmd_vel", cmd)

def logger_tick(node):
    """Low-frequency logging (10Hz)"""
    if node.has_msg("cmd_vel"):
        msg = node.recv("cmd_vel")
        latency = (horus.timestamp_ns() - msg["timestamp_ns"]) / 1e6
        node.log_info(f"Command latency: {latency:.1f}ms")

# Create nodes — rate and execution order live on the constructor
sensor = horus.Node(name="imu", pubs="imu_data", tick=sensor_tick, order=0, rate=100)
controller = horus.Node(name="ctrl", subs="imu_data", pubs="cmd_vel", tick=control_tick, order=1, rate=50)
logger = horus.Node(name="log", subs="cmd_vel", tick=logger_tick, order=2, rate=10)

# add() takes the node and nothing else, and returns the scheduler so calls chain
scheduler = horus.Scheduler()
scheduler.add(sensor).add(controller).add(logger)

scheduler.run(duration=5.0)

# Check statistics
stats = scheduler.get_node_stats("imu")
print(f"Sensor: {stats['total_ticks']} ticks in 5 seconds")

from horus import Topic, CmdVel

# Shared memory, name taken from the message type
local_topic = Topic(CmdVel)

# Shared memory under an explicit name
named_topic = Topic(CmdVel, endpoint="cmdvel")

from horus import Topic, CmdVel

topic = Topic(CmdVel)

# Send and receive typed messages
topic.send(CmdVel(linear=1.0, angular=0.5))
msg = topic.recv()
if msg:
    print(f"linear={msg.linear}, angular={msg.angular}")

# Check topic properties
print(f"Name: {topic.name}")           # "cmd_vel"
print(f"Backend: {topic.backend_type}") # e.g. "SpmcShm"
print(f"Stats: {topic.stats()}")

from horus import Topic, CmdVel

# Generic topic (string name = dynamic typing)
topic = Topic("my_topic")

# Typed topic (class = static typing, better performance)
typed_topic = Topic(CmdVel)

from horus import Topic

topic = Topic("sensor_data")

# Send dict, list, or any JSON-serializable data
topic.send({"temperature": 25.5, "humidity": 60.0})
topic.send([1.0, 2.0, 3.0, 4.0])
topic.send("status: OK")

# Receive (returns Python object)
msg = topic.recv()  # {"temperature": 25.5, "humidity": 60.0}

# Check if generic
print(topic.is_generic())  # True

from horus import Topic, CmdVel

# Backend is chosen for you
topic = Topic(CmdVel)

# Custom ring buffer capacity
big = Topic(CmdVel, capacity=4096)

# On each machine: the replicator runs alongside your nodes and mirrors topics over UDP
horus run --net

import horus
import numpy as np

def process_array(node):
    if node.has_msg("raw_data"):
        data = node.recv("raw_data")
        # Convert to NumPy array
        arr = np.array(data)
        # Process with NumPy
        result = np.fft.fft(arr)
        # np.fft returns complex128 — a generic topic cannot carry Python
        # `complex` ("unsupported type complex"), so send magnitudes instead
        node.send("fft_result", np.abs(result).tolist())

processor = horus.Node(
    subs="raw_data",
    pubs="fft_result",
    tick=process_array
)

import horus
import cv2
import numpy as np

def process_image(node):
    if node.has_msg("camera"):
        frame = node.recv("camera")          # horus.Image
        img = frame.to_numpy()               # zero-copy view, (H, W, 3) uint8

        # Apply OpenCV processing
        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
        edges = cv2.Canny(gray, 50, 150)

        # Publish result as an Image, not a list
        node.send("edges", horus.Image.from_numpy(edges))

vision = horus.Node(
    subs={"camera": {"type": horus.Image}},
    pubs={"edges": {"type": horus.Image}},
    tick=process_image,
    rate=30
)

import horus
from sklearn.linear_model import LinearRegression
import numpy as np

model = LinearRegression()

def train_model(node):
    if node.has_msg("training_data"):
        data = node.recv("training_data")
        X = np.array(data['features'])
        y = np.array(data['labels'])

        # Train model
        model.fit(X, y)
        score = model.score(X, y)

        node.send("model_score", score)

trainer = horus.Node(
    subs="training_data",
    pubs="model_score",
    tick=train_model
)

import horus

class RobotState:
    def __init__(self):
        self.position = {"x": 0.0, "y": 0.0}
        self.velocity = 0.0
        self.last_update = 0

state = RobotState()

def update_state(node):
    if node.has_msg("velocity"):
        state.velocity = node.recv("velocity")

    if node.has_msg("position"):
        state.position = node.recv("position")

    # Publish combined state
    node.send("robot_state", {
        "pos": state.position,
        "vel": state.velocity
    })

state_manager = horus.Node(
    subs=["velocity", "position"],
    pubs="robot_state",
    tick=update_state
)

import horus
import time

class RateLimiter:
    def __init__(self, min_interval):
        self.min_interval = min_interval
        self.last_send = 0

limiter = RateLimiter(min_interval=0.1)  # 100ms minimum

def rate_limited_publish(node):
    current_time = time.time()

    if current_time - limiter.last_send >= limiter.min_interval:
        node.send("output", "data")
        limiter.last_send = current_time

node = horus.Node(
    pubs="output",
    tick=rate_limited_publish,
    rate=100  # Node runs at 100Hz, but publishes at max 10Hz
)

import horus

def safe_processing(node):
    try:
        if node.has_msg("input"):
            data = node.recv("input")
            result = risky_operation(data)
            node.send("output", result)
    except Exception as e:
        node.send("errors", str(e))
        print(f"Error: {e}")

processor = horus.Node(
    subs="input",
    pubs=["output", "errors"],
    tick=safe_processing
)

# Per-node rate and order are set on the Node constructor
sensor     = horus.Node(name="sensor", tick=sensor_tick, order=0, rate=100)  # 100Hz
controller = horus.Node(name="ctrl", tick=control_tick, order=1, rate=50)    # 50Hz
logger     = horus.Node(name="log", tick=logger_tick, order=2, rate=10)      # 10Hz

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

# Monitor performance with get_node_stats()
stats = scheduler.get_node_stats("sensor")
print(f"Sensor executed {stats['total_ticks']} ticks")

# "sensor_data" is a typed topic and its publisher set timestamp_ns
def control_tick(node):
    data = node.recv("sensor_data")
    if data is None:
        return

    # Skip stale data to maintain real-time performance
    age = (horus.timestamp_ns() - data.timestamp_ns) / 1e9
    if age > 0.1:
        node.log_warning("Skipping stale sensor data")
        return

    # Process fresh data only
    process(data)

# Send messages as Python dicts (serialized via MessagePack)
cmd = {"linear": 1.5, "angular": 0.8}
node.send("cmd_vel", cmd)

# Use node.recv_all() to process all available messages at once
def batch_processor(node):
    messages = node.recv_all("input")
    if messages:
        results = [process(msg) for msg in messages]
        for result in results:
            node.send("output", result)

# GOOD: Fast tick
def good_tick(node):
    if node.has_msg("input"):
        data = node.recv("input")
        result = quick_operation(data)
        node.send("output", result)

# BAD: Slow tick
def bad_tick(node):
    time.sleep(1)  # Don't block!
    data = requests.get("http://api.example.com")  # Don't do I/O!

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

def heavy_processing_node(node):
    if node.has_msg("input"):
        data = node.recv("input")
        # Offload to thread pool
        future = executor.submit(expensive_operation, data)
        # Don't block - check result later or use callback

# Isolate heavy processing in separate processes
horus run sensor.py heavy_vision.py light_controller.py

# Each node gets its own CPU core

# Debug build (fast compile, slow runtime)
cd horus_py
maturin develop

# Release build (slow compile, fast runtime)
maturin develop --release

# Build wheel for distribution
maturin build --release

# Install test dependencies
pip install pytest

# Run all tests
pytest tests/

# Run specific feature tests
pytest tests/test_rate_control.py            # Per-node rates
pytest tests/test_timestamps.py              # Timestamps
pytest tests/test_message_integrity.py       # Typed messages
pytest tests/test_cross_language_parity.py   # Typed messages across languages

# With coverage
pytest --cov=horus tests/

# If Rust bindings aren't available, automatically falls back to mock
# You'll see: RuntimeWarning: horus Rust extension not available — running in mock
# mode. Most functionality will not work. Install with: maturin develop

# Use for unit testing Python logic without HORUS running

# Add nodes to a scheduler (all per-node config lives on Node(...))
scheduler = horus.Scheduler()
scheduler.add(my_node)

# Check node statistics
stats = scheduler.get_node_stats("my_node")
print(f"Ticks: {stats['total_ticks']}, Errors: {stats['errors_count']}")

# Monitor message timestamps (typed topic; the publisher must have set timestamp_ns)
msg = node.recv("topic")
if msg:
    age = (horus.timestamp_ns() - msg.timestamp_ns) / 1e9
    print(f"Message age: {age*1000:.1f}ms")

# Python node with typed topic
from horus import Topic, CmdVel

cmd_topic = Topic(CmdVel)  # Typed topic
cmd_topic.send(CmdVel(linear=1.0, angular=0.5))

// Rust node receives
use horus::prelude::*;

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

# Generic Topic - for custom topics
from horus import Topic

topic = Topic("my_topic")  # Pass string for generic topic
topic.send({"linear": 1.0, "angular": 0.5})  # JSON-shaped value, MessagePack on the wire

# Producer
producer = horus.Node(
    pubs="queue",
    tick=lambda n: n.send("queue", generate_work())
)

# Consumer
consumer = horus.Node(
    subs="queue",
    tick=lambda n: process_work(n.recv("queue")) if n.has_msg("queue") else None
)

horus.run(producer, consumer)

def request_node(node):
    node.send("requests", {"id": 1, "query": "data"})

def response_node(node):
    if node.has_msg("requests"):
        req = node.recv("requests")
        response = handle_request(req)
        node.send("responses", response)

req = horus.Node(pubs="requests", tick=request_node)
res = horus.Node(subs="requests", pubs="responses", tick=response_node)

import time

class PeriodicTask:
    def __init__(self, interval):
        self.interval = interval
        self.last_run = 0

task = PeriodicTask(interval=5.0)  # Every 5 seconds

def periodic_tick(node):
    current = time.time()
    if current - task.last_run >= task.interval:
        node.send("periodic", "task_executed")
        task.last_run = current

node = horus.Node(pubs="periodic", tick=periodic_tick, rate=10)

# If you see: ModuleNotFoundError: No module named 'horus'
# Rebuild and install:
cd horus_py
maturin develop --release

# Use release build (not debug)
maturin develop --release

# Check tick rate isn't too high
node = horus.Node(tick=fn, rate=30)  # 30Hz is reasonable

# Avoid accumulating data in closures
# BAD:
all_data = []
def bad_tick(node):
    all_data.append(node.recv("input"))  # Memory leak!

# GOOD:
def good_tick(node):
    data = node.recv("input")
    process_and_discard(data)  # Process immediately

def tick(node):
    node.log_info("Processing sensor data")
    node.log_warning("Sensor reading is stale")
    node.log_error("Failed to process data")
    node.log_debug("Debug information")

def tick(node):
    if node.has_msg("input"):
        data = node.recv("input")
        print(f"[{node.name}] Received: {data}")
        result = process(data)  # your processing
        node.send("output", result)
        print(f"[{node.name}] Published: {result}")

scheduler = horus.Scheduler()
scheduler.add(node)
scheduler.run(duration=10)

# Get stats after running
stats = scheduler.get_node_stats("my_node")
print(f"Ticks: {stats['total_ticks']}")
print(f"Errors: {stats['errors_count']}")