hilspy

HILS (hardware-in-the-loop simulation) I/O for Python on a Rust core, built to be embedded in a scenario library (such as autoware_carla_scenario).

One call opens a channel: a transport (how packets travel) composed with a codec (what they mean). Both are pluggable, in Rust or in Python.

import hilspy

with hilspy.open("can://vcan0", "dbc:vehicle.dbc") as ecu:
    ecu.send("VehicleSpeed", {"Speed": 12.5, "Accel": 0.0, "Gear": "D", "Counter": 0})
    for msg in ecu:  # until ecu.close()
        print(msg.name, msg.fields)  # "SteeringCmd" {"SteeringAngle": -3.5, ...}

Install

Python 3.10+. One abi3 wheel per platform serves every Python version; wheels are built for Linux (x86_64, aarch64), macOS (arm64, x86_64) and Windows x64. can:// (SocketCAN) is Linux-only; every other transport and codec works everywhere.

pip install hilspy
uv add hilspy
pip install git+https://github.com/hakuturu583/hilspy   # from source (needs Rust)

From a checkout:

uv sync                             # builds the extension with maturin
uv run pytest

Versions follow semver and are released by labelling the PR (bump patch / bump minor / bump major); see docs/releasing.md.

API

The whole public API:

Synchronous simulation (virtual clock)

Devices that do I/O on a fixed period (cyclic CAN frames, a GNSS receiver at 10 Hz, a LiDAR streaming packets through each revolution) can run on simulated time instead of the wall clock. Open their channels with a shared hilspy.Clock; every send then happens inside clock.advance(dt), in time order, when the simulation steps past it:

clock = hilspy.Clock(start=carla_world_time)
ecu = hilspy.open("can://vcan0", "dbc:vehicle.dbc", clock=clock)
lidar = hilspy.open("udp://192.168.1.100:2368", "lidar:pandar_xt32", clock=clock)
speed = ecu.every(0.01, "VehicleSpeed", {"Speed": 0.0, "Gear": "D"})
scans = lidar.every(0.1, "scan", first_scan)

while running:
    world.tick()  # e.g. CARLA synchronous mode, 50 ms
    speed.update("VehicleSpeed", {"Speed": ego_speed(), "Gear": "D"})
    scans.update("scan", raycast_points())
    clock.advance(0.05)  # 5 speed frames, half a revolution of packets

On a clock:

• every() sends at once, then each period as the clock passes it (no thread, nothing between steps).
• A paced message's packets (a LiDAR revolution) go out as the clock reaches their times, so a 50 ms step sends the packets of 50 ms.
• Messages are stamped with the clock's time. Codecs write that time (the SBF TOW, a LiDAR scan's timestamps), and received messages carry the time they arrived at.

asyncio and several buses at once

Every channel also has await ch.asend(...), await ch.arecv(timeout), async for msg in ch and async with. hilspy.merge(*channels) receives from many buses at once, yielding (channel, message) as messages arrive, with for or async for:

async with (
    hilspy.open("can://can0", "dbc:powertrain.dbc") as can0,
    hilspy.open("can://can1", "dbc:chassis.dbc") as can1,
    hilspy.open("tcp://192.168.3.1:28784", "sbf") as gnss,
    hilspy.merge(can0, can1, gnss) as buses,
):
    async for bus, msg in buses:
        if bus is gnss and msg.name == "PVTGeodetic":
            await can1.asend("GnssPosition", {"Lat": msg["latitude"], "Lon": msg["longitude"]})

Each bus is received on its own Rust thread, so awaiting never blocks the event loop and a quiet or slow bus does not delay the others. A channel has one consumer at a time: once received with arecv/async for/merge, its synchronous recv refuses.

A message with a name goes through the codec; a nameless one (Message(data=b"...", id=0x123)) is sent as is. Messages the codec does not know (a CAN id missing from the DBC) arrive nameless.

GNSS receiver simulation (SBF over TCP)

with hilspy.open("tcp://0.0.0.0:28784?listen", "sbf") as receiver:
    receiver.send(
        "PVTGeodetic", {"mode": 4, "latitude": lat_rad, "longitude": lon_rad, "height": 40.0}
    )
    receiver.send("AttEuler", {"heading": 90.0, "pitch": 0.0, "roll": 0.0})
    # tow/wnc default to the current GPS time

