Decode a bus and export it

Capture a bus, run a protocol decoder over it, look at what came out, and write it to a CSV file for a spreadsheet, pandas or a report.

The program

Python
NodeJS
Rust
C

import sys
from collections import Counter

from ikalogic_scanastudio import ScanaStudio, Trigger

SCRIPT = "i2c.js"


def main(device: str = "se254") -> int:
    with ScanaStudio.connect() as server:
        print(server.info)
        for available in server.devices():
            print(" ", available)

        workspace = server.create(device)
        print(f"\nopened {workspace.device_name} "
              f"with {len(workspace.channels)} channels")

        last = workspace.capture.run(
            samples=1_000_000,
            sample_rate=25_000_000,
            trigger=Trigger.rising(channel=0, position=0.1),
        )
        seconds = workspace.capture.seconds(last)
        print(f"captured {last} samples ({seconds:.3f} s) "
              f"at {workspace.capture.sample_rate} S/s")

        if server.scripts.find(SCRIPT) is None:
            print(f"\n{SCRIPT} is not installed; skipping the decode")
            return 0

        # What does this script take? Ask it.
        print(f"\n{SCRIPT} takes:")
        for option in workspace.decoders.options(SCRIPT):
            print("  ", option)

        # Channels by name, so the same script works on any device that has them.
        i2c = workspace.decoders.add(
            SCRIPT,
            {
                "ch_scl": workspace.channel("I2C SCL"),
                "ch_sda": workspace.channel("I2C SDA"),
            },
            wait=True,
        )

        packets = list(i2c.packets())
        print(f"\n{i2c.name} produced {len(packets)} packets:")
        for packet in packets[:10]:
            at = workspace.capture.seconds(packet.start)
            content = f": {packet.content}" if packet.content else ""
            print(f"  {at:.6f}s  {packet.title}{content}")

        counts = Counter(p.title for p in packets)
        print("  " + ", ".join(f"{t} x{n}" for t, n in counts.most_common()))

        # The byte stream behind those packets.
        payload = bytes(entry.byte for entry in i2c.hex_bytes())
        print(f"\n{len(payload)} bytes decoded; first 16: {payload[:16].hex(' ')}")

        workspace.export_csv(
            "packets.csv", source="packets", instances=[i2c.instance_id])
        print("\nwrote packets.csv (on the server's machine)")

        workspace.close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main(*sys.argv[1:]))

import { ScanaStudio, describe_option, Trigger } from '@ikalogic/scanastudio';

const SCRIPT = 'i2c.js';

async function main(device = 'se254') {
  const server = await ScanaStudio.connect();
  try {
    console.log(server.info);
    for (const available of await server.devices()) console.log(' ', available.key);

    const workspace = await server.create(device);
    console.log(`opened ${workspace.device_name} with ${workspace.channels.length} channels`);

    const last = await workspace.capture.run({
      samples: 1_000_000,
      sample_rate: 25_000_000,
      trigger: Trigger.rising(0, { position: 0.1 }),
    });
    console.log(`captured ${last} samples `
      + `(${workspace.capture.seconds(last).toFixed(3)} s) `
      + `at ${workspace.capture.sample_rate} S/s`);

    if (!server.scripts.find(SCRIPT)) {
      console.log(`${SCRIPT} is not installed; skipping the decode`);
      return 0;
    }

    // What does this script take? Ask it.
    for (const option of await workspace.decoders.options(SCRIPT)) {
      console.log('  ', describe_option(option));
    }

    // Channels by name, so the same script works on any device that has them.
    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);
    console.log(`${i2c.name} produced ${packets.length} packets:`);
    for (const packet of packets.slice(0, 10)) {
      const at = workspace.capture.seconds(packet.start).toFixed(6);
      console.log(`  ${at}s  ${packet.title}${packet.content ? `: ${packet.content}` : ''}`);
    }

    // The byte stream behind those packets.
    const payload = [];
    for await (const entry of i2c.hex_bytes()) payload.push(entry.byte);
    console.log(`${payload.length} bytes decoded`);

    await workspace.export_csv('packets.csv',
      { source: 'packets', instances: [i2c.instance_id] });
    console.log('wrote packets.csv (on the server\'s machine)');

    workspace.close();
    return 0;
  } finally {
    server.close();
  }
}

process.exit(await main(process.argv[2]));

use std::time::Duration;

use scanastudio_client::{
    decoded, proto::RowFilter, values, CaptureRequest, ScanaStudio,
    Trigger, TriggerExt, DEFAULT_URL, Result,
};

const SCRIPT: &str = "i2c.js";

#[tokio::main]
async fn main() -> Result<()> {
    let server = ScanaStudio::connect(DEFAULT_URL).await?;
    println!("{:?}", server.info());

    let workspace = server.create("se254").await?;
    let last = workspace.capture().run(
        &CaptureRequest::new(1_000_000, 25_000_000)
            .trigger(Trigger::rising(0).at(0.1)),
        Duration::from_secs(300),
    ).await?;
    println!("captured {last} samples at {} S/s", workspace.capture().sample_rate());

    if server.scripts().find(SCRIPT).is_none() {
        println!("{SCRIPT} is not installed; skipping the decode");
        return Ok(());
    }

    // What does this script take? Ask it.
    for option in workspace.decoders().options(SCRIPT).await? {
        println!("  {option}");
    }

    // Channels by name, so the same script works on any device that has them.
    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?;

    // Only this decoder's packets; the filter narrows the read on the server.
    let filter = RowFilter { sources: vec![i2c.instance_id], ..Default::default() };
    let packets = workspace.data().packets(&filter).await?;
    println!("{} produced {} packets", i2c.name, packets.len());
    for packet in packets.iter().take(10) {
        println!("  {:.6}s  {}: {}",
                 workspace.capture().seconds(packet.start), packet.title, packet.content);
    }

    workspace.export_csv(
        "packets.csv", &decoded("packets", vec![i2c.instance_id]),
        Duration::from_secs(600)).await?;

    workspace.close()?;
    Ok(())
}

