Real-time events

The AT1000 broadcasts every resource state change over a WebSocket. Instead of polling the device in a loop, you connect once and react to state changes as they happen - ideal for live dashboards, monitoring, and reacting to knob input.

Concept

The device pushes an event whenever a resource changes state: a relay flips, a power rail changes, a GPIO pin's state or configuration updates, the knob is turned, or the session holder changes. The recommended pattern is fetch then listen: read the initial state via the normal REST calls, then subscribe for live updates. Events need no session token, so a read-only client can monitor a device that another controller is driving.

Connecting

Open the event stream with the events API on an opened tester. connect() resolves once the WebSocket is open:

NodeJS

await tester.events.connect();   // resolves when the stream is open

Python

tester.events.connect()   # blocks until the stream is open

Subscribing

Subscribe to a specific event type, or to '*' for every event. Both languages return an unsubscribe function - call it to stop listening. Note the method name differs between the two SDKs: JavaScript uses events.on(...), Python uses events.subscribe(...).

NodeJS

// Listen to one event type; returns an unsubscribe function.
const off = tester.events.on('relays.state', (event) => {
  console.log('Relay', event.id, '→', event.data);
});

// Listen to every event. Some events (like `gpio.reset`) carry no `data` field,
// so log the whole event object here.
tester.events.on('*', (event) => console.log(event.type, event));

off();  // stop listening to relays.state

Python

# Listen to one event type; returns an unsubscribe callable.
off = tester.events.subscribe("relays.state", lambda event:
    print("Relay", event.id, "→", event.data))

# Listen to every event.
tester.events.subscribe("*", lambda event: print(event.type, event.data))

off()  # stop listening to relays.state

Each event carries type, an optional id (the resource index - e.g. the relay or GPIO number, omitted for non-indexed events), a ts timestamp (milliseconds since the Unix epoch), and a data payload whose shape matches the corresponding REST GET response.

Event catalogue

type	id	data
gpio.state	io id (0–31)	pin state
gpio.config	io id (0–31)	pin configuration
relays.state	relay id (0–7)	single relay state
relays.states	-	all 8 relay states
power.dut	dut id (0)	DUT power-supply state
power.usb	usb id (0–1)	USB power state
com.can_rx	can id	CAN frames received this tick
com.uart_rx	uart id (0–1)	bytes received this tick
com.rs232_rx	rs232 id (0)	bytes received this tick
com.rs485_rx	rs485 id (0–1)	bytes received this tick
hmi.knob	-	knob event (rotation delta, button state)
session.state	-	current session holder snapshot

RX events require arming

The com.*_rx events only flow while the corresponding bus is armed via start_rx (e.g. tester.com.uart(0).start_rx()). Subscribing alone does not arm reception.

Connection lifecycle

Auto-reconnect arms only after the first successful connect(): a failed initial connect (for example against old firmware without the events endpoint) rejects without scheduling retries. After a successful open, any unexpected socket close triggers exponential-backoff reconnects (500 ms doubling to an 8 s cap), and your subscriptions survive across reconnects. To observe connection state and to shut the stream down:

NodeJS

tester.events.onStateChange((connected) => {
  console.log(connected ? 'events online' : 'events offline');
});

tester.events.close();  // stop the stream and cancel reconnects

Python

tester.events.subscribe_state(lambda connected:
    print("events online" if connected else "events offline"))

tester.events.close()  # stop the stream and cancel reconnects

Worked example: react to a relay change

Connect, subscribe to relays.state, then toggle relay 0 and watch the event arrive:

NodeJS

import { AT1000 } from '@ikalogic/at1000';

const hosts = await AT1000.findDevices();
const tester = await AT1000.open(hosts[0]);

await tester.events.connect();
tester.events.on('relays.state', (event) => {
  console.log(`Relay ${event.id} changed to`, event.data);
});

await tester.relays.relay(0).close();  // triggers a relays.state event
await tester.relays.relay(0).open();   // triggers another

Python

from ikalogic_at1000 import AT1000

hosts = AT1000.find_devices()
tester = AT1000.open(hosts[0])

tester.events.connect()
tester.events.subscribe("relays.state", lambda event:
    print(f"Relay {event.id} changed to", event.data))

tester.relays.relay(0).close()  # triggers a relays.state event
tester.relays.relay(0).open()   # triggers another