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:

FIBEX networks

fibex:FILE sends and receives the CAN and CAN FD frames of one channel of an ASAM FIBEX (MCD-2 NET) file, like dbc: does a DBC's:

with hilspy.open("can://can0", "fibex:network.xml", channel="Chassis") as ecu:
    ecu.send("VehicleStatus", {"Speed": 12.5, "Gear": "D", "BrakePressure": 3.0, "BrakeActive": 0})
    wheels = {"Front.Left": 10.0, "Front.Right": 10.0, "Rear.Left": 10.0, "Rear.Right": 10.0}
    cyclic = ecu.every(0.01, "WheelSpeeds", wheels)  # fields of two PDUs, qualified

• channel selects a channel by short name or ID; it is optional when the file has one channel (otherwise the error lists them).
• Messages are named by frame. Identifiers come from the channel's frame triggerings (EXTENDED-ADDRESSING="true" or an identifier over 0x7FF for extended frames, CAN-FRAME-TX-BEHAVIOR/RX-BEHAVIOR CAN-FD for CAN FD).
• Fields are the signals of all the frame's PDUs (at byte offsets); a name two PDUs share becomes PDU.Signal. A signal's BIT-POSITION is read as a DBC start bit: the least significant bit for Intel, the most significant for Motorola (IS-HIGH-LOW-BYTE-ORDER).
• Codings: IDENTICAL, LINEAR and SCALE-LINEAR (one set of coefficients, plus texttable entries) give physical values, TEXTTABLE names work like DBC value tables, and physical constraints are the range strict checks.
• strict, scaling, decode_choices and e2e work as for dbc:; frames the channel lacks decode as nameless messages. Multiplexed PDUs, high-low byte order PDU instances, texttable ranges and piecewise or non-linear scaling are rejected when the file is opened.

SOME/IP services

someip:FILE sends and receives the SOME/IP messages of the services an ASAM FIBEX 4 file describes, over udp:// or tcp://:

origin = {"latitude": 35.68, "longitude": 139.69, "altitude": 40.0, "quality": "RTK_FIXED"}
route = {"origin": origin, "waypoints": [{"id": 1, "speed": 20}], "name": "Route 1"}
with hilspy.open("udp://192.168.0.20:30490", "someip:services.xml", client_id=0x42) as ecu:
    ecu.send("VehicleService.SetRoute", route)
    for msg in ecu:  # responses, events and field notifications
        print(msg.name, msg["message_type"], msg["session_id"], msg.fields)

• Messages are named Service.Method and Service.Event (FIBEX short names); a field's accessors are Service.get_F_field, Service.set_F_field and its notifier Service.changed_F_field.
• Fields: the header (service_id, method_id, length, client_id, session_id, protocol_version, interface_version, message_type, return_code) and the parameters: input parameters for requests and notifications, return parameters for responses ("message_type": "RESPONSE"). Structures and unions are dicts, arrays lists or array.arrays, enums their names (decode_choices=False for numbers).
• Sending fills the header: message_type REQUEST (methods, accessors) or NOTIFICATION (events, notifiers), return_code E_OK, client_id from the option (default 0; notifications 0, as PRS_SOMEIP_00925 requires), session_id counting per method from 1. Pass any of them to override (a response echoes the request's client_id and session_id). A received message sends back as the same bytes.
• Serialization follows the FIBEX attributes: byte order, bit lengths, length fields of strings, structures, unions and dynamic arrays (LENGTH-FIELD-SIZE, ARRAY-LENGTH-FIELD-SIZE, TYPE-FIELD-SIZE), fixed arrays, UTF-8/UTF-16 and fixed-length strings.
• A datagram may carry several messages; a TCP stream is cut by the length field (a datagram's incomplete remainder is an error, never joined to the next one; stream=True treats source-less packets such as a file:// replay as a stream). Unsigned 64-bit values above 2^63 are Python ints too. Unknown services or methods, SOME/IP-TP segments and SOME/IP-SD (service discovery) arrive as nameless messages with their bytes.

AUTOSAR E2E (alive counter + CRC)

The dbc codec protects chosen messages with an AUTOSAR E2E profile (autosar-e2e: profiles 4, 5, 6, 7, 8, 11, 22). Sending fills the counter (counting per message) and CRC; receiving checks them and adds an e2e field (Ok, CrcError, Repeated, WrongSequence, ...):

e2e = {
    "BrakeCmd": {"profile": 11, "data_id": 0x123, "signals": ["Crc", "Counter", "DataIdNibble"]},
    "SafetyStatus": {"profile": 5, "data_id": 0x4242, "offset": 0, "signals": ["Crc", "Counter"]},
}
with hilspy.open("can://can0", "dbc:chassis.dbc", e2e=e2e) as ecu:
    brake = ecu.every(0.01, "BrakeCmd", {"Pressure": 12.5, "Request": 1})  # counter + CRC per cycle
    for msg in ecu:
        if "e2e" in msg and msg["e2e"] != "Ok":
            print("E2E", msg.name, msg["e2e"])

signals names the signals the protection writes, so they may be left out even with strict. Offsets are AUTOSAR's (bits); profile 11 takes mode (nibble/both), crc_offset, counter_offset, nibble_offset; 22 data_id_list (16 values); all take max_delta_counter.

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

GNSS receivers (NMEA, UBX, RTCM) over serial

clock = hilspy.Clock(start=time.time())
gps = hilspy.open("serial:///dev/ttyUSB0?baud=115200", "nmea", talker="GN", clock=clock)
fix = gps.every(
    0.1, "GGA", {"latitude": lat, "longitude": lon, "fix_type": "Rtk", "fix_satellites": 14}
)

ublox = hilspy.open("serial:///dev/ttyACM0", "ubx")
for msg in ublox:  # interleaved NMEA is skipped
    if msg.name == "NAV-PVT":
        print(msg["lat"], msg["lon"], msg["fix_type"])

Decoded fields are the parsing crate's (names and units, e.g. NAV-PVT in degrees, metres and m/s); encoding takes the same fields, so a received sentence or packet can be sent on, edited or not. Times default to the message's time (the channel's clock). Repeated records (GSV satellites, MSM signal data) arrive as lists of dicts.

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