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

Rust

// Rust has no separate `connect()`: `listen` opens the WebSocket and returns
// once the stream is live.
let stream = tester.events.listen(&[])?;   // an empty slice accepts every event

Subscribing

Subscribe to a specific event type, or to '*' for every event. JavaScript and Python return an unsubscribe function - call it to stop listening. Note how the SDKs differ: JavaScript uses events.on(...), Python events.subscribe(...); Rust returns an iterator from events.listen(&[types]) (an empty slice means every event) and stops it through stop_handle().close().

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

Rust

use std::thread;

// Listen to one event type on a worker thread; closing its stop handle takes
// the place of an unsubscribe function.
let relays = tester.events.listen(&["relays.state"])?;
let off = relays.stop_handle();
let worker = thread::spawn(move || {
    for event in relays {
        match event {
            Ok(event) => println!("Relay {:?} → {:?}", event.id, event.data),
            Err(err) => eprintln!("event stream error: {err}"),
        }
    }
});

// Listen to every event: an empty slice accepts all of them. Some events (like
// `gpio.reset`) carry no `data` field, so log the whole event here.
for event in tester.events.listen(&[])?.take(10) {
    let event = event?;
    println!("{} {:?}", event.r#type, event);
}

off.close();          // stop listening to relays.state
let _ = worker.join();

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
gpio.sync	-	{ hold_state } (observed latch level; also fires on an external master's hold/release)
gpio.reset	-	none
relays.state	relay id (0–7)	single relay state
relays.states	-	all 8 relay states
relays.reset	-	none
power.dut	dut id (0)	DUT power-supply state
power.usb	usb id (0–1)	USB power state
power.reset	-	none
com.can_config / rs232_config / rs485_config / spi_config / i2c_config / uart_config	interface id	the interface configuration
com.can_rx	can id (0)	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
com.reset	-	none
hmi.text / hmi.colors / hmi.progress / hmi.play	-	text / colors / progress (absent when hidden) / playback configuration
hmi.knob	-	raw knob event (button_pressed, button_held, rotation_delta)
hmi.prompt	-	PromptState, including string data.id and positional data.selection.index
hmi.clear / hmi.reset	-	none
session.state	-	current session holder snapshot

hmi.prompt events occur on creation, highlight changes, and completion. Their envelope has no top-level numeric resource id. The string data.id identifies the prompt, not a choice.

While a prompt owns knob input, the device does not publish raw hmi.knob events for its navigation. Prompt events remain observable without a token, even though prompt REST reads require ownership.

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() - events.listen(...) in Rust: a failed initial connect (for example against old firmware without the events endpoint) fails without scheduling retries. After a successful open, any unexpected socket close triggers exponential-backoff reconnects (500 ms doubling to an 8 s cap) in all three SDKs, and your subscriptions - the Rust iterator's type filter included - 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

Rust

use std::thread;
use std::time::Duration;

// Rust has no connection-state callback: a dropped connection surfaces as one
// error item on the iterator, then the stream reconnects on its own.
let stream = tester.events.listen(&[])?;
let stop = stream.stop_handle();   // cloneable, closable from any thread

let worker = thread::spawn(move || {
    for event in stream {
        match event {
            Ok(event) => println!("events online: {}", event.r#type),
            Err(err) => println!("events offline: {err}"),
        }
    }
});

thread::sleep(Duration::from_secs(10));
stop.close();         // stop the stream and cancel reconnects
let _ = worker.join();

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

Rust

use ikalogic_at1000::{AT1000, RelayState};
use std::thread;
use std::time::Duration;

let devices = AT1000::find_devices(Duration::from_millis(500))?;
let tester = AT1000::open(&devices[0])?;

let stream = tester.events.listen(&["relays.state"])?;
let stop = stream.stop_handle();
let worker = thread::spawn(move || {
    for event in stream {
        let event = match event {
            Ok(event) => event,
            Err(err) => {
                eprintln!("event stream error: {err}");
                continue;
            }
        };
        match event.data_as::<RelayState>() {
            Ok(state) => println!("Relay {:?} changed to {state:?}", event.id),
            Err(err) => eprintln!("unexpected relay payload: {err}"),
        }
    }
});

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

thread::sleep(Duration::from_millis(200));
stop.close();
let _ = worker.join();