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)

def Node(
    name: str = "",              # Node name (auto-generated from tick function if empty)
    subs: str | list = "",       # Topics to subscribe (string, list of strings, or list of message types)
    pubs: str | list = "",       # Topics to publish (string, list of strings, or list of message types)
    tick: Callable = None,       # Function called every tick: tick(node) -> None
    init: Callable = None,       # Optional: called once on startup: init(node) -> None
    shutdown: Callable = None,   # Optional: called on graceful shutdown: shutdown(node) -> None
    rate: int = 30,              # Tick rate in Hz
    order: int = 100,            # Execution priority (lower = earlier)
    budget: float = None,        # Max tick duration in seconds (None = auto from rate)
    deadline: float = None,      # Hard deadline in seconds (None = auto from rate)
    on_miss: str = None,         # "warn", "skip", "safe_mode", "stop" (None = "warn")
) -> Node

from horus import Node, CmdVel, LaserScan, Imu

node = Node(
    name="controller",
    pubs=[CmdVel],             # Typed — fast Pod zero-copy (~1.5μs)
    subs=[LaserScan, Imu],     # Auto-named: "scan", "imu"
    tick=control_fn,
    rate=100,
    order=0,
)

# FAST — typed (Pod zero-copy, ~2.7μs send+recv)
pubs=[CmdVel]                 # auto-name from type: "cmd_vel"
pubs=[CmdVel, Pose2D]         # multiple typed topics
pubs={"motor": CmdVel}        # custom name + type

# GENERIC — string (MessagePack, ~10μs send+recv)
pubs=["data"]                 # GenericMessage — for dicts and custom data
pubs="single_topic"           # shorthand for single string topic

import horus

class SensorState:
    def __init__(self):
        self.reading = 0.0

    def tick(self, node):
        self.reading += 0.1
        node.send("temperature", self.reading)

    def init(self, node):
        print("Sensor initialized!")

    def shutdown(self, node):
        print("Sensor shutting down!")

# Use it
state = SensorState()
sensor = horus.Node(name="sensor", tick=state.tick, init=state.init,
                    shutdown=state.shutdown, pubs=["temperature"], rate=10)
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})

def send(topic: str, data: Any) -> None

node.send("cmd_vel", {"linear": 1.0, "angular": 0.0})  # dict
node.send("cmd_vel", horus.CmdVel(1.0, 0.0))           # typed message

def recv(topic: str) -> Optional[Any]

msg = node.recv("scan")
if msg is not None:
    print(f"Got {len(msg.ranges)} ranges")

def tick(node):
    # Process all queued commands (don't drop any)
    commands = node.recv_all("commands")
    for cmd in commands:
        execute_command(cmd)
    node.log_debug(f"Processed {len(commands)} commands this tick")

def tick(node):
    if node.has_msg("emergency_stop"):
        stop = node.recv("emergency_stop")
        node.log_warning("Emergency stop received!")
        node.request_stop()

def tick(node):
    if horus.tick() >= 1000:
        node.log_info("Reached 1000 ticks, stopping")
        node.request_stop()

    error = check_safety()
    if error:
        node.log_error(f"Safety violation: {error}")
        node.request_stop()

def init(node):
    node.log_info(f"Publishing to: {node.publishers()}")
    node.log_info(f"Subscribing to: {node.subscribers()}")

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(f"Raw value: {value}")

# Outside scheduler — message is dropped with RuntimeWarning:
node = horus.Node(name="test", tick=tick)
node.log_info("This will be dropped!")  # RuntimeWarning

# Rust node: microsecond budget
Node(tick=motor_ctrl, rate=1000, budget=300 * us)  # 300μs — Rust can do this

# Python ML node: millisecond budget — detects when inference is too slow
Node(tick=run_model, rate=30, budget=50 * ms)  # 50ms — triggers on_miss if exceeded

def run(*nodes: Node, duration: float = None) -> None

# Single node — runs until Ctrl+C
horus.run(node)

# Multiple nodes for 10 seconds
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.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.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)

