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

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.key, available.name)

        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:]))

NodeJS

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]));

Rust

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(())
}

C

#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.23 (cd798ee), protocol 8
  se254 SE254 (demo)
  sq25 SQ25 (demo)
  sq50 SQ50 (demo)
  sq100 SQ100 (demo)
  sq200 SQ200 (demo)
  sp209 SP209 (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 111 packets:
  0.000009s  I2C: CH4
  0.000007s  START
  0.000014s  STOP
  0.000029s  I2C: CH4
  ...
  I2C x33, START x33, STOP x33, RE-START x12

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
0.000009160,I2C,CH4
0.000007360,START,
0.000013960,STOP,

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.

Any other source raises ValueError in Python and RangeError in NodeJS before anything is sent to the server.

Write several files one after the other. Call export_csv once per file. Each call returns when its own export has finished, so the calls can follow each other directly, as the C example does. If another client is exporting from the same workspace at that moment, the call raises Busy.

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.

Capture only the frames you want. The decoder that reads the bus can also start the capture: i2c.trigger({"trig_addr": "0x41"}) waits for a frame to 0x41 before recording, instead of capturing whatever the bus does after a rising edge. See Trigger on a protocol frame.

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.