zelos-can

Rust-first CAN codec for the Zelos ecosystem. DBC-driven signal decode + trace emission at wire speed, with optional python-can–compatible bus implementations for live SocketCAN and in-memory testing.

Install

pip install zelos-can
# python-can compat layer and runtime (optional; only needed for
# can.Bus(interface="zelos-socketcan", ...) or the python-can runtime below):
pip install 'zelos-can[python-can]'

Prebuilt wheels ship for Linux, macOS, and Windows. Headless decode (CanDecoder, VirtualBus, CanCodec) works on all three; the live zelos-socketcan bus is Linux-only (see the feature matrix).

Three ways to use it

1. Drop-in python-can bus (live capture on Linux)

import can
bus = can.Bus(interface="zelos-socketcan", channel="can0")
# Kernel-level filters, SO_RCVBUF tuning, SO_TIMESTAMPNS, reading on
# through link drops and unplugs — all handled in Rust.

Compatible with can.Notifier, bus.send_periodic, bus.state, and the bus.socket escape hatch. Drop-in for interface="socketcan".

2. Bring your own frame source (this is the main pattern)

CanDecoder is a thread-safe handle that takes raw frames and emits decoded signals into the Zelos trace pipeline. Where frames come from is your problem — anything that can produce arbitration_id + data + timestamp works.

The DBC is optional. With no database_file / database_bytes the decoder skips signal decode and just traces the raw frames (log_raw_frames defaults on in that case) — handy for capturing a bus you don't have a DBC for yet:

from zelos_can import CanDecoder
raw = CanDecoder(source_name="can")          # no DBC → raw frames only
raw.decode_frame(arbitration_id=0x123, data=b"\x01\x02")

from zelos_can import CanDecoder
decoder = CanDecoder(database_file="vehicle.dbc", source_name="can")

# From python-can (PEAK/Vector/Kvaser/socketcan/whatever):
for msg in my_bus:
    decoder.decode_message(msg)

# From a network gateway (cannelloni/SLCAN/your protocol):
decoder.decode_frame(
    arbitration_id=0x123,
    data=b"\x01\x02",
    timestamp_ns=time.time_ns(),
    is_extended=False,
    is_fd=False,
)

# From a .asc/.blf/.trc/.mf4 file via python-can readers:
for msg in can.BLFReader("capture.blf"):
    decoder.decode_message(msg)

decode_frame and decode_message release the GIL for the duration of the Rust decode + trace emit — the caller thread stays free.

Complete examples in examples/:

• file_replay.py — .asc/.blf/.trc/.mf4/.log/.csv → .trz
• python_can_forward.py — any python-can interface → live trace + .trz
• udp_source.py — skeleton for a network-CAN gateway
• virtual.py — in-memory VirtualBus + native Rust codec
• socketcan.py — native Rust SocketCAN codec (Linux)

3. Native Rust capture pipeline (fastest)

For the tightest hot path (no Python callbacks per frame), use the native CanCodec with either a VirtualBus or a SocketCAN channel:

from zelos_can import CanCodec, VirtualBus
bus = VirtualBus(channel="sim")
codec = CanCodec(database_file="vehicle.dbc", source_name="sim", bus=bus)
# codec reads frames in Rust; Python never sees individual frames.

Same applies to real hardware via channel="can0" (Linux SocketCAN):

codec = CanCodec(
    database_file="vehicle.dbc",
    source_name="can",
    channel="can0",
    log_raw_frames=True,
    timestamp_mode="hardware",
)

python-can runtime

The bus layer the Zelos CAN, XCP and UDS extensions share: config to a running bus on any python-can interface, DBC decode and transmit, ssh-bridged SocketCAN, offline log conversion. Needs the python-can extra; importing a module without it raises an ImportError naming the extra. zelos_can.bus imports without it; open_python_can_bus needs it.

from pathlib import Path

from zelos_can.bus import open_python_can_bus, prepare_bus_config

entry = {"name": "pt", "interface": "pcan", "channel": "PCAN_USBBUS1", "bitrate": 500_000}
config = prepare_bus_config(entry, Path())  # 2nd arg: the DBC an `interface: demo` bus uses
bus = open_python_can_bus(config, "pt")  # python-can BusABC, init retried 3 times

Trace pipeline

Every decode path funnels through zelos-sdk's TraceNamespace. If zelos_sdk.init() has been called the decoded signals stream live to the Zelos agent over gRPC; wrap your code in a zelos_sdk.TraceWriter(...) context to also record to a .trz file. Both work simultaneously.

Feature matrix

Transmitted frames are traced like any other when the bus is opened with receive_own_messages=True — TX frames echo back through the RX path (with is_rx=False) and flow into the decode/trace pipeline, so a recording captures exactly what hit the wire (including kernel BCM periodics).

