Python Message Library

from horus import CmdVel, Pose2D, Imu, Odometry, LaserScan
from horus import Twist, Transform, MotorCommand, BatteryState, NavGoal, OccupancyGrid

from horus import Pose2D

# Create pose (positional or keyword args)
pose = Pose2D(1.0, 2.0, 0.5)
pose = Pose2D(x=1.0, y=2.0, theta=0.5)

# Origin
origin = Pose2D(0.0, 0.0, 0.0)

# Properties (read/write)
pose.x = 1.5
pose.y = 2.5
pose.theta = 0.785

print(pose)  # Pose2D(x=1.500, y=2.500, theta=0.785, timestamp_ns=0)

from horus import Topic, Pose2D

topic = Topic(Pose2D)  # Typed topic
topic.send(Pose2D(x=1.0, y=2.0, theta=0.5))
pose = topic.recv()  # Returns Pose2D or None

from horus import Twist

twist = Twist(
    linear_x=1.0, linear_y=0.0, linear_z=0.0,   # m/s
    angular_x=0.0, angular_y=0.0, angular_z=0.5  # rad/s
)

# Positional args work too, in the same order
twist = Twist(1.0, 0.0, 0.0, 0.0, 0.0, 0.5)

print(twist)  # Twist(lin=[1.000, 0.000, 0.000], ang=[0.000, 0.000, 0.500], timestamp_ns=0)

from horus import Transform

tf = Transform(
    translation=[1.0, 2.0, 0.0],  # [x, y, z]
    rotation=[0.0, 0.0, 0.0, 1.0]  # Quaternion [x, y, z, w]
)

from horus import Point3, Vector3, Quaternion

point = Point3(x=1.0, y=2.0, z=3.0)
vec = Vector3(x=1.0, y=0.0, z=0.0)
quat = Quaternion(x=0.0, y=0.0, z=0.0, w=1.0)

from horus import CmdVel

# Create velocity command (positional or keyword args)
cmd = CmdVel(1.5, 0.3)
cmd = CmdVel(linear=1.0, angular=0.5)

# Stop command
stop = CmdVel(0.0, 0.0)

# Properties (read/write)
cmd.linear = 1.5   # m/s
cmd.angular = 0.3  # rad/s

print(cmd)  # CmdVel(linear=1.500, angular=0.300, timestamp_ns=0)

from horus import Topic, CmdVel

topic = Topic(CmdVel)  # Typed topic
topic.send(CmdVel(linear=1.0, angular=0.5))
cmd = topic.recv()  # Returns CmdVel or None

from horus import LaserScan

# Create laser scan with parameters
scan = LaserScan(
    angle_min=-3.14159,       # Start angle (radians)
    angle_max=3.14159,        # End angle (radians)
    angle_increment=0.01745,  # Angular resolution (radians)
    range_min=0.1,            # Minimum valid range (meters)
    range_max=10.0,           # Maximum valid range (meters)
    ranges=[1.0, 1.1, 1.2],  # Distance measurements (zero-padded to 360 entries)
)

# Properties (read/write)
scan.ranges = [1.0] * 360  # Always stored as 360 floats
scan.angle_min = -1.57

# Length = readings inside [range_min, range_max], not the array size
length = len(scan)

print(scan)  # LaserScan(angle=[-1.57, 3.14], range=[0.10, 10.00], 360 points, timestamp_ns=0)

from horus import Topic, LaserScan

topic = Topic(LaserScan)  # Typed topic

# Send
scan = LaserScan(angle_min=-3.14, angle_max=3.14, ranges=read_lidar())
topic.send(scan)

# Receive
scan = topic.recv()  # Returns LaserScan or None
if scan:
    print(f"Got {len(scan)} valid range readings")

from horus import Imu

# Create with positional or keyword args
imu = Imu(0.0, 0.0, 9.81, 0.0, 0.0, 0.0)
imu = Imu(
    accel_x=0.0, accel_y=0.0, accel_z=9.81,  # m/s²
    gyro_x=0.0, gyro_y=0.0, gyro_z=0.1,       # rad/s
)

