Common Mistakes (Python)

New to HORUS with Python? Here are the most common mistakes beginners make and how to fix them.

Prerequisites

HORUS installed (Installation Guide)
Quick Start (Python) completed

1. Using Slashes in Topic Names

The Problem:

node.send("sensors/lidar", scan_data)

Why: On Linux, slashes create subdirectories in the shared memory filesystem which works fine. On macOS, shm_open() does not support slashes in names, so this will fail with an OS error.

The Fix:

# CORRECT — Use dots for cross-platform compatibility
node.send("sensors.lidar", scan_data)

Use dot-separated names ("sensors.lidar", "camera.rgb") for portable topic names that work on all platforms.

2. Blocking the tick() Function

The Problem:

import time
import requests

def tick(node):
    # WRONG — blocks the entire scheduler!
    time.sleep(0.1)
    response = requests.get("http://api.example.com/data")
    node.send("data", response.json())

Why: All nodes run in a single tick cycle. A blocking call in one node delays every other node. Worse, Python's GIL means time.sleep() and synchronous HTTP calls hold up the entire tick cycle. If you have a motor controller at 1kHz, a 100ms sleep in any node will cause it to miss hundreds of deadlines.

The Fix:

import horus

# Option 1: Use an async tick for I/O-bound work
import aiohttp

async def tick(node):
    async with aiohttp.ClientSession() as session:
        async with session.get("http://api.example.com/data") as resp:
            data = await resp.json()
            node.send("data", data)

# async def is auto-detected — the node runs on the async executor
fetcher = horus.Node(name="fetcher", pubs=["data"], tick=tick, rate=1)

# Option 2: For CPU-bound work, use compute=True
def tick(node):
    if node.has_msg("image"):
        frame = node.recv("image")
        result = run_heavy_inference(frame)  # Runs on thread pool
        node.send("detections", result)

detector = horus.Node(name="detector", pubs=["detections"], subs=["image"],
                      tick=tick, rate=30, compute=True)

Rule of thumb: Keep synchronous tick functions under 1ms. Use async def for network I/O, compute=True for CPU-heavy work.

3. Using Dicts When You Need Typed Topics

The Problem:

def controller_tick(node):
    # Works, but slow — ~5-50μs per message via JSON serialization
    node.send("cmd_vel", {"linear": 0.5, "angular": 0.1})

controller = horus.Node(name="nav", pubs=["cmd_vel"], tick=controller_tick, rate=1000)

Why: String topics (pubs=["cmd_vel"]) use GenericMessage with MessagePack serialization, which costs 5-50μs per message. At 1kHz, that is 5-50ms per second spent just on serialization. Typed topics use zero-copy POD transport at ~1.5μs. Additionally, dict topics cannot cross to Rust nodes — a Rust subscriber using Topic<CmdVel> will never see Python dicts.

The Fix:

import horus

def controller_tick(node):
    # Zero-copy POD transport — ~1.5μs, visible to Rust nodes
    cmd = horus.CmdVel(linear=0.5, angular=0.1)
    node.send("cmd_vel", cmd)

controller = horus.Node(name="nav", pubs=[horus.CmdVel], tick=controller_tick, rate=1000)

When to use each:

Use case	Topic type	Latency
Control loops (motor commands, sensor fusion)	Typed (`horus.CmdVel`)	~1.5μs
Debug/logging/telemetry	String (dicts)	~5-50μs
Cross-language (Python to Rust)	Typed (required)	~1.5μs
Rapid prototyping	String (dicts)	~5-50μs

4. Forgetting to Handle None from recv()

The Problem:

def tick(node):
    # WRONG — crashes with AttributeError when no message is available!
    temp = node.recv("temperature")
    print(f"Temperature: {temp['value']:.1f}°C")

Why: node.recv() returns None when no message is available. On the first tick, or if the publisher is slower than the subscriber, there will be no message. Accessing attributes or keys on None raises AttributeError or TypeError.

The Fix:

# Option 1: Guard with None check
def tick(node):
    temp = node.recv("temperature")
    if temp is not None:
        print(f"Temperature: {temp['value']:.1f}°C")

# Option 2: Check first with has_msg()
def tick(node):
    if node.has_msg("temperature"):
        temp = node.recv("temperature")
        print(f"Temperature: {temp['value']:.1f}°C")

# Option 3: Cache last known value
last_temp = [None]

def tick(node):
    temp = node.recv("temperature")
    if temp is not None:
        last_temp[0] = temp

    if last_temp[0] is not None:
        print(f"Temperature: {last_temp[0]['value']:.1f}°C")

has_msg() internally peeks by calling recv() and buffering the result. The next recv() returns that buffered value, so the has_msg() + recv() pattern has zero overhead.

