Roflin/rebbarb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
a12e4af68bf1cd0b3c1a66359a9f07281013a781
rebbarb/exi_bba/w5100_parallel_master.py
T
Roflin c16afb6eea Add integrated UART UDP bring-up shell (socket 3) + capture critical-path fix 
An interactive UART command shell drives a UDP send/receive test on the
W5100's socket 3, running ALONGSIDE the live BBA (socket-0 MACRAW) with
EXI keeping bus priority. MACRAW+UDP coexistence is W5100S-datasheet
confirmed (S4.6 + "4 independent SOCKETs"). Now the default flash build
(--console selects the old event-log console).

New / changed gateware:
- uart_shell.py (new): rebbarb> shell over FT2232H channel B. Commands:
  help; udp unicast <ip> [msg]; udp broadcast [msg]. After each send it
  waits (bounded, else "timeout") for a reply and prints "rx <payload>".
  Line buffer + message ROM live in block RAM with a sequential parser
  (LC-efficient); 1-deep RX holding reg keeps pastes intact. 8 sim tests.
- w5100_parallel_master.py: configurable UDP socket (default 3) with UDP
  send AND receive (IP-stack init, runtime dest IP, WIZnet UDP RX header
  + payload). Gated by enable_udp_test so the MACRAW path is unchanged
  when off. Tests U1-U4 + MACRAW T1-T5.
- exi_capture.py: CAPTURE-DOMAIN CRITICAL-PATH FIX. The TX byte-FIFO
  read-enable was gated by its own gray-coded ready
  (r_en = ... | (flushing & r_rdy)), forming a consume_ptr -> gray ->
  r_rdy -> flush -> r_en -> consume_ptr loop that capped capture_clk.
  Replaced the r_rdy-based "drain until empty" flush with a fixed-length
  drain counter (FIFO is only tx_depth deep), removing the pointer
  feedback from r_en. Path 24.3 -> 19.6 ns; flush behavior preserved.
- bba_top.py: wire shell <-> W5100 UDP (send + rx); shell additive.
- synth.py: shell default build; env-var UDP network config; documents a
  reverted PNR-timing-priority experiment.

Timing (--seeds 8, default shell build, 67% LC): capture closes on 4/8
seeds (best seed 4 = 58.36 MHz, +8%), clk passes on all. This is BETTER
than the pre-shell 2/8 baseline because the flush fix improved the
capture domain intrinsically. Flash build/seed4/top.bin.

Bring-up caveats (unchanged): W5100 socket register addresses / UDP
header format are datasheet-derived (confirm on hardware); UDP_SRC_IP /
subnet / gateway must match the LAN for unicast ARP.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-02 14:36:47 +00:00

