applets.interface.better_la: add

Author: anuejn

software/.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+    "python.testing.pytestArgs": [
+        "."
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
+}

software/glasgow/applet/interface/better_la/__init__.py

@@ -0,0 +1,137 @@
+from collections import defaultdict
+import logging
+import argparse
+from vcd import VCDWriter
+from amaranth import *
+from amaranth.lib.cdc import FFSynchronizer
+
+from ....gateware.pads import *
+from ....gateware.analyzer import *
+from ... import *
+from .signal_compressor import SignalCompressor
+from .arbeiter import LAArbeiter
+
+# This LA uses a simple protocol for sending compressed values over the FIFO:
+# Each packet starts with a 8 bit size word. The size can be 0, then the word only consists of that
+# word. If the size is n != 0, the packet is n*2 bytes long. Each 16bit word is encoded acording
+# to the format described in the SignalCompressor value. The packets are round-robin for each pin.
+
+class BetterLASubtarget(Elaboratable):
+    def __init__(self, pads, in_fifo):
+        self.pads    = pads
+        self.in_fifo = in_fifo
+
+        self.la = LAArbeiter(in_fifo)
+
+    def elaborate(self, platform):
+        m = Module()
+        m.submodules += self.la
+
+        pins_i = Signal.like(self.pads.i_t.i)
+        m.submodules += FFSynchronizer(self.pads.i_t.i, pins_i)
+        m.d.comb += self.la.input.eq(pins_i)
+
+        return m
+
+
+class BetterLAApplet(GlasgowApplet):
+    logger = logging.getLogger(__name__)
+    help = "capture logic waveforms"
+    description = """
+    A somewhat better logic analyzer applet that allows for the capture of traces as VCD files.
+    """
+
+    @classmethod
+    def add_build_arguments(cls, parser, access):
+        super().add_build_arguments(parser, access)
+
+        access.add_pin_set_argument(parser, "i", width=range(1, 17), default=1)
+
+    def build(self, target, args):
+        self.mux_interface = iface = target.multiplexer.claim_interface(self, args)
+        iface.add_subtarget(BetterLASubtarget(
+            pads=iface.get_pads(args, pin_sets=("i",)),
+            in_fifo=iface.get_in_fifo(depth=512*16),
+        ))
+
+        self._sample_freq = target.sys_clk_freq
+        self._pins = getattr(args, "pin_set_i")
+
+    @classmethod
+    def add_run_arguments(cls, parser, access):
+        super().add_run_arguments(parser, access)
+
+        g_pulls = parser.add_mutually_exclusive_group()
+        g_pulls.add_argument(
+            "--pull-ups", default=False, action="store_true",
+            help="enable pull-ups on all pins")
+        g_pulls.add_argument(
+            "--pull-downs", default=False, action="store_true",
+            help="enable pull-downs on all pins")
+
+    async def run(self, device, args):
+        pull_low  = set()
+        pull_high = set()
+        if args.pull_ups:
+            pull_high = set(args.pin_set_i)
+        if args.pull_downs:
+            pull_low = set(args.pin_set_i)
+        iface = await device.demultiplexer.claim_interface(self, self.mux_interface, args,
+                                                           pull_low=pull_low, pull_high=pull_high)
+        return iface
+
+    @classmethod
+    def add_interact_arguments(cls, parser):
+        parser.add_argument(
+            "file", metavar="VCD-FILE", type=argparse.FileType("w"),
+            help="write VCD waveforms to VCD-FILE")
+
+    async def interact(self, device, args, iface):
+        pins = defaultdict(list)
+        overrun = False
+
+        zero_chunks = 0
+        chunks = 0
+        try:  # this try catches Ctrl+C for being able to manually interrupt capture
+            while not overrun:
+                for p in self._pins:
+                    pkgs = await LAArbeiter.read_chunk(iface.read)
+                    if len(pkgs) == 0:
+                        zero_chunks += 1
+                    chunks += 1
+                    pins[p].extend(pkgs)
+                    if len(pkgs) > 255 - len(self._pins):
+                        overrun = True
+                        print("overrun")
+        finally:
+            events = []
+            cycles = 0
+            for p, pkgs in pins.items():
+                cycle = 0
+                for pkg in pkgs:
+                    for value, duration in SignalCompressor.decode_pkg(pkg):
+                        timestamp = cycle * 1_000_000_000 // self._sample_freq
+                        events.append((timestamp, p, value))
+                        cycle += duration
+                cycles = max(cycle, cycles)
+            events.sort(key=lambda e: e[0])
+
+            total_pkgs = sum(len(pkgs) for pkgs in pins.values())
+            total_bytes = chunks + total_pkgs * 2
+
+            print(f"captured {cycles} cycles")
+            print(f"chunking overhead: {chunks / total_bytes * 100}%")
+            print(f"zero chunks overhead: {zero_chunks / total_bytes * 100}%")
+            print(f"compression gain: {100 - (total_bytes * 8 / cycle * 100)}%")
+            
+
+            vcd_writer = VCDWriter(args.file, timescale="1 ns", check_values=False)
+            vcd_signals = {
+                p: vcd_writer.register_var(scope="", name="pin[{}]".format(p), var_type="wire", 
+                                           size=1, init=0)
+                for p in pins.keys()
+            }
+            for timestamp, p, value in events:
+                signal = vcd_signals[p]
+                vcd_writer.change(signal, timestamp, value)
+            vcd_writer.close(timestamp)

