add swapbank + test

src/groovylight/bitslicer.py

@@ -1,8 +1,10 @@
-from amaranth import Array, Module, Cat, Signal, Assert, unsigned
+from amaranth import Array, Module, Cat, Mux, ShapeLike, Signal, Assert, unsigned
 from amaranth.build import Platform
 from amaranth.lib import wiring, data
 from amaranth.lib.wiring import In, Out
-from amaranth.lib.memory import Memory, WritePort
+from amaranth.lib.memory import Memory, ReadPort, WritePort
+from amaranth.sim import Simulator
+from amaranth.utils import ceil_log2
 
 
 class RGBLayout(data.StructLayout):
@@ -16,9 +18,9 @@ class RGBLayout(data.StructLayout):
         )
 
 
-rgb888_layout = RGBLayout(8, 8, 8)
+Rgb888Layout = RGBLayout(8, 8, 8)
 
-rgb111_hub75 = RGBLayout(1, 1, 1)
+Rgb111Layout = RGBLayout(1, 1, 1)
 
 
 class Hub75Stream(wiring.Signature):
@@ -63,16 +65,79 @@ class Hub75Data(wiring.Signature):
     def __init__(self):
         super().__init__(
             {
-                "rgb0": Out(rgb111_hub75),
-                "rgb1": Out(rgb111_hub75),
+                "rgb0": Out(Rgb111Layout),
+                "rgb1": Out(Rgb111Layout),
             }
         )
 
+class SwapBuffer(wiring.Component):
+    """A pair of BRAMs for holdling line data that are swapped between using an external signal.
 
+    Contains one write port, and one read port. They are attached to separate BRAMs to allow for
+    one to be read at the same time another is being updated.
 
-def get_lineram() -> Memory:
-    return Memory(shape=rgb888_layout, depth=512, init=[])
+    Parameters:
+        depth: i32, depth of each memory buffer
+        shape: ShapeLike, the underlying shape of the memory.
 
+    Signals:
+        write_port: WritePort.Signature(width, shape)
+        read_port: ReadPort.Signature(width, shape)
+        selector: In(1)
+
+    """
+
+    def __init__(self, shape: ShapeLike, depth: int):
+        super().__init__(
+            {
+                "selector": In(1),
+                "read_port": In(
+                    ReadPort.Signature(addr_width=ceil_log2(depth), shape=shape)
+                ),
+                "write_port": In(
+                    WritePort.Signature(addr_width=ceil_log2(depth), shape=shape)
+                ),
+            }
+        )
+        self.data_shape = shape
+        self.depth = depth
+
+    def elaborate(self, platform: Platform) -> Module:
+        m = Module()
+
+        m.submodules.bram0 = self.bram0 = Memory(
+            shape=self.data_shape, depth=self.depth, init=[]
+        )
+        m.submodules.bram1 = self.bram1 = Memory(
+            shape=self.data_shape, depth=self.depth, init=[]
+        )
+
+        read0 = self.bram0.read_port()
+        write0 = self.bram0.write_port()
+        read1 = self.bram1.read_port()
+        write1 = self.bram1.write_port()
+
+        # for name, member in self.write_port.signature.members.items():
+        #     m.d.comb += self.write_port.members[name].eq(Mux(self.selector, write0[name], write1[name]))
+        #
+
+        m.d.comb += [
+            write0.addr.eq(self.write_port.addr),
+            write1.addr.eq(self.write_port.addr),
+            write0.en.eq(~self.selector),
+            write1.en.eq(self.selector),
+            # self.write_port.data.eq(Mux(self.selector, write0.data, write1.data)),
+            write0.data.eq(self.write_port.data),
+            write1.data.eq(self.write_port.data),
+
+            read0.addr.eq(self.read_port.addr),
+            read1.addr.eq(self.read_port.addr),
+            read0.en.eq(~self.selector),
+            read1.en.eq(self.selector),
+            self.read_port.data.eq(Mux(self.selector, read1.data, read0.data)),
+        ]
+
+        return m
 
 
 class Hub75StringDriver(wiring.Component):
@@ -82,35 +147,41 @@ class Hub75StringDriver(wiring.Component):
     done: Out(1)
     start: In(1)
     active_bank: In(1)  # the bank to use
-    bram_port: WritePort.Signature(addr_width=9, shape=rgb888_layout)
-    out: Out(Hub75Data())
-
+    bram_port: Out(WritePort.Signature(
+        addr_width=9, shape=Rgb888Layout
+    )) # the other line to be swapped to.
+    display_out: Out(Hub75Data())  # data signal output.
 
     def elaborate(self, platform: Platform) -> Module:
         m = Module()
 
-
         # add two memories