LiDAR (Hesai, Velodyne)

lidar:MODEL decodes a sensor's UDP data packets into point clouds, and encodes point clouds into the packets a sensor would send (to stand in for one in the loop). Sensors are profiles, not code:

Names ignore case, - and _ (lidar:VLP-16, lidar:XT32).

# Driver side: one message per revolution.
with hilspy.open("udp://0.0.0.0:2368?listen", "lidar:pandar_xt32") as lidar:
    for scan in lidar:
        xyz = np.column_stack([np.frombuffer(scan[k], np.float32) for k in "xyz"])

# Sensor side: send one revolution of a simulated point cloud.
with hilspy.open("udp://192.168.1.100:2368", "lidar:vlp16") as sensor:
    sensor.send("scan", {"x": xs, "y": ys, "z": zs, "intensity": 100})

A scan has arrays (array.array, readable by numpy without copying) x, y, z (metres; x forward, y left, z up), intensity, channel, azimuth (degrees, clockwise from forward), distance, return and time (seconds after timestamp), plus timestamp, model, return_mode, rpm and packets. Sending takes x/y/z (each point goes to the channel of the nearest elevation) or distance/azimuth/channel, as lists, tuples, array.array or numpy arrays; a received scan can be sent back as is. Sending one revolution spreads its packets over the revolution's duration (100 ms at 600 rpm), as the sensor does, so a driver's socket buffer is not flooded; send returns when the last packet is out (pace=False sends them at once). Codec options: frame="packet" (a message per packet), phase (where scans start, degrees), rpm, return_mode and pace (encoding), profiles=FILE.

Profiles are protobuf messages (schema: lidar_profile.proto): packet layout, per-channel elevation / azimuth offset / firing time, return modes, timing and range. The defaults are written as text in lidar/profiles/ and committed compiled as lidar/profiles.binpb, which hilspy embeds. To describe another sensor, write a ProfileSet in text format, compile it with any protobuf tool (protoc --encode=hilspy.lidar.ProfileSet lidar_profile.proto < my_lidars.txtpb > my_lidars.binpb) and pass it with profiles="my_lidars.binpb"; it is searched before the defaults. Channel angles are nominal values from tier4/nebula's calibration files; per-unit calibration is not applied.

Plugins

A transport is any object with send(packet) and recv(timeout) (and optionally close()); a codec has encode(message) and decode(packet). Packets are nameless Messages.

class SerialTransport:
    def __init__(self, address: str, options: dict[str, str]):
        self.port = serial.Serial(address, int(options.get("baud", 115200)))

    def send(self, packet):
        self.port.write(packet.data)

    def recv(self, timeout):
        self.port.timeout = timeout
        data = self.port.read(self.port.in_waiting or 1)
        return hilspy.Message(data=data) if data else None


hilspy.register_transport("serial", SerialTransport)
gnss = hilspy.open("serial:///dev/ttyACM0?baud=921600", "sbf")

Rust crates register through hilspy_core::transport::register and hilspy_core::codec::register.

A virtual CAN interface for local testing:

sudo modprobe vcan
sudo ip link add dev vcan0 type vcan
sudo ip link set vcan0 mtu 72        # CAN FD
sudo ip link set up vcan0

Examples

uv run python examples/virtual_ecu.py                              # cyclic DBC message, in-process
uv run python examples/virtual_ecu.py --url can://vcan0
uv run python examples/gnss_simulator.py "tcp://127.0.0.1:28784?listen"
uv run python examples/gnss_simulator.py "file://sim.sbf?write" --seconds 5
uv run python examples/sbf_dump.py file://sim.sbf
uv run python examples/multi_bus.py                                # asyncio: CAN + GNSS at once

Development

uv sync                                 # build + dev dependencies (cantools for cross-checks)
cargo test -p hilspy-core               # Rust unit tests
uv run pytest                           # Python tests
HILSPY_TEST_VCAN=vcan0 uv run pytest tests/test_socketcan.py   # needs a vcan interface
uv run pre-commit run --all-files       # ruff, mypy, cargo fmt, clippy

Design notes: docs/design.md.

License

Apache-2.0