from horus import NodeState

# Values:
NodeState.IDLE       # Created but not yet running
NodeState.RUNNING    # Actively ticking
NodeState.PAUSED     # Temporarily suspended
NodeState.STOPPED    # Clean shutdown complete
NodeState.ERROR      # Error state

if node_state == "running":
    print("Node is active")

def Scheduler(
    tick_rate: int = 100,        # Global tick rate in Hz
    rt: bool = False,            # Enable real-time scheduling (prefer_rt)
    watchdog_ms: int = 0,        # Watchdog timeout in ms (0 = disabled)
    deterministic: bool = False, # Deterministic execution mode
    verbose: bool = True,        # Enable verbose logging
) -> Scheduler

# All config on Node(), scheduler.add() takes only the node
scheduler = horus.Scheduler()
scheduler.add(horus.Node(tick=motor_fn, rate=1000, order=0, budget=200))
scheduler.add(horus.Node(tick=planner_fn, order=5, compute=True))
scheduler.add(horus.Node(tick=telemetry_fn, rate=1, order=10))

sensor = horus.Node(name="sensor", tick=sensor_fn, rate=100, order=0)
controller = horus.Node(name="ctrl", tick=ctrl_fn, rate=100, order=1)
logger = horus.Node(name="logger", tick=log_fn, rate=10, order=2)

scheduler.add(sensor)
scheduler.add(controller)
scheduler.add(logger)

with horus.Scheduler(tick_rate=100) as sched:
    sched.add(horus.Node(tick=sensor_fn, rate=100, order=0))
    sched.add(horus.Node(tick=ctrl_fn, rate=100, order=1))
    sched.run(duration=10.0)
# stop() called automatically on exit, even if an exception occurs

stats = scheduler.get_node_stats("motor_ctrl")
print(f"Total ticks: {stats['total_ticks']}")
print(f"Avg tick: {stats.get('avg_tick_duration_ms', 0):.2f} ms")
print(f"Errors: {stats['errors_count']}")

# Slow down logging when battery is low
if battery_low:
    scheduler.set_node_rate("logger", 1)  # 1 Hz
else:
    scheduler.set_node_rate("logger", 10)  # 10 Hz

scheduler.run()              # Run forever (until Ctrl+C)
scheduler.run(duration=30.0) # Run for 30 seconds, then stop

# From another thread or a node's tick:
scheduler.stop()

for node in scheduler.get_all_nodes():
    print(f"{node['name']}: order={node.get('order', '?')}")

print(f"Running {scheduler.get_node_count()} nodes")

if scheduler.has_node("motor_ctrl"):
    stats = scheduler.get_node_stats("motor_ctrl")

print(f"Nodes: {scheduler.get_node_names()}")

if scheduler.remove_node("debug_logger"):
    print("Debug logger removed")

caps = scheduler.capabilities()
print(f"RT priority: {caps.get('max_priority', 'N/A')}")
print(f"Memory lock: {caps.get('memory_locking', False)}")

if not scheduler.has_full_rt():
    print("Warning: running with degraded RT — check capabilities()")

stats = scheduler.safety_stats()
if stats:
    print(f"Deadline misses: {stats.get('deadline_misses', 0)}")
    print(f"Budget overruns: {stats.get('budget_overruns', 0)}")

if scheduler.is_recording():
    paths = scheduler.stop_recording()
    print(f"Saved recordings: {paths}")

recordings = horus.Scheduler.list_recordings()
for r in recordings:
    print(f"  {r}")

scheduler = horus.Scheduler(tick_rate=100)
scheduler.add(sensor)
scheduler.add(controller)

# Test: run one tick, check output
scheduler.tick(["Sensor", "Controller"])
stats = scheduler.get_node_stats("Controller")
assert stats["total_ticks"] == 1

# Benchmark sensor processing for 5 seconds
scheduler.tick_for(["SensorProcessor"], 5.0)
stats = scheduler.get_node_stats("SensorProcessor")
print(f"Processed {stats['total_ticks']} ticks in 5s")

