Python Tensors for ML

HORUS does not load models and does not run inference. There is no horus.ml_utils module, no horus.ai, and no horus.cuda_is_available() — importing any of them raises ModuleNotFoundError / AttributeError. Earlier versions of this page documented those names; they were never part of the shipped package.

What HORUS gives an ML workload is the data path: horus.Tensor, a shared-memory array with numpy interop, and horus.TensorPool, the arena those arrays live in. Your model stays your own code — PyTorch, ONNX Runtime, whatever you already use — and HORUS carries its inputs and outputs between nodes.

Creating tensors

import numpy as np
import horus

# From an existing numpy array — copies once into the shared-memory pool.
frame = np.zeros((480, 640, 3), dtype=np.uint8)
t = horus.Tensor.from_numpy(frame)

# Or allocate directly, without a numpy array to copy from.
scores = horus.Tensor([1, 1000], dtype="float32")          # zero-filled
scratch = horus.Tensor.empty([8, 3, 224, 224], dtype="float32")

print(t.shape, t.dtype, t.device)      # [480, 640, 3] uint8 cpu
print(t.numel, t.nbytes)               # 921600 921600

Tensor(shape, dtype), Tensor.zeros(...) and Tensor.empty(...) all do the same thing today: they take a slot from pool 1 and hand it back zeroed (pool pages come from the OS already zeroed).

Accepted dtype strings: float32, float64, float16, bfloat16, int8, int16, int32, int64, uint8, uint16, uint32, uint64, bool. Aliases such as f32, u8, int, half work too. bfloat16 allocates, but numpy has no bfloat16 type, so .numpy() on one gives you a raw 2-byte void array — keep bf16 conversion inside your framework.

numpy interop

from_numpy() copies into the pool once. numpy() goes the other way with no copy at all: you get an ndarray pointing at the same bytes.

import horus

t = horus.Tensor([2, 3], dtype="float32")

view = t.numpy()          # ndarray sharing the tensor's memory
view[0, 0] = 42.0         # writes straight into the pool slot
print(t.numpy()[0, 0])    # 42.0

print(t.is_cpu(), t.is_contiguous())   # True True
print(t[0])                            # indexing returns a numpy view: [42. 0. 0.]

Because the ndarray aliases pool memory, the tensor keeps an extra reference on its slot for as long as it may be viewed — t.refcount reads 2 after the first numpy() call. The slot is returned when the Tensor is garbage collected. t.release() drops the tensor's own reference immediately, but if a numpy() / torch() view was ever exported, the extra reference taken for that view holds the slot until the Tensor object itself is collected — so release() frees the slot right away only when no view has been handed out.

Shape, slices, reductions

import numpy as np
import horus

t = horus.Tensor.from_numpy(np.arange(12, dtype=np.float32).reshape(3, 4))

print(t.reshape(4, 3).shape)     # [4, 3]
print(t.flatten().shape)         # [12]
print(t.unsqueeze(0).shape)      # [1, 3, 4]  (add a batch dimension)
print(t.slice(0, 2).shape)       # [2, 4]     (first dimension only)
print(t.T.shape)                 # (4, 3)     numpy array, not a Tensor

print(t.sum().tolist())          # [66.0]
print(t.mean(dim=0).tolist())    # [4.0, 5.0, 6.0, 7.0]
print(t.max().tolist())          # [11.0]

reshape, flatten, squeeze, unsqueeze, view and slice return Tensors over the same memory. t.T and t[...] hand you numpy arrays instead, because they delegate to numpy for the general case. Reductions (sum, mean, min, max, with an optional dim=) return a Tensor; call .tolist() to read the value in Python.

Changing dtype

astype() converts directly, and the to_float32() / to_float16() / to_int32() / to_uint8() shorthands wrap it:

import numpy as np
import horus

t = horus.Tensor.from_numpy(np.zeros((2, 2), dtype=np.uint8))

