Rate & Params

Two utility types for timing control and dynamic configuration outside the scheduler's node lifecycle.

Rate — Fixed-Frequency Loop

horus.Rate provides drift-compensated rate limiting for loops that need to run at a fixed frequency. Use it for standalone scripts and background threads — inside nodes, the scheduler handles timing automatically.

# simplified
import horus

rate = horus.Rate(100)  # 100 Hz

while True:
    do_work()
    rate.sleep()  # Blocks until next tick (drift-compensated)

Constructor

# simplified
horus.Rate(hz: float)  # Target frequency in Hz

Methods

Method	Returns	Description
`rate.sleep()`	—	Block until next tick (compensates for work time)
`rate.reset()`	—	Reset cycle start to now (use after a long pause)
`rate.actual_hz()`	`float`	Actual achieved frequency in Hz (exponentially smoothed)
`rate.target_hz()`	`float`	Configured target frequency in Hz
`rate.period()`	`float`	Target period in seconds
`rate.is_late()`	`bool`	Whether the current cycle has exceeded the target period

Example: Hardware Driver Thread

# simplified
import horus
import threading

def imu_reader_thread():
    rate = horus.Rate(100)  # 100 Hz
    topic = horus.Topic(horus.Imu)

    while True:
        reading = read_imu_hardware()
        topic.send(reading)
        rate.sleep()

thread = threading.Thread(target=imu_reader_thread, daemon=True)
thread.start()

When to Use Rate vs Scheduler

Use Case	Use
Nodes with tick callbacks	Scheduler (handles timing, RT, safety)
Standalone scripts	`Rate`
Background threads alongside scheduler	`Rate`
One-shot tools	Neither — just run once

Params — Runtime Parameters

horus.Params is a typed key-value store for dynamic configuration. Change gains, thresholds, or feature flags at runtime without restarting nodes.

# simplified
import horus

params = horus.Params()
params["pid.kp"] = 1.5
params["pid.ki"] = 0.01

kp = params.get("pid.kp")  # 1.5

Constructor

# simplified
horus.Params()  # Empty parameter store

Methods

Method	Returns	Description
`params.get(key, default=None)`	value	Get value, return `default` (`None`) if missing
`params[key]`	value	Get value, raise `KeyError` if missing
`params[key] = value`	—	Set parameter value
`params.has(key)`	`bool`	Check if key exists
`params.keys()`	`list[str]`	All parameter names
`params.remove(key)`	`bool`	Remove a parameter, returns `True` if it existed
`params.reset()`	—	Reset all parameters to built-in defaults
`params.save()`	—	Persist parameters to disk

Example: Dynamic PID Tuning

# simplified
import horus

params = horus.Params()
params["kp"] = 2.0
params["ki"] = 0.1
params["kd"] = 0.05

def controller_tick(node):
    kp = params.get("kp", 2.0)
    ki = params.get("ki", 0.1)
    kd = params.get("kd", 0.05)

    error = get_setpoint() - get_measured()
    output = kp * error + ki * integral + kd * derivative
    node.send("control", {"output": output})

# Another thread or monitoring tool can update params at runtime:
# params["kp"] = 3.0  # Takes effect on next tick

Example: Feature Flags

# simplified
params = horus.Params()
params["enable_slam"] = True
params["max_speed"] = 1.0

def tick(node):
    if params.get("enable_slam", False):
        run_slam()

    speed = min(velocity, params.get("max_speed", 1.0))
    node.send("cmd_vel", horus.CmdVel(linear=speed, angular=0.0))

Rate & Params

Rate — Fixed-Frequency Loop

Constructor

Methods

Example: Hardware Driver Thread

When to Use Rate vs Scheduler

Params — Runtime Parameters

Constructor

Methods

Example: Dynamic PID Tuning

Example: Feature Flags

See Also