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, name, cores, max_deadline_misses, verbose, telemetry, 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

Detection, BoundingBox2D, TrackedObject and Landmark each name two different classes. from horus import Detection is the wire message, which is what a topic carries; from horus.perception import Detection is the helper listed above, which is what the analysis methods live on. They are not interchangeable, and the failure reads as a contradiction: horus.Detection is horus.perception.Detection is False, and passing one where the other is expected raises TypeError: 'Detection' object is not an instance of 'Detection'. DetectionList accepts either.

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>=0.4.0"

The floor is not optional. The newest horus-robotics on PyPI is 0.1.9, and it speaks an older shared-memory format than a 0.4.x CLI — so a bare pip install horus-robotics gets you a package that imports, runs, and whose nodes publish topics the CLI cannot see. Nothing reports it; horus topic list simply shows nothing. The floor turns that silence into a resolver error.

Until 0.4.x reaches PyPI, install from the tree the HORUS installer already cached — same tag as your CLI, by construction:

pip install ~/.horus/cache/horus@0.4.0/horus_py

horus --version prints the version to substitute. See Python Support for the build-from-source route.

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)