"""BBA register file — EXI domain. Decodes EXI transactions (2-byte header + N data bytes), reads/writes the BBA register space, and owns all AsyncFIFO / PulseSynchronizer CDC primitives. Transaction header format -------------------------- Byte 0 [7] write_flag [6:0] addr[12:6] Byte 1 [7:2] addr[5:0] [1:0] xfer_len−1 (0=1B, 1=2B, 2=3B, 3=4B) Addresses 0x0000–0x00FF : register file (sparse individual Signals, exi domain). Addresses 0x0100–0x1FFF : SPRAM ring buffer (sync domain, prefetch FIFOs). """ from amaranth import * from amaranth.lib.fifo import SyncFIFO __all__ = ["BBARegisterFile"] # Register addresses _NCRA = 0x00 _IMR = 0x08 _IR = 0x09 _RWP_LO = 0x16 _RWP_HI = 0x17 _RRP_LO = 0x18 _RRP_HI = 0x19 _PAR0 = 0x20 _PAR1 = 0x21 _PAR2 = 0x22 _PAR3 = 0x23 _PAR4 = 0x24 _PAR5 = 0x25 _NWAYS = 0x31 _HIPR = 0x3A _TWD_LO = 0x34 _TWD_HI = 0x35 _TXDATA = 0x48 # Read-only hardcoded values _NWAYS_VAL = 0x17 _HIPR_VAL = 0x01 # Device ID returned on first 4-byte read of addr 0x0000 _DEVICE_ID = [0x04, 0x02, 0x02, 0x00] class BBARegisterFile(Elaboratable): """EXI transaction decoder and BBA register file with CDC bridges. Sync-domain FIFO/pulse ports are wired by BBATop to the sync-domain modules. """ def __init__(self): # ── EXI byte-stream interface (exi domain, from/to ExiCapture) ──── # RX: received bytes (header + write data + read dummies) — FWFT read # side of ExiCapture's rx_fifo. self.rx_data = Signal(8) self.rx_rdy = Signal() self.rx_en = Signal() # TX: response bytes pushed proactively into ExiCapture's tx_fifo. self.tx_data = Signal(8) self.tx_en = Signal() self.tx_rdy = Signal() # High while an EXI transaction is in progress (from ExiCapture). # SPRAM reads stream until this deasserts → supports variable-length # (DMA) bulk reads, not just ≤4-byte immediate transfers. self.cs_active = Signal() # ── Interrupt (exi domain) ──────────────────────────────────────── self.exi_int_n = Signal(init=1) # ── PAR output (for forwarding to W5500 as source MAC) ─────────── self.par = Signal(48) # PAR0-5 packed: PAR0 in low byte par[0:8] # NCRA[3] = SR (start receive) bit — gates the RX ring-buffer path. self.ncra_sr = Signal() # ── CDC FIFO sync-domain sides (wired by BBATop) ────────────────── # SPRAM request exi→sync: sync reads these self.spram_req_r_data = Signal(16) self.spram_req_r_en = Signal() self.spram_req_r_rdy = Signal() # SPRAM response sync→exi: sync writes these self.spram_rsp_w_data = Signal(8) self.spram_rsp_w_en = Signal() self.spram_rsp_w_rdy = Signal() # TX bytes exi→sync: sync reads these self.tx_bytes_r_data = Signal(8) self.tx_bytes_r_en = Signal() self.tx_bytes_r_rdy = Signal() # TX ctrl (frame length) exi→sync: sync reads these self.tx_ctrl_r_data = Signal(16) self.tx_ctrl_r_en = Signal() self.tx_ctrl_r_rdy = Signal() # RX write-pointer update sync→exi: sync writes these self.rx_wptr_w_data = Signal(8) self.rx_wptr_w_en = Signal() self.rx_wptr_w_rdy = Signal() # RX read-pointer update exi→sync: sync reads these self.rx_rptr_r_data = Signal(8) self.rx_rptr_r_en = Signal() self.rx_rptr_r_rdy = Signal() # PulseSynchronizer ports (exi↔sync) self.ncra_rst_o = Signal() # exi→sync self.rx_irq_i = Signal() # sync→exi self.tx_irq_i = Signal() # sync→exi def elaborate(self, platform): m = Module() # ── CDC FIFOs ──────────────────────────────────────────────────── # Plain SyncFIFOs: `exi` and `sync` were always the SAME clock net # (bba_top wired ClockSignal("sync") to ClockSignal("exi")), so the # AsyncFIFOs between them bought nothing and cost ~3 cycles each way. # On the SPRAM path that latency exceeded the protocol's turnaround # byte and shifted every DMA read by one byte. One slow domain now. spram_req = SyncFIFO(width=16, depth=4) spram_rsp = SyncFIFO(width=8, depth=4) tx_bytes = SyncFIFO(width=8, depth=16) tx_ctrl = SyncFIFO(width=16, depth=4) rx_wptr = SyncFIFO(width=8, depth=4) rx_rptr = SyncFIFO(width=8, depth=4) m.submodules.spram_req = spram_req m.submodules.spram_rsp = spram_rsp m.submodules.tx_bytes = tx_bytes m.submodules.tx_ctrl = tx_ctrl m.submodules.rx_wptr = rx_wptr m.submodules.rx_rptr = rx_rptr # Expose sync-domain FIFO sides m.d.comb += [ self.spram_req_r_data .eq(spram_req.r_data), spram_req.r_en .eq(self.spram_req_r_en), self.spram_req_r_rdy .eq(spram_req.r_rdy), spram_rsp.w_data .eq(self.spram_rsp_w_data), spram_rsp.w_en .eq(self.spram_rsp_w_en), self.spram_rsp_w_rdy .eq(spram_rsp.w_rdy), self.tx_bytes_r_data .eq(tx_bytes.r_data), tx_bytes.r_en .eq(self.tx_bytes_r_en), self.tx_bytes_r_rdy .eq(tx_bytes.r_rdy), self.tx_ctrl_r_data .eq(tx_ctrl.r_data), tx_ctrl.r_en .eq(self.tx_ctrl_r_en), self.tx_ctrl_r_rdy .eq(tx_ctrl.r_rdy), rx_wptr.w_data .eq(self.rx_wptr_w_data), rx_wptr.w_en .eq(self.rx_wptr_w_en), self.rx_wptr_w_rdy .eq(rx_wptr.w_rdy), self.rx_rptr_r_data .eq(rx_rptr.r_data), rx_rptr.r_en .eq(self.rx_rptr_r_en), self.rx_rptr_r_rdy .eq(rx_rptr.r_rdy), ] # ── PulseSynchronizers ─────────────────────────────────────────── # Single slow domain, so these are plain wires rather than # PulseSynchronizers. A pulse generated here is already in the only # domain that exists. ncra_rst_pulse = Signal() m.d.comb += self.ncra_rst_o.eq(ncra_rst_pulse) # ── Register file (sparse individual Signals, exi domain) ──────── # Only the registers actually read/written by the GC or sync domain. # Writes to unknown addresses are silently ignored; reads return 0. r_ncra = Signal(8) r_imr = Signal(8) r_ir = Signal(8) r_rwp_lo = Signal(8) r_rrp_lo = Signal(8) # PAR0–5 reset to a valid Nintendo OUI MAC (00:09:BF:00:00:01) so the # device has a sane source MAC even before the GC driver programs its # own. PAR0 is the first MAC octet. _par_reset = [0x00, 0x09, 0xBF, 0x00, 0x00, 0x01] r_par = Array([Signal(8, name=f"par{i}", init=_par_reset[i]) for i in range(6)]) r_twd_lo = Signal(8) r_twd_hi = Signal(8) # PAR packed output: PAR0 in the LOW byte (par[0:8]). The W5500 master # reads mac_shadow[i] = par[i*8:(i+1)*8], so this puts PAR0 first in the # SHAR write — i.e. PAR0 is the first MAC octet on the wire. m.d.comb += self.par.eq(Cat( r_par[0], r_par[1], r_par[2], r_par[3], r_par[4], r_par[5], )) m.d.comb += self.ncra_sr.eq(r_ncra[3]) # start-receive bit # ── Transaction state ──────────────────────────────────────────── # Real GC<->BBA header (see docs/exi_protocol_real.md), verified # against libogc and Swiss. Byte 0 bit7 selects the address space and # bit6 is the write flag; there is NO length field -- the data phase # runs until CS. # command (bit7=0): [reg&0x3F, 00, 00, data] one 4-byte transfer # register (bit7=1): [0x80|w<<6, ah, al, 00] + data phase until CS addr = Signal(16) is_write = Signal() cmd_space = Signal() # this transaction is a command access cmd_reg = Signal(6) byte_ctr = Signal(3) # response-byte index (device-ID sequence) cmd_wr = Signal() # command-space write pending prefetched = Signal() # first response byte already pushed cmd_wdata = Signal(8) # EXI-side command registers. Swiss uses exactly two: 0x02 is the EXI # interrupt mask and 0x03 the EXI interrupt status (write-1-to-clear). r_cmd_imr = Signal(8) r_cmd_ir = Signal(8) tx_frame_len = Signal(16) # TX trigger bookkeeping. `twd_armed` fires one cycle AFTER TWD_HI is # written, so tx_frame_len has both halves by then; `tx_queued` records # that the drain already has a length for this frame. twd_armed = Signal() tx_queued = Signal() # True until first NCRA reset write: return device ID on addr=0 reads id_phase = Signal(init=1) # Per-byte SPRAM read handshake (register-read path): sp_req marks a # request in flight; drain_ctr counts the read-phase dummy bytes. sp_req = Signal() drain_ctr = Signal(2) # SPRAM streaming-read state (DMA / variable-length reads): # sp_addr — next SPRAM byte address to request (auto-increments) # outstanding — SPRAM requests issued but whose responses are not yet # popped (bounds prefetch and is drained at end) sp_addr = Signal(16) outstanding = Signal(4) SP_LIMIT = 4 # max prefetch depth in flight # Effective address of the current data byte — a REGISTERED running # pointer (set to the base in HEADER1, incremented per byte). Keeping # it registered keeps the 13-bit adder off the combinational path that # feeds the read-response mux → tx_fifo write data. eff_addr = Signal(16) rd_sel = eff_addr[0:8] # ── Combinational read-response value (non-SPRAM) ──────────────── reg_rdval = Signal(8) with m.Switch(rd_sel): with m.Case(_NCRA): m.d.comb += reg_rdval.eq(r_ncra) with m.Case(_IMR): m.d.comb += reg_rdval.eq(r_imr) with m.Case(_IR): m.d.comb += reg_rdval.eq(r_ir) with m.Case(_RWP_LO): m.d.comb += reg_rdval.eq(r_rwp_lo) with m.Case(_RRP_LO): m.d.comb += reg_rdval.eq(r_rrp_lo) with m.Case(_PAR0, _PAR1, _PAR2, _PAR3, _PAR4, _PAR5): m.d.comb += reg_rdval.eq(r_par[eff_addr[0:3]]) with m.Case(_TWD_LO): m.d.comb += reg_rdval.eq(r_twd_lo) with m.Case(_TWD_HI): m.d.comb += reg_rdval.eq(r_twd_hi) with m.Case(_NWAYS): m.d.comb += reg_rdval.eq(_NWAYS_VAL) with m.Case(_HIPR): m.d.comb += reg_rdval.eq(_HIPR_VAL) with m.Default(): m.d.comb += reg_rdval.eq(0) # Device-ID bytes (addr 0 read while id_phase): 0x04 0x02 0x02 0x00 devid = Signal(8) with m.Switch(byte_ctr): with m.Case(0): m.d.comb += devid.eq(0x04) with m.Case(1): m.d.comb += devid.eq(0x02) with m.Case(2): m.d.comb += devid.eq(0x02) with m.Default(): m.d.comb += devid.eq(0x00) cmd_rdval = Signal(8) with m.Switch(cmd_reg): with m.Case(0x02): m.d.comb += cmd_rdval.eq(r_cmd_imr) with m.Case(0x03): m.d.comb += cmd_rdval.eq(r_cmd_ir) with m.Default(): m.d.comb += cmd_rdval.eq(0) rd_val = Signal(8) # response for the current non-SPRAM read byte with m.If(cmd_space): m.d.comb += rd_val.eq(cmd_rdval) with m.Else(): m.d.comb += rd_val.eq(reg_rdval) # ── Default strobes ────────────────────────────────────────────── m.d.sync += [ tx_bytes.w_en .eq(0), tx_ctrl.w_en .eq(0), rx_rptr.w_en .eq(0), rx_wptr.r_en .eq(0), ncra_rst_pulse .eq(0), ] # The SPRAM request write port is COMBINATIONAL (see RHDR_LO): a # registered w_en costs a 24 MHz cycle the turnaround byte cannot # spare. Default it off here and let the states assert it. m.d.comb += [ spram_req.w_en .eq(0), spram_req.w_data.eq(0), self.rx_en .eq(0), self.tx_en .eq(0), self.tx_data.eq(0), # Combinational so the FIFO advances in the SAME cycle as the pop — # a registered r_en would let `pop` re-fire on the same byte. spram_rsp.r_en.eq(0), ] # ── Transaction FSM (proactive push/pull over byte FIFOs) ──────── # The SPI bit cadence lives in the capture domain; here we just consume # received bytes and, for reads, push response bytes into tx_fifo during # the EXI clock-idle gap before the GC clocks the data phase. with m.FSM(domain="sync", name="exi_fsm"): with m.State("HEADER0"): # Byte 0 selects the space and direction. The REAL protocol # (docs/exi_protocol_real.md) has no length field: the GC # clocks as many data bytes as it wants and ends with CS. with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) m.d.sync += [is_write.eq(self.rx_data[6]), prefetched.eq(0), byte_ctr.eq(0), sp_req.eq(0), outstanding.eq(0)] with m.If(self.rx_data[7]): # register space m.d.sync += cmd_space.eq(0) m.next = "RHDR_HI" with m.Else(): # command space m.d.sync += [cmd_space.eq(1), cmd_reg.eq(self.rx_data[0:6]), addr.eq(self.rx_data[0:6]), eff_addr.eq(self.rx_data[0:6])] # EXI_GetID (byte 0 == 0x00) gives only ONE header # byte of warning before the GC starts clocking the # reply, so push the first ID byte the instant byte 0 # identifies the query -- the earliest point possible. with m.If((self.rx_data == 0x00) & id_phase & self.tx_rdy): m.d.comb += [self.tx_data.eq(0x04), self.tx_en.eq(1)] m.d.sync += byte_ctr.eq(1) m.next = "CHDR_1" # ── register space: [0x80|w<<6, ah, al, 0x00] ────────────────── with m.State("RHDR_HI"): with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) m.d.sync += addr[8:16].eq(self.rx_data) m.next = "RHDR_LO" with m.State("RHDR_LO"): # The address completes HERE, one whole byte before the data # phase. That trailing 0x00 byte is the device's prefetch # window -- the old 2-byte-plus-length format spent those bits # on the length field and left zero slack, which is why the # first on-console read came back 22 bit-times late. Kick the # fetch off now so the answer is ready for the first data slot. with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) full = Cat(self.rx_data, addr[8:16]) # 16-bit address m.d.sync += [addr.eq(full), eff_addr.eq(full), sp_addr.eq(full)] with m.If(~is_write & (full >= 0x100) & spram_req.w_rdy): # COMBINATIONAL write: the turnaround byte gives only # ~296 ns and the SPRAM round trip needs most of it, so # a registered w_en (one more 24 MHz cycle, ~42 ns) is # enough to miss the window and shift the whole DMA # read by a byte. Same-domain SyncFIFO now, so this is # legal. m.d.comb += [spram_req.w_data.eq(full), spram_req.w_en.eq(1)] m.d.sync += [sp_addr.eq(full + 1), outstanding.eq(1)] m.next = "RHDR_PAD" with m.State("RHDR_PAD"): # Trailing header byte. For a register read, push the first # response byte during this byte-time so it has crossed the # tx CDC before the GC clocks data slot 0. with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) with m.If(is_write): m.next = "WRITE" with m.Elif(addr >= 0x100): m.next = "SPRAM_STREAM" with m.Else(): m.next = "REG_READ" with m.Elif(~is_write & (addr < 0x100) & self.tx_rdy & ~prefetched): # Prefetch the FIRST response byte (once!) while the pad # byte is still on the wire, so it has crossed the tx CDC # before the GC clocks data slot 0. Must be one-shot: this # branch is re-evaluated every cycle the pad byte has not # arrived, and repeating it would consume the register # stream early (symptom: reads shifted by a byte). m.d.comb += [self.tx_data.eq(rd_val), self.tx_en.eq(1)] m.d.sync += [eff_addr.eq(eff_addr + 1), prefetched.eq(1)] # ── command space: [reg, 00, 00, data] in ONE 4-byte transfer ── with m.State("CHDR_1"): with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) # EXI_GetID is its own shape: the GC writes 0x0000 (two # bytes, so it lands here as command reg 0) and then clocks # FOUR read bytes. Hand it the ID stream rather than the # single-byte command response. with m.If(~is_write & (cmd_reg == 0) & id_phase): # byte_ctr is NOT reset here: HEADER0 already pushed # ID byte 0 and advanced it. Resetting would restart # the sequence and duplicate the first byte. m.next = "DEVID" with m.Else(): m.next = "CHDR_2" with m.Elif(~is_write & (cmd_reg == 0) & id_phase & self.tx_rdy): # Device-ID query: keep feeding the ID while byte 1 is # still on the wire so the tx FIFO stays ahead of the GC. # NOT one-shot -- the FIFO is only 2 deep, and stopping # after one byte leaves a hole in the next slot. m.d.comb += [self.tx_data.eq(devid), self.tx_en.eq(1)] m.d.sync += byte_ctr.eq(byte_ctr + 1) with m.Elif(~is_write & self.tx_rdy & ~prefetched): # Ordinary command read: response lands in byte slot 3. m.d.comb += [self.tx_data.eq(rd_val), self.tx_en.eq(1)] m.d.sync += prefetched.eq(1) with m.State("CHDR_2"): with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) m.next = "CHDR_3" with m.State("CHDR_3"): with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) with m.If(is_write): m.d.sync += cmd_wr.eq(1) # applied below m.d.sync += cmd_wdata.eq(self.rx_data) m.next = "TRAILER" with m.State("TRAILER"): # Soak up anything else the GC clocks, then wait for CS to rise. with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) with m.If(~self.cs_active): m.next = "HEADER0" with m.State("WRITE"): # Consume xfer_len+1 data bytes, writing the register file. with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) with m.Switch(rd_sel): with m.Case(_NCRA): m.d.sync += r_ncra.eq(self.rx_data) with m.If(self.rx_data[0]): m.d.sync += r_ncra[0].eq(0) # RESET self-clears m.d.sync += ncra_rst_pulse.eq(1) m.d.sync += id_phase.eq(0) # Fallback trigger: TWD_HI normally queues the # frame (see _TWD_HI). Only fire here if it did # not, so software that skips TWD still transmits # and we never double-queue one frame. with m.If(self.rx_data[1:3].any() & ~tx_queued): with m.If(tx_ctrl.w_rdy): m.d.sync += tx_ctrl.w_data.eq(tx_frame_len) m.d.sync += tx_ctrl.w_en.eq(1) m.d.sync += tx_queued.eq(1) with m.Case(_IMR): m.d.sync += r_imr.eq(self.rx_data) with m.Case(_IR): m.d.sync += r_ir.eq(r_ir & ~self.rx_data) # write-1-clear with m.Case(_RRP_LO): m.d.sync += r_rrp_lo.eq(self.rx_data) with m.If(rx_rptr.w_rdy): m.d.sync += rx_rptr.w_data.eq(self.rx_data) m.d.sync += rx_rptr.w_en.eq(1) with m.Case(_PAR0, _PAR1, _PAR2, _PAR3, _PAR4, _PAR5): m.d.sync += r_par[eff_addr[0:3]].eq(self.rx_data) with m.Case(_TWD_LO): m.d.sync += r_twd_lo.eq(self.rx_data) m.d.sync += tx_frame_len[0:8].eq(self.rx_data) with m.Case(_TWD_HI): m.d.sync += r_twd_hi.eq(self.rx_data) m.d.sync += tx_frame_len[8:16].eq(self.rx_data) # Hand the length to TXFrameDrain NOW, not at the ST # trigger. The drain blocks on tx_bytes_r_rdy, so # starting it here lets it consume bytes AS the GC # streams them and keeps tx_bytes a rate-matching # buffer. Waiting for ST meant the whole frame had # to fit in the 16-deep FIFO, and anything past 16 # bytes was silently dropped (hardware 2026-09-30: # frames truncated at exactly 16 bytes on the wire). m.d.sync += twd_armed.eq(1) with m.Case(_TXDATA): with m.If(tx_bytes.w_rdy): m.d.sync += tx_bytes.w_data.eq(self.rx_data) m.d.sync += tx_bytes.w_en.eq(1) # All other addresses silently ignored # No length field in the real protocol: consume data # bytes until CS rises. m.d.sync += eff_addr.eq(eff_addr + 1) with m.If(~self.cs_active): m.next = "HEADER0" with m.State("REG_READ"): # Register read (addr < 0x100): stream until CS, like the SPRAM # path. The real protocol has no length field. with m.If(self.tx_rdy & self.cs_active): m.d.comb += [self.tx_data.eq(rd_val), self.tx_en.eq(1)] m.d.sync += eff_addr.eq(eff_addr + 1) with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) # drain dummies with m.If(~self.cs_active): m.next = "HEADER0" with m.State("DEVID"): # Stream 04 02 02 00 until CS rises. with m.If(self.tx_rdy & self.cs_active): m.d.comb += [self.tx_data.eq(devid), self.tx_en.eq(1)] m.d.sync += byte_ctr.eq(byte_ctr + 1) with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) with m.If(~self.cs_active): m.next = "HEADER0" with m.State("SPRAM_STREAM"): # Stream SPRAM bytes until CS deasserts — handles both ≤4-byte # immediate reads and arbitrary-length DMA reads uniformly. # Issue read requests ahead (prefetch, bounded by SP_LIMIT) and # push responses into tx_fifo; the capture domain pops them as # the GC clocks. Drain rx dummies as they arrive. issue = Signal() pop = Signal() m.d.comb += issue.eq(self.cs_active & spram_req.w_rdy & (outstanding < SP_LIMIT)) m.d.comb += pop.eq(spram_rsp.r_rdy & self.tx_rdy) with m.If(issue): m.d.comb += spram_req.w_data.eq(sp_addr) m.d.comb += spram_req.w_en.eq(1) m.d.sync += sp_addr.eq(sp_addr + 1) with m.If(pop): m.d.comb += self.tx_data.eq(spram_rsp.r_data) m.d.comb += self.tx_en.eq(1) m.d.comb += spram_rsp.r_en.eq(1) m.d.sync += outstanding.eq(outstanding + issue - pop) with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) # drain dummy bytes with m.If(~self.cs_active): m.next = "SPRAM_END" with m.State("SPRAM_END"): # CS deasserted: drain in-flight SPRAM responses and rx dummies, # then idle. Leftover prefetch in tx_fifo is flushed by # ExiCapture on the next CS assertion. with m.If(spram_rsp.r_rdy): m.d.comb += spram_rsp.r_en.eq(1) m.d.sync += outstanding.eq(outstanding - 1) with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) with m.If((outstanding == 0) & ~self.rx_rdy & ~spram_rsp.r_rdy): m.next = "HEADER0" # ── Command-space write, applied the cycle after CHDR_3 ────────── with m.If(cmd_wr): m.d.sync += cmd_wr.eq(0) with m.Switch(cmd_reg): with m.Case(0x02): m.d.sync += r_cmd_imr.eq(cmd_wdata) with m.Case(0x03): m.d.sync += r_cmd_ir.eq(r_cmd_ir & ~cmd_wdata) # ── Interrupt output ───────────────────────────────────────────── m.d.sync += self.exi_int_n.eq(~(r_ir & r_imr).any()) # ── Consume RWP updates from sync domain ────────────────────────── with m.If(rx_wptr.r_rdy): m.d.sync += rx_wptr.r_en.eq(1) m.d.sync += r_rwp_lo.eq(rx_wptr.r_data) # ── PulseSynchronizer arrivals ──────────────────────────────────── with m.If(self.rx_irq_i): m.d.sync += r_ir[1].eq(1) # RI bit # ── Deferred TX queue (armed by a TWD_HI write) ────────────────── # One cycle after TWD_HI so tx_frame_len holds both bytes. Held until # tx_ctrl has room so a full control FIFO cannot drop the trigger. with m.If(twd_armed & tx_ctrl.w_rdy): m.d.sync += tx_ctrl.w_data.eq(tx_frame_len) m.d.sync += tx_ctrl.w_en.eq(1) m.d.sync += twd_armed.eq(0) m.d.sync += tx_queued.eq(1) with m.If(self.tx_irq_i): m.d.sync += r_ir[2].eq(1) # TI bit m.d.sync += r_ncra[1:3].eq(0) # clear ST bits m.d.sync += tx_queued.eq(0) # ready for the next frame return m # ── Testbench ───────────────────────────────────────────────────────────── if __name__ == "__main__": import sys, os sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) from amaranth.sim import Simulator, Period reg = BBARegisterFile() # Drive the byte-stream interface directly (the SPI bit cadence and FIFOs # live in ExiCapture; here we model the byte producer/consumer). # Responses can be pushed WHILE the header is still being clocked -- the # device prefetches during the header's turnaround byte, and the device-ID # reply starts from header byte 0. So every helper that advances time has # to record tx pushes, or the early ones are silently lost and reads look # shifted. tx_log is that shared record; collect_tx drains it first. tx_log = [] def sample_tx(ctx): if ctx.get(reg.tx_en): tx_log.append(ctx.get(reg.tx_data)) async def push_rx(ctx, b): """Present one received byte and wait for the register file to pop it.""" ctx.set(reg.rx_data, b) ctx.set(reg.rx_rdy, 1) while True: en = ctx.get(reg.rx_en) sample_tx(ctx) await ctx.tick("sync") if en: break ctx.set(reg.rx_rdy, 0) async def collect_tx(ctx, n): """Collect n response bytes, including any pushed during the header.""" out = [] while tx_log and len(out) < n: out.append(tx_log.pop(0)) for _ in range(3000): if len(out) >= n: break if ctx.get(reg.tx_en): out.append(ctx.get(reg.tx_data)) await ctx.tick("sync") return out # Real protocol (docs/exi_protocol_real.md): register space is a 4-byte # header [0x80|w<<6, ah, al, 0x00] followed by a data phase that runs # until CS rises -- there is no length field. async def exi_read(ctx, addr, length=1): tx_log.clear() ctx.set(reg.cs_active, 1) await push_rx(ctx, 0x80) await push_rx(ctx, (addr >> 8) & 0xFF) await push_rx(ctx, addr & 0xFF) await push_rx(ctx, 0x00) # turnaround / prefetch window result = await collect_tx(ctx, length) ctx.set(reg.cs_active, 0) await ctx.tick("sync").repeat(6) return result async def exi_write(ctx, addr, data): ctx.set(reg.cs_active, 1) await push_rx(ctx, 0xC0) await push_rx(ctx, (addr >> 8) & 0xFF) await push_rx(ctx, addr & 0xFF) await push_rx(ctx, 0x00) for b in data: await push_rx(ctx, b) ctx.set(reg.cs_active, 0) await ctx.tick("sync").repeat(6) async def exi_devid(ctx): """EXI_GetID: write 0x0000 (2 bytes), then clock 4 read bytes.""" tx_log.clear() ctx.set(reg.cs_active, 1) await push_rx(ctx, 0x00) # command space, reg 0, read await push_rx(ctx, 0x00) result = await collect_tx(ctx, 4) ctx.set(reg.cs_active, 0) await ctx.tick("sync").repeat(6) return result # SPRAM contents the streaming-read test reads back (byte i = 0xA0+i). spram_mem = {0x100 + i: (0xA0 + i) & 0xFF for i in range(64)} async def spram_model(ctx): """Model the SPRAM (sync side): answer spram_req with mem[addr]. One request at a time, with cleanly-pulsed r_en/w_en so the FIFO pop and the response push stay in lock-step (no double-response races). """ state = "POP" held = 0 async for vals in ctx.tick("sync").sample( reg.spram_req_r_rdy, reg.spram_req_r_data, reg.spram_rsp_w_rdy): rdy, addr, rsp_rdy = vals[-3:] ctx.set(reg.spram_req_r_en, 0) ctx.set(reg.spram_rsp_w_en, 0) if state == "POP": if rdy: held = spram_mem.get(addr, 0) ctx.set(reg.spram_req_r_en, 1) # consume the request state = "RESP" else: # RESP if rsp_rdy: ctx.set(reg.spram_rsp_w_data, held) ctx.set(reg.spram_rsp_w_en, 1) # deliver the response state = "POP" errors = [] async def testbench(ctx): ctx.set(reg.tx_rdy, 1) # tx_fifo always has room in this model await ctx.tick("sync").repeat(8) # T1: Device ID (addr=0, 4-byte read) result = await exi_devid(ctx) if result != _DEVICE_ID: errors.append(f"T1 device ID: expected {_DEVICE_ID}, got {result}") print(f"T1 device ID: {[f'0x{b:02X}' for b in result]}") await ctx.tick("sync").repeat(4) # T2: Write and read back PAR0-PAR3 await exi_write(ctx, _PAR0, [0xDE, 0xAD, 0xBE, 0xEF]) await ctx.tick("sync").repeat(4) result = await exi_read(ctx, _PAR0, length=4) if result != [0xDE, 0xAD, 0xBE, 0xEF]: errors.append(f"T2 PAR readback: {result}") print(f"T2 PAR0-3: {[f'0x{b:02X}' for b in result]}") await ctx.tick("sync").repeat(4) # T3: NWAYS hardcoded 0x17 result = await exi_read(ctx, _NWAYS, length=1) if result != [0x17]: errors.append(f"T3 NWAYS: expected 0x17, got {result}") print(f"T3 NWAYS: 0x{result[0]:02X}") await ctx.tick("sync").repeat(4) # T4: HIPR hardcoded 0x01 result = await exi_read(ctx, _HIPR, length=1) if result != [0x01]: errors.append(f"T4 HIPR: expected 0x01, got {result}") print(f"T4 HIPR: 0x{result[0]:02X}") await ctx.tick("sync").repeat(4) # T5: IMR write, rx_irq pulse, INT_N asserts, then IR clear await exi_write(ctx, _IMR, [0x02]) # enable RI (bit 1) await ctx.tick("sync").repeat(4) ctx.set(reg.rx_irq_i, 1) await ctx.tick("sync").repeat(1) ctx.set(reg.rx_irq_i, 0) await ctx.tick("sync").repeat(12) # wait for PS propagation int_n = ctx.get(reg.exi_int_n) if int_n != 0: errors.append(f"T5 INT_N after RI: expected 0, got {int_n}") print(f"T5 INT_N after RI pulse: {int_n} (want 0)") await exi_write(ctx, _IR, [0x02]) # write-1-to-clear RI await ctx.tick("sync").repeat(4) int_n = ctx.get(reg.exi_int_n) if int_n != 1: errors.append(f"T5 INT_N after clear: expected 1, got {int_n}") print(f"T5 INT_N after IR clear: {int_n} (want 1)") # T6: streaming SPRAM read (DMA) — the real protocol has no length # field, so the data phase simply runs for as long as CS is held. # Swiss reads the packet buffer exactly this way (bba_ins -> DMA). N = 12 tx_log.clear() ctx.set(reg.cs_active, 1) await push_rx(ctx, 0x80) # register space, read await push_rx(ctx, 0x01) # addr[15:8] -> 0x0100 await push_rx(ctx, 0x00) # addr[7:0] await push_rx(ctx, 0x00) # turnaround byte got = [] for _ in range(5000): if ctx.get(reg.tx_en): got.append(ctx.get(reg.tx_data)) if len(got) >= N: break await ctx.tick("sync") ctx.set(reg.cs_active, 0) # end the transaction await ctx.tick("sync").repeat(40) # let SPRAM_END drain/clean up want = [spram_mem[0x100 + i] for i in range(N)] print(f"T6 DMA read {N}B: {[f'0x{b:02X}' for b in got]}") if got != want: errors.append(f"T6 streaming SPRAM read: got {got}, want {want}") # T7: a normal register read still works after the streaming transaction # (FSM cleaned up and returned to HEADER0) result = await exi_read(ctx, _NWAYS, length=1) if result != [0x17]: errors.append(f"T7 NWAYS after DMA: got {result}") print(f"T7 NWAYS after DMA read: 0x{result[0]:02X}") sim = Simulator(reg) sim.add_clock(Period(MHz=24), domain="sync") sim.add_testbench(testbench) sim.add_process(spram_model) sim.run() if errors: print("\nFAILURES:") for e in errors: print(" ", e) sys.exit(1) else: print("\nAll tests passed.")