Encoder not working yet

Author: jeandet

README.md

@@ -18,12 +18,19 @@ This project leverages the Amaranth HDL to test a particle detector ASIC design.
 
 ## Getting Started
 
+You first need install Lattice iCE40 toolchain (yosys, nextpnr, icestorm) and iverilog for simulation. On Fedora, you can install them using:
+```bash
+sudo dnf install iverilog icestorm yosys nextpnr python3-pip
+```
+
+You can also have a look to the [OSS CAD Suite](https://github.com/YosysHQ/oss-cad-suite-build) project, which provides pre-built toolchain binaries for various platforms.
+
 1. Clone the repository:
     ```bash
     git clone <repository_url>
     cd ASIC_Counter
     ```
-2. Install dependencies using uv:
+2. Install python dependencies using uv:
     ```bash
     uv venv
     uv pip install -e .[dev]

asic_counter/protocol.py

@@ -67,40 +67,47 @@ def __init__(self):
         self.data_in = Signal(8)
         self.in_rdy = Signal()
         self.in_ack = Signal()
+        self.eof = Signal()
+        
+        self.uart_data_out = Signal(8)
+        self.uart_rdy = Signal()
+        self.uart_ack = Signal()
 
-        self.data_out = Signal(8)
-        self.out_rdy = Signal()
-        self.out_ack = Signal()
 
     def elaborate(self, platform):
         m = Module()
 
-        # For simplicity, we just output the command and value when out_rdy is high
         with m.FSM(init="Idle") as fsm:
             with m.State("Idle"):
-                with m.If(self.out_rdy & self.in_rdy):
-                    m.d.sync += self.data_out.eq(0xA5)  # Start word
+                with m.If(self.uart_rdy & self.in_ack):
+                    m.d.sync += self.uart_data_out.eq(0xA5)  # Start word
+                    m.d.sync += self.uart_ack.eq(1)
                     m.next = "Second start word"
+                with m.Else():
+                    m.d.sync += self.uart_ack.eq(0)
+                m.d.sync += self.in_rdy.eq(self.uart_rdy) 
             with m.State("Second start word"):
-                with m.If(self.out_ack):
-                    m.d.sync += self.data_out.eq(0x0F)  # Second start word
-                    m.next = "Output second start word"
-            with m.State("Output second start word"):
-                with m.If(self.out_ack):
-                    # For simplicity, we just output a fixed command and value
-                    m.d.sync += self.data_out.eq(0b101)  # Command
-                    m.next = "Output command"
-            with m.State("Output command"):
-                with m.If(self.out_ack):
-                    m.d.sync += self.data_out.eq(0x34)  # Value LSB
-                    m.next = "Output value LSB"
-            with m.State("Output value LSB"):
-                with m.If(self.out_ack):
-                    m.d.sync += self.data_out.eq(0x12)  # Value MSB
-                    m.next = "Output value MSB"
-            with m.State("Output value MSB"):
-                with m.If(self.out_ack):
-                    m.d.sync += self.out_rdy.eq(0)  # Done
+                with m.If(self.uart_rdy & self.in_ack):
+                    m.d.sync += self.uart_data_out.eq(0x0F)  # Second start word
+                    m.next = "Output values"
+                with m.Else():
+                    m.d.sync += self.uart_ack.eq(0)
+                    m.d.sync += self.in_rdy.eq(self.uart_rdy) 
+            with m.State("Output values"):
+                with m.If(self.uart_rdy & self.in_ack):
+                    m.d.sync += self.uart_data_out.eq(self.data_in)
+                with m.Else():
+                    m.d.sync += self.uart_ack.eq(0)
+                    m.d.sync += self.in_rdy.eq(self.uart_rdy) 
+                with m.If(self.eof):
+                    m.next = "EOF"
+            with m.State("EOF"):
+                m.d.sync += self.uart_data_out.eq(0xFF)  # EOF marker
+                with m.If(self.uart_rdy & self.in_ack):
+                    m.d.sync += self.uart_ack.eq(1)
                     m.next = "Idle"
+                with m.Else():
+                    m.d.sync += self.uart_ack.eq(0)
+                    m.d.sync += self.in_rdy.eq(self.uart_rdy) 
 
         return m

asic_counter/test_runner.py

@@ -1,10 +1,19 @@
 from cocotb_test.simulator import run as _run
+from cocotb.triggers import RisingEdge
 from amaranth.back import verilog
 from amaranth import Signal
 from tempfile import NamedTemporaryFile
 import inspect
 import os
 
+async def reset_dut(dut):
+    dut.rst.value = 1
+    for _ in range(5):  
+        await RisingEdge(dut.clk)
+    dut.rst.value = 0
+    for _ in range(5):  
+        await RisingEdge(dut.clk)
+
 def find_dut_ports(dut):
     # Heuristic: look for Signals that are not internal (e.g., not starting with "_")
     ports = []

tests/protocol/test_protocol.py tests/decoder/test_decoder.pytests/protocol/test_protocol.py renamed to tests/decoder/test_decoder.py

@@ -2,16 +2,9 @@
 
 from cocotb.triggers import Timer, RisingEdge
 from cocotb.clock import Clock
-from asic_counter.test_runner import run_cocotb
+from asic_counter.test_runner import run_cocotb, reset_dut
 from asic_counter.protocol import Decoder
 
-async def reset_dut(dut):
-    dut.rst.value = 1
-    for _ in range(5):  
-        await RisingEdge(dut.clk)
-    dut.rst.value = 0
-    for _ in range(5):  
-        await RisingEdge(dut.clk)
 
 async def send_cmd(dut, cmd, value):
     # Send start sequence

tests/encoder/test_encoder.py

@@ -0,0 +1,51 @@
+import cocotb
+
+from cocotb.triggers import Timer, RisingEdge
+from cocotb.clock import Clock
+from asic_counter.test_runner import run_cocotb, reset_dut
+from asic_counter.protocol import Encoder
+
+class FakeUART:
+    def __init__(self, dut:Encoder):
+        self.dut:Encoder = dut
+        self.data = []
+        self.dut.uart_rdy.value = 1
+        self.task = cocotb.start_soon(self.send())
+    
+    async def send(self):
+        self.dut.uart_rdy.value = 1
+        while True:
+            if self.dut.uart_ack.value == 0:
+                await RisingEdge(self.dut.uart_ack)
+            self.dut.uart_rdy.value = 0
+            self.data.append(int(self.dut.uart_data_out.value))
+            await Timer(100, units="ns")  # Simulate some delay for the UART to process the data
+            self.dut.uart_rdy.value = 1
+
+async def send_data(dut:Encoder, values):
+    dut.in_ack.value = 1
+    for value in values:
+        if dut.in_rdy.value == 0:
+            await dut.in_rdy.rising_edge
+        dut.data_in.value = value
+        await RisingEdge(dut.clk)
+    await RisingEdge(dut.clk)
+    dut.eof.value = 1
+    await RisingEdge(dut.clk)
+    dut.in_ack.value = 0
+    
+    
+@cocotb.test()
+async def test_protocol_simple_cmds(dut:Encoder):
+    # Drive inputs (Amaranth signals become Verilog ports)
+    clk_gen = cocotb.start_soon(Clock(dut.clk, 1000 // 12, units="ns").start())
+    dut.in_ack.value = 0
+    dut.eof.value = 0
+    await reset_dut(dut)
+    uart = FakeUART(dut)
+    await send_data(dut, values=[1, 2, 3, 4])
+    await RisingEdge(dut.clk)
+    
+
+def broken_test_protocol():
+    run_cocotb(Encoder)