Tutorial 9: Record & Replay (Python)

Prefer another language? The same tutorial exists for Rust and C++.

Record your nodes' topic data while your robot runs, then replay it offline for debugging. This is the robotics equivalent of a flight recorder.

What You'll Learn

  • Using horus run --record to capture topic data, and horus record list/info/clean to manage the sessions
  • Using horus record replay to play back a recording, and horus record inject to run live nodes against it
  • Driving record and replay from Python, without the CLI
  • Using BlackBox for crash analysis
  • Designing nodes that work with both live and recorded data

Recording Live Data

Recording is enabled per run with the --record flag — pass a session name and every topic your nodes publish or subscribe to is snapshotted once per tick for the lifetime of the run:

# Record this run's topic data into a named session
horus run --record session_001

# Then manage captured sessions with the `horus record` subcommands:
horus record list                 # list captured sessions
horus record info session_001     # inspect one session
horus record clean --older-than 7 # delete sessions older than 7 days

Each snapshot is stamped with a microsecond wall-clock timestamp and the tick number. Because it is a per-tick sample rather than a per-message log, a topic published twice inside one tick contributes only its latest value.

Recording is a scheduler feature, not a Python one, so nothing in your node code changes: --record sets HORUS_RECORD_SESSION, and the scheduler behind horus.Scheduler reads that variable at construction time. To record unconditionally — a soak test, a field robot that should always keep a flight recording — ask for it in code instead:

import horus

# The same switch `horus run --record <session>` flips, but always on.
sched = horus.Scheduler(tick_rate=100, name="robot", recording=True)
print("recording:", sched.is_recording())

# sched.add(your nodes here)
sched.run()

Sessions land in ~/.local/share/horus/recordings/<session>/ (on macOS, ~/Library/Application Support/horus/recordings/), one <node>@<id>.horus file per node plus a scheduler@<id>.horus index.

📝A topic is captured once a node has used it

Constructing a Topic does not put it in the recording. A topic registers itself against the owning node the first time you send() or recv() on it from inside a callback the scheduler drives — init(), tick() or shutdown() — and the recorder only samples topics that are registered. A topic your node builds but never uses from one of those, or one you drive from a thread of your own that the scheduler does not know about, simply will not be in the session. (Declaring topics in pubs= / subs= on the node is worth doing for the monitoring tools, but it is the first send()/recv() that gets them recorded.)

Replaying Recorded Data

horus record replay re-runs the recording itself: it builds a fresh scheduler containing only replay nodes, which republish the recorded outputs onto the same SHM topics. None of your node code is loaded — this is for inspecting what happened, at whatever speed you like:

# Replay a session at original speed (pass the session name)
horus record replay session_001

# Replay at 2x speed (for faster analysis)
horus record replay session_001 --speed 2.0

# Time-travel: replay a specific tick window
horus record replay session_001 --start-tick 350 --stop-tick 500

To run your own edited nodes against the recorded sensor data, use horus record inject instead. It replays only the nodes you name and launches your source file alongside them with horus run, so live code and recorded data share the same topics:

# Replay the recorded lidar node; run controller.py live against it
horus record inject session_001 --nodes lidar_node --script controller.py

# Same, at half speed, over one tick window
horus record inject session_001 --nodes lidar_node --script controller.py \
    --speed 0.5 --start-tick 350 --stop-tick 500

Replaying from Python

Everything the CLI does is a Scheduler method, so you can assemble the same mix inside a pytest case — recorded nodes beside live ones, with the tick window and speed set in code:

import horus

# One node's recording out of a session directory.
recording = "/home/me/.local/share/horus/recordings/session_001/lidar_node@abc123.horus"

sched = horus.Scheduler(tick_rate=100, deterministic=True)

# Republishes lidar_node's recorded outputs onto the same topics.
sched.add_replay(recording, priority=0)
sched.start_at_tick(350)
sched.stop_at_tick(500)
sched.set_replay_speed(0.5)

# Your own nodes go in beside it, exactly as in a live run:
# sched.add(Controller())

print("Replaying ticks 350-500 at 0.5x")
sched.run()
⚠️Order matters: replay settings after add_replay()

stop_at_tick() and set_replay_speed() write into the replay state that add_replay() (or replay_from()) creates. Called on a scheduler that has no recording loaded yet they are silently ignored — you get a full-speed replay of the whole recording and no error. horus record inject applies them in this same order internally.

horus.Scheduler.replay_from(scheduler_path) is the whole-session version — it loads every node from a scheduler@<id>.horus file, which is what horus record replay does internally. sched.list_recordings(), sched.delete_recording(name) and sched.stop_recording() are the CLI's list, delete and "stop and flush" as plain methods, useful when a regression test wants to clean up after itself.

Designing for Replay

The key pattern: separate your processing nodes from your hardware drivers. Processing nodes subscribe to topics — they don't care if data comes from live hardware or a recording.

import horus
from horus import CmdVel, LaserScan, Topic