# Properties (read/write)
imu.accel_x = 1.0
imu.gyro_z = 0.5

print(imu)  # Imu(accel=[1.000, 0.000, 9.810], gyro=[0.000, 0.000, 0.500], timestamp_ns=0)

from horus import Odometry

# All fields have defaults (0.0)
odom = Odometry()
odom = Odometry(x=1.0, y=2.0, theta=0.5, linear_velocity=0.5, angular_velocity=0.1)

# Properties (read/write)
odom.x = 1.0                # meters
odom.y = 2.0                # meters
odom.theta = 0.5            # radians
odom.linear_velocity = 0.5  # m/s (scalar)
odom.angular_velocity = 0.1 # rad/s (scalar)

print(odom)  # Odometry(x=1.000, y=2.000, theta=0.500, v_lin=0.500, v_ang=0.100, timestamp_ns=0)

from horus import BatteryState

battery = BatteryState(voltage=12.6, percentage=85.0)
battery.current = 2.5        # Amps (positive = discharging)
battery.temperature = 25.0   # Celsius

from horus import NavSatFix

gps = NavSatFix(latitude=37.7749, longitude=-122.4194, altitude=10.0)

from horus import RangeSensor

range_sensor = RangeSensor()
range_sensor.range = 1.5       # meters
range_sensor.min_range = 0.02  # minimum valid range
range_sensor.max_range = 4.0   # maximum valid range
range_sensor.field_of_view = 0.26  # radians (~15°)

from horus import MotorCommand

cmd = MotorCommand()
cmd.motor_id = 0            # Motor index (u8)
cmd.mode = 0                # Control mode (u8: 0=velocity, 1=position, 2=torque, 3=voltage)
cmd.target = 1.0            # Target value (f64, units depend on mode)
cmd.max_velocity = 10.0     # Velocity limit (f64)
cmd.max_acceleration = 5.0  # Acceleration limit (f64)
cmd.feed_forward = 0.0      # Feed-forward term (f64)
cmd.enable = True           # Enable motor (bool — an int raises TypeError)

from horus import DifferentialDriveCommand

cmd = DifferentialDriveCommand()
cmd.left_velocity = 1.0   # Left wheel velocity (m/s or rad/s)
cmd.right_velocity = 1.0  # Right wheel velocity

from horus import ServoCommand

cmd = ServoCommand()
cmd.servo_id = 0          # Servo index (u8)
cmd.position = 1.57       # Target position in radians (f32)
cmd.speed = 1.0           # Normalized speed, clamped to 0.0-1.0 (f32)
cmd.enable = True         # Torque enable (bool — an int raises TypeError)

from horus import PidConfig

pid = PidConfig()
pid.kp = 1.0   # Proportional gain
pid.ki = 0.1   # Integral gain
pid.kd = 0.05  # Derivative gain

from horus import Image

# Create image (allocates from pool). Encoding is a plain lowercase string.
img = Image(height=480, width=640, encoding="rgb8")

# Or wrap an existing array
img = Image.from_numpy(arr)

# Copy pixel data in
img.copy_from(pixel_bytes)

# Access data — height/width/encoding are properties, not methods
arr = img.to_numpy()     # zero-copy (H, W, C) numpy view
h = img.height
w = img.width
enc = img.encoding       # e.g. 'rgb8'
px = img.pixel(10, 20)         # single pixel as bytes
crop = img.roi(0, 0, 320, 240)  # cropped region as bytes

from horus import CompressedImage

img = CompressedImage(
    format="jpeg",     # "jpeg" or "png"
    data=jpeg_bytes,   # Compressed image data
    width=640,
    height=480,
)

img.data = new_jpeg_bytes
print(img.data_len())   # method, not a property

from horus import CameraInfo

info = CameraInfo(
    width=640, height=480,
    fx=525.0, fy=525.0,     # Focal lengths in pixels
    cx=320.0, cy=240.0,     # Principal point in pixels
)

# with_distortion_model returns a NEW CameraInfo — there is no distortion_model attribute
info = info.with_distortion_model("plumb_bob")