Decoder semantics — divergences from cantools

zelos-can decodes per the DBC spec. For well-formed inputs we match cantools exactly.

SIG_VALTYPE_ interpretation

Both decoders treat the : 1 / : 2 marker as advisory and pick the IEEE 754 binary format from the signal length:

For lengths that aren't a real IEEE binary width (e.g. SIG_VALTYPE_:1 + length=10):

• cantools raises at decode time (expected float size of 16, 32, or 64 bits).
• zelos-can preserves is_float so the schema field stays typed as float, but decode_signal returns None per-frame and the decode_errors metric ticks. A parse-time warning surfaces the malformed DBC up front (cantools surfaces nothing at load time).

Storage typing for unsigned signals with negative offset

For an unsigned raw with a DBC offset that drives physical values negative (e.g. (1, -1000) on a u16):

• cantools returns Python int (unbounded — sign is implicit).
• zelos-can promotes the Arrow schema type to the signed sibling (Int{16,32,64}). The typed cache must store a fixed-width Arrow type; the unsigned variant would saturate f64 → uN casts of negative physicals to zero, silently zeroing valid frames.

Everything else

Integer signals, factor/offset on integer raws, big/little endianness, multiplexing, value tables, and well-formed IEEE float widths all decode identically to cantools.

A2L reader (zelos_can.a2l)

Reads an ASAM MCD-2 MC (ASAP2) file into one catalog dict for XCP measurement. Not imported by import zelos_can.

from zelos_can import a2l

cat = a2l.load("ecu.a2l")               # strict=False by default
for m in cat["measurements"]:
    print(m["name"], m["address"], m["datatype"], m["conversion"]["kind"])
print(cat["transports"], cat["events"], cat["warnings"])

The full shape is in the zelos_can.a2l docstring.

• No silent misreads. Parse faults, integers out of range for their field (a2lfile narrows hex literals otherwise) and duplicate names after structure flattening raise a2l.A2lError (a ValueError) with file and line.
• Skipped, with a reason in warnings: calibration components, pointers, MSB_FIRST_MSW_LAST / MSB_LAST_MSW_FIRST, BIT_OPERATION, instance OVERWRITE, missing conversions, transports other than UDP/TCP/CAN.
• Structures: INSTANCEs flatten to inst.member / inst[i].member.
• Conversions: IDENTICAL, LINEAR, RAT_FUNC and TAB_VERB (incl. COMPU_VTAB_RANGE) in full; TAB_INTP, TAB_NOINTP and FORM carried, not evaluated.
• Missing /include: dropped with a warning; strict=True raises instead. XCPlite files reference an XCP_104.aml that is rarely shipped alongside.

J1939 (zelos_can.j1939)

Decoding is the j1939 keyword of CanCodec / CanDecoder ("auto" by default: messages the DBC marks VFrameFormat = J1939PG, or every extended one under a ProtocolType = "J1939" database attribute; "on": every 29-bit message; "off"). The module is the id arithmetic a sender needs, and the reference for what a PG means to the decoder. Not imported by import zelos_can.

from zelos_can import j1939

j1939.pgn(0x0CF00401)                                # 61444, EEC1 from sender 0x01
j1939.can_id(0xF004, source=0x21, priority=3)        # 0x0CF00421
j1939.can_id(0xEF00, source=0x21, destination=0x0B)  # PDU1: the destination is in PS
can_id, data = j1939.request(0xFECA, source=0x21)    # Request DM1 from everyone

Metrics

m = decoder.metrics()
# messages_received / messages_decoded / unknown_messages / decode_errors
# kernel_drops   (SocketCAN only, SO_RXQ_OVFL)
# broadcast_overflows (VirtualBus slow-consumer lag)
# reconnections  (SocketCAN auto-recover)

Tests against nodes (test/hil)

Tiers are pytest markers; a station file says which ones a host provides and the rest skip.

Bench setup: copy test/hil/stations/bench.example.toml to bench.toml (gitignored) and fill in the channels, lock file, firmware dir and flash/reset/console commands. just hil-capture NODE SECONDS records a real capture that replaces the synthetic one.

Benchmarks

Numbers measured on a single apple-silicon core via criterion. Ballpark only — reproduce with cargo bench.

At 9 signals per message the decode path is ~24M full DBC-decoded messages per second in pure Rust, which is well past any realistic CAN bus rate — the GIL crossing on the Python boundary is what actually sets your ceiling, which is why the CanDecoder + Notifier (no per-frame GIL re-entry) path exists.

Broadcast fanout scales sublinearly (16-subscriber case delivers 50M total frames/s vs 21M for one subscriber), so adding listeners costs negligibly in the expected 1-10kHz CAN regime.