RS485 multi-drop network poll
An RS485 bus with a dozen nodes on it is only as good as its wiring, its termination and its biasing. The AT1032S drives the bus directly from a built-in RS485 port, so proving that every node is present and answering is a short script: poll each address in turn with a Modbus RTU request, check the reply, and name whatever stays silent.
Every node polled in turn. The one that does not answer is named in the summary.
Wiring​
| AT1032S | Bus |
|---|---|
DA26 A(+) | RS485 A |
DA27 B(−) | RS485 B |
The test sequence​
- NodeJS
- Python
import { AT1000 } from '@ikalogic/at1000';
// ---- The bus and the nodes on it ------------------------------------------
const BAUD_RATE = 19200;
const TERMINATION = true; // 120 ohm, switched on when we sit at the bus end
const NODES = [0x01, 0x02, 0x03, 0x04];
const START_REGISTER = 0x0000;
const REGISTER_COUNT = 2;
const REPLY_TIMEOUT_MS = 300;
// ---------------------------------------------------------------------------
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
// Modbus RTU frame check, low byte first.
function crc16(bytes) {
let crc = 0xFFFF;
for (const byte of bytes) {
crc ^= byte;
for (let bit = 0; bit < 8; bit++) {
crc = (crc & 1) ? ((crc >> 1) ^ 0xA001) : (crc >> 1);
}
}
return [crc & 0xFF, (crc >> 8) & 0xFF];
}
// Function 3: read holding registers.
function readHoldingRegisters(address, start, count) {
const body = [address, 0x03, (start >> 8) & 0xFF, start & 0xFF,
(count >> 8) & 0xFF, count & 0xFF];
return [...body, ...crc16(body)];
}
const devices = await AT1000.findDevices();
const tester = await AT1000.open(devices[0]);
await tester.reset();
const rs485 = tester.com.rs485(1);
await rs485.enable({
baud_rate: BAUD_RATE,
data_bits: 8,
parity: 'none',
stop_bits: '1',
termination_resistors: TERMINATION,
});
await rs485.start_rx();
// Poll one node and return its registers, or null when it stays silent.
async function poll(address) {
await rs485.rx(); // clear anything left from the last node
await rs485.tx(readHoldingRegisters(address, START_REGISTER, REGISTER_COUNT));
const expected = 5 + REGISTER_COUNT * 2; // addr, fn, count, data, 2 crc bytes
const deadline = Date.now() + REPLY_TIMEOUT_MS;
let frame = [];
while (Date.now() < deadline && frame.length < expected) {
frame = frame.concat(await rs485.rx());
if (frame.length < expected) await sleep(10);
}
if (frame.length < expected) return null;
if (frame[0] !== address || frame[1] !== 0x03) return null;
const body = frame.slice(0, expected - 2);
const [lo, hi] = crc16(body);
if (frame[expected - 2] !== lo || frame[expected - 1] !== hi) return null;
const registers = [];
for (let i = 0; i < REGISTER_COUNT; i++) {
registers.push((frame[3 + i * 2] << 8) | frame[4 + i * 2]);
}
return registers;
}
const missing = [];
for (const address of NODES) {
const registers = await poll(address);
if (registers === null) {
missing.push(address);
console.log(`Node 0x${address.toString(16).padStart(2, '0')}: no valid reply`);
} else {
console.log(`Node 0x${address.toString(16).padStart(2, '0')}: ` +
`${registers.join(', ')}`);
}
}
await rs485.stop_rx();
await rs485.disable();
console.log('');
if (missing.length === 0) {
console.log(`All ${NODES.length} nodes answered.`);
} else {
const list = missing.map((a) => '0x' + a.toString(16).padStart(2, '0'));
console.log(`Missing: ${list.join(', ')}`);
}
await tester.reset();
from ikalogic_at1000 import AT1000
import time
# ---- The bus and the nodes on it ---------------------------------------------
BAUD_RATE = 19200
TERMINATION = True # 120 ohm, switched on when we sit at the bus end
NODES = [0x01, 0x02, 0x03, 0x04]
START_REGISTER = 0x0000
REGISTER_COUNT = 2
REPLY_TIMEOUT_S = 0.3
# -------------------------------------------------------------------------------
def crc16(data):
"""Modbus RTU frame check, low byte first."""
crc = 0xFFFF
for byte in data:
crc ^= byte
for _ in range(8):
crc = (crc >> 1) ^ 0xA001 if crc & 1 else crc >> 1
return [crc & 0xFF, (crc >> 8) & 0xFF]
def read_holding_registers(address, start, count):
"""Function 3: read holding registers."""
body = [address, 0x03, (start >> 8) & 0xFF, start & 0xFF,
(count >> 8) & 0xFF, count & 0xFF]
return body + crc16(body)
devices = AT1000.find_devices()
tester = AT1000.open(devices[0])
tester.reset()
rs485 = tester.com.rs485(1)
rs485.enable(
baud_rate=BAUD_RATE,
data_bits=8,
parity="none",
stop_bits="1",
termination_resistors=TERMINATION,
)
rs485.start_rx()
def poll(address):
"""Poll one node and return its registers, or None when it stays silent."""
rs485.rx() # clear anything left from the last node
rs485.tx(read_holding_registers(address, START_REGISTER, REGISTER_COUNT))
expected = 5 + REGISTER_COUNT * 2 # addr, fn, count, data, 2 crc bytes
deadline = time.monotonic() + REPLY_TIMEOUT_S
frame = []
while time.monotonic() < deadline and len(frame) < expected:
frame += list(rs485.rx())
if len(frame) < expected:
time.sleep(0.01)
if len(frame) < expected:
return None
if frame[0] != address or frame[1] != 0x03:
return None
body = frame[:expected - 2]
if frame[expected - 2:expected] != crc16(body):
return None
return [(frame[3 + i * 2] << 8) | frame[4 + i * 2] for i in range(REGISTER_COUNT)]
missing = []
for address in NODES:
registers = poll(address)
if registers is None:
missing.append(address)
print(f"Node 0x{address:02x}: no valid reply")
else:
print(f"Node 0x{address:02x}: {', '.join(str(r) for r in registers)}")
rs485.stop_rx()
rs485.disable()
print("")
if not missing:
print(f"All {len(NODES)} nodes answered.")
else:
print("Missing: " + ", ".join(f"0x{a:02x}" for a in missing))
tester.reset()
tester.close()
Adapting it to your bus​
NODESis the address list to poll.START_REGISTERandREGISTER_COUNTchoose what each node is asked for.- Baud rates run from 300 bps to 20 Mbps, with 7 or 8 data bits and any parity, so the same script drives a slow legacy bus or a fast one.
- A second RS485 port is available on its own pins, so two independent buses can be polled from one device, or the two can be combined into a full-duplex RS422 link.