buswrapper.py

from nmigen import *
from enum import IntEnum

from nmigen.utils import bits_for

class AccessFlags(IntEnum):
    R  = 1 << 0
    W  = 1 << 1
    RW = R | W

class BusWrapper(Elaboratable):

    def __init__(self, address_width=8, io_width=32, signals_rw=[], signals_r=[], signals_w=[]):
        self.address_width = address_width
        self.io_width = io_width

        self.cs         = Signal()
        self.we         = Signal()
        self.addr       = Signal(address_width)
        self.write_data = Signal(io_width)
        self.read_data  = Signal(io_width)

        self.endpoints  = dict()
        self.add_endpoints(AccessFlags.RW, signals_rw)
        self.add_endpoints(AccessFlags.R,  signals_r)
        self.add_endpoints(AccessFlags.W,  signals_w)

    def add_endpoints(self, flags: AccessFlags, signals: list[Signal]):
        assert(self.address_width >= bits_for(len(self.endpoints) + len(signals) - 1))
        for s in signals:
            assert(self.io_width >= len(s))

        for i, v in enumerate(signals):
            if isinstance(v, Array):
                self.add_endpoints(flags, v)
            else:
                self.endpoints.update({len(self.endpoints) : (flags, v)})

    def __repr__(self):
        lines = [f"Addr\tFlags\tName"]
        for i, v in self.endpoints.items():
            s = v[1]
            if isinstance(s, Signal):
                name = s.name
            else:
                # Create a name from a Slice()
                name = f"{s.value.name}_{s.start}_{s.stop - 1}"
            lines.append(f"{i:02x}\t{v[0].name}\t{name}")
        return "\n".join(lines)

    def elaborate(self, platform):
        m = Module()

        endpoints = self.endpoints
        addr = self.addr
        read_data = self.read_data
        write_data = self.write_data

        with m.If(self.cs):
            with m.If(self.we):
                with m.Switch(addr):
                    for ep in filter(lambda i: endpoints[i][0] & AccessFlags.W, endpoints):
                        with m.Case(ep):
                            m.d.sync += endpoints[ep][1].eq(write_data)
            with m.Else(): # ~we
                with m.Switch(addr):
                    for ep in filter(lambda i: endpoints[i][0] & AccessFlags.R, endpoints):
                        with m.Case(ep):
                            m.d.comb += read_data.eq(endpoints[ep][1])

        return m

top.py

from nmigen import *
from nmcrypto import *
from nmcrypto.util.buswrapper import *
from nmcrypto.platforms.ecp5 import ECP5Platform
from nmcrypto.platforms.artix7 import Artix7Platform
from nmcrypto.platforms.cyclonev import CycloneVPlatform

class ChaCha20Tester(Elaboratable):

    def elaborate(self, platform):
        m = Module()

        cs = platform.request("cs")
        we = platform.request("we")
        addr = platform.request("addr")
        write_data = platform.request("write_data")
        read_data = platform.request("read_data")

        def split_into_words(s, l):
            # Split s into signals that are at most l bits wide
            return list(s[i*l:i*l+l] for i in range((len(s) + l - 1) // l))

        m.submodules.chacha = chacha = ChaChaFSM1()
        m.submodules.wrapper = wrapper = BusWrapper(
            signals_rw=[
                chacha.i_en,
                chacha.i_counter
            ],
            signals_r=[
                chacha.o_data_ready,
                *split_into_words(chacha.o_data, 32),
            ],
            signals_w=[
                *split_into_words(chacha.i_data, 32),
                chacha.i_key,
                chacha.i_nonce,
            ],
        )

        m.d.comb += [
            wrapper.cs.eq(cs),
            wrapper.we.eq(we),
            wrapper.addr.eq(addr),
            wrapper.write_data.eq(write_data),
            read_data.eq(wrapper.read_data),
        ]

        return m

if __name__ == "__main__":
    platform = ECP5Platform()
    platform.build(ChaCha20Tester())