CAN node emulation and conformance
The AT1032S joins a CAN bus as a full node, with its 120 Ω termination switched on in software and acceptance filters applied in hardware. That lets it do two jobs at once: check that the board under test answers requests correctly, and stand in for the ECUs that are not on the bench, so a control board runs as though the whole machine were around it.
One node on the bus, playing tester and missing-ECU at the same time.
Wiring
| AT1032S | Bus |
|---|---|
DA24 CANH | CAN high |
DA25 CANL | CAN low |
The test sequence
- NodeJS
- Python
import { AT1000 } from '@ikalogic/at1000';
// ---- The bus and the frames we care about ---------------------------------
const BAUD_RATE = 500000;
const TERMINATION = true; // switch the 120 ohm on when we are at the bus end
const REQUEST_ID = 0x120; // what we ask the board
const REQUEST_DATA = [0x01, 0x00]; // a "report status" command on this bus
const REPLY_ID = 0x121; // what the board must answer with
const REPLY_BYTES = 4; // how long a good answer is
const REPLY_TIMEOUT_MS = 200;
// Frames a missing ECU would normally put on the bus.
const EMULATED = [
{ id: 0x300, data: [0x01, 0x00, 0x00, 0x00], period_ms: 100, name: 'sensor ECU' },
{ id: 0x310, data: [0x00, 0x80], period_ms: 200, name: 'actuator ECU' },
];
const RUN_SECONDS = 10;
// ---------------------------------------------------------------------------
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const hex = (n) => '0x' + n.toString(16).toUpperCase();
const devices = await AT1000.findDevices();
const tester = await AT1000.open(devices[0]);
await tester.reset();
const can = tester.com.can(0);
// Accept only the reply id. The mask clears the bits that must match exactly.
await can.configure({
enabled: true,
baud_rate: BAUD_RATE,
termination_resistors: TERMINATION,
rx_filter: { mask: 0x7FF, id: REPLY_ID, extended_id: false },
});
await can.start_rx();
// ---- 1. Request and reply -------------------------------------------------
await can.rx(); // drop anything already buffered
await can.tx({ id: REQUEST_ID, data: REQUEST_DATA });
let reply = null;
const deadline = Date.now() + REPLY_TIMEOUT_MS;
while (Date.now() < deadline && reply === null) {
const frames = await can.rx();
reply = frames.find((frame) => frame.id === REPLY_ID) ?? null;
if (reply === null) await sleep(10);
}
if (reply === null) {
console.log(`FAIL: no reply on ${hex(REPLY_ID)} within ${REPLY_TIMEOUT_MS} ms`);
} else {
console.log(`Reply ${hex(reply.id)}: ${reply.data.map((b) => hex(b)).join(' ')}`);
console.log(reply.data.length === REPLY_BYTES
? 'Reply length ok'
: `FAIL: ${reply.data.length} bytes`);
}
// ---- 2. Stand in for the ECUs that are not on the bench --------------------
const names = EMULATED.map((e) => e.name).join(' and ');
console.log(`\nEmulating ${names} for ${RUN_SECONDS} s.`);
const started = Date.now();
const lastSent = EMULATED.map(() => 0);
while (Date.now() - started < RUN_SECONDS * 1000) {
const now = Date.now();
for (let i = 0; i < EMULATED.length; i++) {
if (now - lastSent[i] >= EMULATED[i].period_ms) {
await can.tx({ id: EMULATED[i].id, data: EMULATED[i].data });
lastSent[i] = now;
}
}
// Anything the board sends back while we play the missing ECUs.
for (const frame of await can.rx()) {
console.log(` board sent ${hex(frame.id)}: ` +
`${frame.data.map((b) => hex(b)).join(' ')}`);
}
await sleep(10);
}
await can.stop_rx();
await can.disable();
await tester.reset();
from ikalogic_at1000 import AT1000
import time
# ---- The bus and the frames we care about ------------------------------------
BAUD_RATE = 500000
TERMINATION = True # switch the 120 ohm on when we are at the bus end
REQUEST_ID = 0x120 # what we ask the board
REQUEST_DATA = [0x01, 0x00] # a "report status" command on this bus
REPLY_ID = 0x121 # what the board must answer with
REPLY_BYTES = 4 # how long a good answer is
REPLY_TIMEOUT_S = 0.2
# Frames a missing ECU would normally put on the bus.
EMULATED = [
{"id": 0x300, "data": [0x01, 0x00, 0x00, 0x00], "period_s": 0.1, "name": "sensor ECU"},
{"id": 0x310, "data": [0x00, 0x80], "period_s": 0.2, "name": "actuator ECU"},
]
RUN_SECONDS = 10
# -------------------------------------------------------------------------------
devices = AT1000.find_devices()
tester = AT1000.open(devices[0])
tester.reset()
can = tester.com.can(0)
# Accept only the reply id. The mask clears the bits that must match exactly.
can.configure(
enabled=True,
baud_rate=BAUD_RATE,
termination_resistors=TERMINATION,
rx_filter={"mask": 0x7FF, "id": REPLY_ID, "extended_id": False},
)
can.start_rx()
# ---- 1. Request and reply ------------------------------------------------------
can.rx() # drop anything already buffered
can.tx({"id": REQUEST_ID, "data": REQUEST_DATA})
reply = None
deadline = time.monotonic() + REPLY_TIMEOUT_S
while time.monotonic() < deadline and reply is None:
for frame in can.rx():
if frame.id == REPLY_ID:
reply = frame
break
if reply is None:
time.sleep(0.01)
if reply is None:
print(f"FAIL: no reply on {hex(REPLY_ID)} within {REPLY_TIMEOUT_S * 1000:.0f} ms")
else:
print(f"Reply {hex(reply.id)}: {' '.join(hex(b) for b in reply.data)}")
print("Reply length ok" if len(reply.data) == REPLY_BYTES
else f"FAIL: {len(reply.data)} bytes")
# ---- 2. Stand in for the ECUs that are not on the bench ------------------------
names = " and ".join(e["name"] for e in EMULATED)
print(f"\nEmulating {names} for {RUN_SECONDS} s.")
started = time.monotonic()
last_sent = [0.0] * len(EMULATED)
while time.monotonic() - started < RUN_SECONDS:
now = time.monotonic()
for i, frame in enumerate(EMULATED):
if now - last_sent[i] >= frame["period_s"]:
can.tx({"id": frame["id"], "data": frame["data"]})
last_sent[i] = now
# Anything the board sends back while we play the missing ECUs.
for frame in can.rx():
print(f" board sent {hex(frame.id)}: {' '.join(hex(b) for b in frame.data)}")
time.sleep(0.01)
can.stop_rx()
can.disable()
tester.reset()
tester.close()
Adapting it to your bus
BAUD_RATEcovers 10 kbit/s to 1 Mbit/s, andTERMINATIONswitches the 120 Ω on when the AT1032S sits at the end of the bus.rx_filterdecides what reaches your script. Widen the mask to take a range of ids, or set the mask to0to receive everything on the bus.EMULATEDis the list of frames the AT1032S puts on the bus on behalf of hardware that is not present. Add one entry per missing node.- The ids and payloads above stand in for whatever your bus actually speaks. Replace
REQUEST_DATA,REPLY_BYTESand theEMULATEDframes with your own. A protocol layered on top of CAN, such as UDS or CANopen, adds its own framing rules on top of the raw frames shown here. - For a worked account of this approach on a real installation, see reproducing a factory CANopen and Modbus network on the bench.