Automated bench & end-of-line test
Arm a trigger, capture the board under test, decode the bus, check the result, and exit with a pass/fail code. This is the shape of an end-of-line tester, and of a CI job that checks firmware for regressions.
The run is unattended: no window, no buttons, no operator.
The program​
- Python
- NodeJS
- Rust
- C
"""Pass/fail an I2C board test. Exit code 0 passes, 1 fails, 2 is a bench fault."""
import sys
from collections import Counter
from ikalogic_scanastudio import (
ScanaStudio, ScanaStudioError, Refused, ScriptError, Timeout, Trigger,
)
DEVICE = "se254"
SCRIPT = "i2c.js"
EXPECTED_ADDRESS = "0x4E"
def run() -> int:
with ScanaStudio.connect() as server:
workspace = server.create(DEVICE)
# Check the script library after creating the workspace: the list
# arrives shortly after connect() and may still be empty before that.
if server.scripts.find(SCRIPT) is None:
workspace.close()
print(f"{SCRIPT} is not installed on the server", file=sys.stderr)
return 2
try:
# Arm on the board's reset line going high, keeping 10% of history.
workspace.capture.run(
samples=2_000_000,
sample_rate=25_000_000,
trigger=Trigger.rising(channel=workspace.channel("RESET"), position=0.1),
timeout=30.0,
)
i2c = workspace.decoders.add(
SCRIPT,
{
"ch_scl": workspace.channel("I2C SCL"),
"ch_sda": workspace.channel("I2C SDA"),
},
wait=True,
)
packets = list(i2c.packets())
counts = Counter(p.title for p in packets)
print(f"{len(packets)} packets: "
+ ", ".join(f"{title} x{n}" for title, n in counts.most_common()))
# --- the assertions -------------------------------------------
failures = []
if not packets:
failures.append("the bus was silent")
if not any(p.title == "Address" and p.content == EXPECTED_ADDRESS
for p in packets):
failures.append(f"never saw address {EXPECTED_ADDRESS}")
nacks = counts.get("NACK", 0)
if nacks:
failures.append(f"{nacks} NACK(s) on the bus")
if failures:
for failure in failures:
print(f"FAIL: {failure}", file=sys.stderr)
workspace.save("/data/failures/last-failure.scana")
return 1
print("PASS")
return 0
finally:
workspace.close()
def main() -> int:
try:
return run()
except (Refused, Timeout, ScriptError) as exc:
print(f"bench fault: {exc}", file=sys.stderr)
return 2
except ScanaStudioError as exc:
print(f"bench fault: {exc}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())
import {
ScanaStudio, ScanaStudioError, Refused, ScriptError, Timeout, Trigger,
} from '@ikalogic/scanastudio';
const DEVICE = 'se254';
const SCRIPT = 'i2c.js';
const EXPECTED_ADDRESS = '0x4E';
async function run() {
const server = await ScanaStudio.connect();
try {
const workspace = await server.create(DEVICE);
// Check the script library after creating the workspace: the list
// arrives shortly after connect() and may still be empty before that.
if (!server.scripts.find(SCRIPT)) {
workspace.close();
console.error(`${SCRIPT} is not installed on the server`);
return 2;
}
try {
// Arm on the board's reset line going high, keeping 10% of history.
await workspace.capture.run({
samples: 2_000_000,
sample_rate: 25_000_000,
trigger: Trigger.rising(workspace.channel('RESET'), { position: 0.1 }),
timeout_ms: 30_000,
});
const i2c = await workspace.decoders.add(
SCRIPT,
{
ch_scl: workspace.channel('I2C SCL'),
ch_sda: workspace.channel('I2C SDA'),
},
{ wait: true },
);
const packets = [];
for await (const packet of i2c.packets()) packets.push(packet);
const counts = new Map();
for (const p of packets) counts.set(p.title, (counts.get(p.title) ?? 0) + 1);
console.log(`${packets.length} packets:`,
[...counts].map(([t, n]) => `${t} x${n}`).join(', '));
// --- the assertions -------------------------------------------
const failures = [];
if (packets.length === 0) failures.push('the bus was silent');
if (!packets.some((p) => p.title === 'Address' && p.content === EXPECTED_ADDRESS)) {
failures.push(`never saw address ${EXPECTED_ADDRESS}`);
}
const nacks = counts.get('NACK') ?? 0;
if (nacks) failures.push(`${nacks} NACK(s) on the bus`);
if (failures.length > 0) {
for (const failure of failures) console.error(`FAIL: ${failure}`);
await workspace.save('/data/failures/last-failure.scana');
return 1;
}
console.log('PASS');
return 0;
} finally {
workspace.close();
}
} finally {
server.close();
}
}
let code;
try {
code = await run();
} catch (error) {
if (error instanceof Refused || error instanceof Timeout
|| error instanceof ScriptError || error instanceof ScanaStudioError) {
console.error(`bench fault: ${error.message}`);
code = 2;
} else {
throw error;
}
}
process.exit(code);
use std::collections::HashMap;
use std::process::ExitCode;
use std::time::Duration;
use scanastudio_client::{
proto::RowFilter, values, CaptureRequest, ScanaStudio, Trigger, TriggerExt,
DEFAULT_URL, Result,
};
const DEVICE: &str = "se254";
const SCRIPT: &str = "i2c.js";
const EXPECTED_ADDRESS: &str = "0x4E";
async fn run() -> Result<bool> {
let server = ScanaStudio::connect(DEFAULT_URL).await?;
let workspace = server.create(DEVICE).await?;
// Check the script library after creating the workspace: the list
// arrives shortly after connect() and may still be empty before that.
if server.scripts().find(SCRIPT).is_none() {
eprintln!("{SCRIPT} is not installed on the server");
workspace.close()?;
return Ok(false);
}
// Arm on the board's reset line going high, keeping 10% of history.
let reset = workspace.channel("RESET")?;
workspace.capture().run(
&CaptureRequest::new(2_000_000, 25_000_000)
.trigger(Trigger::rising(reset).at(0.1)),
Duration::from_secs(30),
).await?;
let scl = workspace.channel("I2C SCL")?;
let sda = workspace.channel("I2C SDA")?;
let i2c = workspace.decoders()
.add(SCRIPT, &values([("ch_scl", scl.into()), ("ch_sda", sda.into())]))
.await?;
workspace.decoders()
.wait_for_instances(&[i2c.instance_id], Duration::from_secs(300))
.await?;
let filter = RowFilter { sources: vec![i2c.instance_id], ..Default::default() };
let packets = workspace.data().packets(&filter).await?;
let mut counts: HashMap<&str, usize> = HashMap::new();
for packet in &packets {
*counts.entry(packet.title.as_str()).or_default() += 1;
}
println!("{} packets", packets.len());
// --- the assertions -------------------------------------------
let mut failures = Vec::new();
if packets.is_empty() {
failures.push("the bus was silent".to_string());
}
if !packets.iter().any(|p| p.title == "Address" && p.content == EXPECTED_ADDRESS) {
failures.push(format!("never saw address {EXPECTED_ADDRESS}"));
}
if let Some(nacks) = counts.get("NACK") {
failures.push(format!("{nacks} NACK(s) on the bus"));
}
let passed = failures.is_empty();
if !passed {
for failure in &failures {
eprintln!("FAIL: {failure}");
}
workspace
.save("/data/failures/last-failure.scana", Duration::from_secs(600))
.await?;
} else {
println!("PASS");
}
workspace.close()?;
Ok(passed)
}
#[tokio::main]
async fn main() -> ExitCode {
match run().await {
Ok(true) => ExitCode::from(0),
Ok(false) => ExitCode::from(1),
Err(error) => {
eprintln!("bench fault: {error}");
ExitCode::from(2)
}
}
}
/* Capture, decode, and check the packet JSON.
cc -I/opt/scanastudio-sdk/include bench.c \
-L/opt/scanastudio-sdk/lib -lscanastudio -o bench */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "scanastudio.h"
#define DEVICE "se254"
#define EXPECTED_ADDRESS "0x4E"
int main(void) {
ss_server *server = ss_server_connect(NULL);
if (!server) { fprintf(stderr, "connect: %s\n", ss_last_error()); return 2; }
ss_workspace *workspace = ss_server_create(server, DEVICE);
if (!workspace) { fprintf(stderr, "open: %s\n", ss_last_error()); return 2; }
int reset = ss_workspace_channel(workspace, "RESET");
/* Rising edge on RESET, 10 % pre-trigger, 30 s budget. */
char *trigger = ss_trigger_edge(reset, "rising", 0.1, NULL);
char request[192];
snprintf(request, sizeof request,
"{\"samples\": 2000000, \"sample_rate\": 25000000, \"trigger\": %s}",
trigger);
ss_string_free(trigger);
if (ss_capture_run(workspace, request, 30.0) < 0) {
fprintf(stderr, "capture: %s\n", ss_last_error());
return 2;
}
char options[64];
snprintf(options, sizeof options, "{\"ch_scl\": %d, \"ch_sda\": %d}",
ss_workspace_channel(workspace, "I2C SCL"), ss_workspace_channel(workspace, "I2C SDA"));
uint32_t i2c = ss_decoder_add(workspace, "i2c.js", options);
if (!i2c) { fprintf(stderr, "decoder: %s\n", ss_last_error()); return 2; }
if (ss_decoder_wait(workspace, i2c, 300.0) != ss_status_Ok) return 2;
char filter[64];
snprintf(filter, sizeof filter, "{\"sources\": [%u]}", i2c);
ss_packets *batch = ss_packets_read(workspace, filter);
if (!batch) { fprintf(stderr, "packets: %s\n", ss_last_error()); return 2; }
size_t count = ss_packets_count(batch);
ss_packet *packets = malloc(sizeof *packets * count);
ss_packets_copy(batch, 0, packets, count);
/* --- the assertions ------------------------------------------------
The fields are typed, so there is no JSON to parse here. Every
string points into the batch: read it before ss_packets_free. */
int addressed = 0, nacked = 0;
for (size_t i = 0; i < count; i += 1) {
if (strcmp(packets[i].content, EXPECTED_ADDRESS) == 0) addressed = 1;
if (strcmp(packets[i].content, "NACK") == 0) nacked = 1;
}
int passed = addressed && !nacked;
free(packets);
ss_packets_free(batch);
if (!passed) {
fprintf(stderr, "FAIL\n");
ss_workspace_save(workspace, "/data/failures/last-failure.scana", 600.0);
} else {
printf("PASS\n");
}
ss_workspace_close(workspace);
ss_workspace_free(workspace);
ss_server_disconnect(server);
return passed ? 0 : 1;
}
ss_capture_run takes the same request as ss_capture_start, so a falling
edge, a runt pulse or a sequence is the same code with a different
ss_trigger_* call. Leave trigger out altogether and the capture starts at
once. See Triggers.
What it prints​
A failing run. This is the demo device, which produces noise rather than I²C traffic, so the address assertion never passes:
144 packets: I2C x41, START x41, STOP x41, RE-START x21
FAIL: never saw address 0x4E
The exit code is 1, and /data/failures/last-failure.scana is written so you
can open it in ScanaStudio afterwards.
A passing run against real traffic prints the packet counts and then:
PASS
with exit code 0. A bench fault (no device, a missing script, a timeout)
prints the server's reason to stderr and exits 2:
bench fault: <the reason, as the server phrased it>
Print those reasons as they are rather than paraphrasing them. They tell an operator whether to re-seat a probe or call someone.
Notes for your own bench​
Three exit codes, not two. 0 passed, 1 the board failed, 2 the bench
failed (no device, missing script, a timeout). A CI job that cannot tell those
apart may pass a bad board because the analyser was unplugged.
Trigger on the event. Trigger.rising(...) with a position keeps history
from before the event, so a failure shows you the run-up and not just the
aftermath. See Triggers for pulse
and sequence triggers when a plain edge is too broad.
Save the failures. workspace.save() on the failing path gives you a .scana
you can open in ScanaStudio later to see why a board failed.
Look up channels by name. workspace.channel("I2C SCL") keeps the same
script running on an SE254 at the desk and an SP259 on the line.
Check the script library after creating the workspace. The list arrives
shortly after connect() returns, so reading it too early can show every decoder
as missing. See
Decoders.
Assert on protocol content, not on counts alone. "12 packets" passes for the wrong 12. Check for the address you expect, and for the errors you do not.
se254 generates pseudo-random noise, so the I²C decoder finds packets by chance
and never the address you are looking for. Use the demo device to check that
your script runs; use hardware to check that it judges correctly.
The connection and the workspace are the expensive parts. Open once, then loop
over capture.run() and the assertions per board. A decoder can be relaunched
with relaunch() instead of re-added.