#include <stdio.h>
#include <stdlib.h>

#include "scanastudio.h"

int main(void) {
    ss_server *server = ss_server_connect(NULL);
    if (!server) return 1;

    ss_workspace *workspace = ss_server_create(server, "se254");
    if (!workspace) return 1;

    /* A rising edge on channel 0, keeping 10 % of the capture before it. */
    char *trigger = ss_trigger_edge(0, "rising", 0.1, NULL);
    char request[192];
    snprintf(request, sizeof request,
             "{\"samples\": 1000000, \"sample_rate\": 25000000, \"trigger\": %s}",
             trigger);
    ss_string_free(trigger);

    if (ss_capture_run(workspace, request, 300.0) < 0) {
        fprintf(stderr, "capture: %s\n", ss_last_error());
        return 1;
    }
    printf("captured %lld samples at %llu S/s\n",
           (long long)ss_capture_last_sample(workspace),
           (unsigned long long)ss_capture_sample_rate(workspace));

    /* Channels by name, so the same program works on any device that has them. */
    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) {
        ss_decoder_wait(workspace, i2c, 300.0);

        char filter[64];
        snprintf(filter, sizeof filter, "{\"sources\": [%u]}", i2c);
        ss_packets *batch = ss_packets_read(workspace, filter);
        if (batch) {
            size_t count = ss_packets_count(batch);
            ss_packet *packets = malloc(sizeof *packets * count);
            ss_packets_copy(batch, 0, packets, count);
            for (size_t i = 0; i < count; i += 1) {
                printf("%s: %s\n", packets[i].title, packets[i].content);
            }
            free(packets);
            /* Every title and content printed above dies with the batch. */
            ss_packets_free(batch);
        }
    }

    /* Any of the four sources: "samples", "raw", "packets" or "hex". */
    ss_workspace_export_csv(workspace, "samples.csv",
                  "{\"source\": \"samples\", \"channels\": [0, 1]}", 600.0);
    ss_workspace_export_csv(workspace, "packets.csv",
                  "{\"source\": \"packets\"}", 600.0);

    ss_workspace_close(workspace);
    ss_workspace_free(workspace);
    ss_server_disconnect(server);
    return 0;
}

What it prints

Against the se254 demo device:

ScanaStudio 6.0.13 (1a909ee), protocol 8
  se254 — SE254 (demo) (demo)
  sp209 — SP209 (demo) (demo)
  ...

opened SE254 (demo) with 4 channels
captured 1000000 samples (0.040 s) at 25000000 S/s

i2c.js takes:
   ch_sda: SDA Channel (ch_selector) = None
   ch_scl: SCL Channel (ch_selector) = None
   AUTO0: Advanced options (tab) = False
   address_opt [Advanced options]: Address convention (combo) one of ['7 bit address', '8 bit address (inlcuding R/W flag)'] = '7 bit address'
   ...

I2C produced 133 packets:
  0.000006s  I2C: CH4
  0.000006s  START
  0.000020s  RE-START
  0.000032s  STOP
  I2C x38, START x38, STOP x38, RE-START x19

0 bytes decoded; first 16:

wrote packets.csv (on the server's machine)

0 bytes decoded is the demo device showing through: the decoder found framing in the noise but no data bytes, so the hex view is empty. On real bus traffic that line reports the payload.

And packets.csv on the server looks like:

Time,I2C,DATA
000005600,I2C,CH4
000005600,START,
000020400,RE-START,

Notes for your own bench

Ask the script what it takes. decoders.options(script) prints every setting with its default. Do not guess option names from another decoder: a UART script wants baud, an SPI script wants a chip-select channel, and the names are the script's own.

Pick the right export source. Four are available, and they answer different questions:

`source`	Use it for
`packets`	The protocol view: addresses, commands, ACKs. Best for a report.
`hex`	The payload as bytes. Best for firmware and CRC work.
`raw`	Every element a decoder produced, with its exact span. Best for timing.
`samples`	The logic levels themselves. Big, and the only one that needs no decoder.

Remember where the file lands. The server writes it. Pointing a local script at a remote server puts the CSV on the remote machine, not yours.

Filter on the server, not in your loop. packets(filter=...) narrows the read before the data is sent. See Reading captured data.

The workspace freezes while it exports

A CSV export freezes the workspace: every other command, from every other client, is refused with Busy until it finishes. Export at the end of a run, not in the middle of one.

Do not chain exports back to back

A second export_csv issued immediately after the first is currently refused, because the workspace is still frozen, and the Python and NodeJS SDKs return without raising. No file is written and nothing tells you. If you need several files from one capture, pause between the calls and check that each file exists. See Saving and exporting.

Chain decoders

A high-level decoder can sit on top of a low-level one (a sensor protocol over I²C, say). Add both; each one is an instance with its own packets, and packets() without a filter merges them in time order.

The program​

What it prints​

Notes for your own bench​

The program

What it prints

Notes for your own bench