Python API

Complete Python API documentation for HORUS. The Python bindings are built with PyO3, exposing the core Rust framework to Python with zero-copy shared memory IPC.

Python Bindings

Full reference for the HORUS Python bindings:

• Node — Create nodes with tick, init, and shutdown callbacks
• Topic — Unified pub/sub with typed messages (CmdVel, Pose2D, Imu, etc.)
• Scheduler — Node execution with priority ordering and rate control
• Scheduler configuration — tick_rate, rt, deterministic, watchdog_ms, blackbox_mb, recording, cores, net keyword arguments
• TransformFrame / Transform — Coordinate frame management
• TensorPool / Tensor — Zero-copy shared memory tensors; .numpy() and .torch() share the buffer via the array interface

Custom Messages

Create your own typed messages in Python:

• Dict topics — no build step: node.send('robot.status', {'battery': 85, 'ok': True}), node.recv('robot.status') returns a dict
• Compiled messages — horus.msggen.register_message() + build_messages() (runs maturin develop --release, needs a horus_py source checkout). A real Rust type with named fields, but it still travels over the generic MessagePack path, not the zero-copy backend the built-in types use — see Custom Messages
• YAML schema support for team projects

Async Nodes

Asynchronous Python nodes for non-blocking I/O operations.

Quick Reference

Core Classes

There is no importable horus.NodeInfo or horus.Priority — neither name is part of the public Python API. Runtime context and logging reach you through the node object passed to init / tick / shutdown: node.log_info(), node.log_warning(), node.log_error(), node.log_debug() (these only emit while the scheduler is driving the node). Node(order=...) sets the registration priority, and Node(priority=...) is the SCHED_FIFO real-time priority (1-99, where higher is more urgent).

order does not sequence ticks in Python. Every Python node carries a rate — 30 Hz by default — which makes it real-time and gives it its own thread, so nodes run concurrently regardless of order. It sequences ticks only under deterministic=True.

Runtime Helpers

Exceptions

Message Types

Transform System

Perception Types

These live in the horus.perception submodule, which must be imported explicitly — import horus alone does not bind it, so horus.perception.Detection raises AttributeError unless you import the submodule first:

import horus.perception as perception       # or: from horus.perception import Detection

Tensor System

Networking

The Python bindings do not do networking. Both of the APIs that look like they should are inert:

The net=True warning is a RuntimeWarning, which Python hides by default outside __main__ — so a service can request replication, not get it, and print nothing. Run with -W always::RuntimeWarning if you need to see it.

Cross-machine replication is a Rust-side feature: build with --features net or use horus run --net, and let the replicator mirror the shared-memory topics your Python nodes already publish to. See Network Backends.

Installation

pip install horus-robotics

Minimal Example

import horus

def my_tick(node):
    node.send("greeting", "Hello from Python!")
    msg = node.recv("greeting")
    if msg:
        print(msg)

node = horus.Node(
    name="MyNode",
    pubs=["greeting"],
    subs=["greeting"],
    tick=my_tick,
    rate=10
)

horus.run(node)