rebbarb/exi_bba/exi_capture.py

"""ExiCapture — the EXI front end, clocked by the GameCube's own EXI clock.

Owns the pin-driven clock domains (which is why it creates them rather than
`BBATop`: the byte FIFOs have to name them), wraps the `SPIMode3Slave` bit
engine, and bridges that engine to the 24 MHz `sync` domain through two
`AsyncFIFO`s:

    EXI clock pin (~27 MHz)                  sync (24 MHz, HFOSC)
    ┌─────────────────────┐   rx_fifo  ───►  received bytes (header + data)
    │  SPIMode3Slave      │   (8-bit, exififo_w→sync)
    │  (bit engine)       │   tx_fifo  ◄───  response bytes to drive on MISO
    └─────────────────────┘   (8-bit, sync→exififo_r)

These two FIFOs are the ONLY clock-domain crossings in the EXI path, and each
has a whole byte time (~296 ns at 27 MHz) of slack.  That is the point of
clocking the engine from the pin: oversampling it from a faster domain put
BIT-level timing across a crossing instead, with roughly 55 ns of synchroniser
latency against a 37 ns bit, and MISO came out two bits late.

Domains created here
--------------------
    exiclk     EXI clock pin, posedge, ASYNC-RESET BY CS.  Hosts the bit
               engine: samples MOSI, and launches MISO one bit early.
    exififo_w  EXI clock pin, posedge, RESET-LESS.  rx FIFO write side.
    exififo_r  EXI clock pin, posedge, RESET-LESS.  tx FIFO read side.

All three are the same clock and the same edge; they differ only in reset,
and each difference is forced:

  * The bit engine MUST be reset by CS, because its bit and byte counters
    have to restart at every transaction.
  * The FIFO sides must NOT see CS.  Resetting one side of an `AsyncFIFO`
    desynchronises its gray pointers against the other side and corrupts it.
  * Both FIFO sides must be POSEDGE, because `AsyncFIFO` instantiates a
    posedge-only reset CDC internally and cannot host a negedge side.

TX response timing
------------------
The engine pops `tx_fifo` once per byte boundary, unconditionally -- there is
no header counter and no slot threshold.  The byte popped at the end of byte N
is presented during byte N+1, which gives the scheme two useful properties:

  * pops during a header drain whatever the previous transaction left in the
    FIFO, so no CS-fall flush is needed; and
  * the response the register file pushes during the protocol's turnaround
    byte lands in data slot 0.

⚠️ Command space has only ONE drain slot, because its header is two bytes.  A
command-space responder must therefore push exactly its bytes and stop; see
the device-ID note in `bba_register_file.py`.
"""

from amaranth import *
from amaranth.lib.cdc import FFSynchronizer
from amaranth.lib.fifo import AsyncFIFO

from exi_bba.spi_mode3_slave import SPIMode3Slave

__all__ = ["ExiCapture"]