-        m.submodules.bram0 = bram0 = Memory(shape=rgb888_layout, depth=512, init=[])
-        m.submodules.bram1 = bram1 = Memory(shape=rgb888_layout, depth=512, init=[])
-
+        m.submodules.bram0 = bram0 = Memory(shape=Rgb888Layout, depth=512, init=[])
+        m.submodules.bram1 = bram1 = Memory(shape=Rgb888Layout, depth=512, init=[])
 
         # We use two brams here so we can swap between each - that way we can update the data
         # while displaying the other line. Switching is controlled by the coordinator.
 
-        readports = Array([bram0.read_port(), bram1.read_port()])
+        readports = Mux([bram0.read_port(), bram1.read_port()])
         active_readport = readports[self.active_bank]
-        m.d.comb += self.active_readport.eq(readports[self.active_bank])
+        m.d.comb += active_readport.eq(readports[self.active_bank])
         writeports = Array([bram0.write_port(), bram1.write_port()])
         m.d.comb += self.bram_port.eq(writeports[~self.active_bank])
 
+        # We want to set up the bram ports
+        port0 = bram0.read_port()
+        port1 = bram1.read_port()
+        # the enable for these ports is based on active bank.
+        m.d.comb += port0.en.eq(self.active_bank)
+        m.d.comb += port1.en.eq(~self.active_bank)
+
         counter = Signal(32)
         m.d.sync += counter.eq(counter + 1)
 
         ram_rgb_slice = Cat(
-                active_readport.data['red'].bit_select(self.bcm_shift, 1),
-                active_readport.data['blue'].bit_select(self.bcm_shift, 1),
-                active_readport.data['green'].bit_select(self.bcm_shift, 1),
+            active_readport.data["red"].bit_select(self.bcm_shift, 1),
+            active_readport.data["blue"].bit_select(self.bcm_shift, 1),
+            active_readport.data["green"].bit_select(self.bcm_shift, 1),
         )
 
         pixnum = Signal(8, reset=127)
@@ -134,15 +205,27 @@ class Hub75StringDriver(wiring.Component):
                 with m.Elif(counter == 1):
                     m.d.sync += pixrow.eq(1)
                 with m.Elif(counter == 2):
-                    m.d.sync += self.out.rgb0.eq(ram_rgb_slice)
+                    m.d.sync += self.display_out.rgb0.eq(ram_rgb_slice)
                 with m.Elif(counter == 3):
-                    m.d.sync += [self.out.rgb1.eq(ram_rgb_slice), counter.eq(0)]
+                    m.d.sync += [self.display_out.rgb1.eq(ram_rgb_slice), counter.eq(0)]
                     m.next = "writerow"
 
-
         return m
 
 
+def test_stringdriver():
+    dut = Hub75StringDriver()
+    sim = Simulator(dut)
+    sim.add_clock(1e-6)
+
+    async def testbench(ctx):
+        pass
+
+    sim.add_testbench(testbench)
+    with sim.write_vcd("output.vcd"):
+        sim.run_until(1e-6 * 1000)
+
+
 class Hub75Coordinator(wiring.Component):
     """A shared-control hub75 driver"""
 
@@ -271,7 +354,9 @@ class Hub75EDriver(wiring.Component):
 
         return m
 
+
 if __name__ == "__main__":
     m = Hub75EDriver()
     from amaranth.cli import main
+
     main(m)

src/groovylight/test_swapbank.py

@@ -0,0 +1,33 @@
+from amaranth import Array, Module, Cat, Signal, Assert, unsigned
+from amaranth.build import Platform
+from amaranth.lib import wiring, data
+from amaranth.lib.wiring import In, Out
+from amaranth.lib.memory import Memory, WritePort
+from amaranth.sim import Simulator
+
+from .bitslicer import Hub75StringDriver, Rgb888Layout, SwapBuffer
+
+
+def test_swapbuffer():
+    dut = SwapBuffer(Rgb888Layout, 512)
+    sim = Simulator(dut)
+    sim.add_clock(1e-6)
+
+    async def testbench(ctx):
+        init_color = {"red": 0, "green": 0, "blue": 0}
+        test_color = {"red": 8, "green": 8, "blue": 8}
+        ctx.set(dut.selector, 0)
+        ctx.set(dut.write_port.addr, 1)
+        ctx.set(dut.read_port.addr, 1)
+        ctx.set(dut.write_port.data, test_color)
+        await ctx.tick()
+        # assert that the read port addr 1 = 0
+        assert ctx.get(dut.read_port.data) == init_color
+        # swap buffer
+        ctx.set(dut.selector, 1)
+        await ctx.tick().repeat(2) # takes two clocks after switching selector to output data.
+        assert ctx.get(dut.read_port.data) == test_color
+
+    sim.add_testbench(testbench)
+    with sim.write_vcd("output.vcd"):
+        sim.run_until(1e-6 * 1000)