as_f32 = t.astype("float32")     # or t.to_float32()
print(as_f32.dtype)              # float32

The conversion goes through numpy internally, so a dtype change costs one copy — which is what a dtype change costs anyway. Asking for the dtype the tensor already has skips the copy and hands back a handle to the same memory. CUDA tensors have to go through .cpu() first, the same as for .numpy().

When you need to rescale as well as convert, do it in numpy and re-wrap so both happen in one pass:

as_f32 = horus.Tensor.from_numpy(t.numpy().astype(np.float32) / 255.0)

Passing tensors between nodes

Declare the topic's type as horus.Tensor on both sides. Only the 168-byte descriptor travels through the ring buffer; the pixels stay where they were written.

import numpy as np
import horus


def capture(node):
    frame = np.random.default_rng().integers(0, 255, (240, 320, 3), dtype=np.uint8)
    node.send("camera.frames", horus.Tensor.from_numpy(frame))


def consume(node):
    t = node.recv("camera.frames")     # None when nothing new arrived
    if t is None:
        return
    print("frame", t.shape, t.dtype, t.numpy().mean())


camera = horus.Node(
    name="camera", tick=capture, rate=10, order=0,
    pubs=[horus.Pub("camera.frames", horus.Tensor)],
)
worker = horus.Node(
    name="worker", tick=consume, rate=10, order=1,
    subs=[horus.Sub("camera.frames", horus.Tensor)],
)

horus.run(camera, worker, duration=1.0)   # drop `duration` to run until Ctrl+C

Run the subscriber at the publisher's rate. Unlike ordinary message topics, a tensor topic delivers nothing to a subscriber that ticks slower than the publisher: with the camera at 10 Hz and the worker at 10 Hz the worker sees frames, at 30 Hz / 10 Hz node.recv() returns None on every tick. If your model can only keep up at 10 Hz, publish frames at 10 Hz — do not publish at 30 and hope to sample them.

If you need to move a descriptor yourself (across a process boundary you manage by hand, for instance), t.to_descriptor() gives you those 168 bytes and horus.Tensor.from_descriptor(pool_id, blob) reattaches to the same slot.

The pool

Every tensor comes from a TensorPool. Create one explicitly when you want to bound how much shared memory a pipeline can hold:

import horus

pool = horus.TensorPool(pool_id=7, size_mb=64, max_slots=32)

frame = pool.alloc((240, 320, 3), dtype="uint8")
feats = pool.alloc([16, 16], dtype="float32")

print(pool.stats())
frame.release()                              # drop early instead of waiting for GC
print(pool.stats()["allocated_slots"])       # 1

stats() returns a dict with pool_id, pool_size, max_slots, allocated_slots, total_refcount, used_bytes and free_bytes. Watch allocated_slots against max_slots — that is the limit you hit first in a pipeline that forgets to drop references. used_bytes is a high-water mark for the bump allocator, so it does not fall back to zero when slots are freed and reused.

Pools are keyed by pool_id and live in shared memory, so a second process constructing TensorPool(pool_id=7, ...) opens the existing pool rather than making a new one. horus.Tensor(...), Tensor.zeros/empty/from_numpy all use pool 1, created on first use with default settings.

Devices: CPU today

Tensor carries a device field and the string parser understands cuda:N, but the shipped wheel is built without CUDA, so there is nowhere to put a GPU tensor:

import horus

pool = horus.TensorPool(pool_id=7, size_mb=64, max_slots=32)   # as in "The pool" above

t = horus.Tensor([2, 2])
print(t.device, t.is_gpu)   # cpu False

t.cuda()
# RuntimeError: CUDA support not compiled. Rebuild with --features cuda
#               or use torch.as_tensor(handle).cuda()

pool.alloc((2, 2), device="cuda:0")
# RuntimeError: Allocation failed: ... device mismatch: requested cuda:0
#               but pool backend targets cpu (backend: mmap)

