Sensor Messages

from horus import Imu, LaserScan, Odometry, NavSatFix, BatteryState, RangeSensor

import horus

# All 6 accel/gyro fields are REQUIRED positional arguments
imu = horus.Imu(
    0.0,    # accel_x — m/s², linear acceleration X
    0.0,    # accel_y — m/s², linear acceleration Y
    9.81,   # accel_z — m/s², linear acceleration Z (gravity)
    0.0,    # gyro_x  — rad/s, angular velocity X (roll rate)
    0.0,    # gyro_y  — rad/s, angular velocity Y (pitch rate)
    0.05,   # gyro_z  — rad/s, angular velocity Z (yaw rate)
    timestamp_ns=0,  # optional
)

# Access fields
print(f"Gravity: {imu.accel_z:.2f} m/s²")
print(f"Yaw rate: {imu.gyro_z:.3f} rad/s")

def imu_tick(node):
    imu = read_hardware()

    # IMPORTANT: always validate before publishing
    if not imu.is_valid():
        node.log_warning("Invalid IMU reading (NaN/inf) — skipping")
        return

    node.send("imu.raw", imu)

import horus

scan = horus.LaserScan(
    angle_min=-3.14159,      # rad — start angle
    angle_max=3.14159,       # rad — end angle
    angle_increment=0.01745, # rad — step between readings
    range_min=0.1,           # m — minimum valid range
    range_max=30.0,          # m — maximum valid range
    ranges=[1.5, 2.0, 0.0, 3.2],  # optional list of range measurements
    timestamp_ns=0,          # optional
)

# Access range data
for i in range(len(scan)):
    if scan.is_range_valid(i):
        print(f"Reading {i}: {scan.ranges[i]:.2f}m at {scan.angle_at(i):.2f} rad")

def lidar_tick(node):
    scan = node.recv("scan")
    if scan is None:
        return

    # Find closest obstacle
    min_dist = scan.min_range()
    if min_dist is not None and min_dist < 0.5:
        node.log_warning(f"Obstacle at {min_dist:.2f}m!")
        node.send("cmd_vel", horus.CmdVel(linear=0.0, angular=0.0))
        return

    # Count valid readings — use len(), not .valid_count()
    valid = len(scan)
    node.log_debug(f"{valid}/360 valid readings")

import horus

odom = horus.Odometry(
    x=1.0,                # m — position X
    y=2.0,                # m — position Y
    theta=0.5,            # rad — heading
    linear_velocity=0.5,  # m/s — forward speed
    angular_velocity=0.1, # rad/s — turn rate
    timestamp_ns=0,       # optional
)

print(f"Position: ({odom.x}, {odom.y})")
print(f"Heading: {odom.theta:.2f} rad")
print(f"Speed: {odom.linear_velocity:.2f} m/s")

import horus

fix = horus.NavSatFix(
    latitude=37.7749,      # degrees
    longitude=-122.4194,   # degrees
    altitude=10.5,         # meters above WGS84 ellipsoid
    timestamp_ns=0,        # optional
)

print(f"Position: {fix.latitude}, {fix.longitude}")
print(f"Satellites: {fix.satellites_visible}")
print(f"HDOP: {fix.hdop}")

import horus

battery = horus.BatteryState(
    voltage=12.4,              # V
    percentage=0.75,           # 0.0-1.0
    current=-2.5,              # A (negative = discharging)
    temperature=35.0,          # °C (default: 25.0)
    power_supply_status=1,     # u8 status code
    timestamp_ns=0,            # optional
)

if battery.percentage < 0.2:
    print("LOW BATTERY WARNING")

# Check power supply status (u8), NOT a bool
print(f"Status code: {battery.power_supply_status}")

import horus

dist = horus.RangeSensor(
    range=1.5,           # m — measured distance
    sensor_type=0,       # sensor type code
    field_of_view=0.26,  # rad (~15°)
    min_range=0.02,      # m (default: 0.02)
    max_range=4.0,       # m (default: 4.0)
    timestamp_ns=0,      # optional
)

import horus

state = horus.JointState(
    names=["shoulder", "elbow", "wrist", "gripper", "j5", "j6"],
    positions=[0.0, 0.5, -0.3, 1.2, 0.0, 0.0],
    velocities=[0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
    efforts=[0.1, 0.5, 0.3, 0.8, 0.0, 0.0],
    timestamp_ns=0,  # optional
)

print(f"Joint '{state.names[1]}' at {state.positions[1]:.2f} rad")

temp = horus.Temperature(temperature=22.5, variance=0.1, timestamp_ns=0)

mag = horus.MagneticField(x=0.00002, y=0.00005, z=-0.00004, timestamp_ns=0)  # Tesla

pressure = horus.FluidPressure(pressure=101325.0, variance=10.0, timestamp_ns=0)  # Pascals

lux = horus.Illuminance(illuminance=500.0, variance=5.0, timestamp_ns=0)  # Lux

import horus

def imu_tick(node):
    imu = read_hardware()  # Your I2C/SPI driver

    # Validate before publishing
    if not imu.is_valid():
        return

    node.send("imu.raw", imu)

sensor = horus.Node(
    name="imu_reader",
    pubs=[horus.Imu],
    tick=imu_tick,
    rate=100,
    order=0,
)
horus.run(sensor, rt=True)