class ExiCapture(Elaboratable):
    """The EXI bit engine plus the byte FIFOs that connect it to `sync`.

    Physical SPI pins, from the GC
    ------------------------------
    spi_clk, spi_mosi, spi_cs_n
        Raw inputs.  `spi_clk` becomes the clock of all three domains above
        and `spi_cs_n` the async reset of `exiclk`.
    spi_miso
        Serial output, driven from the fabric shift register.
    miso_next
        The same bit one edge early, for an SB_IO output register; this is
        what the synthesised build drives the pad from.

    RX byte stream, read side in `sync`
    -----------------------------------
    rx_data   the received byte at the head of the FIFO
    rx_rdy    a byte is available
    rx_en     pop; assert for one `sync` cycle to consume `rx_data`

    TX byte stream, write side in `sync`
    ------------------------------------
    tx_data   response byte to enqueue
    tx_en     write strobe
    tx_rdy    the FIFO has room

    Transaction framing
    -------------------
    cs_active
        High in `sync` while a transaction is in progress.  The register file
        uses it to stream a variable-length (DMA) read until CS deasserts,
        since the protocol carries no length field.
    """

    def __init__(self, rx_depth=4, tx_depth=2, phase_meter=False):
        self._rx_depth = rx_depth
        self._tx_depth = tx_depth
        self._phase_meter = phase_meter   # diagnostic; see SPIMode3Slave

        self.spi_clk  = Signal(init=1)
        self.spi_mosi = Signal()
        self.spi_cs_n = Signal(init=1)
        self.spi_miso = Signal()
        self.miso_next = Signal(init=1)

        self.rx_data = Signal(8)
        self.rx_rdy  = Signal()
        self.rx_en   = Signal()

        self.tx_data = Signal(8)
        self.tx_en   = Signal()
        self.tx_rdy  = Signal()

        self.cs_active = Signal()

    def elaborate(self, platform):
        m = Module()
        self._clock_domains(m)
        spi = self._bit_engine(m)
        self._byte_fifos(m, spi)
        self._transaction_framing(m)
        return m

    def _clock_domains(self, m):
        """Create the three pin-clocked domains.

        EXI CLK is pin 44, which sits at die (6,0) on the 5k-sg48 = global
        buffer input net 6, so it drives a real clock network with no board
        change.  See the module docstring for why each domain's reset differs.
        """
        m.domains.exiclk    = cd_p  = ClockDomain(async_reset=True)
        m.domains.exififo_w = cd_fw = ClockDomain(reset_less=True)
        m.domains.exififo_r = cd_fr = ClockDomain(reset_less=True)
        m.d.comb += [
            cd_p .clk.eq(self.spi_clk), cd_p .rst.eq(self.spi_cs_n),
            cd_fw.clk.eq(self.spi_clk),
            cd_fr.clk.eq(self.spi_clk),
        ]

    def _bit_engine(self, m):
        """Instantiate the bit engine and wire it to the pins."""
        spi = SPIMode3Slave(phase_meter=self._phase_meter)
        m.submodules.spi = spi

        # The device-ID fast path is ARMED PERMANENTLY, because real hardware
        # answers EXI_GetID whenever it is asked -- that is how EXI enumerates
        # devices, and a device that stops identifying itself after init reads
        # to the host as absent rather than broken.
        #
        # `id_arm` remains a port only for the PHASE_METER diagnostic build,
        # which must turn the fast path OFF so its constant 0x80 pattern is
        # not replaced by ID bytes in the slots it measures.
        m.d.comb += spi.id_arm.eq(0 if self._phase_meter else 1)

        m.d.comb += [
            spi.spi_clk .eq(self.spi_clk),
            spi.spi_mosi.eq(self.spi_mosi),
            spi.spi_cs_n.eq(self.spi_cs_n),
            self.spi_miso .eq(spi.spi_miso),
            self.miso_next.eq(spi.miso_next),
        ]
        return spi

    def _byte_fifos(self, m, spi):
        """The two AsyncFIFOs that carry bytes across to and from `sync`."""
        rx_fifo = AsyncFIFO(width=8, depth=self._rx_depth,
                            w_domain="exififo_w", r_domain="sync")
        tx_fifo = AsyncFIFO(width=8, depth=self._tx_depth,
                            w_domain="sync", r_domain="exififo_r")
        m.submodules.rx_fifo = rx_fifo
        m.submodules.tx_fifo = tx_fifo

        # RX: every byte the engine completes goes straight in.
        m.d.comb += [
            rx_fifo.w_data.eq(spi.rx_byte),
            rx_fifo.w_en  .eq(spi.rx_valid),
            self.rx_data  .eq(rx_fifo.r_data),
            self.rx_rdy   .eq(rx_fifo.r_rdy),
            rx_fifo.r_en  .eq(self.rx_en),
        ]

        # TX: `sync` pushes responses and the engine pops one per byte time.
        # Those pops are unconditional -- see the module docstring.
        m.d.comb += [
            tx_fifo.w_data.eq(self.tx_data),
            tx_fifo.w_en  .eq(self.tx_en),
            self.tx_rdy   .eq(tx_fifo.w_rdy),
            spi.tx_byte   .eq(tx_fifo.r_data),
            tx_fifo.r_en  .eq(spi.tx_load),
        ]

    def _transaction_framing(self, m):
        """Bring CS into `sync` as a level.

        CS changes once per transaction, which is slow relative to `sync`, so
        a plain two-stage synchroniser is the right primitive.  Note what is
        deliberately NOT done here: the EXI clock and MOSI are not
        synchronised anywhere.  CLK *is* a clock and CS *is* an async reset --
        synchronising them into a fast domain was the original MISO bug, not
        a safety measure.
        """
        cs_n_s = Signal(init=1)
        m.submodules.cs_sync = FFSynchronizer(self.spi_cs_n, cs_n_s,
                                              o_domain="sync", init=1)
        m.d.comb += self.cs_active.eq(~cs_n_s)


# ── Testbench ─────────────────────────────────────────────────────────────