software/glasgow/applet/interface/better_la/arbeiter.py

@@ -0,0 +1,86 @@
+from typing import Callable
+from amaranth import *
+from amaranth.lib.fifo import SyncFIFOBuffered
+
+from . import SignalCompressor
+
+class LAArbeiter(Elaboratable):
+    """This Logic Analyzer Arbeiter instanciates n Signal compressors and n Fifos and arbeites the
+    output of the fifos in a round robin fashion. Its output format is one length byte followed by
+    2*length bytes of compressed channel data. After that the next channel is send with the same
+    format.
+    """
+    def __init__(self, output_fifo: SyncFIFOBuffered, n_channels=16, pressure_threshold=64):
+        self.output_fifo = output_fifo
+        assert output_fifo.width == 8
+        self.input = Signal(n_channels)
+        
+        self._pressure_threshold = pressure_threshold
+
+        self.fifos = [SyncFIFOBuffered(width=16, depth=256) for _ in range(n_channels)]
+        self.compressors = [SignalCompressor(self.input[i]) for i in range(n_channels)]
+
+    def elaborate(self, platform):
+        m = Module()
+
+        to_transfer = Signal(8)
+        enough_pressure = Signal(len(self.fifos))
+        any_enough_pressure = Signal()
+        m.d.comb += any_enough_pressure.eq(enough_pressure.any())
+
+        with m.FSM():
+            for i in range(len(self.input)):
+                fifo = self.fifos[i]
+                compressor = self.compressors[i]
+                m.submodules[f"fifo_{i}"] = fifo
+                m.submodules[f"compressor_{i}"] = compressor
+
+                m.d.comb += fifo.w_en.eq(compressor.valid)
+                m.d.comb += fifo.w_data.eq(compressor.value)
+    
+                m.d.sync += enough_pressure[i].eq(fifo.r_level > self._pressure_threshold)
+
+                def go_to_next(i):
+                    with m.If(any_enough_pressure):
+                        m.next = f"announce_{(i + 1) % len(self.input)}"
+                    with m.Else():
+                        m.next = f"wait_{(i + 1) % len(self.input)}"
+
+
+                with m.State(f"wait_{i}"):
+                    with m.If(any_enough_pressure):
+                        m.next = f"announce_{i}"
+                with m.State(f"announce_{i}"):
+                    m.d.comb += self.output_fifo.w_data.eq(fifo.r_level)
+                    m.d.comb += self.output_fifo.w_en.eq(1)
+                    m.d.sync += to_transfer.eq(fifo.r_level)
+                    with m.If(self.output_fifo.w_rdy):
+                        with m.If(fifo.r_level > 0):
+                            m.next = f"send_{i}_lower"
+                        with m.Else():
+                            go_to_next(i)
+
+                with m.State(f"send_{i}_lower"):
+                    m.d.comb += self.output_fifo.w_data.eq(fifo.r_data[0:8])
+                    m.d.comb += self.output_fifo.w_en.eq(1)
+                    with m.If(self.output_fifo.w_rdy):
+                        m.next = f"send_{i}_upper"
+
+                with m.State(f"send_{i}_upper"):
+                    m.d.comb += self.output_fifo.w_data.eq(fifo.r_data[8:16])
+                    m.d.comb += self.output_fifo.w_en.eq(1)
+                    with m.If(self.output_fifo.w_rdy):
+                        m.d.comb += fifo.r_en.eq(1)
+                        with m.If(to_transfer > 1):
+                            m.next = f"send_{i}_lower"
+                            m.d.sync += to_transfer.eq(to_transfer - 1)
+                        with m.Else():
+                            go_to_next(i)
+
+        return m
+    
+    @staticmethod
+    async def read_chunk(read: Callable[[int], bytes]):
+        length = (await read(1))[0]
+        contents = (await read(2 * length))
+        return [contents[2*i+1] << 8 | contents[2*i] for i in range(length)]