# Read-only properties
info.camera_matrix            # 3x3 intrinsic matrix, 9 floats (row-major)
info.distortion_coefficients  # Up to 8 distortion coefficients
info.rectification_matrix     # 3x3 rectification matrix
info.projection_matrix        # 3x4 projection matrix, 12 floats (row-major)

# width and height stay writable
info.width = 1280
info.height = 720

from horus import Detection, BoundingBox2D

det = Detection(
    class_name="person",  # Class label (truncated to 31 bytes)
    confidence=0.95,      # Detection confidence 0-1 (f32)
    x=100.0, y=50.0,      # Top-left corner in pixels (f32)
    width=200.0,          # Width in pixels (f32)
    height=400.0,         # Height in pixels (f32)
    class_id=0,           # Class index (u32)
    instance_id=1,        # Instance ID for tracking (u32)
)

# The scalar fields are read/write
det.confidence = 0.92
det.class_name = "cyclist"

det.bbox = BoundingBox2D(100.0, 50.0, 200.0, 400.0)  # x, y, width, height

from horus import NavGoal

goal = NavGoal()
goal.x = 10.0                  # Target X position
goal.y = 5.0                   # Target Y position
goal.theta = 0.0               # Target orientation
goal.tolerance_position = 0.1  # Position tolerance (meters)
goal.tolerance_angle = 0.1     # Orientation tolerance (radians)

# Or via the constructor — note the kwargs are named the other way round:
goal = NavGoal(x=10.0, y=5.0, theta=0.0,
               position_tolerance=0.1, angle_tolerance=0.1, timeout=30.0)

from horus import GoalResult

result = GoalResult(goal_id=1, status=2, progress=1.0)  # progress is a 0.0-1.0 fraction
result.status  # 0=Pending, 1=Active, 2=Succeeded, 3=Aborted,
               # 4=Cancelled, 5=Preempted, 6=TimedOut

from horus import NavPath, Waypoint

# Waypoints are appended one at a time (max 256); there is no `waypoints` attribute
path = NavPath()
for wp in [
    Waypoint(x=0.0, y=0.0, theta=0.0),
    Waypoint(x=5.0, y=0.0, theta=0.0),
    Waypoint(x=5.0, y=5.0, theta=1.57),
]:
    path.add_waypoint(wp)

print(path.waypoint_count)   # 3 (read-only property)
print(path.total_length)     # 0.0 — a plain writable field in meters; add_waypoint
                             # does not accumulate it, so set it yourself if you need it
for wp in path.get_waypoints():
    print(wp.pose)

from horus import OccupancyGrid, Pose2D

# Dimensions are fixed at construction; width/height are read-only afterwards
grid = OccupancyGrid(
    width=100,        # Grid width (cells)
    height=100,       # Grid height (cells)
    resolution=0.05,  # Meters per cell
)

grid.origin = Pose2D(-2.5, -2.5, 0.0)   # Map origin (single Pose2D, not origin_x/origin_y)
grid.data = [0] * (100 * 100)           # i8 values: 0=free, 100=occupied, -1=unknown

# Helpers
grid.set_occupancy(10, 10, 100)
print(grid.occupancy(10, 10))        # 100
print(grid.world_to_grid(0.0, 0.0))  # (50, 50)

from horus import CostMap, OccupancyGrid

# A CostMap wraps an OccupancyGrid and inflates obstacles into it
grid = OccupancyGrid(width=100, height=100, resolution=0.05)
costmap = CostMap(grid=grid, inflation_radius=0.55)  # meters

costmap.compute_costs()
c = costmap.cost(1.0, 2.0)   # Cost 0-255 at world coordinates (x, y)

# costmap.costs is read-only (bytes, one u8 per cell).
# Also writable: inflation_radius, cost_scaling_factor, lethal_cost

from horus import PointCloud

# Create point cloud (allocates from pool)
cloud = PointCloud(num_points=1000)             # XYZ float32 by default
cloud = PointCloud(num_points=1000, fields=4)   # XYZI

# Or wrap existing point data ((N, 3) or (N, 4) array)
cloud = PointCloud.from_numpy(arr)

