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".

zelos-virtual (in-memory) and zelos-ssh-socketcan (a remote Linux bus over ssh + the edge's candump/cansend, any OS, channel [user@]host:can0) are registered the same way. Over ssh the edge hears this bus's own transmits like any frame, so they are always received.

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 file via python-can readers (.mf4/.log: zelos_can.LogReader):
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

interface is a python-can interface name, passed as is: zelos-socketcan, zelos-ssh-socketcan (with the remote_host / ssh_* fields), socketcan, pcan, ...

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

A CanCodec sends as a J1939 node on any bus it owns:

codec.j1939_claim(name=0x1000, address=0x80)       # usable 250 ms later, if uncontested
codec.j1939_address()                              # 0x80, or None while claiming / lost
codec.j1939_send(0xFECA, dm1).wait(timeout=1)      # > 8 bytes to everyone: BAM
codec.j1939_send(0xEF00, data, destination=0x21).wait(timeout=5)  # to one node: RTS/CTS
codec.j1939_send(0xFECA, dm1, fd=True).wait(timeout=1)  # J1939-22, on a CanCodec(fd=True)

CANopen (zelos_can.canopen)

Decoding is the canopen / canopen_nodes keywords of CanCodec / CanDecoder, on every lane (native socketcan, ExternalBus, offline convert): "auto" (default) is on with at least one node, "on" decodes the protocol for every node-id, "off". The module holds the EDS/DCF reader and the ids and NMT frame a master sends. Not imported by import zelos_can.

from zelos_can import CanDecoder, canopen

decoder = CanDecoder(canopen_nodes=[{"node_id": 0x20, "name": "pdu", "file": "pdu.eds"}])
decoder.canopen_nodes()      # [{node_id, name, configured, state, heartbeat_lost, ...}]
decoder.canopen_warnings()   # file, construction and learning warnings, each once
cob, data = canopen.nmt_frame(canopen.NMT_START, 0)      # start every node
canopen.cob_id(canopen.TPDO1, 0x20)                      # 0x1A0
canopen.describe_abort(0x06020000)  # "object does not exist in the object dictionary"

Tables are named {cob:04x} ({cob:08x} for a 29-bit PDO), then the node's name, e.g. 0720_pdu_Heartbeat, 05a0_pdu_SDO, 01a0_pdu_TPDO1.

SDO rows: index, subindex, name, direction (upload|download), kind (expedited|segmented|block, null until the side that picks it spoke), size, data (hex, first 64 bytes), truncated, value, text, abort_code, abort_text, duration_ms, resync.

A live CanCodec (native socketcan, VirtualBus, ExternalBus / ssh) is also a client:

codec.sdo_read(0x20, 0x1018, 1)            # (node_id, index, subindex=0, timeout=1.0)
codec.sdo_write(0x20, 0x6200, 1, b"\x01")  # (node_id, index, subindex, data, timeout=1.0)
codec.nmt("start", 0x20)  # start | stop | pre_operational | reset_node | reset_communication; node 0: all

Metrics: canopen_frames, canopen_sdo_completed, canopen_sdo_aborted, canopen_sdo_resync, canopen_sdo_crc_error, canopen_short_frame, canopen_heartbeat_late, canopen_heartbeat_lost, canopen_pdo_short, canopen_pdo_remapped, canopen_pdo_unlearned, canopen_pdo_reverted, canopen_unknown_node, canopen_client_sdo, canopen_client_timeout, canopen_client_stale.

EDS/DCF reader

Reads a CiA 306 EDS or DCF in Rust into one catalog dict. Anything it cannot carry faithfully is a load error or skipped and named in warnings, never a silently different value.

from zelos_can import canopen

cat = canopen.load("pdu.eds", node_id=0x20)   # strict=False by default
for p in cat["pdos"]:
    print(p["kind"], p["n"], hex(p["cob_id"]), [e["name"] for e in p["entries"]])
print(cat["comm"]["sdo"], cat["comm"]["emcy"], cat["warnings"])

• Node id: the node_id argument, else the DCF's NodeID. Resolves $NODEID and the predefined COB-IDs; without one they stay "$NODEID+0x..." and PDO cob_id is None.
• Values: decimal, 0x hex, octal with a leading 0, $NODEID+n or n+$NODEID; whitespace and ; comments (at line start or after whitespace) ignored. Section and key names are case-insensitive.
• Skipped, with a reason in warnings: unknown data or object types, unparsable values. Duplicate sections merge and duplicate keys take the later value, warned.
• Errors: canopen.CanopenError (a ValueError) with file and line for an unreadable file, a line that is neither [section] nor key=value, or a node_id outside 1..127; strict=True raises the first warning.
• Encoding: ISO-8859-1 per CiA 306; a UTF-8 BOM is dropped.

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. The pdu node (CANopen node 0x20, test/hil/nodes/pdu.eds) runs on every tier for decode, and on hil against python-canopen as master for its own behavior (test_pdu_conformance.py).

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.