cores/ila: Add ILA core

Author: Jean THOMAS

lambdalib/cores/ila.py

@@ -0,0 +1,108 @@
+from amaranth import *  # type: ignore
+from lambdalib.interface import stream
+from lambdalib.cores.mem.stream import MemoryStream
+from lambdalib.cores.serial import AsyncSerialTXStream
+
+__all__ = ["ILA"]
+
+
+class ILA(Elaboratable):
+    """Integrated Logic Analyzer (ILA) for capturing and analyzing digital signals.
+
+    This is a simple implementation of an Integrated Logic Analyzer (ILA).
+    It captures data on a trigger signal and allows reading it out later via
+    a serial interface at 115200 baud (or any other baudrate if specified).
+
+    An accompagnying tool can be used to read the captured data from the serial
+    port and generate a VCD file for waveform analysis:
+
+        ila.py --depth 65536 --layout "data:8,valid:1,ready:1" \
+               /dev/tty.usbserial-102 capture.vcd
+
+    The ILA operates in four states:
+    - IDLE: Waiting for a trigger signal
+    - CAPTURE: Recording incoming data into memory
+    - REWIND: Preparing to read out captured data
+    - READOUT: Streaming captured data via serial interface
+
+    Parameters:
+        data_width: Width of the data bus to capture (in bits)
+        depth: Number of samples to capture in memory
+        sys_clk_freq: System clock frequency for serial baud rate calculation
+
+    Attributes:
+        data_in: Input signal to capture
+        trigger: Signal to start data capture
+        tx: Serial output for data readout
+    """
+    def __init__(self, data_width: int, depth: int, sys_clk_freq: int, baudrate: int = 115200):
+        self._data_width = data_width
+        self._depth = depth
+        self._sys_clk_freq = sys_clk_freq
+        self._baudrate = baudrate
+
+        self.data_in = Signal(data_width)
+        self.trigger = Signal()
+        self.tx      = Signal()
+
+    def elaborate(self, platform) -> Module:
+        m = Module()
+
+        # Calculate upper multiple of 8 for better data alignment
+        aligned_width = ((self._data_width + 7) // 8) * 8
+        
+        m.submodules.mem = mem = MemoryStream(dw=aligned_width, depth=self._depth)
+        
+        # Pad input data to aligned width
+        padded_data = Signal(aligned_width)
+        m.d.comb += padded_data[:self._data_width].eq(self.data_in)
+        m.d.comb += [
+            mem.sink.data.eq(padded_data),
+        ]
+        
+        m.submodules.downconverter = downconverter = stream._DownConverter(
+            nbits_from=aligned_width,
+            nbits_to=8,
+            ratio=aligned_width // 8,
+            reverse=False
+        )
+
+        m.submodules.tx = tx = AsyncSerialTXStream(
+            o=self.tx,
+            divisor=self._sys_clk_freq // self._baudrate,
+        )
+        m.d.comb += downconverter.source.connect(tx.sink)
+
+        count = Signal(range(self._depth))
+
+        with m.FSM():
+            with m.State("IDLE"):
+                m.d.sync += count.eq(0)
+
+                with m.If(self.trigger):
+                    m.d.comb += mem.sink.valid.eq(1)
+                    m.d.sync += count.eq(count + 1)
+                    m.next = "CAPTURE"
+
+            with m.State("CAPTURE"):
+                m.d.comb += mem.sink.valid.eq(1)
+                m.d.sync += count.eq(count + 1)
+                with m.If(count == self._depth - 1):
+                    m.next = "REWIND"
+
+            with m.State("REWIND"):
+                m.d.comb += mem.rewind.eq(1)
+                m.d.sync += count.eq(0)
+                m.next = "READOUT"
+
+            with m.State("READOUT"):
+                m.d.comb += mem.source.connect(downconverter.sink)
+                with m.If(mem.source.valid & downconverter.sink.ready):
+                    m.d.sync += count.eq(count + 1)
+                with m.If(count == self._depth - 1):
+                    m.next = "STUCK"
+
+            with m.State("STUCK"):
+                pass
+
+        return m

lambdalib/software/ila.py

@@ -0,0 +1,141 @@
+"""Host-side software for ILA"""
+from argparse import ArgumentParser
+from serial import Serial
+from vcd import VCDWriter
+
+
+def parse_layout(layout_str):
+    """Parse signal layout string into list of (name, width) tuples.
+    
+    Args:
+        layout_str: String like 'data_in:10,trigger:1,address:8'
+        
+    Returns:
+        List of (signal_name, width) tuples
+        
+    Raises:
+        ValueError: If layout string is malformed
+    """
+    signals = []
+    if not layout_str.strip():
+        raise ValueError("Layout string cannot be empty")
+        
+    for signal_def in layout_str.split(','):
+        signal_def = signal_def.strip()
+        if ':' not in signal_def:
+            raise ValueError(f"Invalid signal definition '{signal_def}'. Expected format 'name:width'")
+            
+        name, width_str = signal_def.split(':', 1)
+        name = name.strip()
+        width_str = width_str.strip()
+        
+        if not name:
+            raise ValueError(f"Signal name cannot be empty in '{signal_def}'")
+            
+        try:
+            width = int(width_str)
+            if width <= 0:
+                raise ValueError(f"Signal width must be positive, got {width} for '{name}'")
+        except ValueError as e:
+            if "invalid literal" in str(e):
+                raise ValueError(f"Invalid width '{width_str}' for signal '{name}'. Width must be a positive integer")
+            raise
+            
+        signals.append((name, width))
+    
+    return signals
+
+
+def extract_signal_value(data_value, bit_offset, width):
+    """Extract a signal value from the packed data.
+    
+    Args:
+        data_value: Full packed data value
+        bit_offset: Starting bit position (LSB = 0)
+        width: Number of bits to extract
+        
+    Returns:
+        Extracted signal value
+    """
+    mask = (1 << width) - 1
+    return (data_value >> bit_offset) & mask
+
+
+if __name__ == "__main__":
+    parser = ArgumentParser(description="ILA Capture Tool")
+    parser.add_argument("port", help="Serial port for ILA data")
+    parser.add_argument("output", help="Output VCD file")
+    parser.add_argument("--baudrate", type=int, default=115200, help="Baud rate for serial communication (default: 115200)")
+    parser.add_argument("--depth", type=int, required=True, help="Number of samples captured by ILA")
+    parser.add_argument("--layout", type=str, required=True, help="Signal layout description (e.g. 'data_in:10,trigger:1,address:8')")
+    args = parser.parse_args()
+
+    # Parse and validate the layout
+    try:
+        signals = parse_layout(args.layout)
+    except ValueError as e:
+        print(f"Error parsing layout: {e}")
+        exit(1)
+    
+    # Calculate data width from layout
+    data_width = sum(width for _, width in signals)
+    
+    print(f"Parsed layout: {signals}")
+    print(f"Inferred data width: {data_width} bits")
+    print(f"Waiting for {args.depth} samples of {data_width}-bit data from {args.port}")
+
+    try:
+        with Serial(args.port, args.baudrate, timeout=None) as ser, open(args.output, "w") as vcd_file:
+            with VCDWriter(vcd_file, timescale="1 ns") as vcd:
+                # Register clock signal
+                clk_signal = vcd.register_var("ila", "clk", "wire", size=1)
+                
+                # Register all signals in the VCD
+                vcd_signals = []
+                for name, width in signals:
+                    vcd_signal = vcd.register_var("ila", name, "wire", size=width)
+                    vcd_signals.append(vcd_signal)
+                
+                # Initialize clock signal to low
+                vcd.change(clk_signal, 0, 0)
+                
+                # Capture and decode samples
+                bytes_to_read = (data_width + 7) // 8  # Round up to nearest byte
+                print(f"Waiting for data on {args.port}... (Press Ctrl+C to abort)")
+                
+                for sample_index in range(args.depth):
+                    # Read raw data from serial port - this will block until data is available
+                    raw_data = ser.read(bytes_to_read)
+                    
+                    if len(raw_data) < bytes_to_read:
+                        print(f"Incomplete data received at sample {sample_index}. Expected {bytes_to_read} bytes, got {len(raw_data)}. Exiting.")
+                        break
+                    
+                    # Convert bytes to integer (little-endian)
+                    data_value = int.from_bytes(raw_data, byteorder='little')
+                    
+                    # Extract and log individual signal values
+                    bit_offset = 0
+                    timestamp = sample_index * 1000  # 1ns timestep, 1us per sample
+                    
+                    # Generate clock signal - rising edge at start of each sample
+                    vcd.change(clk_signal, timestamp, 1)
+                    
+                    for (name, width), vcd_signal in zip(signals, vcd_signals):
+                        signal_value = extract_signal_value(data_value, bit_offset, width)
+                        vcd.change(vcd_signal, timestamp, signal_value)
+                        bit_offset += width
+
+                    # Falling edge at middle of sample period
+                    vcd.change(clk_signal, timestamp + 500, 0)
+                        
+                    if sample_index % 100 == 0:  # Progress indicator
+                        print(f"Processed {sample_index + 1}/{args.depth} samples")
+                        
+        print(f"Data capture complete. Output written to {args.output}")
+        
+    except KeyboardInterrupt:
+        print(f"\nCapture interrupted by user. Partial data written to {args.output}")
+    except Exception as e:
+        print(f"Error during capture: {e}")
+        exit(1)