import threading

def monitor_thread(sched):
    while sched.is_running():
        print(f"Tick: {sched.current_tick()}")
        time.sleep(1.0)

t = threading.Thread(target=monitor_thread, args=(scheduler,), daemon=True)
t.start()
scheduler.run(duration=10.0)

order = scheduler.get_node_info("motor_ctrl")
if order is not None:
    print(f"motor_ctrl runs at order {order}")

scheduler = horus.Scheduler(tick_rate=1000, rt=True)
# ... add nodes and run ...

for d in scheduler.degradations():
    print(f"Degraded: {d.get('feature')} — {d.get('reason')}")

from horus import Node, run, us

sensor = Node(tick=read_lidar, rate=10, order=0, pubs=["scan"])
ctrl = Node(tick=navigate, rate=30, order=1, subs=["scan"], pubs=["cmd"])
motor = Node(tick=drive, rate=1000, order=2, budget=300*us, subs=["cmd"])

# All scheduler config as kwargs
run(sensor, ctrl, motor, rt=True, watchdog_ms=500)
run(node, duration=10, deterministic=True)

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

# Config on Node, then add
motor = horus.Node(name="motor", tick=motor_fn, rate=500, order=0, budget=200, on_miss="safe_mode")
scheduler.add(motor)

from horus import Scheduler, Node

# Development — simple
scheduler = Scheduler()

# Production — watchdog + RT
scheduler = Scheduler(tick_rate=1000, rt=True, watchdog_ms=500)

# With blackbox + telemetry
scheduler = Scheduler(
    tick_rate=1000,
    watchdog_ms=500,
    blackbox_mb=64,
    telemetry="http://localhost:9090",
    verbose=True,
)

# Deterministic mode for simulation/testing
scheduler = Scheduler(tick_rate=100, deterministic=True)

scheduler = Scheduler()
scheduler.add(Node(name="sensor", tick=sensor_fn, rate=100, order=0))
scheduler.add(Node(name="ctrl", tick=ctrl_fn, rate=100, order=1))

# Run for a short duration
scheduler.run(duration=0.1)

import horus
import time

def control_tick(node):
    if node.has_msg("sensor_data"):
        msg = node.recv("sensor_data")

        # Use message-level timestamps for latency checks
        if hasattr(msg, 'timestamp_ns') and msg.timestamp_ns:
            age_s = (time.time_ns() - msg.timestamp_ns) / 1e9
            if age_s > 0.1:  # More than 100ms old
                node.log_warning(f"Stale data: {age_s*1000:.1f}ms old")
                return

            latency = age_s
            print(f"Latency: {latency*1000:.1f}ms")

        # Process fresh data
        process(msg)

# 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
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)
    node.log_info("Published IMU data")

def control_tick(node):
    """Medium-frequency control (50Hz)"""
    if node.has_msg("imu_data"):
        imu = node.recv("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 = node.recv("cmd_vel")
        node.log_info(f"Command received: {msg}")

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

# Add nodes to scheduler
scheduler = horus.Scheduler()
scheduler.add(sensor)
scheduler.add(controller)
scheduler.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

# 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

# Just create a topic — HORUS picks the fastest path automatically:
# Same-thread:     ~3ns  (when pub+sub are in the same node)
# Same-process:    ~18-36ns  (when pub+sub are in different nodes, same process)
# Cross-process:   ~85-167ns  (when pub+sub are in different processes)
topic = Topic("cmd_vel", CmdVel)

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.recv("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.recv("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.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
)

# Configure rate and order on the Node, then add to scheduler
sensor = horus.Node(name="sensor", tick=sensor_fn, rate=100, order=0)
controller = horus.Node(name="ctrl", tick=ctrl_fn, rate=50, order=1)
logger = horus.Node(name="logger", tick=log_fn, rate=10, order=2)

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

scheduler.run()

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

import time

def control_tick(node):
    if node.has_msg("sensor_data"):
        data = node.recv("sensor_data")
        # Use message-level timestamps for staleness checks
        if hasattr(data, 'timestamp_ns') and data.timestamp_ns:
            age_s = (time.time_ns() - data.timestamp_ns) / 1e9
            if age_s > 0.1:
                node.log_warning("Skipping stale sensor data")
                return
        process(data)

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

# For staleness checks, use typed messages with timestamp_ns
# or track send time at the application level

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

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

# Monitor message timestamps via message-level fields
msg = node.recv("topic")
if msg and hasattr(msg, 'timestamp_ns') and msg.timestamp_ns:
    age = (time.time_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})  # Uses JSON serialization