# Access data
arr = cloud.to_numpy()       # zero-copy (N, 3) numpy view
n = cloud.point_count
xyz = cloud.is_xyz()
p = cloud.point_at(0)

from horus import DepthImage

# Create depth image (allocates from pool)
depth = DepthImage(height=480, width=640)   # float32 meters
depth = DepthImage(height=480, width=640, dtype="uint16")  # uint16 millimeters

# Or wrap existing data
depth = DepthImage.from_numpy(arr)

# Access data and parameters (height/width/depth_scale are properties, not calls)
arr = depth.to_numpy()            # zero-copy (H, W) numpy view
h, w = depth.height, depth.width
scale = depth.depth_scale         # Scale factor (f32)
d = depth.get_depth(320, 240)     # Single depth reading in meters
stats = depth.depth_statistics()  # (min, max, mean) in meters over valid pixels, or None

from horus import BoundingBox3D

box = BoundingBox3D(
    cx=1.0, cy=2.0, cz=0.5,             # Center position [x, y, z]
    length=0.5, width=0.5, height=1.8,  # Dimensions in meters
    yaw=0.0,                            # Heading in radians
)

# Full 3-axis rotation (roll/pitch/yaw), positional only:
box = BoundingBox3D.with_rotation(1.0, 2.0, 0.5, 0.5, 0.5, 1.8, 0.0, 0.0, 0.0)

print(box.volume)  # 0.44999998... (volume is computed in f32)

from horus import WrenchStamped

wrench = WrenchStamped(
    fx=10.0, fy=0.0, fz=-9.81,   # Force [fx, fy, fz] in Newtons
    tx=0.0, ty=0.5, tz=0.0,      # Torque [tx, ty, tz] in Nm
)

print(wrench.force_magnitude())   # 14.008...
print(wrench.torque_magnitude())  # 0.5

from horus import TactileArray

tactile = TactileArray(rows=4, cols=4)   # array is sized by the constructor
tactile.forces = [0.0] * 16              # Per-taxel pressure values (f32)

tactile.set_force(1, 2, 3.5)             # Set one taxel
print(tactile.get_force(1, 2))

from horus import ForceCommand

cmd = ForceCommand(
    fx=0.0, fy=0.0, fz=-10.0,   # Desired force [fx, fy, fz] in Newtons
    tx=0.0, ty=0.0, tz=0.0,     # Desired torque [tx, ty, tz] in Nm
)

# Individual scalar properties (read/write)
cmd.fz = -12.0
cmd.timeout_seconds = 0.5

# Python sender
from horus import Topic, CmdVel

topic = Topic(CmdVel)  # Typed topic
topic.send(CmdVel(linear=1.0, angular=0.5))

// Rust receiver
use horus::prelude::*;

let topic: Topic<CmdVel> = Topic::new("cmd_vel")?;
if let Some(cmd) = topic.recv() {
    println!("Received: linear={}, angular={}", cmd.linear, cmd.angular);
}

from horus import Node, Topic, CmdVel, LaserScan

# Create topics outside tick function
scan_topic = Topic(LaserScan)
cmd_topic = Topic(CmdVel)

def controller_tick(node):
    scan = scan_topic.recv()
    if scan:
        # Simple obstacle avoidance — min_range() skips the zero padding
        min_dist = scan.min_range()  # nearest valid reading, or None
        if min_dist and min_dist < 0.5:
            # Too close - stop
            cmd_topic.send(CmdVel(0.0, 0.0))
        else:
            # Safe - move forward
            cmd_topic.send(CmdVel(linear=0.5, angular=0.0))

node = Node(name="controller", tick=controller_tick, rate=10)

import math
from horus import Topic, Pose2D

# Track robot pose
pose_topic = Topic(Pose2D)

current_pose = Pose2D(0.0, 0.0, 0.0)

def update_pose(delta_x, delta_y, delta_theta):
    global current_pose
    current_pose.x += delta_x
    current_pose.y += delta_y
    current_pose.theta += delta_theta

    # Normalize angle to [-pi, pi]
    current_pose.theta = math.atan2(
        math.sin(current_pose.theta),
        math.cos(current_pose.theta)
    )

    pose_topic.send(current_pose)