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; a name two PDUs share becomes PDU.Signal. A signal's BIT-POSITION is read as a DBC start bit within its PDU: the least significant bit for Intel, the most significant for Motorola (IS-HIGH-LOW-BYTE-ORDER). A PDU instance's BIT-POSITION is the PDU's least significant bit in the frame, at any bit; a high-low instance holds the PDU big-endian, its BIT-POSITION the PDU's most significant bit as for a Motorola signal (so the PDU's last byte comes first, and its Intel signals are Motorola in the frame and vice versa). Off a byte boundary, a Motorola signal that crosses a byte of its PDU is split in the frame and rejected, as is a signal past the end of the frame.
• Multiplexed PDUs (MULTIPLEXER) work like DBC multiplexing: the SWITCH is the multiplexor field, the signals of each switched PDU of the DYNAMIC-PART are present for its SWITCH-CODEs, and the STATIC-PART's signals always. Switched and static PDUs use the multiplexed PDU's bit positions (as AUTOSAR copies their segments, and canmatrix writes them).
• Coded types: unsigned, two's complement (2C, SIGNED), sign-magnitude (SM: the top bit is the sign) and one's complement (1C) integers of up to 64 bits, and IEEE floats; a negative zero decodes to 0.
• Codings: IDENTICAL, LINEAR, SCALE-LINEAR, TEXTTABLE, RAT-FUNC and SCALE-RAT-FUNC compu methods; physical constraints are the range strict checks. A texttable range decodes any raw value in it to its name and encodes the name to its lower limit. Piecewise scales decode by the scale whose raw interval holds the value and encode by inverting the first scale that reaches the physical value; a rational function encodes where it is linear or strictly monotonic over its scale (otherwise sending the signal fails, naming it). A single linear function with exact texttable entries is a plain DBC factor, offset and value table.
• strict, scaling, decode_choices and e2e work as for dbc:; frames the channel lacks decode as nameless messages. FORMULA compu methods, nested multiplexing, switched or static PDU instances off bit 0 or high-low, and encodings other than those above 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