1355 lines
62 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.
"""W5100 parallel-bus master — sync domain.
A drop-in alternative to `W5500SPIMaster` that talks to a WIZnet **W5100** over
its **indirect parallel bus** instead of SPI. The external streaming interface
(init_req/init_done/par, tx_*, rx_*) is identical, so BBATop wiring is unchanged;
only the physical pins differ (a parallel bus instead of 4 SPI wires).
Why parallel
------------
SPI serialises 8 bits per byte, so on this UP5K (whose W5500-operating logic
closes only ~40 MHz) the SPI byte rate caps at ~12 Mbit/s. A parallel bus moves
a whole byte per access, so the same ~24 MHz sync logic clears the 27 Mbit/s EXI
ceiling — the real hard limit — with margin. See CLAUDE.md.
W5100 indirect bus interface (IDM)
----------------------------------
Only two address lines A[1:0] are wired (the upper address lines are tied to 0
on the board, so a power-up *direct*-mode access at A=00 still lands on MR):
A[1:0] register
00 MR (Mode Register — also reachable directly at power-up)
01 IDM_AR0 (indirect address, high byte)
10 IDM_AR1 (indirect address, low byte)
11 IDM_DR (indirect data — accesses mem[IDM_AR]; auto-increments
IDM_AR when MR.AI is set)
So a register/buffer access is: write IDM_AR0/AR1 with the 16-bit address, then
read/write IDM_DR. With MR.AI=1 a multi-byte block is one address-set followed
by a burst of IDM_DR accesses (the chip auto-increments) — used for SHAR and for
streaming frame data.
A bus cycle drives A + (for writes) D with /CS and /RD or /WR asserted for
`strobe_cycles` sync clocks (≥ the W5100's ~80 ns access time at 24 MHz).
Phase status
------------
Phase 1 (this file): bus access engine + transaction engine + init sequence,
verified against a W5100 bus model. TX/RX MACRAW (with socket-buffer ring
wraparound) land in phases 2–3.
"""
from amaranth import *
__all__ = ["W5100ParallelMaster"]
# ── W5100 register addresses (indirect 16-bit address space) ────────────────
_MR = 0x0000 # Mode register (common)
_GAR0 = 0x0001 # Gateway IP, 4 bytes
_SUBR0 = 0x0005 # Subnet mask, 4 bytes
_SHAR0 = 0x0009 # Source MAC, 6 bytes
_SIPR0 = 0x000F # Source IP, 4 bytes
_IR = 0x0015 # Interrupt register
_IMR = 0x0016 # Interrupt mask
_RMSR = 0x001A # RX memory size (2 bits/socket)
_TMSR = 0x001B # TX memory size
_S0_MR = 0x0400 # Socket 0 mode
_S0_CR = 0x0401 # Socket 0 command
_S0_IR = 0x0402 # Socket 0 interrupt
_S0_SR = 0x0403 # Socket 0 status
_S0_TX_FSR = 0x0420 # Socket 0 TX free size (2 bytes)
_S0_TX_RD = 0x0422 # Socket 0 TX read pointer
_S0_TX_WR = 0x0424 # Socket 0 TX write pointer
_S0_RX_RSR = 0x0426 # Socket 0 RX received size (2 bytes)
_S0_RX_RD = 0x0428 # Socket 0 RX read pointer
# Per-socket register/buffer geometry (2 KB per socket, RMSR/TMSR=0x55).
# The UDP-test socket number is configurable; addresses are computed from it in
# __init__ (see _socket_addrs). Socket-n register block base = 0x0400+n*0x100,
# TX buffer base = 0x4000+n*0x800, RX buffer base = 0x6000+n*0x800.
_TX_BASE = 0x4000 # Socket 0 TX buffer base (default 2 KB window)
_RX_BASE = 0x6000 # Socket 0 RX buffer base
_S0_TX_MASK = 0x07FF # 2 KB ring mask
_S0_RX_MASK = 0x07FF
_SN_MASK = 0x07FF # 2 KB ring mask (any socket)
def _socket_addrs(n):
"""Return the register/buffer addresses for socket `n` (0..3)."""
base = 0x0400 + n * 0x0100
return dict(
MR=base + 0x00, CR=base + 0x01, IR=base + 0x02, SR=base + 0x03,
PORT=base + 0x04, DIPR=base + 0x0C, DPORT=base + 0x10,
TX_WR=base + 0x24, RX_RSR=base + 0x26, RX_RD=base + 0x28,
TX_BASE=0x4000 + n * 0x0800, RX_BASE=0x6000 + n * 0x0800,
)
# MR bits / command / mode values
_MR_RST = 0x80
_MR_AI = 0x02 # address auto-increment (indirect mode)
_MR_IND = 0x01 # indirect bus interface mode
_S0_MR_MACRAW = 0x04
_S1_MR_UDP = 0x02 # socket UDP mode
_CR_OPEN = 0x01
_CR_SEND = 0x20
_CR_RECV = 0x40
def _ip_bytes(dotted):
"""'192.168.1.100' → [192, 168, 1, 100] (big-endian, network order)."""
parts = [int(x) for x in dotted.split(".")]
if len(parts) != 4 or any(not 0 <= p <= 255 for p in parts):
raise ValueError(f"invalid IPv4 address: {dotted!r}")
return parts
def _port_bytes(port):
"""6464 → [0x19, 0x40] (big-endian)."""
return [(port >> 8) & 0xFF, port & 0xFF]
# Indirect-mode address selects (A[1:0])
_A_MR = 0b00
_A_AR0 = 0b01 # IDM_AR high byte
_A_AR1 = 0b10 # IDM_AR low byte
_A_DR = 0b11 # IDM_DR (data)
class W5100ParallelMaster(Elaboratable):
"""W5100 master over the indirect parallel bus, sync clock domain.
Physical bus pins
-----------------
bus_addr : A[1:0] output
bus_data_o : D[7:0] output value (drive when bus_data_oe=1)
bus_data_oe: data-bus output enable (1=FPGA drives D, 0=W5100 drives D)
bus_data_i : D[7:0] input value (sampled during reads)
cs_n / rd_n / wr_n : bus control (active low)
w5100_int_n : W5100 INT_N input (active low)
w5100_rst_n : W5100 hardware reset (active low)
Init / TX / RX interfaces are identical to W5500SPIMaster.
"""
def __init__(self, strobe_cycles=3, reset_cycles=24000,
enable_udp_test=False, udp_socket=3,
src_ip="192.168.1.123", subnet="255.255.255.0",
gateway="192.168.1.1", dst_ip="192.168.1.100",
src_port=6464, dst_port=6464,
udp_payload=b"REBBARB-UDP-TEST\r\n"):
# /RD//WR strobe width in sync cycles (≥ W5100 access time).
self._strobe = strobe_cycles
# MR-reset settle wait; testbench overrides with a small value.
self._reset_cycles = reset_cycles
# ── Optional socket-1 UDP bring-up test ───────────────────────────────
# When enabled, init also configures the W5100's IP stack (GAR/SUBR/
# SIPR) and opens socket 1 in UDP mode; pulsing `udp_send_req` then
# emits one UDP datagram to `udp_dst_ip`:dst_port with the payload
# supplied on the `udp_pl_*` stream. MACRAW socket 0 is untouched.
# Everything below is elaborated only when enabled, so a plain BBA
# build (and the existing MACRAW testbench) is bit-for-bit unchanged.
#
# Network config is build-time (the board's own address); the
# destination and payload are RUNTIME so the UART shell can drive them
# from a typed command. `dst_ip` here is only a power-on default for
# the udp_dst_ip input.
if not 1 <= udp_socket <= 3:
raise ValueError("udp_socket must be 1..3 (socket 0 is MACRAW)")
self._enable_udp = enable_udp_test
self._sn = _socket_addrs(udp_socket)
self._src_ip = _ip_bytes(src_ip)
self._subnet = _ip_bytes(subnet)
self._gateway = _ip_bytes(gateway)
self._src_port = _port_bytes(src_port)
self._dst_port = _port_bytes(dst_port)
_dst_default = _ip_bytes(dst_ip)
# Runtime UDP-send control / datapath.
self.udp_send_req = Signal() # pulse: emit one datagram
self.udp_test_busy = Signal() # level: send in progress
self.udp_dst_ip = Signal(32, # destination IP (big-endian:
init=int.from_bytes(bytes(_dst_default), "big")) # [0]=MSB octet)
# Payload byte stream (driven by the shell; consumed during the send).
self.udp_pl_data = Signal(8)
self.udp_pl_valid = Signal()
self.udp_pl_last = Signal()
self.udp_pl_ready = Signal()
# UDP receive (poll-driven): pulse udp_rx_req to check the socket's RX
# buffer for a datagram. Either udp_rx_none pulses (nothing waiting) or
# the payload streams out on udp_rx_* (sof/eof frame the datagram);
# udp_rx_busy is high while a check/read is in progress.
self.udp_rx_req = Signal()
self.udp_rx_busy = Signal()
self.udp_rx_none = Signal()
self.udp_rx_data = Signal(8)
self.udp_rx_valid = Signal()
self.udp_rx_sof = Signal()
self.udp_rx_eof = Signal()
self.udp_rx_ready = Signal()
# Physical parallel bus
self.bus_addr = Signal(2)
self.bus_data_o = Signal(8)
self.bus_data_oe = Signal()
self.bus_data_i = Signal(8)
self.cs_n = Signal(init=1)
self.rd_n = Signal(init=1)
self.wr_n = Signal(init=1)
self.w5100_int_n = Signal(init=1)
self.w5100_rst_n = Signal(init=1)
# Init control
self.init_req = Signal()
self.init_done = Signal()
self.par = Signal(48) # MAC address (PAR0..5 packed)
# TX stream
self.tx_data = Signal(8)
self.tx_valid = Signal()
self.tx_ready = Signal()
self.tx_sof = Signal()
self.tx_eof = Signal()
# RX stream
self.rx_data = Signal(8)
self.rx_valid = Signal()
self.rx_ready = Signal()
self.rx_sof = Signal()
self.rx_eof = Signal()
def elaborate(self, platform):
m = Module()
STROBE = self._strobe
sn = self._sn # UDP-test socket register/buffer addresses
# ── Bus access engine: one indirect-bus read or write cycle ──────────
bus_go = Signal()
bus_rw = Signal() # 1 = write, 0 = read
bus_a = Signal(2)
bus_wdata = Signal(8)
bus_rdata = Signal(8)
bus_done = Signal()
bus_ctr = Signal(range(STROBE + 2))
rw_r = Signal()
# registered physical outputs
a_o = Signal(2)
d_o = Signal(8)
d_oe = Signal()
cs_r = Signal(init=1)
rd_r = Signal(init=1)
wr_r = Signal(init=1)
m.d.comb += [
self.bus_addr .eq(a_o),
self.bus_data_o .eq(d_o),
self.bus_data_oe.eq(d_oe),
self.cs_n .eq(cs_r),
self.rd_n .eq(rd_r),
self.wr_n .eq(wr_r),
]
m.d.sync += bus_done.eq(0)
with m.FSM(domain="sync", name="bus_fsm"):
with m.State("IDLE"):
m.d.sync += [cs_r.eq(1), rd_r.eq(1), wr_r.eq(1), d_oe.eq(0)]
with m.If(bus_go):
m.d.sync += [a_o.eq(bus_a), rw_r.eq(bus_rw),
cs_r.eq(0), bus_ctr.eq(0)]
with m.If(bus_rw):
m.d.sync += [d_o.eq(bus_wdata), d_oe.eq(1), wr_r.eq(0)]
with m.Else():
m.d.sync += rd_r.eq(0)
m.next = "STROBE"
with m.State("STROBE"):
m.d.sync += bus_ctr.eq(bus_ctr + 1)
with m.If(bus_ctr == STROBE - 1):
with m.If(~rw_r):
m.d.sync += bus_rdata.eq(self.bus_data_i) # sample read
m.d.sync += [rd_r.eq(1), wr_r.eq(1)]
m.next = "FINISH"
with m.State("FINISH"):
m.d.sync += [cs_r.eq(1), d_oe.eq(0), bus_done.eq(1)]
m.next = "IDLE"
# ── Transaction engine: address-set + payload over the bus engine ────
WBUF = 8
xfer_start = Signal()
xfer_direct = Signal() # 1 = single A=00 access (MR), addr ignored
xfer_addr = Signal(16)
xfer_rw = Signal() # payload direction: 1=write, 0=read
xfer_len = Signal(range(WBUF + 1))
xfer_stream = Signal() # stream-write payload from s_*
xfer_sread = Signal() # stream-read payload to r_*
xfer_rcount = Signal(16)
xfer_done = Signal()
wbuf = Array([Signal(8, name=f"wbuf{i}") for i in range(WBUF)])
rbuf = Array([Signal(8, name=f"rbuf{i}") for i in range(WBUF)])
s_count = Signal(16) # bytes streamed-written (advances pointers)
xfer_idx = Signal(range(WBUF + 1))
s_last_r = Signal()
r_idx = Signal(16)
# Streaming payload interfaces.
s_data, s_valid, s_last, s_consume = Signal(8), Signal(), Signal(), Signal()
r_data, r_valid, r_first, r_last, r_ready = (
Signal(8), Signal(), Signal(), Signal(), Signal())
# TX stream-write source mux: during a UDP-test send the payload comes
# from the external `udp_pl_*` stream (the shell); otherwise the normal
# MACRAW TX interface feeds it. `udp_streaming` is raised only while the
# UDP payload is being written to the socket-1 TX buffer.
udp_streaming = Signal()
if self._enable_udp:
with m.If(udp_streaming):
m.d.comb += [s_data.eq(self.udp_pl_data),
s_valid.eq(self.udp_pl_valid),
s_last.eq(self.udp_pl_last),
self.udp_pl_ready.eq(s_consume)]
with m.Else():
m.d.comb += [s_data.eq(self.tx_data), s_valid.eq(self.tx_valid),
s_last.eq(self.tx_eof), self.tx_ready.eq(s_consume)]
else:
m.d.comb += [s_data.eq(self.tx_data), s_valid.eq(self.tx_valid),
s_last.eq(self.tx_eof), self.tx_ready.eq(s_consume)]
# RX stream-read sink mux: during a socket-N UDP receive the payload
# streams out on `udp_rx_*` (to the shell); otherwise it goes to the
# MACRAW rx interface (to the frame assembler). `udp_rx_streaming` is
# raised only while the UDP payload is being read out.
udp_rx_streaming = Signal()
if self._enable_udp:
with m.If(udp_rx_streaming):
m.d.comb += [self.udp_rx_data.eq(r_data),
self.udp_rx_valid.eq(r_valid),
self.udp_rx_sof.eq(r_first),
self.udp_rx_eof.eq(r_last),
r_ready.eq(self.udp_rx_ready)]
with m.Else():
m.d.comb += [self.rx_data.eq(r_data), self.rx_valid.eq(r_valid),
self.rx_sof.eq(r_first), self.rx_eof.eq(r_last),
r_ready.eq(self.rx_ready)]
else:
m.d.comb += [self.rx_data.eq(r_data), self.rx_valid.eq(r_valid),
self.rx_sof.eq(r_first), self.rx_eof.eq(r_last),
r_ready.eq(self.rx_ready)]
# Socket-buffer ring wraparound. Unlike the W5500, the W5100's IDM
# address does NOT auto-wrap at the socket-buffer boundary — it just
# increments linearly into the next region. So when a streamed access
# reaches `xfer_wend`, the engine re-sets IDM_AR back to `xfer_wbase`.
xfer_wrap = Signal()
xfer_wbase = Signal(16)
xfer_wend = Signal(16)
cur_addr = Signal(16)
m.d.comb += [bus_go.eq(0), bus_rw.eq(0), bus_a.eq(0), bus_wdata.eq(0)]
m.d.comb += [s_consume.eq(0), r_valid.eq(0), r_data.eq(0),
r_first.eq(0), r_last.eq(0)]
m.d.sync += xfer_done.eq(0)
def bus_write(a, data):
m.d.comb += [bus_go.eq(1), bus_rw.eq(1), bus_a.eq(a), bus_wdata.eq(data)]
def bus_read(a):
m.d.comb += [bus_go.eq(1), bus_rw.eq(0), bus_a.eq(a)]
with m.FSM(domain="sync", name="xfer_fsm"):
with m.State("IDLE"):
with m.If(xfer_start):
m.d.sync += [xfer_idx.eq(0), s_count.eq(0), r_idx.eq(0),
cur_addr.eq(xfer_addr)]
with m.If(xfer_direct):
m.next = "DIRECT"
with m.Else():
m.next = "AR_HI"
# Direct MR write (A=00)
with m.State("DIRECT"):
bus_write(_A_MR, wbuf[0])
m.next = "DIRECT_W"
with m.State("DIRECT_W"):
with m.If(bus_done):
m.next = "FINISH"
# Set indirect address IDM_AR (high then low)
with m.State("AR_HI"):
bus_write(_A_AR0, xfer_addr[8:16])
m.next = "AR_HI_W"
with m.State("AR_HI_W"):
with m.If(bus_done):
m.next = "AR_LO"
with m.State("AR_LO"):
bus_write(_A_AR1, xfer_addr[0:8])
m.next = "AR_LO_W"
with m.State("AR_LO_W"):
with m.If(bus_done):
with m.If(xfer_stream):
m.next = "SW_LOAD"
with m.Elif(xfer_sread):
m.next = "SR_LOAD"
with m.Elif(xfer_rw):
m.next = "WB_ISSUE"
with m.Else():
m.next = "RB_ISSUE"
# Fixed-length write from wbuf (IDM_DR burst, auto-increment)
with m.State("WB_ISSUE"):
bus_write(_A_DR, wbuf[xfer_idx])
m.next = "WB_WAIT"
with m.State("WB_WAIT"):
with m.If(bus_done):
m.d.sync += xfer_idx.eq(xfer_idx + 1)
with m.If(xfer_idx + 1 == xfer_len):
m.next = "FINISH"
with m.Else():
m.next = "WB_ISSUE"
# Fixed-length read into rbuf (with ring wrap, for the length header)
with m.State("RB_ISSUE"):
with m.If(xfer_wrap & (cur_addr == xfer_wend)):
m.next = "RB_WRAP_HI"
with m.Else():
bus_read(_A_DR)
m.next = "RB_WAIT"
with m.State("RB_WAIT"):
with m.If(bus_done):
m.d.sync += rbuf[xfer_idx].eq(bus_rdata)
m.d.sync += [xfer_idx.eq(xfer_idx + 1), cur_addr.eq(cur_addr + 1)]
with m.If(xfer_idx + 1 == xfer_len):
m.next = "FINISH"
with m.Else():
m.next = "RB_ISSUE"
with m.State("RB_WRAP_HI"):
bus_write(_A_AR0, xfer_wbase[8:16])
m.next = "RB_WRAP_HI_W"
with m.State("RB_WRAP_HI_W"):
with m.If(bus_done):
m.next = "RB_WRAP_LO"
with m.State("RB_WRAP_LO"):
bus_write(_A_AR1, xfer_wbase[0:8])
m.next = "RB_WRAP_LO_W"
with m.State("RB_WRAP_LO_W"):
with m.If(bus_done):
m.d.sync += cur_addr.eq(xfer_wbase)
m.next = "RB_ISSUE"
# Stream-write payload from s_* until s_last (with ring wrap)
with m.State("SW_LOAD"):
with m.If(xfer_wrap & (cur_addr == xfer_wend)):
m.next = "SW_WRAP_HI"
with m.Elif(s_valid):
bus_write(_A_DR, s_data)
m.d.sync += s_last_r.eq(s_last)
m.next = "SW_WAIT"
with m.State("SW_WAIT"):
with m.If(bus_done):
m.d.comb += s_consume.eq(1)
m.d.sync += [s_count.eq(s_count + 1), cur_addr.eq(cur_addr + 1)]
with m.If(s_last_r):
m.next = "FINISH"
with m.Else():
m.next = "SW_LOAD"
with m.State("SW_WRAP_HI"):
bus_write(_A_AR0, xfer_wbase[8:16])
m.next = "SW_WRAP_HI_W"
with m.State("SW_WRAP_HI_W"):
with m.If(bus_done):
m.next = "SW_WRAP_LO"
with m.State("SW_WRAP_LO"):
bus_write(_A_AR1, xfer_wbase[0:8])
m.next = "SW_WRAP_LO_W"
with m.State("SW_WRAP_LO_W"):
with m.If(bus_done):
m.d.sync += cur_addr.eq(xfer_wbase)
m.next = "SW_LOAD"
# Stream-read payload to r_* for rcount bytes (with ring wrap)
with m.State("SR_LOAD"):
with m.If(r_idx == xfer_rcount):
m.next = "FINISH"
with m.Elif(xfer_wrap & (cur_addr == xfer_wend)):
m.next = "SR_WRAP_HI"
with m.Else():
bus_read(_A_DR)
m.next = "SR_WAIT"
with m.State("SR_WAIT"):
with m.If(bus_done):
m.next = "SR_PUSH"
with m.State("SR_PUSH"):
m.d.comb += [r_data.eq(bus_rdata), r_valid.eq(1),
r_first.eq(r_idx == 0),
r_last.eq(r_idx + 1 == xfer_rcount)]
with m.If(r_ready):
m.d.sync += [r_idx.eq(r_idx + 1), cur_addr.eq(cur_addr + 1)]
m.next = "SR_LOAD"
with m.State("SR_WRAP_HI"):
bus_write(_A_AR0, xfer_wbase[8:16])
m.next = "SR_WRAP_HI_W"
with m.State("SR_WRAP_HI_W"):
with m.If(bus_done):
m.next = "SR_WRAP_LO"
with m.State("SR_WRAP_LO"):
bus_write(_A_AR1, xfer_wbase[0:8])
m.next = "SR_WRAP_LO_W"
with m.State("SR_WRAP_LO_W"):
with m.If(bus_done):
m.d.sync += cur_addr.eq(xfer_wbase)
m.next = "SR_LOAD"
with m.State("FINISH"):
m.d.sync += xfer_done.eq(1)
m.next = "IDLE"
# ── Control regs ─────────────────────────────────────────────────────
mac_shadow = Array([Signal(8, name=f"mac{i}") for i in range(6)])
wait_ctr = Signal(range(self._reset_cycles + 2))
tx_wr = Signal(16)
rx_rsr = Signal(16)
rx_rd = Signal(16)
pkt_len = Signal(16)
s1_tx_wr = Signal(16) # UDP-socket TX write pointer (send)
sn_rx_rsr = Signal(16) # UDP-socket RX received size
sn_rx_rd = Signal(16) # UDP-socket RX read pointer
sn_pkt_len = Signal(16) # UDP datagram payload length (from header)
def write_reg(name, addr, payload, nxt, direct=False):
"""Emit a 2-state block that writes `payload` (a list) to `addr`."""
with m.State(name):
m.d.sync += [xfer_addr.eq(addr), xfer_rw.eq(1),
xfer_stream.eq(0), xfer_sread.eq(0), xfer_wrap.eq(0),
xfer_direct.eq(1 if direct else 0),
xfer_len.eq(len(payload))]
for i, b in enumerate(payload):
m.d.sync += wbuf[i].eq(b)
m.d.sync += xfer_start.eq(1)
m.next = name + "_W"
with m.State(name + "_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.next = nxt
# ── Main control FSM (Phase 1: init only) ────────────────────────────
if self._enable_udp:
m.d.sync += self.udp_rx_none.eq(0) # pulse: default low
with m.FSM(domain="sync", name="main_fsm"):
with m.State("IDLE"):
m.d.sync += self.init_done.eq(0)
with m.If(self.init_req):
for i in range(6):
m.d.sync += mac_shadow[i].eq(self.par[i*8:(i+1)*8])
m.next = "MR_RST"
with m.Elif(~self.w5100_int_n):
m.next = "RX_CHECK"
with m.Elif(self.tx_valid & self.tx_sof):
m.next = "TX_START"
if self._enable_udp:
with m.Elif(self.udp_send_req):
m.next = "UDP_DIPR"
with m.Elif(self.udp_rx_req):
m.next = "UDP_RX_RSR"
# MR = 0x80 software reset (direct A=00), then settle.
write_reg("MR_RST", _MR, [_MR_RST], "MR_WAIT", direct=True)
with m.State("MR_WAIT"):
with m.If(wait_ctr == self._reset_cycles):
m.d.sync += wait_ctr.eq(0)
m.next = "MR_MODE"
with m.Else():
m.d.sync += wait_ctr.eq(wait_ctr + 1)
# MR = indirect + auto-increment (direct A=00).
write_reg("MR_MODE", _MR, [_MR_IND | _MR_AI], "SHAR", direct=True)
# SHAR = source MAC (6-byte auto-increment burst).
with m.State("SHAR"):
m.d.sync += [xfer_addr.eq(_SHAR0), xfer_rw.eq(1),
xfer_stream.eq(0), xfer_sread.eq(0),
xfer_direct.eq(0), xfer_len.eq(6)]
for i in range(6):
m.d.sync += wbuf[i].eq(mac_shadow[i])
m.d.sync += xfer_start.eq(1)
m.next = "SHAR_W"
with m.State("SHAR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.next = "MEMSZ"
# RMSR/TMSR = 0x55 (2 KB per socket — default; socket 0 used).
write_reg("MEMSZ", _RMSR, [0x55, 0x55], "S0_MODE") # RMSR then TMSR
# Socket 0: MACRAW mode, OPEN, enable interrupt.
write_reg("S0_MODE", _S0_MR, [_S0_MR_MACRAW], "S0_OPEN")
write_reg("S0_OPEN", _S0_CR, [_CR_OPEN], "S0_IMR")
# After S0 IMR, either finish (plain BBA) or configure the IP stack
# and open socket 1 in UDP mode (bring-up test build).
write_reg("S0_IMR", _IMR, [0x01],
"UDP_CFG_GAR" if self._enable_udp else "INIT_DONE")
if self._enable_udp:
# IP stack: gateway, subnet, source IP (SHAR already set above).
write_reg("UDP_CFG_GAR", _GAR0, self._gateway, "UDP_CFG_SUBR")
write_reg("UDP_CFG_SUBR", _SUBR0, self._subnet, "UDP_CFG_SIPR")
write_reg("UDP_CFG_SIPR", _SIPR0, self._src_ip, "UDP_CFG_MR")
# Socket 1: UDP mode, source port, OPEN, fixed dest port.
write_reg("UDP_CFG_MR", sn["MR"], [_S1_MR_UDP], "UDP_CFG_PORT")
write_reg("UDP_CFG_PORT", sn["PORT"], self._src_port, "UDP_CFG_OPEN")
write_reg("UDP_CFG_OPEN", sn["CR"], [_CR_OPEN], "UDP_CFG_DPORT")
write_reg("UDP_CFG_DPORT", sn["DPORT"], self._dst_port, "INIT_DONE")
with m.State("INIT_DONE"):
m.d.sync += self.init_done.eq(1)
m.next = "IDLE"
# ── TX MACRAW ────────────────────────────────────────────────────
# read S0_TX_WR → stream frame into the TX buffer at that offset
# (ring-wrapping at the 2 KB boundary) → advance S0_TX_WR → SEND.
with m.State("TX_START"): # read S0_TX_WR (2 bytes)
m.d.sync += [xfer_addr.eq(_S0_TX_WR), xfer_rw.eq(0),
xfer_stream.eq(0), xfer_sread.eq(0), xfer_wrap.eq(0),
xfer_direct.eq(0), xfer_len.eq(2)]
m.d.sync += xfer_start.eq(1)
m.next = "TX_RDPTR_W"
with m.State("TX_RDPTR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += tx_wr.eq(Cat(rbuf[1], rbuf[0])) # big-endian
m.next = "TX_DATA"
with m.State("TX_DATA"): # stream frame → TX buffer
m.d.sync += [xfer_addr.eq(_TX_BASE + (tx_wr & _S0_TX_MASK)),
xfer_rw.eq(1), xfer_stream.eq(1), xfer_sread.eq(0),
xfer_direct.eq(0), xfer_wrap.eq(1),
xfer_wbase.eq(_TX_BASE),
xfer_wend.eq(_TX_BASE + _S0_TX_MASK + 1)]
m.d.sync += xfer_start.eq(1)
m.next = "TX_DATA_W"
with m.State("TX_DATA_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += [xfer_stream.eq(0), xfer_wrap.eq(0),
tx_wr.eq(tx_wr + s_count)] # advanced pointer
m.next = "TX_UPDPTR"
with m.State("TX_UPDPTR"): # write back S0_TX_WR
m.d.sync += [xfer_addr.eq(_S0_TX_WR), xfer_rw.eq(1),
xfer_stream.eq(0), xfer_sread.eq(0), xfer_wrap.eq(0),
xfer_direct.eq(0), xfer_len.eq(2)]
m.d.sync += [wbuf[0].eq(tx_wr[8:16]), wbuf[1].eq(tx_wr[0:8])]
m.d.sync += xfer_start.eq(1)
m.next = "TX_UPDPTR_W"
with m.State("TX_UPDPTR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.next = "TX_SEND"
# S0_CR = SEND
write_reg("TX_SEND", _S0_CR, [_CR_SEND], "IDLE")
# ── RX MACRAW ────────────────────────────────────────────────────
# On W5100 INT: read RX_RSR; if non-zero read RX_RD, read the 2-byte
# MACRAW length, stream (length−2) frame bytes out (ring-wrapping),
# advance RX_RD by the length, issue RECV, clear the RECV interrupt.
with m.State("RX_CHECK"): # read S0_RX_RSR (2 bytes)
m.d.sync += [xfer_addr.eq(_S0_RX_RSR), xfer_rw.eq(0),
xfer_stream.eq(0), xfer_sread.eq(0), xfer_wrap.eq(0),
xfer_direct.eq(0), xfer_len.eq(2)]
m.d.sync += xfer_start.eq(1)
m.next = "RX_RSR_W"
with m.State("RX_RSR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += rx_rsr.eq(Cat(rbuf[1], rbuf[0]))
m.next = "RX_RSR_CHK"
with m.State("RX_RSR_CHK"):
with m.If(rx_rsr == 0):
m.next = "IDLE" # nothing received
with m.Else():
m.next = "RX_RDPTR"
with m.State("RX_RDPTR"): # read S0_RX_RD (2 bytes)
m.d.sync += [xfer_addr.eq(_S0_RX_RD), xfer_rw.eq(0),
xfer_stream.eq(0), xfer_sread.eq(0), xfer_wrap.eq(0),
xfer_direct.eq(0), xfer_len.eq(2)]
m.d.sync += xfer_start.eq(1)
m.next = "RX_RDPTR_W"
with m.State("RX_RDPTR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += rx_rd.eq(Cat(rbuf[1], rbuf[0]))
m.next = "RX_LEN"
with m.State("RX_LEN"): # read 2-byte MACRAW length (wrap)
m.d.sync += [xfer_addr.eq(_RX_BASE + (rx_rd & _S0_RX_MASK)),
xfer_rw.eq(0), xfer_stream.eq(0), xfer_sread.eq(0),
xfer_direct.eq(0), xfer_len.eq(2), xfer_wrap.eq(1),
xfer_wbase.eq(_RX_BASE),
xfer_wend.eq(_RX_BASE + _S0_RX_MASK + 1)]
m.d.sync += xfer_start.eq(1)
m.next = "RX_LEN_W"
with m.State("RX_LEN_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += pkt_len.eq(Cat(rbuf[1], rbuf[0]))
m.next = "RX_FRAME"
with m.State("RX_FRAME"): # stream (pkt_len−2) frame bytes
m.d.sync += [xfer_addr.eq(_RX_BASE + ((rx_rd + 2) & _S0_RX_MASK)),
xfer_rw.eq(0), xfer_stream.eq(0), xfer_sread.eq(1),
xfer_direct.eq(0), xfer_rcount.eq(pkt_len - 2),
xfer_wrap.eq(1), xfer_wbase.eq(_RX_BASE),
xfer_wend.eq(_RX_BASE + _S0_RX_MASK + 1)]
m.d.sync += xfer_start.eq(1)
m.next = "RX_FRAME_W"
with m.State("RX_FRAME_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += [xfer_sread.eq(0), xfer_wrap.eq(0)]
m.next = "RX_UPDRD"
with m.State("RX_UPDRD"): # S0_RX_RD += pkt_len, write back
m.d.sync += [xfer_addr.eq(_S0_RX_RD), xfer_rw.eq(1),
xfer_stream.eq(0), xfer_sread.eq(0), xfer_wrap.eq(0),
xfer_direct.eq(0), xfer_len.eq(2)]
m.d.sync += [wbuf[0].eq((rx_rd + pkt_len)[8:16]),
wbuf[1].eq((rx_rd + pkt_len)[0:8])]
m.d.sync += xfer_start.eq(1)
m.next = "RX_UPDRD_W"
with m.State("RX_UPDRD_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.next = "RX_RECV"
# S0_CR = RECV, then clear the RECV interrupt bit (S0_IR[2]).
write_reg("RX_RECV", _S0_CR, [_CR_RECV], "RX_CLR_IR")
write_reg("RX_CLR_IR", _S0_IR, [0x04], "IDLE")
# ── UDP test send (configurable socket) ──────────────────────────
# Runtime dest IP → Sn_DIPR, read Sn_TX_WR, stream the shell-
# supplied payload into the socket TX buffer (ring-wrapping),
# advance Sn_TX_WR, SEND. DPORT + socket open were done at init.
if self._enable_udp:
with m.State("UDP_DIPR"): # write Sn_DIPR (runtime, 4 B)
m.d.sync += self.udp_test_busy.eq(1)
m.d.sync += [xfer_addr.eq(sn["DIPR"]), xfer_rw.eq(1),
xfer_stream.eq(0), xfer_sread.eq(0),
xfer_wrap.eq(0), xfer_direct.eq(0),
xfer_len.eq(4)]
m.d.sync += [wbuf[0].eq(self.udp_dst_ip[24:32]),
wbuf[1].eq(self.udp_dst_ip[16:24]),
wbuf[2].eq(self.udp_dst_ip[8:16]),
wbuf[3].eq(self.udp_dst_ip[0:8])]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_DIPR_W"
with m.State("UDP_DIPR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.next = "UDP_TXWR"
with m.State("UDP_TXWR"): # read Sn_TX_WR (2 B)
m.d.sync += [xfer_addr.eq(sn["TX_WR"]), xfer_rw.eq(0),
xfer_stream.eq(0), xfer_sread.eq(0),
xfer_wrap.eq(0), xfer_direct.eq(0),
xfer_len.eq(2)]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_TXWR_W"
with m.State("UDP_TXWR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += s1_tx_wr.eq(Cat(rbuf[1], rbuf[0]))
m.next = "UDP_DATA"
with m.State("UDP_DATA"): # stream payload → Sn TX buffer
m.d.sync += [xfer_addr.eq(sn["TX_BASE"] + (s1_tx_wr & _SN_MASK)),
xfer_rw.eq(1), xfer_stream.eq(1),
xfer_sread.eq(0), xfer_direct.eq(0),
xfer_wrap.eq(1), xfer_wbase.eq(sn["TX_BASE"]),
xfer_wend.eq(sn["TX_BASE"] + _SN_MASK + 1),
udp_streaming.eq(1)]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_DATA_W"
with m.State("UDP_DATA_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += [xfer_stream.eq(0), xfer_wrap.eq(0),
udp_streaming.eq(0),
s1_tx_wr.eq(s1_tx_wr + s_count)]
m.next = "UDP_UPDPTR"
with m.State("UDP_UPDPTR"): # write back Sn_TX_WR (2 B)
m.d.sync += [xfer_addr.eq(sn["TX_WR"]), xfer_rw.eq(1),
xfer_stream.eq(0), xfer_sread.eq(0),
xfer_wrap.eq(0), xfer_direct.eq(0),
xfer_len.eq(2)]
m.d.sync += [wbuf[0].eq(s1_tx_wr[8:16]),
wbuf[1].eq(s1_tx_wr[0:8])]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_UPDPTR_W"
with m.State("UDP_UPDPTR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.next = "UDP_SEND"
write_reg("UDP_SEND", sn["CR"], [_CR_SEND], "UDP_DONE")
with m.State("UDP_DONE"):
m.d.sync += self.udp_test_busy.eq(0)
m.next = "IDLE"
# ── UDP test receive (configurable socket) ───────────────────────
# Poll-driven (the shell pulses udp_rx_req). Read Sn_RX_RSR; if 0,
# pulse udp_rx_none. Otherwise read Sn_RX_RD, read the 8-byte WIZnet
# UDP header ([srcIP 4][srcPort 2][len 2]), stream `len` payload
# bytes out on udp_rx_* (ring-wrapping), advance Sn_RX_RD by 8+len,
# RECV, clear the socket IR. MACRAW socket-0 RX is untouched.
if self._enable_udp:
with m.State("UDP_RX_RSR"): # read Sn_RX_RSR (2 B)
m.d.sync += self.udp_rx_busy.eq(1)
m.d.sync += [xfer_addr.eq(sn["RX_RSR"]), xfer_rw.eq(0),
xfer_stream.eq(0), xfer_sread.eq(0),
xfer_wrap.eq(0), xfer_direct.eq(0),
xfer_len.eq(2)]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_RX_RSR_W"
with m.State("UDP_RX_RSR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += sn_rx_rsr.eq(Cat(rbuf[1], rbuf[0]))
m.next = "UDP_RX_CHK"
with m.State("UDP_RX_CHK"):
with m.If(sn_rx_rsr == 0):
m.d.sync += [self.udp_rx_none.eq(1),
self.udp_rx_busy.eq(0)]
m.next = "IDLE"
with m.Else():
m.next = "UDP_RX_RD"
with m.State("UDP_RX_RD"): # read Sn_RX_RD (2 B)
m.d.sync += [xfer_addr.eq(sn["RX_RD"]), xfer_rw.eq(0),
xfer_stream.eq(0), xfer_sread.eq(0),
xfer_wrap.eq(0), xfer_direct.eq(0),
xfer_len.eq(2)]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_RX_RD_W"
with m.State("UDP_RX_RD_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += sn_rx_rd.eq(Cat(rbuf[1], rbuf[0]))
m.next = "UDP_RX_HDR"
with m.State("UDP_RX_HDR"): # read 8-byte UDP header (wrap)
m.d.sync += [xfer_addr.eq(sn["RX_BASE"] + (sn_rx_rd & _SN_MASK)),
xfer_rw.eq(0), xfer_stream.eq(0), xfer_sread.eq(0),
xfer_direct.eq(0), xfer_len.eq(8), xfer_wrap.eq(1),
xfer_wbase.eq(sn["RX_BASE"]),
xfer_wend.eq(sn["RX_BASE"] + _SN_MASK + 1)]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_RX_HDR_W"
with m.State("UDP_RX_HDR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
# header[6:8] = payload length (big-endian)
m.d.sync += sn_pkt_len.eq(Cat(rbuf[7], rbuf[6]))
m.next = "UDP_RX_FRAME"
with m.State("UDP_RX_FRAME"): # stream `len` payload bytes out
m.d.sync += [xfer_addr.eq(sn["RX_BASE"] + ((sn_rx_rd + 8) & _SN_MASK)),
xfer_rw.eq(0), xfer_stream.eq(0), xfer_sread.eq(1),
xfer_direct.eq(0), xfer_rcount.eq(sn_pkt_len),
xfer_wrap.eq(1), xfer_wbase.eq(sn["RX_BASE"]),
xfer_wend.eq(sn["RX_BASE"] + _SN_MASK + 1),
udp_rx_streaming.eq(1)]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_RX_FRAME_W"
with m.State("UDP_RX_FRAME_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += [xfer_sread.eq(0), xfer_wrap.eq(0),
udp_rx_streaming.eq(0)]
m.next = "UDP_RX_UPDRD"
with m.State("UDP_RX_UPDRD"): # Sn_RX_RD += 8 + len, write back
m.d.sync += [xfer_addr.eq(sn["RX_RD"]), xfer_rw.eq(1),
xfer_stream.eq(0), xfer_sread.eq(0),
xfer_wrap.eq(0), xfer_direct.eq(0), xfer_len.eq(2)]
m.d.sync += [wbuf[0].eq((sn_rx_rd + 8 + sn_pkt_len)[8:16]),
wbuf[1].eq((sn_rx_rd + 8 + sn_pkt_len)[0:8])]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_RX_UPDRD_W"
with m.State("UDP_RX_UPDRD_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.next = "UDP_RX_RECV"
write_reg("UDP_RX_RECV", sn["CR"], [_CR_RECV], "UDP_RX_CLR_IR")
write_reg("UDP_RX_CLR_IR", sn["IR"], [0x04], "UDP_RX_DONE")
with m.State("UDP_RX_DONE"):
m.d.sync += self.udp_rx_busy.eq(0)
m.next = "IDLE"
return m
# ── Testbench ─────────────────────────────────────────────────────────────
if __name__ == "__main__":
import sys
from amaranth.sim import Simulator, Period
dut = W5100ParallelMaster(strobe_cycles=3, reset_cycles=10)
errors = []
MAC = [0x11, 0x22, 0x33, 0x44, 0x55, 0x66]
PAR = sum(b << (8 * i) for i, b in enumerate(MAC))
# Expected indirect-address writes captured by the model (addr, value).
# MR is written directly (A=00) → captured as ('MR', value).
EXPECTED = [
("MR", _MR_RST),
("MR", _MR_IND | _MR_AI),
(_SHAR0 + 0, MAC[0]), (_SHAR0 + 1, MAC[1]), (_SHAR0 + 2, MAC[2]),
(_SHAR0 + 3, MAC[3]), (_SHAR0 + 4, MAC[4]), (_SHAR0 + 5, MAC[5]),
(_RMSR + 0, 0x55), (_RMSR + 1, 0x55),
(_S0_MR, _S0_MR_MACRAW),
(_S0_CR, _CR_OPEN),
(_IMR, 0x01),
]
writes = [] # captured (addr-or-'MR', value) — IDM_DR + MR writes
model_mem = {} # W5100 memory image (registers + TX/RX buffers)
async def w5100_model(ctx):
"""W5100 indirect-bus slave model: tracks MR/IDM_AR, records IDM_DR and
MR writes, and drives bus_data_i for reads. Mode-0 timing: a write is
latched on /WR rising while /CS low; reads driven while /RD low."""
idm_ar = 0
mr = 0
prev_cs = prev_rd = prev_wr = 1
async for vals in ctx.tick("sync").sample(
dut.cs_n, dut.rd_n, dut.wr_n,
dut.bus_addr, dut.bus_data_o, dut.bus_data_oe):
cs, rd, wr, a, do, doe = vals[-6:]
ai = (mr >> 1) & 1 # MR.AI
# Drive read data while /RD asserted (combinational, before sample).
if cs == 0 and rd == 0:
if a == _A_MR:
val = mr
elif a == _A_AR0:
val = (idm_ar >> 8) & 0xFF
elif a == _A_AR1:
val = idm_ar & 0xFF
else:
val = model_mem.get(idm_ar, 0)
ctx.set(dut.bus_data_i, val)
# Latch write on /WR rising edge.
if cs == 0 and prev_wr == 0 and wr == 1:
if a == _A_MR:
mr = do
writes.append(("MR", do))
elif a == _A_AR0:
idm_ar = (idm_ar & 0x00FF) | (do << 8)
elif a == _A_AR1:
idm_ar = (idm_ar & 0xFF00) | do
else: # IDM_DR
model_mem[idm_ar] = do
writes.append((idm_ar, do))
# RECV command consumes the RX data: clear RSR (mirrors HW).
if idm_ar == _S0_CR and do == _CR_RECV:
model_mem[_S0_RX_RSR] = 0
model_mem[_S0_RX_RSR + 1] = 0
if ai:
idm_ar = (idm_ar + 1) & 0xFFFF
# Auto-increment after a data read (/RD rising, A=DR).
if cs == 0 and prev_rd == 0 and rd == 1 and a == _A_DR and ai:
idm_ar = (idm_ar + 1) & 0xFFFF
prev_cs, prev_rd, prev_wr = cs, rd, wr
async def testbench(ctx):
ctx.set(dut.par, PAR)
await ctx.tick("sync").repeat(2)
# T1: trigger init, wait for init_done.
ctx.set(dut.init_req, 1)
await ctx.tick("sync").repeat(1)
ctx.set(dut.init_req, 0)
done = False
for _ in range(4000):
await ctx.tick("sync").repeat(1)
if ctx.get(dut.init_done):
done = True
break
if not done:
errors.append("init_done never asserted")
print(f"T1 init captured {len(writes)} writes")
if writes != EXPECTED:
errors.append("init write sequence mismatch")
for i in range(max(len(writes), len(EXPECTED))):
g = writes[i] if i < len(writes) else None
e = EXPECTED[i] if i < len(EXPECTED) else None
mark = "" if g == e else " <-- MISMATCH"
gs = f"({g[0]:#06x},{g[1]:#04x})" if g and isinstance(g[0], int) else str(g)
es = f"({e[0]:#06x},{e[1]:#04x})" if e and isinstance(e[0], int) else str(e)
print(f" [{i:2}] got {gs:20} exp {es:20}{mark}")
else:
print("T1 init sequence matches expected (MR, SHAR, mem sizes, "
"S0 MACRAW/OPEN, IMR)")
# ── helper: stream one TX frame through the external tx interface ─────
async def feed_frame(ctx, frame):
for i, b in enumerate(frame):
ctx.set(dut.tx_data, b)
ctx.set(dut.tx_valid, 1)
ctx.set(dut.tx_sof, 1 if i == 0 else 0)
ctx.set(dut.tx_eof, 1 if i == len(frame) - 1 else 0)
got = False
for _ in range(400):
await ctx.tick("sync").repeat(1)
if ctx.get(dut.tx_ready):
got = True
break
if not got:
errors.append(f"feed_frame: byte {i} never consumed")
return
ctx.set(dut.tx_valid, 0)
ctx.set(dut.tx_sof, 0)
ctx.set(dut.tx_eof, 0)
# let TX_UPDPTR + SEND complete
for _ in range(200):
await ctx.tick("sync").repeat(1)
if model_mem.get(_S0_CR) == _CR_SEND:
break
# ── T2: TX MACRAW frame (TX_WR=0, no wrap) ───────────────────────────
FRAME = [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF, 0x10, 0x20]
await feed_frame(ctx, FRAME)
buf = [model_mem.get(_TX_BASE + i, None) for i in range(len(FRAME))]
if buf != FRAME:
errors.append(f"T2 TX buffer mismatch: {buf} != {FRAME}")
tx_wr_hi = model_mem.get(_S0_TX_WR, 0)
tx_wr_lo = model_mem.get(_S0_TX_WR + 1, 0)
adv = (tx_wr_hi << 8) | tx_wr_lo
if adv != len(FRAME):
errors.append(f"T2 S0_TX_WR advance: got {adv}, want {len(FRAME)}")
if model_mem.get(_S0_CR) != _CR_SEND:
errors.append("T2 SEND command not issued")
print(f"T2 TX: buffer={['0x%02X' % b for b in buf]} "
f"TX_WR={adv} SEND={model_mem.get(_S0_CR)==_CR_SEND}")
# ── T3: TX MACRAW with ring wraparound (TX_WR near 2 KB boundary) ─────
# Pre-load S0_TX_WR = 0x07FE so a 6-byte frame straddles the boundary:
# offsets 0x7FE,0x7FF then wraps to 0x000,0x001,0x002,0x003.
model_mem[_S0_TX_WR] = 0x07
model_mem[_S0_TX_WR + 1] = 0xFE
model_mem[_S0_CR] = 0x00 # clear so we can detect the new SEND
WFRAME = [0x41, 0x42, 0x43, 0x44, 0x45, 0x46]
await feed_frame(ctx, WFRAME)
# expected physical layout
exp = {
_TX_BASE + 0x7FE: WFRAME[0],
_TX_BASE + 0x7FF: WFRAME[1],
_TX_BASE + 0x000: WFRAME[2],
_TX_BASE + 0x001: WFRAME[3],
_TX_BASE + 0x002: WFRAME[4],
_TX_BASE + 0x003: WFRAME[5],
}
for addr, want in exp.items():
got = model_mem.get(addr)
if got != want:
errors.append(f"T3 wrap byte @0x{addr:04X}: got {got}, want 0x{want:02X}")
adv2 = (model_mem.get(_S0_TX_WR, 0) << 8) | model_mem.get(_S0_TX_WR + 1, 0)
want_wr = (0x07FE + len(WFRAME)) & 0xFFFF
if adv2 != want_wr:
errors.append(f"T3 wrap S0_TX_WR: got 0x{adv2:04X}, want 0x{want_wr:04X}")
ok = all(model_mem.get(a) == v for a, v in exp.items())
print(f"T3 TX wrap: bytes_placed_ok={ok} TX_WR=0x{adv2:04X} (want 0x{want_wr:04X})")
# ── helper: drive an RX event and collect the streamed-out frame ─────
def load_rx(rx_rd_off, frame):
"""Place a MACRAW packet [len_hi,len_lo,frame...] in the RX buffer at
offset rx_rd_off (ring), set RX_RSR/RX_RD, return the 16-bit length."""
plen = len(frame) + 2
payload = [(plen >> 8) & 0xFF, plen & 0xFF] + list(frame)
for i, b in enumerate(payload):
off = (rx_rd_off + i) & _S0_RX_MASK
model_mem[_RX_BASE + off] = b
model_mem[_S0_RX_RSR] = (plen >> 8) & 0xFF
model_mem[_S0_RX_RSR + 1] = plen & 0xFF
model_mem[_S0_RX_RD] = (rx_rd_off >> 8) & 0xFF
model_mem[_S0_RX_RD + 1] = rx_rd_off & 0xFF
return plen
async def do_rx(ctx, rx_rd_off, frame):
plen = load_rx(rx_rd_off, frame)
ctx.set(dut.rx_ready, 1)
collected = []
ctx.set(dut.w5100_int_n, 0) # assert RX interrupt
for _ in range(1500):
await ctx.tick("sync").repeat(1)
if ctx.get(dut.rx_valid) and ctx.get(dut.rx_ready):
collected.append(ctx.get(dut.rx_data))
if model_mem.get(_S0_CR) == _CR_RECV:
break
ctx.set(dut.w5100_int_n, 1) # deassert; let it finish + idle
for _ in range(300):
await ctx.tick("sync").repeat(1)
ctx.set(dut.rx_ready, 0)
return collected, plen
# ── T4: RX MACRAW frame (RX_RD=0, no wrap) ───────────────────────────
model_mem[_S0_CR] = 0x00
RX_FRAME = [0xDE, 0xAD, 0xBE, 0xEF, 0x01, 0x02, 0x03]
got, plen = await do_rx(ctx, 0x0000, RX_FRAME)
if got != RX_FRAME:
errors.append(f"T4 RX frame mismatch: {['0x%02X'%b for b in got]} != "
f"{['0x%02X'%b for b in RX_FRAME]}")
new_rd = (model_mem.get(_S0_RX_RD, 0) << 8) | model_mem.get(_S0_RX_RD + 1, 0)
if new_rd != plen:
errors.append(f"T4 RX_RD advance: got 0x{new_rd:04X}, want 0x{plen:04X}")
print(f"T4 RX: frame={['0x%02X'%b for b in got]} RX_RD=0x{new_rd:04X} "
f"RECV={model_mem.get(_S0_CR)==_CR_RECV}")
# ── T5: RX MACRAW with ring wraparound (RX_RD near 2 KB boundary) ─────
model_mem[_S0_CR] = 0x00
RX_FRAME2 = [0x51, 0x52, 0x53, 0x54, 0x55]
# rx_rd = 0x07FD: [len_hi@7FD][len_lo@7FE][f0@7FF][f1@000][f2@001]...
got2, plen2 = await do_rx(ctx, 0x07FD, RX_FRAME2)
if got2 != RX_FRAME2:
errors.append(f"T5 RX wrap frame mismatch: {['0x%02X'%b for b in got2]} != "
f"{['0x%02X'%b for b in RX_FRAME2]}")
new_rd2 = (model_mem.get(_S0_RX_RD, 0) << 8) | model_mem.get(_S0_RX_RD + 1, 0)
want_rd2 = (0x07FD + plen2) & 0xFFFF
if new_rd2 != want_rd2:
errors.append(f"T5 RX wrap RX_RD: got 0x{new_rd2:04X}, want 0x{want_rd2:04X}")
print(f"T5 RX wrap: frame={['0x%02X'%b for b in got2]} "
f"RX_RD=0x{new_rd2:04X} (want 0x{want_rd2:04X})")
sim = Simulator(dut)
sim.add_clock(Period(MHz=24), domain="sync")
sim.add_testbench(testbench)
sim.add_process(w5100_model)
sim.run()
# ── UDP-test path (configurable socket, enable_udp_test=True) ───────────
# Reuses the same address-agnostic bus model (it reads/writes model_mem at
# whatever IDM_AR is set), so it transparently covers the IP-config and
# socket registers. Verifies: init programs GAR/SUBR/SIPR + opens the
# socket in UDP mode; a udp_send_req writes Sn_DIPR to the runtime dest,
# streams the payload into the socket TX buffer, advances Sn_TX_WR, issues
# SEND; and a udp_rx_req reads a datagram (header + payload) back out.
SRC_IP, SUBNET, GATEWAY = "10.0.0.7", "255.255.255.0", "10.0.0.1"
DEF_DST, DST_PORT, SRC_PORT = "192.168.1.100", 6464, 40000
UDP_SOCK = 3
S = _socket_addrs(UDP_SOCK)
dut2 = W5100ParallelMaster(
strobe_cycles=3, reset_cycles=10, enable_udp_test=True,
udp_socket=UDP_SOCK, src_ip=SRC_IP, subnet=SUBNET, gateway=GATEWAY,
dst_ip=DEF_DST, src_port=SRC_PORT, dst_port=DST_PORT)
writes2, model_mem2 = [], {}
async def w5100_model2(ctx):
idm_ar = 0
mr = 0
prev_wr = prev_rd = 1
async for vals in ctx.tick("sync").sample(
dut2.cs_n, dut2.rd_n, dut2.wr_n,
dut2.bus_addr, dut2.bus_data_o):
cs, rd, wr, a, do = vals[-5:]
ai = (mr >> 1) & 1
if cs == 0 and rd == 0:
if a == _A_MR: val = mr
elif a == _A_AR0: val = (idm_ar >> 8) & 0xFF
elif a == _A_AR1: val = idm_ar & 0xFF
else: val = model_mem2.get(idm_ar, 0)
ctx.set(dut2.bus_data_i, val)
if cs == 0 and prev_wr == 0 and wr == 1:
if a == _A_MR:
mr = do; writes2.append(("MR", do))
elif a == _A_AR0: idm_ar = (idm_ar & 0x00FF) | (do << 8)
elif a == _A_AR1: idm_ar = (idm_ar & 0xFF00) | do
else:
model_mem2[idm_ar] = do; writes2.append((idm_ar, do))
if ai: idm_ar = (idm_ar + 1) & 0xFFFF
if cs == 0 and prev_rd == 0 and rd == 1 and a == _A_DR and ai:
idm_ar = (idm_ar + 1) & 0xFFFF
prev_wr, prev_rd = wr, rd
def _u16(addr):
return (model_mem2.get(addr, 0) << 8) | model_mem2.get(addr + 1, 0)
async def testbench2(ctx):
ctx.set(dut2.par, PAR)
ctx.set(dut2.udp_pl_valid, 0)
await ctx.tick("sync").repeat(2)
# U1: init → IP stack + socket-1 UDP open.
ctx.set(dut2.init_req, 1)
await ctx.tick("sync").repeat(1)
ctx.set(dut2.init_req, 0)
done = False
for _ in range(6000):
await ctx.tick("sync").repeat(1)
if ctx.get(dut2.init_done):
done = True; break
if not done:
errors.append("UDP: init_done never asserted")
checks = {
_GAR0: _ip_bytes(GATEWAY), _SUBR0: _ip_bytes(SUBNET),
_SIPR0: _ip_bytes(SRC_IP),
}
for base, octets in checks.items():
got = [model_mem2.get(base + i) for i in range(4)]
if got != octets:
errors.append(f"UDP init {base:#06x}: got {got}, want {octets}")
if model_mem2.get(S["MR"]) != _S1_MR_UDP:
errors.append(f"UDP init Sn_MR: got {model_mem2.get(S['MR'])}, want UDP")
if _u16(S["PORT"]) != SRC_PORT:
errors.append(f"UDP init Sn_PORT: {_u16(S['PORT'])} != {SRC_PORT}")
if _u16(S["DPORT"]) != DST_PORT:
errors.append(f"UDP init Sn_DPORT: {_u16(S['DPORT'])} != {DST_PORT}")
if (S["CR"], _CR_OPEN) not in writes2:
errors.append("UDP init: socket OPEN not issued")
print(f"U1 init (socket {UDP_SOCK}): GAR/SUBR/SIPR set, Sn_MR=UDP, "
f"Sn_PORT={_u16(S['PORT'])} Sn_DPORT={_u16(S['DPORT'])} OPEN issued")
# U2: send one datagram to a runtime dest IP with a payload.
DST = "192.168.1.55"
PAYLOAD = list(b"HELLO-ETH")
ctx.set(dut2.udp_dst_ip, int.from_bytes(bytes(_ip_bytes(DST)), "big"))
idx = 0
ctx.set(dut2.udp_pl_data, PAYLOAD[0])
ctx.set(dut2.udp_pl_last, 1 if len(PAYLOAD) == 1 else 0)
ctx.set(dut2.udp_pl_valid, 1)
ctx.set(dut2.udp_send_req, 1)
await ctx.tick("sync").repeat(1)
ctx.set(dut2.udp_send_req, 0)
sent = False
for _ in range(4000):
await ctx.tick("sync").repeat(1)
if ctx.get(dut2.udp_pl_ready) and ctx.get(dut2.udp_pl_valid):
idx += 1
if idx < len(PAYLOAD):
ctx.set(dut2.udp_pl_data, PAYLOAD[idx])
ctx.set(dut2.udp_pl_last, 1 if idx == len(PAYLOAD) - 1 else 0)
else:
ctx.set(dut2.udp_pl_valid, 0)
if model_mem2.get(S["CR"]) == _CR_SEND:
sent = True; break
ctx.set(dut2.udp_pl_valid, 0)
if not sent:
errors.append("UDP send: SEND command never issued")
for _ in range(50):
await ctx.tick("sync").repeat(1)
if ctx.get(dut2.udp_test_busy) == 0:
break
dip = [model_mem2.get(S["DIPR"] + i) for i in range(4)]
if dip != _ip_bytes(DST):
errors.append(f"UDP send DIPR: got {dip}, want {_ip_bytes(DST)}")
buf = [model_mem2.get(S["TX_BASE"] + i) for i in range(len(PAYLOAD))]
if buf != PAYLOAD:
errors.append(f"UDP send payload: {buf} != {PAYLOAD}")
if _u16(S["TX_WR"]) != len(PAYLOAD):
errors.append(f"UDP send Sn_TX_WR: {_u16(S['TX_WR'])} != {len(PAYLOAD)}")
print(f"U2 send: DIPR={dip} payload={bytes(buf)!r} "
f"Sn_TX_WR={_u16(S['TX_WR'])} SEND={sent}")
# ── U3: receive a UDP datagram from the socket RX buffer ─────────────
# WIZnet UDP RX format: [srcIP(4)][srcPort(2)][len(2)][payload].
model_mem2[S["CR"]] = 0x00
SRC = _ip_bytes("192.168.1.9")
SPORT = 6464
RPAYLOAD = list(b"WORLD!")
rlen = len(RPAYLOAD)
rx_rd0 = 0x0000
hdr = SRC + [SPORT >> 8, SPORT & 0xFF, (rlen >> 8) & 0xFF, rlen & 0xFF]
packet = hdr + RPAYLOAD
for i, b in enumerate(packet):
model_mem2[S["RX_BASE"] + ((rx_rd0 + i) & _SN_MASK)] = b
total = len(packet)
model_mem2[S["RX_RSR"]] = (total >> 8) & 0xFF
model_mem2[S["RX_RSR"] + 1] = total & 0xFF
model_mem2[S["RX_RD"]] = (rx_rd0 >> 8) & 0xFF
model_mem2[S["RX_RD"] + 1] = rx_rd0 & 0xFF
ctx.set(dut2.udp_rx_ready, 1)
rx_got = []
got_none = [False]
ctx.set(dut2.udp_rx_req, 1)
# hold req until the master goes busy, then drop it
for _ in range(200):
await ctx.tick("sync").repeat(1)
if ctx.get(dut2.udp_rx_busy):
ctx.set(dut2.udp_rx_req, 0); break
recvd = False
for _ in range(3000):
await ctx.tick("sync").repeat(1)
if ctx.get(dut2.udp_rx_valid) and ctx.get(dut2.udp_rx_ready):
rx_got.append(ctx.get(dut2.udp_rx_data))
if ctx.get(dut2.udp_rx_none):
got_none[0] = True
if model_mem2.get(S["CR"]) == _CR_RECV:
recvd = True; break
for _ in range(30):
await ctx.tick("sync").repeat(1)
ctx.set(dut2.udp_rx_ready, 0)
if rx_got != RPAYLOAD:
errors.append(f"U3 RX payload: {bytes(rx_got)!r} != {bytes(RPAYLOAD)!r}")
new_rd = _u16(S["RX_RD"])
if new_rd != rx_rd0 + total:
errors.append(f"U3 RX_RD advance: got {new_rd}, want {rx_rd0 + total}")
if model_mem2.get(S["CR"]) != _CR_RECV:
errors.append("U3 RX: RECV command not issued")
if got_none[0]:
errors.append("U3 RX: udp_rx_none pulsed despite data present")
print(f"U3 recv: payload={bytes(rx_got)!r} RX_RD={new_rd} "
f"RECV={model_mem2.get(S['CR'])==_CR_RECV}")
# ── U4: RX poll with an empty buffer → udp_rx_none, no stream ────────
model_mem2[S["RX_RSR"]] = 0
model_mem2[S["RX_RSR"] + 1] = 0
model_mem2[S["CR"]] = 0x00
none_seen = False
streamed = False
ctx.set(dut2.udp_rx_req, 1)
for _ in range(200):
await ctx.tick("sync").repeat(1)
if ctx.get(dut2.udp_rx_busy):
ctx.set(dut2.udp_rx_req, 0); break
for _ in range(200):
await ctx.tick("sync").repeat(1)
if ctx.get(dut2.udp_rx_none):
none_seen = True
if ctx.get(dut2.udp_rx_valid):
streamed = True
if none_seen and not ctx.get(dut2.udp_rx_busy):
break
if not none_seen:
errors.append("U4 RX-empty: udp_rx_none never pulsed")
if streamed:
errors.append("U4 RX-empty: streamed data despite empty buffer")
if model_mem2.get(S["CR"]) == _CR_RECV:
errors.append("U4 RX-empty: RECV issued on empty buffer")
print(f"U4 recv-empty: none={none_seen} streamed={streamed}")
sim2 = Simulator(dut2)
sim2.add_clock(Period(MHz=24), domain="sync")
sim2.add_testbench(testbench2)
sim2.add_process(w5100_model2)
sim2.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