software/glasgow/applet/interface/better_la/signal_compressor.py

@@ -0,0 +1,70 @@
+from itertools import chain
+from typing import List, Tuple
+from amaranth import *
+
+class SignalCompressor(Elaboratable):
+    """The SignalCompressor converts information about value changes into an efficient compressed 
+    format. It outputs a 16bit stream that is encoded in one of three ways:
+
+    0b0: plain, no compression [15 bit value dump]
+    0b10: constant 0 for the following n [14 bit] cycles
+    0b11: constant 1 for the following n [14 bit] cycles
+    """
+    def __init__(self, signal):
+        self.signal = signal
+
+        self.valid = Signal()
+        self.value = Signal(16)
+
+    def elaborate(self, platform):
+        m = Module()
+        
+        last = Signal()
+        m.d.sync += last.eq(self.signal)
+        change = Signal()
+        m.d.comb += change.eq(self.signal ^ last)
+
+
+        counter = Signal(14)
+        m.d.sync += counter.eq(counter + 1)
+
+        buffer = Signal(15)
+        m.d.sync += buffer.eq((buffer << 1) | self.signal)
+
+        plain_mode = Signal()
+
+        with m.If(change):
+            with m.If(counter < 15):
+                m.d.sync += plain_mode.eq(1)
+            with m.Elif(~plain_mode):
+                m.d.comb += self.valid.eq(1)
+                m.d.comb += self.value.eq(Cat(1, last, counter))
+                m.d.sync += counter.eq(0)
+                m.d.sync += plain_mode.eq(0)
+
+        with m.If(counter == 2**len(counter) - 1):
+            m.d.comb += self.valid.eq(1)
+            m.d.comb += self.value.eq(Cat(1, last, counter))
+            m.d.sync += counter.eq(0)
+            m.d.sync += plain_mode.eq(0)
+        
+        with m.If(plain_mode & (counter == 14)):
+            m.d.comb += self.valid.eq(1)
+            m.d.comb += self.value.eq(Cat(0, buffer))
+            m.d.sync += counter.eq(0)
+            m.d.sync += plain_mode.eq(0)
+
+        return m
+
+    @staticmethod
+    def decode_pkg(pkg) -> List[Tuple[int, int]]:
+        if pkg & 0b01:
+            value = pkg >> 1 & 0b01
+            duration = pkg >> 2
+            return [(value, duration + 1)]
+        else:
+            return [(int(x), 1) for x in list('{0:015b}'.format(pkg >> 1))]
+
+    @staticmethod
+    def expand_duration_list(duration_list: List[Tuple[int, int]]) -> List[int]:
+        return list(chain(*[[value] * duration for value, duration in duration_list]))