SPI-06: Transaction Decoding Lab

def parse_trace(filename):
    """Parse a signal trace file. Returns (signals, samples).
    signals: list of signal names (excluding 'time')
    samples: list of dicts {signal_name: int_value, '_time': float}
    """
    with open(filename, 'r') as f:
        lines = [line.strip() for line in f if line.strip()]

    n_samples  = int(lines[0])
    sig_names  = lines[1].split()          # includes 'time' at index 0
    # lines[2] = bit widths (ignored — all 1-bit in this lab)
    data_lines = lines[3:3 + n_samples]

    signals = sig_names[1:]              # drop 'time' column
    samples = []
    for line in data_lines:
        parts = line.split()
        entry = {'_time': float(parts[0])}
        for i, name in enumerate(signals):
            entry[name] = int(parts[i + 1])
        samples.append(entry)
    return signals, samples


def get_signal_at(samples, time, signal):
    """Return the value of 'signal' at the most recent sample ≤ time."""
    val = 0
    for s in samples:
        if s['_time'] <= time:
            val = s[signal]
        else:
            break
    return val

def decode_exchanges(samples, cpol, cpha):
    """Yield (mosi_byte, miso_byte) for each complete exchange detected."""
    # Sample on rising if CPOL==CPHA, falling otherwise
    sample_on_rising = (cpol == cpha)

    prev_ss   = 1
    prev_sclk = cpol          # SCLK idle level = CPOL
    mosi_bits = []
    miso_bits = []

    for s in samples:
        ss   = s['ss']
        sclk = s['sclk']
        mosi = s['mosi']
        miso = s['miso']

        # Detect SS falling edge (exchange start)
        if prev_ss == 1 and ss == 0:
            mosi_bits = []
            miso_bits = []

        # Detect SCLK sample edge while SS is low
        if ss == 0:
            rising  = (prev_sclk == 0 and sclk == 1)
            falling = (prev_sclk == 1 and sclk == 0)
            if (sample_on_rising and rising) or (not sample_on_rising and falling):
                mosi_bits.append(mosi)
                miso_bits.append(miso)

        # Detect SS rising edge (exchange complete)
        if prev_ss == 0 and ss == 1:
            if len(mosi_bits) == 8:
                # MSB-first: first bit collected = bit 7
                mosi_byte = int(''.join(str(b) for b in mosi_bits), 2)
                miso_byte = int(''.join(str(b) for b in miso_bits), 2)
                yield mosi_byte, miso_byte

        prev_ss   = ss
        prev_sclk = sclk


def decode_part1(samples, cpol=0, cpha=0):
    exchanges = list(decode_exchanges(samples, cpol, cpha))

    i = 0
    while i + 1 < len(exchanges):
        header_mosi, _ = exchanges[i]
        data_mosi, data_miso = exchanges[i + 1]

        addr   = (header_mosi >> 2) & 0x3F
        rw     = (header_mosi >> 1) & 1
        # stream bit = header_mosi & 1  (always 0 in Part 1)

        if rw == 1:    # write
            print(f"WR {addr:02x} {data_mosi:02x}")
        else:           # read
            print(f"RD {addr:02x} {data_miso:02x}")

        i += 2         # advance 2 exchanges per transaction

def decode_part2(samples, cpol=0, cpha=0):
    exchanges = list(decode_exchanges(samples, cpol, cpha))

    i = 0
    while i < len(exchanges):
        header_mosi, _ = exchanges[i]

        addr   = (header_mosi >> 2) & 0x3F
        rw     = (header_mosi >> 1) & 1
        stream = header_mosi & 1

        if stream:
            # Next exchange carries the length N
            n = exchanges[i + 1][0]    # MOSI of length exchange
            data_exchanges = exchanges[i + 2 : i + 2 + n]

            if rw == 1:
                values = [f"{e[0]:02x}" for e in data_exchanges]  # MOSI
                print(f"WR STREAM {addr:02x} " + " ".join(values))
            else:
                values = [f"{e[1]:02x}" for e in data_exchanges]  # MISO
                print(f"RD STREAM {addr:02x} " + " ".join(values))

            i += 2 + n                   # header + length + N data

        else:
            # Same as Part 1: one data exchange follows
            data_mosi, data_miso = exchanges[i + 1]
            if rw == 1:
                print(f"WR {addr:02x} {data_mosi:02x}")
            else:
                print(f"RD {addr:02x} {data_miso:02x}")
            i += 2