if __name__ == "__main__":
    import sys
    from amaranth.sim import Simulator, Period

    dut = ExiCapture()
    errors = []

    # SPI half-period, in units of HP below.  At 27 MHz EXI the real
    # ratio is ~2; use 4 here for a clean, well-oversampled functional check.
    HALF = 4

    # The bit engine is clocked by the EXI pin, so there is no faster domain
    # to count ticks against; the SPI clock is driven with real delays.
    HP = Period(MHz=54)   # the unit of delay; half-periods are HP * HALF

    async def spi_byte(ctx, mosi_val):
        """Clock one SPI Mode 3 byte; return the assembled MISO byte."""
        miso = 0
        for bit in range(7, -1, -1):
            ctx.set(dut.spi_mosi, (mosi_val >> bit) & 1)
            ctx.set(dut.spi_clk, 0)
            await ctx.delay(HP * (HALF))
            miso = (miso << 1) | ctx.get(dut.spi_miso)
            ctx.set(dut.spi_clk, 1)
            await ctx.delay(HP * (HALF))
        return miso

    async def core_drain_rx(ctx, into):
        """Pop one byte from the core RX side if available."""
        if ctx.get(dut.rx_rdy):
            into.append(ctx.get(dut.rx_data))
            ctx.set(dut.rx_en, 1)
            await ctx.tick("sync").repeat(1)
            ctx.set(dut.rx_en, 0)
            return True
        return False

    async def push_tx(ctx, b):
        ctx.set(dut.tx_data, b)
        ctx.set(dut.tx_en, 1)
        await ctx.tick("sync").repeat(1)
        ctx.set(dut.tx_en, 0)

    async def do_txn(ctx, hdr, responses, n_data, rx_seen):
        """One EXI transaction, modelling the REAL protocol shape.

        The engine pops tx_fifo at every byte boundary, so a response must be
        in the FIFO before the LAST header byte ends to land in data slot 0.
        That byte is the protocol's turnaround byte (docs/exi_protocol_real.md)
        and is exactly when the register file pushes.  Pushing after the whole
        header -- as this bench used to -- is one slot too late.
        """
        ctx.set(dut.spi_cs_n, 0)
        ctx.set(dut.spi_clk, 1)
        await ctx.delay(HP * (HALF))
        for h in hdr[:-1]:
            await spi_byte(ctx, h)
            await core_drain_rx(ctx, rx_seen)
        for r in responses:                       # pushed during the turnaround
            await push_tx(ctx, r)
        await spi_byte(ctx, hdr[-1])
        await core_drain_rx(ctx, rx_seen)
        miso = [await spi_byte(ctx, 0x00) for _ in range(n_data)]
        ctx.set(dut.spi_cs_n, 1)
        await ctx.delay(HP * (HALF))
        for _ in range(20):                       # drain data-phase dummies
            await core_drain_rx(ctx, rx_seen)
            await ctx.tick("sync").repeat(1)
        return miso

    async def testbench(ctx):
        rx_seen = []
        await ctx.delay(HP * (2))

        # Real register-space header: [0x80, ah, al, 0x00] -- the trailing
        # byte is the turnaround slot.
        HDR = [0x80, 0x00, 0x12, 0x00]

        # ── T1: header + 2 data bytes read back ──────────────────────────
        miso = await do_txn(ctx, HDR, [0xA5, 0x5A], 2, rx_seen)
        print(f"T1 rx={[hex(b) for b in rx_seen[:4]]}  MISO={[f'0x{b:02X}' for b in miso]}")
        if rx_seen[:4] != HDR:
            errors.append(f"T1 header rx wrong: {[hex(b) for b in rx_seen[:4]]}")
        if miso != [0xA5, 0x5A]:
            errors.append(f"T1 MISO wrong: {[hex(b) for b in miso]}")

        # ── T2: over-push must NOT leak into the next transaction ────────
        # Txn A pushes 2 responses but the GC clocks only 1 data byte, leaving
        # a stale byte in tx_fifo.  Txn B must read its OWN responses.  There
        # is no CS-fall flush any more: the engine's pops during the header
        # drain whatever the previous transaction left behind.
        rx_seen.clear()
        await do_txn(ctx, HDR, [0xA5, 0x5A], 1, rx_seen)   # leaves 0x5A
        rx_seen.clear()
        misoB = await do_txn(ctx, HDR, [0x11, 0x22], 2, rx_seen)
        print(f"T2 MISO after over-push: {[f'0x{b:02X}' for b in misoB]}  (want 0x11 0x22)")
        if misoB != [0x11, 0x22]:
            errors.append(f"T2 stale byte leaked: {[hex(b) for b in misoB]}")

    sim = Simulator(dut)
    sim.add_clock(Period(MHz=24), domain="sync")
    sim.add_testbench(testbench)

    with sim.write_vcd("ExiCapture.vcd"):
        sim.run()

    if errors:
        print("\nFAILURES:")
        for e in errors:
            print(" ", e)
        sys.exit(1)
    else:
        print("\nAll tests passed.")