HORUS Python Bindings

# From HORUS root directory
./install.sh

# 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.get("input")  # Get one message

    # Get all messages
    all_msgs = node.get_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.get("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.get("raw")
        result = data * 2.0
        node.send("processed", result)

def display(node):
    if node.has_msg("processed"):
        value = node.get("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.get("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.get("lidar")

    if node.has_msg("camera"):
        camera_data = node.get("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)

# Fluent API — configure nodes inline with chained calls
scheduler = horus.Scheduler()
scheduler.node(motor_ctrl).order(0).budget(200).rate(1000.0).build()
scheduler.node(planner).order(5).compute().build()
scheduler.node(telemetry).order(10).async_io().rate(1.0).build()

scheduler.add(sensor, order=0, rate=100.0)
scheduler.add(controller, order=1, budget=200, on_miss="safe_mode", deadline=5.0)
scheduler.add(logger, order=2, rate=10.0)
scheduler.add(monitor, order=3, failure_policy="skip")

# Quick helper — creates a Scheduler, auto-assigns execution order, and runs
horus.run(sensor, controller, logger, duration=10)
horus.run(node)  # Single node, no duration limit

from horus import Miss

# Available policies
Miss.WARN        # Log warning and continue (default)
Miss.SKIP        # Skip the node for this tick
Miss.SAFE_MODE   # Call enter_safe_state() on the node
Miss.STOP        # Stop the entire scheduler

# Via add()
scheduler.add(motor, order=0, budget=200, on_miss="safe_mode")

# Via fluent builder
scheduler.node(motor).order(0).budget(200).on_miss("safe_mode").build()

from horus import Scheduler, SchedulerConfig

# Development — lightweight
scheduler = Scheduler()

# Production — watchdog + monitoring
from horus._horus import SchedulerConfig
cfg = SchedulerConfig.with_watchdog()
cfg.tick_rate = 1000.0
scheduler = Scheduler(config=cfg)

# With blackbox
cfg = SchedulerConfig.with_watchdog()
cfg.tick_rate = 1000.0
cfg = cfg.blackbox_mb(64)
scheduler = Scheduler(config=cfg)

# Single-tick execution for unit tests and simulation
scheduler = Scheduler()
scheduler.add(sensor, order=0)
scheduler.add(controller, order=1)

# Run exactly one tick cycle — no loop, no sleep
scheduler.tick_once()

# Or tick specific nodes
scheduler.tick(["sensor"])

import horus
import time

def control_tick(node):
    if node.has_msg("sensor_data"):
        # Check message age
        age = node.get_message_age("sensor_data")
        if age > 0.1:  # More than 100ms old
            node.log_warning(f"Stale data: {age*1000:.1f}ms old")
            return

        # Or use built-in staleness detection
        if node.is_stale("sensor_data", max_age=0.1):
            return  # Skip stale data

        # Get message with timestamp
        msg, timestamp = node.get_with_timestamp("sensor_data")
        latency = time.time() - timestamp
        print(f"Latency: {latency*1000:.1f}ms")

        # Process fresh data
        data = node.get("sensor_data")
        process(data)

# 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.get("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
import time

def sensor_tick(node):
    """High-frequency sensor (100Hz)"""
    imu = {"accel_x": 1.0, "accel_y": 0.0, "accel_z": 9.8}
    node.send("imu_data", imu)
    age = node.get_message_age("imu_data")
    node.log_info(f"Published IMU (age: {age:.3f}s)")

def control_tick(node):
    """Medium-frequency control (50Hz)"""
    if node.has_msg("imu_data"):
        # Check for stale data
        if node.is_stale("imu_data", max_age=0.05):
            node.log_warning("Stale IMU data!")
            return

        imu = node.get("imu_data")
        cmd = {"linear": 1.0, "angular": 0.0}
        node.send("cmd_vel", cmd)

def logger_tick(node):
    """Low-frequency logging (10Hz)"""
    if node.has_msg("cmd_vel"):
        msg, timestamp = node.get_with_timestamp("cmd_vel")
        latency = (time.time() - timestamp) * 1000
        node.log_info(f"Command latency: {latency:.1f}ms")

# Create nodes
sensor = horus.Node(name="imu", pubs="imu_data", tick=sensor_tick)
controller = horus.Node(name="ctrl", subs="imu_data", pubs="cmd_vel", tick=control_tick)
logger = horus.Node(name="log", subs="cmd_vel", tick=logger_tick)

# Configure with different rates and priorities
scheduler = horus.Scheduler()
scheduler.add(sensor, order=0, rate=100.0)
scheduler.add(controller, order=1, rate=50.0, budget=500)
scheduler.add(logger, order=2, rate=10.0)

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

# Local (shared memory) - default
local_topic = Topic(CmdVel)

# Network (UDP direct)
network_topic = Topic(CmdVel, endpoint="cmdvel@192.168.1.100:8000")

# Router (TCP broker for WAN/NAT traversal)
router_topic = Topic(CmdVel, endpoint="cmdvel@router")

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. "mpmc_shm"
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

# Auto-detected (default) — HORUS picks the best backend
topic = Topic(CmdVel)

# Explicit backend selection
fast = Topic(CmdVel, backend="direct")      # ~3ns, same-thread callbacks
spsc = Topic(CmdVel, backend="spsc")        # ~18ns, single producer/consumer
mpmc = Topic(CmdVel, backend="mpmc")        # ~36ns, multiple producers/consumers
shm  = Topic(CmdVel, backend="mpmc_shm")   # ~167ns, cross-process MPMC

from horus import RouterClient, Topic, CmdVel

# Create router client for explicit connection management
router = RouterClient("192.168.1.100", 7777)

# Build endpoints through the router
cmd_endpoint = router.endpoint("cmdvel")  # Returns "cmdvel@192.168.1.100:7777"
pose_endpoint = router.endpoint("pose")

# Use endpoints with Topic
topic = Topic(CmdVel, endpoint=cmd_endpoint)

# Router properties
print(f"Address: {router.address}")        # "192.168.1.100:7777"
print(f"Connected: {router.is_connected}") # True
print(f"Topics: {router.topics}")          # ["cmdvel", "pose"]
print(f"Uptime: {router.uptime_seconds}s")

from horus import default_router_endpoint, router_endpoint

# Default router (localhost:7777)
ep1 = default_router_endpoint("cmdvel")  # "cmdvel@router"

# Custom router address
ep2 = router_endpoint("cmdvel", "192.168.1.100", 7777)  # "cmdvel@192.168.1.100:7777"

from horus import RouterServer

# Start a local router (for development/testing)
server = RouterServer(port=7777)
server.start()

# For production, use CLI instead:
# $ horus router start --port 7777

# === ROBOT (192.168.1.50) ===
from horus import Topic, CmdVel, Imu, Odometry

# Local: Critical flight control (ultra-fast)
imu_topic = Topic(Imu)  # ~85ns local shared memory

# Network: Telemetry to ground station
telemetry = Topic(Odometry, endpoint="telem@192.168.1.100:8000")

# Network: Commands from ground station
commands = Topic(CmdVel, endpoint="cmd@0.0.0.0:8001")


# === GROUND STATION (192.168.1.100) ===
from horus import Topic, CmdVel, Odometry

# Receive telemetry from robot
telemetry_sub = Topic(Odometry, endpoint="telem@0.0.0.0:8000")

# Send commands to robot
command_pub = Topic(CmdVel, endpoint="cmd@192.168.1.50:8001")

import horus
import numpy as np

def process_array(node):
    if node.has_msg("raw_data"):
        data = node.get("raw_data")
        # Convert to NumPy array
        arr = np.array(data)
        # Process with NumPy
        result = np.fft.fft(arr)
        node.send("fft_result", 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"):
        img_data = node.get("camera")
        # Convert to OpenCV format
        img = np.array(img_data, dtype=np.uint8).reshape((480, 640, 3))

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

        # Publish result
        node.send("edges", edges.flatten().tolist())

vision = horus.Node(
    subs="camera",
    pubs="edges",
    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.get("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.get("velocity")

    if node.has_msg("position"):
        state.position = node.get("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.get("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
)

# NEW: Use scheduler with per-node rates for optimal performance
scheduler = horus.Scheduler()

# High-frequency sensor (100Hz)
scheduler.add(sensor, order=0, rate=100.0)

# Medium-frequency control (50Hz)
scheduler.add(controller, order=1, rate=50.0)

# Low-frequency logging (10Hz)
scheduler.add(logger, order=2, rate=10.0)

scheduler.run()

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

def control_tick(node):
    # Skip stale data to maintain real-time performance
    if node.is_stale("sensor_data", max_age=0.1):
        node.log_warning("Skipping stale sensor data")
        return

    # Process fresh data only
    data = node.get("sensor_data")
    process(data)

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

# Check message age using node method
if node.get_message_age("cmd_vel") > 0.1:
    print("Message is stale")

# Use node.get_all() to process all available messages at once
def batch_processor(node):
    messages = node.get_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.get("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.get("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
horus run tests/test_rate_control.py    # Phase 1: Per-node rates
horus run tests/test_timestamps.py      # Phase 2: Timestamps
horus run tests/test_typed_messages.py  # Phase 3: Typed messages

# With coverage
pytest --cov=horus tests/

# Test multiprocess execution (Phase 4)
horus run tests/multiprocess_publisher.py tests/multiprocess_subscriber.py

# If Rust bindings aren't available, automatically falls back to mock
# You'll see: "Warning: Rust bindings not available. Running in mock mode."

# Use for unit testing Python logic without HORUS running

# Check node statistics
scheduler = horus.Scheduler()
scheduler.add(my_node, order=0)

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

# Monitor message timestamps
msg, timestamp = node.get_with_timestamp("topic")
age = time.time() - timestamp
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})  # Uses JSON serialization

# 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.get("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.get("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.get("input"))  # Memory leak!

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

# Python nodes use standard print() for logging
print("Debug message")  # Visible in console, not in monitor

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")

# These print to console, not monitor

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

scheduler = horus.Scheduler()
scheduler.add(node, order=0)
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']}")

from horus import HorusNotFoundError, HorusTransformError, HorusTimeoutError

try:
    result = some_horus_operation()
except HorusNotFoundError:
    print("Resource not found")
except HorusTransformError:
    print("Transform computation failed")
except HorusTimeoutError:
    print("Operation timed out")

try:
    tf = tf_tree.tf("nonexistent", "world")
except HorusNotFoundError as e:
    print(e)  # "Frame not found: nonexistent"

try:
    img = Image(height=-1, width=640, encoding="rgb8")
except ValueError as e:
    print(e)  # "Invalid input: height must be positive"

try:
    result = horus_operation()
except HorusNotFoundError:
    pass  # Specific: missing resource
except HorusTransformError:
    pass  # Specific: TF failure
except HorusTimeoutError:
    pass  # Specific: deadline exceeded
except (ValueError, TypeError):
    pass  # Bad input or serialization
except (IOError, MemoryError):
    pass  # System-level failures
except RuntimeError:
    pass  # Catch-all for internal errors