5. Not Calling horus.run()

The Problem:

import horus

def sensor_tick(node):
    node.send("data", 42.0)

sensor = horus.Node(name="sensor", pubs=["data"], tick=sensor_tick, rate=10)

# Script ends here — nothing happens!

Why: Creating a horus.Node only defines the node. It does not start the scheduler or begin ticking. Without horus.run() or a Scheduler, nothing executes.

The Fix:

import horus

def sensor_tick(node):
    node.send("data", 42.0)

sensor = horus.Node(name="sensor", pubs=["data"], tick=sensor_tick, rate=10)

# MUST call run() to start the scheduler
horus.run(sensor)

For more control, use horus.Scheduler directly:

sched = horus.Scheduler(tick_rate=100, watchdog_ms=500)
sched.add(sensor)
sched.run()

6. Import Errors: _horus Module Not Found

The Problem:

$ python3 -c "import horus"
ModuleNotFoundError: No module named '_horus'

Why: The horus Python package has two layers: _horus (compiled Rust PyO3 bindings) and horus (Python wrapper). If _horus is missing, the Rust native extension was not installed or was installed for a different Python version.

The Fix:

# Install from PyPI (recommended)
pip install horus-robotics

# Verify it works
python3 -c "import horus; print('OK')"

If you are building from source:

cd horus_py
maturin develop --release

Common causes:

Symptom	Cause	Fix
`No module named '_horus'`	Native extension not installed	`pip install horus-robotics`
`No module named 'horus'`	Package not in current environment	Check `which python3` and `pip list`
Works in terminal, fails in script	Different Python environment	Use `horus run` instead of `python3 script.py`
`ImportError: ... ABI`	Python version mismatch	Reinstall with the correct Python version

Always use horus run to execute your scripts, not python3 src/main.py directly. horus run sets up the SHM namespace, environment variables, and ensures the correct Python environment is used.

7. Topic Name Typos

The Problem:

def sensor_tick(node):
    node.send("temperture", 22.5)   # Typo: "temperture"

def monitor_tick(node):
    temp = node.recv("temperature")  # Correct name — but never receives!
    if temp is not None:
        print(f"Temp: {temp}")

sensor = horus.Node(name="sensor", pubs=["temperture"], tick=sensor_tick, rate=1)
monitor = horus.Node(name="monitor", subs=["temperature"], tick=monitor_tick, rate=1)

Why: Python has no compile-time type checking for topic names. Mismatched names silently create two separate topics — the publisher writes to one, the subscriber reads from another. The subscriber's recv() always returns None, with no error.

The Fix:

import horus

# Define topic names as constants
TOPIC_TEMPERATURE = "sensor.temperature"

def sensor_tick(node):
    node.send(TOPIC_TEMPERATURE, 22.5)

def monitor_tick(node):
    temp = node.recv(TOPIC_TEMPERATURE)
    if temp is not None:
        print(f"Temp: {temp}")

sensor = horus.Node(name="sensor", pubs=[TOPIC_TEMPERATURE], tick=sensor_tick, rate=1)
monitor = horus.Node(name="monitor", subs=[TOPIC_TEMPERATURE], tick=monitor_tick, rate=1)

Better yet, use typed topics which enforce the topic name automatically:

# Typed topics include their canonical name — no string to mistype
sensor = horus.Node(name="sensor", pubs=[horus.Imu], tick=sensor_tick, rate=100)
monitor = horus.Node(name="monitor", subs=[horus.Imu], tick=monitor_tick, rate=100)

Debugging tip: Use horus topic list in another terminal to see all active topics and spot mismatches.

8. Using Global State Instead of Closure/Class State

The Problem:

import horus

# WRONG — global mutable state shared between ALL nodes
counter = 0

def tick_a(node):
    global counter
    counter += 1
    node.send("count_a", counter)

def tick_b(node):
    global counter
    counter += 1  # Both nodes mutate the same variable!
    node.send("count_b", counter)

node_a = horus.Node(name="a", pubs=["count_a"], tick=tick_a, rate=10)
node_b = horus.Node(name="b", pubs=["count_b"], tick=tick_b, rate=10)
horus.run(node_a, node_b)

Why: Global variables are shared across all nodes. When two nodes mutate the same global, the values interleave unpredictably. This also breaks deterministic mode where execution order must produce reproducible results.

The Fix:

import horus

# Option 1: Closure state (simple, recommended for small state)
def make_counter_node(name, topic):
    counter = [0]  # Mutable via list — each node gets its own

    def tick(node):
        counter[0] += 1
        node.send(topic, counter[0])

    return horus.Node(name=name, pubs=[topic], tick=tick, rate=10)

