Roflin/rebbarb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2458884160f1a925633b9bc76dc937073d0db4b6
rebbarb/exi_bba/bba_register_file.py
T
Roflin 2458884160 Rework the EXI decoder to the real GC<->BBA protocol 
Replaces the invented 2-byte header (addr split across both bytes, 2-bit
length field) with the protocol libogc and Swiss actually speak:

  command  (bit7=0, 6-bit reg)   [reg, 00, 00, data]  one 4-byte transfer
  register (bit7=1, 16-bit reg)  [0x80|w<<6, ah, al, 00] + data until CS

This was not an optimisation: the old decoder read Swiss's NWAYS read
(80 00 31 00) as 'write 0x31 to NCRA', and bit 0 of 0x31 is RESET, so every
register read would have reset the device.

Changes:
  * 16-bit address; no length field -- register reads and writes now stream
    until CS, like the SPRAM path already did
  * RHDR_LO starts the SPRAM fetch as soon as the address completes, using
    the trailing header byte as the prefetch window the protocol provides
  * command space serves EXI regs 0x02/0x03 (interrupt mask/status), the only
    two Swiss uses
  * EXI_GetID gets its own path: it offers just ONE header byte of warning, so
    the first ID byte is pushed the instant byte 0 identifies the query

Two timing subtleties, both found by simulation and commented in place:
  * MISO is driven LIVE from the tx FIFO head, so a byte pushed while the pad
    byte is on the wire appears in the pad slot.  A filler byte is parked
    there (MISO is don't-care through the header) so the first real response
    lands in data slot 0.
  * the prefetch must be one-shot; re-firing while waiting for the pad byte
    consumed the register stream early and shifted reads by a byte.

All testbenches updated to the real format and passing: bba_register_file
(7), bba_top (8 incl. the full W5100->SPRAM->GC loop and TX), exi_j4,
exi_capture, spi_mode3_slave, tx_frame_drain, spram_arbiter,
rx_frame_assembler, w5100_parallel_master.  gc_bba_test rebuilt against the
new protocol.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-10-03 11:58:52 +02:00

810 lines
37 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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.cdc import PulseSynchronizer
from amaranth.lib.fifo import AsyncFIFO
__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 ────────────────────────────────────────────────────
spram_req = AsyncFIFO(width=16, depth=4, w_domain="exi", r_domain="sync")
spram_rsp = AsyncFIFO(width=8, depth=4, w_domain="sync", r_domain="exi")
tx_bytes = AsyncFIFO(width=8, depth=16, w_domain="exi", r_domain="sync")
tx_ctrl = AsyncFIFO(width=16, depth=4, w_domain="exi", r_domain="sync")
rx_wptr = AsyncFIFO(width=8, depth=4, w_domain="sync", r_domain="exi")
rx_rptr = AsyncFIFO(width=8, depth=4, w_domain="exi", r_domain="sync")
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 ───────────────────────────────────────────
ncra_rst_ps = PulseSynchronizer(i_domain="exi", o_domain="sync")
rx_irq_ps = PulseSynchronizer(i_domain="sync", o_domain="exi")
tx_irq_ps = PulseSynchronizer(i_domain="sync", o_domain="exi")
m.submodules.ncra_rst_ps = ncra_rst_ps
m.submodules.rx_irq_ps = rx_irq_ps
m.submodules.tx_irq_ps = tx_irq_ps
m.d.comb += [
self.ncra_rst_o .eq(ncra_rst_ps.o),
rx_irq_ps.i .eq(self.rx_irq_i),
tx_irq_ps.i .eq(self.tx_irq_i),
]
# ── 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.exi += [
spram_req.w_en .eq(0),
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_ps.i .eq(0),
]
m.d.comb += [
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="exi", 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.exi += [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.exi += cmd_space.eq(0)
m.next = "RHDR_HI"
with m.Else(): # command space
m.d.exi += [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.exi += 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.exi += 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.exi += [addr.eq(full), eff_addr.eq(full),
sp_addr.eq(full)]
with m.If(~is_write & (full >= 0x100) & spram_req.w_rdy):
m.d.exi += [spram_req.w_data.eq(full),
spram_req.w_en.eq(1),
sp_addr.eq(full + 1),
outstanding.eq(1)]
# MISO is driven LIVE from the tx FIFO head, so a byte
# pushed now would show up during the pad slot rather than
# the first data slot. Park a filler byte there -- MISO is
# don't-care for the whole header -- so the real first
# response byte lands one slot later, where it belongs.
with m.If(~is_write & self.tx_rdy):
m.d.comb += [self.tx_data.eq(0x00), self.tx_en.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.exi += [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.exi += 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.exi += 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.exi += cmd_wr.eq(1) # applied below
m.d.exi += 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.exi += r_ncra.eq(self.rx_data)
with m.If(self.rx_data[0]):
m.d.exi += r_ncra[0].eq(0) # RESET self-clears
m.d.exi += ncra_rst_ps.i.eq(1)
m.d.exi += 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.exi += tx_ctrl.w_data.eq(tx_frame_len)
m.d.exi += tx_ctrl.w_en.eq(1)
m.d.exi += tx_queued.eq(1)
with m.Case(_IMR):
m.d.exi += r_imr.eq(self.rx_data)
with m.Case(_IR):
m.d.exi += r_ir.eq(r_ir & ~self.rx_data) # write-1-clear
with m.Case(_RRP_LO):
m.d.exi += r_rrp_lo.eq(self.rx_data)
with m.If(rx_rptr.w_rdy):
m.d.exi += rx_rptr.w_data.eq(self.rx_data)
m.d.exi += rx_rptr.w_en.eq(1)
with m.Case(_PAR0, _PAR1, _PAR2, _PAR3, _PAR4, _PAR5):
m.d.exi += r_par[eff_addr[0:3]].eq(self.rx_data)
with m.Case(_TWD_LO):
m.d.exi += r_twd_lo.eq(self.rx_data)
m.d.exi += tx_frame_len[0:8].eq(self.rx_data)
with m.Case(_TWD_HI):
m.d.exi += r_twd_hi.eq(self.rx_data)
m.d.exi += 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.exi += twd_armed.eq(1)
with m.Case(_TXDATA):
with m.If(tx_bytes.w_rdy):
m.d.exi += tx_bytes.w_data.eq(self.rx_data)
m.d.exi += 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.exi += 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.exi += 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.exi += 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.exi += spram_req.w_data.eq(sp_addr)
m.d.exi += spram_req.w_en.eq(1)
m.d.exi += 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.exi += 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.exi += 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.exi += cmd_wr.eq(0)
with m.Switch(cmd_reg):
with m.Case(0x02): m.d.exi += r_cmd_imr.eq(cmd_wdata)
with m.Case(0x03): m.d.exi += r_cmd_ir.eq(r_cmd_ir & ~cmd_wdata)
# ── Interrupt output ─────────────────────────────────────────────
m.d.exi += 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.exi += rx_wptr.r_en.eq(1)
m.d.exi += r_rwp_lo.eq(rx_wptr.r_data)
# ── PulseSynchronizer arrivals ────────────────────────────────────
with m.If(rx_irq_ps.o):
m.d.exi += 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.exi += tx_ctrl.w_data.eq(tx_frame_len)
m.d.exi += tx_ctrl.w_en.eq(1)
m.d.exi += twd_armed.eq(0)
m.d.exi += tx_queued.eq(1)
with m.If(tx_irq_ps.o):
m.d.exi += r_ir[2].eq(1) # TI bit
m.d.exi += r_ncra[1:3].eq(0) # clear ST bits
m.d.exi += 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("exi")
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("exi")
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
# The device parks a filler byte in the pad slot (MISO is don't-care
# for the whole header) so the first REAL response byte lands in data
# slot 0. This bench counts pushes rather than wire slots, so drop it.
result = (await collect_tx(ctx, length + 1))[1:]
ctx.set(reg.cs_active, 0)
await ctx.tick("exi").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("exi").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("exi").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("exi").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("exi").repeat(4)
# T2: Write and read back PAR0-PAR3
await exi_write(ctx, _PAR0, [0xDE, 0xAD, 0xBE, 0xEF])
await ctx.tick("exi").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("exi").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("exi").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("exi").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("exi").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("exi").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("exi").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("exi")
ctx.set(reg.cs_active, 0) # end the transaction
await ctx.tick("exi").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="exi")
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.")
Reference in New Issue View Git Blame Copy Permalink