# This controller works identically with live OR recorded data
class Controller(horus.Node):
    def __init__(self):
        super().__init__(
            name="controller",
            rate=100,
            order=10,
            subs=["lidar.scan"],
            pubs=["cmd_vel"],
        )
        self.scan = Topic(LaserScan, endpoint="lidar.scan")
        self.cmd = Topic(CmdVel, endpoint="cmd_vel")

    def tick(self, info=None):
        # Drain to the freshest scan on the topic; do nothing if none arrived.
        scan = None
        while True:
            msg = self.scan.recv()
            if msg is None:
                break
            scan = msg
        if scan is None:
            return

        # This code runs the same whether lidar.scan comes from
        # a real LiDAR or from `horus record inject`
        min_range = min((r for r in scan.ranges if r > 0.01), default=999.0)

        linear = 0.3 if min_range > 0.5 else 0.0
        self.cmd.send(CmdVel(linear=linear, angular=0.0))


sched = horus.Scheduler(tick_rate=100, name="replay_demo", deterministic=True)
sched.add(Controller())

print("Controller running at 100 Hz (Ctrl+C to stop)")
sched.run()

The node names matter here: horus record inject --nodes lidar_node matches on the name= a node was constructed with, and that is also the filename the session is written under. Rename a node between the recording and the replay and the injector will not find it.

📝LaserScan.ranges is a Python list

Python hands you scan.ranges as a 360-element list, where Rust and C++ see a [f32; 360] array. Reading it is the same; writing it is not — mutating one element in place does not reach shared memory, so a Python driver node has to assign the whole list back (scan.ranges = values) before send(). The wire format is identical either way, so the recording a Python node produces replays into a Rust or C++ node unchanged.

BlackBox for Crash Analysis

Python has no horus.blackbox module — but it does not need one. The C++ blackbox::record() helper is a thin wrapper that publishes a Warning-level entry into the shared-memory log ring buffer, and self.log_warning() on a node already does exactly that. The buffer lives outside your process, so the entries are still there after a crash:

import horus
from horus import EmergencyStop, LaserScan, Topic


class SafetyMonitor(horus.Node):
    def __init__(self):
        super().__init__(name="safety", rate=50, order=0)
        self.scan = Topic(LaserScan, endpoint="lidar.scan")
        self.estop = Topic(EmergencyStop, endpoint="safety.estop")

    def tick(self, info=None):
        # Record important events during operation. The node's name is the
        # category the entry is filed under.
        scan = self.scan.recv()
        if scan is not None:
            closest = min((r for r in scan.ranges if r > 0.01), default=999.0)
            if closest < 0.35:
                self.log_warning(f"Obstacle detected at {closest:.2f}m")

        stop = self.estop.recv()
        if stop is not None and stop.engaged:
            self.log_error("E-stop triggered")


# blackbox_mb enables the scheduler's own flight recorder.
sched = horus.Scheduler(tick_rate=50, name="safety_demo", blackbox_mb=16)
sched.add(SafetyMonitor())
sched.run()

stop.engaged is a bool here, where Rust and C++ read the same field as a u8 compared against 1 — identical bytes on the wire, different binding.

After a crash, inspect the log ring buffer with the CLI:

horus log safety            # filter by node name — the category you logged under
horus log --level warn -n 100
horus log --since 5m
📝Category vs. node name

The C++ helper takes a free-form category string and files the entry under it, so blackbox::record("motor", ...) can appear under motor from any node. log_warning() has no such argument: entries are attributed to whichever node the scheduler is currently ticking, and they are dropped entirely if you call them outside init() / tick() / shutdown(). In practice that is the same thing done more strictly — one category per node — but if you were relying on cross-node categories in C++, put the category in the message text and grep for it.

horus blackbox --last 100 reads a separate store: it shows the events the scheduler itself writes to the flight recorder (deadline misses, budget violations, node errors, emergency stops), not the ones you log from Python. That store is off by default and is switched on with blackbox_mb=, as above; it persists to .horus/blackbox/ in the working directory, so it also survives the crash.

Workflow

1. Run robot        → horus run --record test
2. Bug happens      → stop the run (Ctrl+C)
3. Fix controller   → edit code (no rebuild — it's Python)
4. Test with data   → horus record inject test --nodes lidar_node --script controller.py
5. Verify fix       → no bug? ship it

No need to set up hardware again. No need to reproduce the exact scenario. The recording has everything.

Key Takeaways

  • horus run --record <session> captures topic data per tick; horus record list/info/clean manage the captured sessions
  • A topic is only captured once a node has actually send()/recv()'d on it from a scheduler-driven callback
  • horus record replay re-runs the recording itself; horus record inject feeds recorded data to your live nodes
  • The same thing from Python: recording=True, add_replay(), Scheduler.replay_from(), start_at_tick() / stop_at_tick() / set_replay_speed()
  • Design nodes as pure topic processors — they work with live AND recorded data
  • self.log_warning() is Python's blackbox::record(): the entry lands in the shared-memory log ring buffer, which outlives the crashed process — read it back with horus log
  • Replay at different speeds for fast iteration

Next Steps