node_a = make_counter_node("a", "count_a")
node_b = make_counter_node("b", "count_b")
horus.run(node_a, node_b)

# Option 2: Class state (better for complex state)
class CounterNode:
    def __init__(self):
        self.counter = 0

    def tick(self, node):
        self.counter += 1
        node.send("count", self.counter)

state = CounterNode()
node = horus.Node(name="counter", pubs=["count"], tick=state.tick, rate=10)
horus.run(node)

9. Not Checking for Slow Subscribers

The Problem:

def monitor_tick(node):
    data = node.recv("fast.sensor")
    if data is not None:
        # Process only the latest message — silently losing all others
        expensive_analysis(data)

Why: If the publisher runs at 1kHz and your subscriber ticks at 10Hz, recv() only returns the latest message. The other ~99 messages per tick are silently overwritten in the ring buffer. For many applications (motor commands, camera frames) this is fine — you want the latest data. But for event streams (button presses, error reports, waypoints), losing messages is a bug.

The Fix:

# Use recv_all() to drain all queued messages
def monitor_tick(node):
    messages = node.recv_all("events")
    for msg in messages:
        process_event(msg)

# Or increase the ring buffer capacity for bursty topics
event_node = horus.Node(
    name="event_handler",
    subs={"events": {"type": "events", "capacity": 4096}},
    tick=monitor_tick,
    rate=10,
)

Use node.info to monitor health during development:

def tick(node):
    data = node.recv("sensor")
    if data is not None:
        process(data)

    # Periodically check metrics
    if node.info and node.info.tick_count() % 1000 == 0:
        metrics = node.info.get_metrics()
        node.log_info(f"Avg tick: {node.info.avg_tick_duration_ms():.2f}ms")

10. Mixing Python async and HORUS tick

The Problem:

import asyncio
import horus

async def tick(node):
    # WRONG — calling asyncio.run() or loop.run_until_complete() inside a tick
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(fetch_data())
    node.send("data", result)

# Also WRONG — using asyncio.sleep() in a synchronous tick
import asyncio

def tick(node):
    asyncio.run(asyncio.sleep(0.01))  # Creates a new event loop every tick!
    node.send("heartbeat", True)

Why: HORUS has its own async executor for async def tick functions. Manually creating or running asyncio event loops inside a tick conflicts with the scheduler's execution model, can deadlock, and creates a new event loop on every tick (which is expensive).

The Fix:

import aiohttp
import horus

# CORRECT — just make tick async and await directly
async def tick(node):
    async with aiohttp.ClientSession() as session:
        async with session.get("http://api.example.com/data") as resp:
            data = await resp.json()
            node.send("data", data)

# HORUS auto-detects async def and runs it on its async executor
fetcher = horus.Node(name="fetcher", pubs=["data"], tick=tick, rate=1)
horus.run(fetcher)

Rules for async nodes:

Declare tick (and optionally init/shutdown) as async def — HORUS detects this automatically
Use await for all I/O operations inside the tick
Do not use asyncio.run(), asyncio.get_event_loop(), or loop.run_until_complete() inside tick functions
Async nodes cannot use compute=True or on="topic" — these are mutually exclusive execution modes

Quick Reference

Mistake	Fix
Slashes in topic names	Use dots: `sensors.lidar`
Blocking tick() with sleep/HTTP	Use `async def` tick or `compute=True`
Dicts in control loops	Use typed topics: `pubs=[horus.CmdVel]`
Not handling `None` from `recv()`	Check `if temp is not None` or use `has_msg()`
Forgetting `horus.run()`	Always call `horus.run(*nodes)` to start execution
`_horus` import error	`pip install horus-robotics` or `maturin develop --release`
Topic name typos	Use string constants or typed topics
Global mutable state	Use closure state (list) or class instances
Missing messages from fast publishers	Use `recv_all()` or increase `capacity`
Manual asyncio inside tick	Use `async def tick(node)` — HORUS runs the executor

Still Having Issues?

Check Troubleshooting for error messages
See Python Examples for working code
Run horus topic list to see active topics and debug connectivity
Run horus monitor to watch node tick rates and health in real time

Common Mistakes (Python)

Prerequisites

1. Using Slashes in Topic Names

2. Blocking the tick() Function

3. Using Dicts When You Need Typed Topics

4. Forgetting to Handle None from recv()

5. Not Calling horus.run()

6. Import Errors: _horus Module Not Found

7. Topic Name Typos

8. Using Global State Instead of Closure/Class State

9. Not Checking for Slow Subscribers

10. Mixing Python async and HORUS tick

Quick Reference

Still Having Issues?

See Also