def tick(node):
    t = horus.now()
    node.send("timestamp", t)

def tick(node):
    # PID controller using dt() for correct integration
    error = target - current
    integral += error * horus.dt()
    derivative = (error - prev_error) / horus.dt()
    output = kp * error + ki * integral + kd * derivative

def tick(node):
    if horus.elapsed() > 30.0:
        node.log_info("Running for 30+ seconds, stabilized")

def tick(node):
    if horus.tick() % 100 == 0:
        node.log_info(f"Tick {horus.tick()}: system healthy")

def tick(node):
    # Always do critical work
    process_sensor_data()

    # Only do expensive work if budget allows
    if horus.budget_remaining() > 0.001:  # >1ms remaining
        run_expensive_optimization()

def tick(node):
    # Simulated sensor noise (reproducible in deterministic mode)
    noise = (horus.rng_float() - 0.5) * 0.1
    reading = true_value + noise

def tick(node):
    ts = horus.timestamp_ns()
    transform = tf.lookup("camera", "base_link", ts)

import horus

params = horus.Params()

def controller_tick(node):
    # Read gains from params — change at runtime via CLI or monitor
    kp = params.get("pid_kp", 1.0)
    ki = params.get("pid_ki", 0.1)
    kd = params.get("pid_kd", 0.01)
    max_speed = params.get("max_speed", 1.5)

    error = target - current
    output = min(kp * error, max_speed)
    node.send("cmd_vel", output)

controller = horus.Node(name="PIDController", tick=controller_tick,
                        rate=100, pubs=["cmd_vel"])
horus.run(controller)

horus param set pid_kp 2.5
horus param set max_speed 0.8
horus param list

import threading
from horus import Rate, Topic, Image

def camera_loop():
    rate = Rate(30)  # 30 FPS target
    topic = Topic(Image)

    while running:
        frame = capture_camera()
        topic.send(frame)

        if rate.is_late():
            print(f"Camera behind: {rate.actual_hz():.1f} Hz (target {rate.target_hz():.0f})")

        rate.sleep()

thread = threading.Thread(target=camera_loop, daemon=True)
thread.start()

# horus.toml
[drivers]
arm = { type = "dynamixel", port = "/dev/ttyUSB0", baudrate = 1000000 }
lidar = { type = "rplidar", port = "/dev/ttyUSB1" }

import horus

hw = horus.drivers.load()

if hw.has("arm"):
    arm = hw.dynamixel("arm")
    port = arm.get_or("port", "/dev/ttyUSB0")
    baud = arm.get_or("baudrate", 115200)
    print(f"Arm on {port} at {baud}")

print(f"Available drivers: {hw.list()}")

from horus import us, ms

us  # 1e-6 — microseconds to seconds
ms  # 1e-3 — milliseconds to seconds

# Use with budget/deadline for readability
node = horus.Node(tick=fn, rate=1000, budget=300 * us, deadline=900 * us)

from horus import HorusNotFoundError, HorusTransformError, HorusTimeoutError

try:
    tf.tf("missing_frame", "base")
except HorusTransformError as e:
    print(f"Transform failed: {e}")

try:
    tf.wait_for_transform("src", "dst", timeout_sec=1.0)
except HorusTimeoutError:
    print("Timed out waiting for transform")

# 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.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

# 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.recv("input")
        print(f"[{node.name}] Received: {data}")
        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']}")

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