def decode_part3(samples):
    # Read CPOL and CPHA from the first sample
    cpol = samples[0]['cpol']
    cpha = samples[0]['cpha']
    # Delegate to Part 2 decoder with the correct mode
    decode_part2(samples, cpol=cpol, cpha=cpha)

#!/usr/bin/env python3
"""SPI Protocol Decoder — solves Parts 1, 2, and 3 of the lab.
Usage:  python3 spi_decoder.py <trace_file>
"""
import sys


# ─── 1. PARSE TRACE FILE ───────────────────────────────────────
def parse_trace(filename):
    with open(filename) as f:
        lines = [l.strip() for l in f if l.strip()]
    n          = int(lines[0])
    sig_names  = lines[1].split()[1:]   # drop 'time'
    # lines[2] = bit widths (ignored)
    samples = []
    for line in lines[3 : 3 + n]:
        parts = line.split()
        entry = {'_time': float(parts[0])}
        for i, name in enumerate(sig_names):
            entry[name] = int(parts[i + 1])
        samples.append(entry)
    return samples


# ─── 2. EXTRACT EXCHANGES ──────────────────────────────────────
def get_exchanges(samples, cpol, cpha):
    """Yield (mosi_byte, miso_byte) for each complete 8-bit exchange."""
    sample_on_rising = (cpol == cpha)
    prev_ss = 1;  prev_sclk = cpol
    bits_m, bits_s = [], []

    for s in samples:
        ss = s['ss'];  sclk = s['sclk']

        # SS asserted → start fresh
        if prev_ss == 1 and ss == 0:
            bits_m, bits_s = [], []

        # Sample edge while exchange is active
        if ss == 0:
            rising  = prev_sclk == 0 and sclk == 1
            falling = prev_sclk == 1 and sclk == 0
            if (sample_on_rising and rising) or (not sample_on_rising and falling):
                bits_m.append(s['mosi'])
                bits_s.append(s['miso'])

        # SS deasserted → emit exchange
        if prev_ss == 0 and ss == 1 and len(bits_m) == 8:
            mosi = int(''.join(map(str, bits_m)), 2)
            miso = int(''.join(map(str, bits_s)), 2)
            yield (mosi, miso)

        prev_ss = ss;  prev_sclk = sclk


# ─── 3. DECODE TRANSACTIONS ────────────────────────────────────
def decode(samples):
    # Part 3: read CPOL and CPHA from the trace (constant throughout)
    cpol = samples[0]['cpol']
    cpha = samples[0]['cpha']

    exch = list(get_exchanges(samples, cpol, cpha))
    i = 0

    while i < len(exch):
        hdr, _  = exch[i]
        addr    = (hdr >> 2) & 0x3F
        rw      = (hdr >> 1) & 1
        stream  =  hdr        & 1
        verb    = "WR" if rw else "RD"

        if stream:
            n      = exch[i + 1][0]   # length byte (on MOSI)
            data_e = exch[i + 2 : i + 2 + n]
            vals   = [f"{(e[0] if rw else e[1]):02x}" for e in data_e]
            print(f"{verb} STREAM {addr:02x} " + " ".join(vals))
            i += 2 + n
        else:
            dm, ds = exch[i + 1]
            val    = dm if rw else ds
            print(f"{verb} {addr:02x} {val:02x}")
            i += 2


# ─── 4. ENTRY POINT ────────────────────────────────────────────
if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: python3 spi_decoder.py <trace_file>")
        sys.exit(1)
    samples = parse_trace(sys.argv[1])
    decode(samples)

# Test against the simple two-exchange example (case000)
$ python3 spi_decoder.py test_cases/case000.txt
WR 30 d5
WR 0b 9a
RD 0d 15

# Test against the streaming example (case001)
$ python3 spi_decoder.py test_cases/case001.txt
RD STREAM 16 a5 bf 4d
RD 1d 24
RD STREAM 3d 9a ba

# Run all Part 1 test cases
$ sh grade.sh --only $(cd test_cases; echo part1_*)

# Run all Part 2 test cases
$ sh grade.sh --only $(cd test_cases; echo part2_*)

# Run all Part 3 test cases
$ sh grade.sh --only $(cd test_cases; echo part3_*)