So GPU placement is your framework's job: keep the HORUS tensor on the CPU as the transport, and move the batch onto the device inside torch or ONNX Runtime.

Plugging in your model

Here is the whole shape of it — HORUS moves frames in, your session runs, HORUS moves results out.

import numpy as np
import onnxruntime as ort
import horus

session = ort.InferenceSession("model.onnx", providers=["CPUExecutionProvider"])
input_name = session.get_inputs()[0].name


def capture(node):
    """Stand-in for a real camera driver."""
    frame = np.random.default_rng().integers(0, 255, (240, 320, 3), dtype=np.uint8)
    node.send("camera.frames", horus.Tensor.from_numpy(frame))


def detect(node):
    t = node.recv("camera.frames")
    if t is None:
        return

    frame = t.numpy()                                  # zero-copy HWC uint8 view
    batch = frame.astype(np.float32) / 255.0           # your preprocessing
    batch = np.ascontiguousarray(batch.transpose(2, 0, 1)[None])

    out = session.run(None, {input_name: batch})[0]    # your model, your rules

    node.send("detections", horus.Tensor.from_numpy(np.ascontiguousarray(out)))


def plan(node):
    d = node.recv("detections")
    if d is not None:
        print("detections", d.shape, d.dtype)


camera = horus.Node(
    name="camera", tick=capture, rate=10, order=0,
    pubs=[horus.Pub("camera.frames", horus.Tensor)],
)
detector = horus.Node(
    name="detector", tick=detect, rate=10, order=1,
    subs=[horus.Sub("camera.frames", horus.Tensor)],
    pubs=[horus.Pub("detections", horus.Tensor)],
)
planner = horus.Node(
    name="planner", tick=plan, rate=10, order=2,
    subs=[horus.Sub("detections", horus.Tensor)],
)

horus.run(camera, detector, planner, duration=1.0)

Two details worth copying: build the session once at import/startup, never inside tick; and pass np.ascontiguousarray(...) to from_numpy when your preprocessing produced a transposed or strided view, since the pool stores C- contiguous data.

PyTorch

Tensor.torch() is the same bridge as numpy() — it calls torch.as_tensor(handle), so you get a CPU torch tensor over the pool slot with no copy. torch must be installed in your environment; HORUS does not depend on it.

import torch

t = node.recv("camera.frames")
shared = t.torch()                       # CPU tensor aliasing shared memory
batch = shared.permute(2, 0, 1).unsqueeze(0).float().div(255).cuda()

with torch.no_grad():
    out = model(batch)

Use out-of-place ops (.float(), .div()) on that first view. An in-place op such as .div_(255) would write through into the pool slot, corrupting the frame every other subscriber is reading.

Measuring inference latency

There is no PerformanceMonitor class; the scheduler already prints a per-node timing report (avg / p99 / max / overruns / deadline misses) on shutdown. For latency of the model call alone, time it yourself:

import statistics
import time
from collections import deque

latencies = deque(maxlen=100)


def timed_infer(node):
    t = node.recv("camera.frames")
    if t is None:
        return

    start = time.perf_counter()
    result = session.run(None, {input_name: preprocess(t.numpy())})
    latencies.append((time.perf_counter() - start) * 1e3)

    if len(latencies) == latencies.maxlen:
        window = sorted(latencies)
        print(f"avg {statistics.mean(window):.2f} ms  "
              f"p95 {window[int(0.95 * len(window)) - 1]:.2f} ms  "
              f"max {window[-1]:.2f} ms")
        latencies.clear()

To have the scheduler police the budget instead of just reporting it, give the node a budget= and a deadline= (both in seconds — budget=5 * horus.ms reads well); overruns then show up in the timing report and can trigger the configured on_miss policy.

API summary

horus.Tensor

horus.TensorPool

See Also

• Python Bindings - Core Python API
• AI Integration - End-to-end ML pipeline patterns
• Tensor Pool (Rust) - The pool implementation these tensors sit on