Roflin/rebbarb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add integrated UART UDP bring-up shell (socket 3) + capture critical-path fix

Browse Source 
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>
This commit is contained in:
Dennis Brentjes
2026-08-02 14:36:47 +00:00
parent a72e4ab1d4
commit c16afb6eea
6 changed files with 1458 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CLAUDE.md
+58 -11
View File
@@ -78,22 +78,33 @@ All modules elaborate without errors and pass their unit tests. The full design
synthesizes, places and routes on the iCE40UP5K, but **capture-domain timing is synthesizes, places and routes on the iCE40UP5K, but **capture-domain timing is
seed-dependent and only closes on a minority of seeds** — you MUST sweep. seed-dependent and only closes on a minority of seeds** — you MUST sweep.
Measured 2026-07-31, full build (`BBATopSynth(status_panel=True, Measured 2026-08, DEFAULT build (`BBATopSynth(status_panel=True,
uart_console=True)`, 47% LC), `python -m exi_bba.synth --seeds 8`: uart_shell=True)` — the interactive UART shell + socket-3 UDP test, ~67% LC),
`python -m exi_bba.synth --seeds 8`:
| domain | target | result | | domain | target | result |
|---|---|---| |---|---|---|
| `clk` (exi/sync) | 24 MHz | 35.4–38.4 MHz — **PASS on every seed** | | `clk` (exi/sync) | 24 MHz | 27–30 MHz — **PASS on every seed** |
| `capture_clk` | 54.02 MHz | 49.7–55.7 MHz — **PASS on only 2 of 8 seeds** (3 and 6) | | `capture_clk` | 54.02 MHz | 49.6–58.4 MHz — **PASS on 4 of 8 seeds** (3, 4, 7, 8) |
Best seed 3 = 55.69 MHz, margin +1.67 MHz (3%). Seed 1 — the default when you Best seed 4 = 58.36 MHz, margin +4.3 MHz (8%). Still seed-dependent — `python -m
run without `--seeds` — **FAILS at 53.08 MHz**. So `python -m exi_bba.synth` exi_bba.synth` with no arguments (seed 1) FAILS capture (50.96 MHz); always pass
with no arguments produces a bitstream that does not meet timing; always pass
`--seeds 8` (or more) and flash the reported best seed from `--seeds 8` (or more) and flash the reported best seed from
`build/seed<N>/top.bin`. `build/seed<N>/top.bin`.
There is essentially no margin: assume any added logic breaks capture timing **Capture-domain critical-path fix (2026-08):** capture used to close on only
until a sweep proves otherwise. The earlier "~70 MHz, both PASS" figure in this 2 of 8 seeds at ~55 MHz *without* the shell. The binding path was the TX byte
FIFO's read-enable gated by its own gray-coded ready
(`tx_fifo.r_en = ... | (flushing & r_rdy)`) — a `consume_ptr → gray → r_rdy →
flush → r_en → consume_ptr` loop in `exi_capture.py`. Replacing the r_rdy-based
"drain until empty" flush with a FIXED-length drain counter (the FIFO is only
`tx_depth` deep) removed the pointer feedback from `r_en`, dropping the path
from 24.3 → 19.6 ns and lifting capture to 4/8 passing (best 58.4) EVEN with the
shell integrated. **Do not re-introduce any `tx_fifo.r_rdy` dependence into
`r_en` or the flush deassert** — see the comment in `exi_capture.py`.
There is little margin: assume added logic in (or near) the capture domain may
break capture timing until a sweep proves otherwise. The earlier "~70 MHz, both PASS" figure in this
file was wrong — it came from a sweep that silently never ran (see the file was wrong — it came from a sweep that silently never ran (see the
`_prepared` note in `synth.py`) and from misreading nextpnr's PRE-routing `_prepared` note in `synth.py`) and from misreading nextpnr's PRE-routing
placement estimate, which runs ~8 MHz optimistic. placement estimate, which runs ~8 MHz optimistic.
@@ -109,9 +120,11 @@ placement estimate, which runs ~8 MHz optimistic.
| `SPRAMArbiter` | `exi_bba/spram_arbiter.py` | ✅ 3 tests | | `SPRAMArbiter` | `exi_bba/spram_arbiter.py` | ✅ 3 tests |
| `RXFrameAssembler` | `exi_bba/rx_frame_assembler.py` | ✅ 3 tests | | `RXFrameAssembler` | `exi_bba/rx_frame_assembler.py` | ✅ 3 tests |
| `TXFrameDrain` | `exi_bba/tx_frame_drain.py` | ✅ 2 tests | | `TXFrameDrain` | `exi_bba/tx_frame_drain.py` | ✅ 2 tests |
| `W5100ParallelMaster` | `exi_bba/w5100_parallel_master.py` | ✅ 5 tests (init/TX/RX vs bus model, incl. ring wrap) — **default eth back-end** | | `W5100ParallelMaster` | `exi_bba/w5100_parallel_master.py` | ✅ MACRAW init/TX/RX (T1–T5) + socket-N UDP send/receive (U1–U4) vs bus model, incl. ring wrap — **default eth back-end** |
| `W5500SPIMaster` | `exi_bba/w5500_spi_master.py` | ✅ init/TX/RX vs SPI-slave model (alt back-end) | | `W5500SPIMaster` | `exi_bba/w5500_spi_master.py` | ✅ init/TX/RX vs SPI-slave model (alt back-end) |
| `StatusPanel` | `exi_bba/status_panel.py` | ✅ 6 tests (heartbeat, stretched activity LEDs, debounced buttons, freeze) | | `StatusPanel` | `exi_bba/status_panel.py` | ✅ 6 tests (heartbeat, stretched activity LEDs, debounced buttons, freeze) |
| `UARTConsole` | `exi_bba/uart_console.py` | ✅ 7 tests (event log + 'r' reinit) — event-logger console (`--console` build) |
| `UARTShell` | `exi_bba/uart_shell.py` | ✅ 8 tests (help/unicast/broadcast/reply-print/timeout/badip/backspace) — interactive UDP bring-up shell (**default build**) |
| `EEPROMModel` | `exi_bba/eeprom_model.py` | ✅ 4 tests | | `EEPROMModel` | `exi_bba/eeprom_model.py` | ✅ 4 tests |
**Bring-up status panel (optional):** `BBATop(status_panel=True)` adds a **Bring-up status panel (optional):** `BBATop(status_panel=True)` adds a
@@ -140,13 +153,47 @@ python -m exi_bba.bba_register_file
python -m exi_bba.spram_arbiter python -m exi_bba.spram_arbiter
python -m exi_bba.rx_frame_assembler python -m exi_bba.rx_frame_assembler
python -m exi_bba.tx_frame_drain python -m exi_bba.tx_frame_drain
python -m exi_bba.w5100_parallel_master # 5 tests: init, TX(+wrap), RX(+wrap) python -m exi_bba.w5100_parallel_master # T1-5 MACRAW + U1-4 socket-N UDP tx/rx
python -m exi_bba.w5500_spi_master python -m exi_bba.w5500_spi_master
python -m exi_bba.status_panel # 6 tests: heartbeat/activity/buttons python -m exi_bba.status_panel # 6 tests: heartbeat/activity/buttons
python -m exi_bba.uart_console # 7 tests: event log + 'r' reinit
python -m exi_bba.uart_shell # 8 tests: shell cmds + reply/timeout
python -m exi_bba.eeprom_model python -m exi_bba.eeprom_model
python -m exi_bba.bba_top # end-to-end EXI integration test (W5100 RX loop) python -m exi_bba.bba_top # end-to-end EXI integration test (W5100 RX loop)
``` ```
### UART shell + UDP bring-up test (default flash build)
`BBATop(uart_shell=True)` adds an **interactive UART command shell** that shares
the FT2232H channel-B UART pins with (and replaces) the event-log console, and
drives a **UDP send/receive test on a second W5100 socket** (default socket 3)
alongside the BBA's socket-0 MACRAW path. `synth.py` builds it **by default**
(pass `--console` for the old event-logger instead). EXI/BBA keeps priority: the
UDP test is the lowest-priority branch in the W5100 master's bus arbiter, so a
GC frame always preempts a queued send/poll.
Shell (115200 8N1, `rebbarb> ` prompt):
- `help` — list commands
- `udp unicast <ip> [msg]` — send a UDP datagram to `<ip>:dst_port` (the W5100
ARPs the target), then wait (bounded) for a reply
- `udp broadcast [msg]` — same to 255.255.255.255 (limited broadcast; no ARP)
- after a send, the shell prints `sent`, waits up to `reply_timeout_cycles`
(~1 s) for a UDP reply on the socket, and prints `rx <payload>` or `timeout`
(so it never hangs). Example: `udp broadcast Hello World` → a responder
unicasts back → `rx <reply>` on the console.
Config is build-time via `synth.py` env vars (the board's own IP identity —
match your LAN): `UDP_SOCKET` (1–3, default 3), `UDP_SRC_IP`, `UDP_SUBNET`,
`UDP_GATEWAY`, `UDP_DST_IP`, `UDP_SRC_PORT`, `UDP_DST_PORT`. Receive on the PC
with `nc -ul <port>` or `sudo tcpdump -n udp port <port>`.
**MACRAW + UDP coexistence is datasheet-confirmed** (W5100S §4.6: *"MACRAW Mode
SOCKET 0 does not receive any Data Packet for other SOCKET"*; feature list:
*"Support 4 independent SOCKETs simultaneously"*). All 4 sockets get a 2 KB
RX + 2 KB TX buffer at reset (`RMSR/TMSR=0x55`), so socket-3 frames are stored
in the chip independently — no extra FPGA buffer needed. Register addresses are
still datasheet-from-memory — **confirm at hardware bring-up**.
### Pending work ### Pending work
- **Synthesis/timing**: ⚠️ partially — synthesizes and P&Rs, `clk` closes with - **Synthesis/timing**: ⚠️ partially — synthesizes and P&Rs, `clk` closes with
wide margin, but `capture_clk` closes on only **2 of 8 seeds** and the default wide margin, but `capture_clk` closes on only **2 of 8 seeds** and the default
exi_bba/bba_top.py
+49 -3
View File
@@ -24,6 +24,7 @@ from exi_bba.w5500_spi_master import W5500SPIMaster
from exi_bba.w5100_parallel_master import W5100ParallelMaster from exi_bba.w5100_parallel_master import W5100ParallelMaster
from exi_bba.status_panel import StatusPanel from exi_bba.status_panel import StatusPanel
from exi_bba.uart_console import UARTConsole from exi_bba.uart_console import UARTConsole
from exi_bba.uart_shell import UARTShell
from amaranth.lib.cdc import FFSynchronizer from amaranth.lib.cdc import FFSynchronizer
@@ -48,7 +49,11 @@ class BBATop(Elaboratable):
""" """
def __init__(self, eth="w5100", reset_cycles=24000, def __init__(self, eth="w5100", reset_cycles=24000,
status_panel=False, uart_console=False): status_panel=False, uart_console=False, uart_shell=False,
udp_socket=3, udp_src_ip="192.168.1.123",
udp_subnet="255.255.255.0", udp_gateway="192.168.1.1",
udp_dst_ip="192.168.1.100",
udp_src_port=40000, udp_dst_port=6464):
# Ethernet back-end: "w5100" (indirect parallel bus, reaches the EXI # Ethernet back-end: "w5100" (indirect parallel bus, reaches the EXI
# ceiling) or "w5500" (SPI, ~12 Mbit/s). Both expose the identical # ceiling) or "w5500" (SPI, ~12 Mbit/s). Both expose the identical
# tx/rx/init/par interface, so only the physical pins differ. # tx/rx/init/par interface, so only the physical pins differ.
@@ -64,6 +69,19 @@ class BBATop(Elaboratable):
# uart_rx ← FT2232H Channel B (net UART_TXD, pin 18) — net names are # uart_rx ← FT2232H Channel B (net UART_TXD, pin 18) — net names are
# FTDI-perspective, so this is not a naming mismatch, see synth.py. # FTDI-perspective, so this is not a naming mismatch, see synth.py.
self._uart_console = uart_console self._uart_console = uart_console
# Optional interactive UART command shell (bring-up UDP test on the
# W5100's socket 1). Shares the UART pins with the console, so only one
# may be enabled; the UDP path needs the W5100 back-end. See uart_shell.
self._uart_shell = uart_shell
if uart_console and uart_shell:
raise ValueError("uart_console and uart_shell share the UART pins; "
"enable only one")
if uart_shell and eth != "w5100":
raise ValueError("uart_shell UDP test requires eth='w5100'")
self._udp_cfg = dict(udp_socket=udp_socket, src_ip=udp_src_ip,
subnet=udp_subnet, gateway=udp_gateway,
dst_ip=udp_dst_ip, src_port=udp_src_port,
dst_port=udp_dst_port)
# EXI (GC side) # EXI (GC side)
self.exi_clk = Signal(init=1) self.exi_clk = Signal(init=1)
@@ -99,7 +117,7 @@ class BBATop(Elaboratable):
self.panel_led = Signal(5) # to onboard LEDs (see StatusPanel) self.panel_led = Signal(5) # to onboard LEDs (see StatusPanel)
self.panel_btn = Signal(3) # from onboard button(s) self.panel_btn = Signal(3) # from onboard button(s)
if uart_console: if uart_console or uart_shell:
self.uart_tx = Signal(init=1) # FPGA → PC (FT2232H Channel B) self.uart_tx = Signal(init=1) # FPGA → PC (FT2232H Channel B)
self.uart_rx = Signal(init=1) # PC → FPGA self.uart_rx = Signal(init=1) # PC → FPGA
@@ -161,7 +179,9 @@ class BBATop(Elaboratable):
drain = TXFrameDrain() drain = TXFrameDrain()
eth = (W5500SPIMaster(reset_cycles=self._reset_cycles) eth = (W5500SPIMaster(reset_cycles=self._reset_cycles)
if self._eth == "w5500" if self._eth == "w5500"
else W5100ParallelMaster(reset_cycles=self._reset_cycles)) else W5100ParallelMaster(reset_cycles=self._reset_cycles,
enable_udp_test=self._uart_shell,
**self._udp_cfg))
m.submodules.cap = cap m.submodules.cap = cap
m.submodules.reg = reg m.submodules.reg = reg
@@ -335,6 +355,32 @@ class BBATop(Elaboratable):
console.uart_rx .eq(self.uart_rx), console.uart_rx .eq(self.uart_rx),
] ]
if self._uart_shell:
# Interactive command shell driving the W5100 socket-1 UDP test.
# `eth` is the W5100 master built with enable_udp_test above.
shell = UARTShell()
m.submodules.shell = shell
m.d.comb += [
self.uart_tx .eq(shell.uart_tx),
shell.uart_rx .eq(self.uart_rx),
# send
eth.udp_send_req .eq(shell.udp_send_req),
eth.udp_dst_ip .eq(shell.udp_dst_ip),
eth.udp_pl_data .eq(shell.udp_pl_data),
eth.udp_pl_valid .eq(shell.udp_pl_valid),
eth.udp_pl_last .eq(shell.udp_pl_last),
shell.udp_pl_ready .eq(eth.udp_pl_ready),
shell.udp_test_busy .eq(eth.udp_test_busy),
# receive (reply)
eth.udp_rx_req .eq(shell.udp_rx_req),
shell.udp_rx_busy .eq(eth.udp_rx_busy),
shell.udp_rx_none .eq(eth.udp_rx_none),
shell.udp_rx_data .eq(eth.udp_rx_data),
shell.udp_rx_valid .eq(eth.udp_rx_valid),
shell.udp_rx_eof .eq(eth.udp_rx_eof),
eth.udp_rx_ready .eq(shell.udp_rx_ready),
]
if need_ready: if need_ready:
with m.If(eth.init_done): with m.If(eth.init_done):
m.d.sync += ready.eq(1) m.d.sync += ready.eq(1)
exi_bba/exi_capture.py
+19 -7
View File
@@ -148,14 +148,26 @@ class ExiCapture(Elaboratable):
# clock for DMA reads, so when CS deasserts mid-stream a few unsent # clock for DMA reads, so when CS deasserts mid-stream a few unsent
# bytes remain. On CS-fall (frame_start) drain tx_fifo to empty before # bytes remain. On CS-fall (frame_start) drain tx_fifo to empty before
# the new transaction's data phase, so stale bytes never reach MISO. # the new transaction's data phase, so stale bytes never reach MISO.
flushing = Signal() # The flush must NOT read `tx_fifo.r_rdy`. r_rdy is derived from the
m.d.comb += tx_fifo.r_en.eq( # gray-coded FIFO pointers that `r_en` advances, so any r_rdy → r_en
(spi.tx_load & (txld_cnt >= 2)) | (flushing & tx_fifo.r_rdy) # dependence (either gating the drain with `& r_rdy`, or deasserting a
) # `flushing` flag on `~r_rdy`) closes a long capture-domain loop
# (consume_ptr → gray → r_rdy → flush → r_en → consume_ptr) — the
# measured critical path capping capture_clk. Instead, drain for a
# FIXED number of cycles: the FIFO is only `tx_depth` deep, so pulsing
# r_en for `tx_depth + 1` cycles empties it regardless of occupancy
# (the FIFO advances its read pointer only on r_en & r_rdy internally,
# so pulses past empty are harmless). This removes the pointer
# feedback from r_en entirely; the only remaining r_en source is the
# legitimate data-byte pop.
drain = Signal(range(self._tx_depth + 2))
with m.If(spi.frame_start): with m.If(spi.frame_start):
m.d.capture += flushing.eq(1) m.d.capture += drain.eq(self._tx_depth + 1)
with m.Elif(~tx_fifo.r_rdy): with m.Elif(drain != 0):
m.d.capture += flushing.eq(0) m.d.capture += drain.eq(drain - 1)
m.d.comb += tx_fifo.r_en.eq(
(spi.tx_load & (txld_cnt >= 2)) | (drain != 0)
)
with m.If(spi.frame_start): with m.If(spi.frame_start):
m.d.capture += txld_cnt.eq(0) m.d.capture += txld_cnt.eq(0)
exi_bba/synth.py
+52 -7
View File
@@ -62,6 +62,20 @@ from exi_bba.bba_top import BBATop
# iCE40UP5K's dedicated SB_RGBA_DRV pins 39/40/41 — fixed by the chip # iCE40UP5K's dedicated SB_RGBA_DRV pins 39/40/41 — fixed by the chip
# package on any board, not board-specific, so no resource needed here. # package on any board, not board-specific, so no resource needed here.
# nextpnr P&R options. The binding constraint is the isolated 54 MHz capture
# domain (the SPI Mode-3 bit engine); the 24 MHz sync domain has wide margin.
#
# TRIED (2026-08 shell build) and REVERTED — prioritising routing toward the
# critical capture paths did NOT help: `--router router2 --tmg-ripup
# --placer-heap-timingweight 30 --placer-heap-critexp 4` left capture at
# 41-49 MHz (best 49.07, still < 54.02) across seeds AND eroded the slow-clock
# margin (24.4-25.9 MHz vs the 28-33 that plain --opt-timing gives). The
# capture shortfall is congestion + inherent path delay at this LC level, not a
# routing-priority tuning problem, so weighting P&R toward it only robs the
# sync domain. Plain --opt-timing is the better baseline.
_PNR_TIMING_OPTS = "--opt-timing"
class IceBreakerPlatform(LatticeICE40Platform): class IceBreakerPlatform(LatticeICE40Platform):
device = "iCE40UP5K" device = "iCE40UP5K"
package = "SG48" package = "SG48"
@@ -185,8 +199,10 @@ class BBATopSynth(BBATop):
o_RGB2=Signal(name="rgb_b"), o_RGB2=Signal(name="rgb_b"),
) )
# ── UART debug console → FT2232H Channel B ───────────────────── # ── UART debug console/shell → FT2232H Channel B ───────────────
if self._uart_console: # Both the event-log console and the interactive command shell use
# the same two UART pins (only one may be enabled at a time).
if self._uart_console or self._uart_shell:
uart = platform.request("uart", 0) uart = platform.request("uart", 0)
m.d.comb += [ m.d.comb += [
uart.tx.o .eq(self.uart_tx), uart.tx.o .eq(self.uart_tx),
@@ -205,13 +221,42 @@ class BBATopSynth(BBATop):
# build/top.bin is the result of the last (or best) seed tried. # build/top.bin is the result of the last (or best) seed tried.
if __name__ == "__main__": if __name__ == "__main__":
import os
do_flash = "--flash" in sys.argv do_flash = "--flash" in sys.argv
# The interactive UART shell + UDP bring-up test is the default build;
# pass --console to flash the old event-log console instead (they share the
# UART pins, so only one can be built).
use_shell = "--console" not in sys.argv
n_seeds = next((int(sys.argv[i+1]) for i, a in enumerate(sys.argv) n_seeds = next((int(sys.argv[i+1]) for i, a in enumerate(sys.argv)
if a == "--seeds"), 1) if a == "--seeds"), 1)
# UDP bring-up test config (only used with --shell). Set these env vars to
# match your LAN — src_ip/subnet/gateway are the board's own identity for
# the W5100 IP stack; the unicast destination is typed at runtime.
udp_kw = dict(
udp_socket = int(os.environ.get("UDP_SOCKET", "3")),
udp_src_ip = os.environ.get("UDP_SRC_IP", "192.168.1.123"),
udp_subnet = os.environ.get("UDP_SUBNET", "255.255.255.0"),
udp_gateway = os.environ.get("UDP_GATEWAY", "192.168.1.1"),
udp_dst_ip = os.environ.get("UDP_DST_IP", "192.168.1.100"),
udp_src_port = int(os.environ.get("UDP_SRC_PORT", "40000")),
udp_dst_port = int(os.environ.get("UDP_DST_PORT", "6464")),
)
def make_dut():
if use_shell:
return BBATopSynth(status_panel=True, uart_shell=True, **udp_kw)
return BBATopSynth(status_panel=True, uart_console=True)
print(f"Synthesizing BBATop for {IceBreakerPlatform.device}-" print(f"Synthesizing BBATop for {IceBreakerPlatform.device}-"
f"{IceBreakerPlatform.package} " f"{IceBreakerPlatform.package} "
f"(do_program={do_flash}, seeds=1..{n_seeds})") f"(do_program={do_flash}, seeds=1..{n_seeds}, "
f"{'shell+UDP' if use_shell else 'console'})")
if use_shell:
print(f" UDP: socket {udp_kw['udp_socket']}, "
f"src {udp_kw['udp_src_ip']}:{udp_kw['udp_src_port']} "
f"gw {udp_kw['udp_gateway']} mask {udp_kw['udp_subnet']} "
f"dst-port {udp_kw['udp_dst_port']}")
best_seed = 1 best_seed = 1
best_fmax = 0.0 best_fmax = 0.0
@@ -220,7 +265,7 @@ if __name__ == "__main__":
print(f"\n{'='*60}") print(f"\n{'='*60}")
print(f" Seed {seed}/{n_seeds}") print(f" Seed {seed}/{n_seeds}")
print(f"{'='*60}") print(f"{'='*60}")
opts = (f"--opt-timing --seed {seed} --timing-allow-fail") opts = f"{_PNR_TIMING_OPTS} --seed {seed} --timing-allow-fail"
# A Platform instance can only be built ONCE — amaranth's # A Platform instance can only be built ONCE — amaranth's
# TemplatedPlatform.prepare() does `assert not self._prepared`. Reusing # TemplatedPlatform.prepare() does `assert not self._prepared`. Reusing
@@ -244,7 +289,7 @@ if __name__ == "__main__":
build_ok = True build_ok = True
try: try:
platform.build(BBATopSynth(status_panel=True, uart_console=True), do_program=False, platform.build(make_dut(), do_program=False,
verbose=True, nextpnr_opts=opts, build_dir=build_dir) verbose=True, nextpnr_opts=opts, build_dir=build_dir)
except Exception as exc: except Exception as exc:
# nextpnr exits non-zero even with --timing-allow-fail on some # nextpnr exits non-zero even with --timing-allow-fail on some
@@ -334,12 +379,12 @@ if __name__ == "__main__":
"miss bits. Re-run with more seeds (--seeds 16) or reduce logic.") "miss bits. Re-run with more seeds (--seeds 16) or reduce logic.")
elif do_flash: elif do_flash:
print(f"\nFlashing with seed {best_seed}...") print(f"\nFlashing with seed {best_seed}...")
opts = f"--opt-timing --seed {best_seed} --timing-allow-fail" opts = f"{_PNR_TIMING_OPTS} --seed {best_seed} --timing-allow-fail"
# Fresh platform again — the sweep above already consumed one per seed. # Fresh platform again — the sweep above already consumed one per seed.
# Reuse the best seed's own build dir so the flashed bitstream is the # Reuse the best seed's own build dir so the flashed bitstream is the
# one that was actually measured. # one that was actually measured.
IceBreakerPlatform().build( IceBreakerPlatform().build(
BBATopSynth(status_panel=True, uart_console=True), do_program=True, make_dut(), do_program=True,
verbose=True, nextpnr_opts=opts, build_dir=f"build/seed{best_seed}") verbose=True, nextpnr_opts=opts, build_dir=f"build/seed{best_seed}")
print("Done.") print("Done.")
exi_bba/uart_shell.py
+757
View File
@@ -0,0 +1,757 @@
"""Interactive UART command shell (sync domain, 24 MHz).
A tiny line-oriented console for bring-up over the FT2232H channel-B UART
(115200 8N1). Prints a ``rebbarb> `` prompt, echoes typed characters (with
backspace editing), and on Enter parses one command:
help list commands
udp unicast <ip> [msg] send a UDP datagram to <ip>:DST_PORT (W5100 ARPs)
udp broadcast [msg] send a UDP datagram to 255.255.255.255:DST_PORT
`msg` is optional; when omitted a built-in default payload is sent. The shell
drives the W5100 master's runtime UDP-send interface (`udp_send_req`,
`udp_dst_ip`, and the `udp_pl_*` payload stream) — it does not touch the MACRAW
BBA path.
Grammar notes (it's a debug shell, not bash):
* commands and keywords are case-insensitive;
* tokens are separated by exactly one space;
* the destination is parsed as a dotted IPv4 literal;
* everything after the address token (unicast) / keyword (broadcast) is the
raw payload, preserving case and embedded spaces, to end of line.
On the PC:
udp unicast → nc -ul 6464 (or: sudo tcpdump -n udp port 6464)
udp broadcast → nc -ul 6464 (limited broadcast reaches the segment)
"""
from amaranth import *
from amaranth.lib.cdc import FFSynchronizer
from amaranth.lib.memory import Memory
__all__ = ["UARTShell"]
# ── Printable message ROM ──────────────────────────────────────────────────
_MSGS = {
"BANNER": b"\r\nre-bba-rb ethernet test\r\n",
"PROMPT": b"rebbarb> ",
"HELP": (b"commands:\r\n"
b" help show this\r\n"
b" udp unicast <ip> [msg] send UDP to <ip>\r\n"
b" udp broadcast [msg] send UDP broadcast\r\n"),
"SENT": b"sent\r\n",
"ERR": b"? (try 'help')\r\n",
"BADIP": b"bad ip\r\n",
"BUSY": b"busy\r\n",
"CRLF": b"\r\n",
"BKSP": b"\b \b",
"RXPFX": b"rx ",
"TIMEOUT": b"timeout\r\n",
}
_ORDER = list(_MSGS)
_ROM = b"".join(_MSGS[k] for k in _ORDER)
_OFF = {}
_acc = 0
for _k in _ORDER:
_OFF[_k] = (_acc, _acc + len(_MSGS[_k]))
_acc += len(_MSGS[_k])
_DEFAULT_PAYLOAD = b"rebbarb-udp-test"
# return codes for the shared print routine
_RET_PROMPT = 0 # after printing, (re)issue the prompt then read input
_RET_INPUT = 1 # after printing, go straight back to reading input
_RET_PARSE = 2 # after printing (the CR echo), parse the line
_RET_RXWAIT = 3 # after printing, wait for a UDP reply (with timeout)
_RET_RXBODY = 4 # after printing "rx ", stream the reply payload
class UARTShell(Elaboratable):
def __init__(self, clk_freq=24_000_000, baud_rate=115_200,
default_payload=_DEFAULT_PAYLOAD, lbuf_len=64,
reply_timeout_cycles=24_000_000, poll_gap_cycles=24_000):
self._div = round(clk_freq / baud_rate)
self._defpl = list(default_payload)
self._lbuf_n = lbuf_len
# After a send, wait this many sync cycles for a reply before giving up
# (default ~1 s at 24 MHz) so the console never hangs; poll the socket
# RX every `poll_gap_cycles` (~1 ms) in between.
self._to_cyc = reply_timeout_cycles
self._gap_cyc = poll_gap_cycles
# UART pins
self.uart_tx = Signal(init=1)
self.uart_rx = Signal(init=1)
# W5100 UDP-send interface (drive the W5100ParallelMaster)
self.udp_send_req = Signal()
self.udp_dst_ip = Signal(32)
self.udp_pl_data = Signal(8)
self.udp_pl_valid = Signal()
self.udp_pl_last = Signal()
self.udp_pl_ready = Signal()
self.udp_test_busy = Signal()
# W5100 UDP-receive interface (poll for a reply after each send)
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_eof = Signal()
self.udp_rx_ready = Signal()
def elaborate(self, platform):
m = Module()
div = self._div
N = self._lbuf_n
# Message ROM lives in a block RAM (SB_RAM40_4K) instead of LUTs — the
# ~230-byte string table was the single biggest LUT consumer (a wide
# addressed mux). PRINT already streams one byte per UART-byte-time, so
# the block RAM's 1-cycle read latency is hidden (one prime cycle on
# entry). The tiny default payload stays in logic (cheap, and keeps the
# payload path combinational — no streaming-latency hazard).
m.submodules.rom_mem = rom_mem = Memory(
shape=unsigned(8), depth=len(_ROM), init=list(_ROM))
rom_rd = rom_mem.read_port() # synchronous (block RAM)
defrom = Array([Const(b, 8) for b in self._defpl])
DEF_LEN = len(self._defpl)
# ── UART TX (8N1) ─────────────────────────────────────────────────
tx_cnt = Signal(range(div))
tx_bits = Signal(range(11))
tx_shreg = Signal(10, init=0b1111111111)
tx_busy = Signal()
tx_load = Signal()
tx_byte = Signal(8)
m.d.comb += [tx_busy.eq(tx_bits != 0), self.uart_tx.eq(tx_shreg[0])]
with m.If(tx_load & ~tx_busy):
m.d.sync += [tx_shreg.eq(Cat(Const(0, 1), tx_byte, Const(1, 1))),
tx_bits.eq(10), tx_cnt.eq(div - 1)]
with m.Elif(tx_busy):
with m.If(tx_cnt == 0):
m.d.sync += [tx_shreg.eq(Cat(tx_shreg[1:], Const(1, 1))),
tx_bits.eq(tx_bits - 1), tx_cnt.eq(div - 1)]
with m.Else():
m.d.sync += tx_cnt.eq(tx_cnt - 1)
# ── UART RX (8N1, mid-bit sampling) ───────────────────────────────
rx_sync = Signal(init=1)
rx_prev = Signal(init=1)
rx_cnt = Signal(range(div))
rx_bits = Signal(range(9))
rx_shr = Signal(8)
rx_byte = Signal(8)
rx_valid = Signal()
m.submodules += FFSynchronizer(self.uart_rx, rx_sync, init=1)
m.d.sync += rx_valid.eq(0)
with m.FSM(name="rx"):
with m.State("IDLE"):
m.d.sync += rx_prev.eq(rx_sync)
with m.If(rx_prev & ~rx_sync):
m.d.sync += rx_cnt.eq(div // 2 - 1)
m.next = "START"
with m.State("START"):
with m.If(rx_cnt == 0):
with m.If(~rx_sync):
m.d.sync += [rx_cnt.eq(div - 1), rx_bits.eq(7)]
m.next = "DATA"
with m.Else():
m.next = "IDLE"
with m.Else():
m.d.sync += rx_cnt.eq(rx_cnt - 1)
with m.State("DATA"):
with m.If(rx_cnt == 0):
m.d.sync += [rx_shr.eq(Cat(rx_shr[1:], rx_sync)),
rx_cnt.eq(div - 1)]
with m.If(rx_bits == 0):
m.next = "STOP"
with m.Else():
m.d.sync += rx_bits.eq(rx_bits - 1)
with m.Else():
m.d.sync += rx_cnt.eq(rx_cnt - 1)
with m.State("STOP"):
with m.If(rx_cnt == 0):
with m.If(rx_sync):
m.d.sync += [rx_byte.eq(rx_shr), rx_valid.eq(1)]
m.next = "IDLE"
with m.Else():
m.d.sync += rx_cnt.eq(rx_cnt - 1)
# 1-deep RX holding register: decouples byte capture from the shell's
# echo/print activity so a character arriving while the shell is busy
# echoing (or printing a short response) isn't dropped. This keeps a
# back-to-back paste of a command line intact — the incoming and echo
# rates match, so at most one byte is ever in flight during an echo.
# (A byte arriving during a long print, e.g. 'help' output, can still
# be lost, but you don't type into a response.)
rx_pending = Signal()
rx_hold = Signal(8)
with m.If(rx_valid):
m.d.sync += [rx_hold.eq(rx_byte), rx_pending.eq(1)]
# ── Line buffer (block RAM) ───────────────────────────────────────
# Stored in an SB_RAM40 rather than flip-flops; the parser reads it
# SEQUENTIALLY (one byte per cycle) through a single registered read
# port. This is what lets the 64-byte buffer AND its wide read muxes
# leave the LC fabric (they were the dominant cost). `la` drives the
# read address; `lbuf_rd.data` is valid the cycle after `la` settles.
m.submodules.lbuf_mem = lbuf_mem = Memory(
shape=unsigned(8), depth=N, init=[])
lbuf_wr = lbuf_mem.write_port()
lbuf_rd = lbuf_mem.read_port()
llen = Signal(range(N + 1))
la = Signal(range(N))
m.d.comb += lbuf_rd.addr.eq(la)
m.d.comb += la.eq(0) # default; read states override per-cycle
# write port: addr/data always presented; a store pulses `en` in INPUT.
m.d.comb += [lbuf_wr.addr.eq(llen), lbuf_wr.data.eq(rx_hold),
lbuf_wr.en.eq(0)]
def low(c):
return Mux((c >= ord('A')) & (c <= ord('Z')), c | 0x20, c)
# Command templates, matched sequentially in the SCAN state. Padded to
# a common length so a runtime index is always in range (guarded by the
# per-template length so padding bytes are never actually compared).
_TH, _TU, _TB, _TA = "help", "udp unicast ", "udp broadcast", "udp bcast"
_TMAX = 14
def _tmpl(s):
return Array([Const(ord(c), 8) for c in s.ljust(_TMAX, "\x00")])
THarr, TUarr, TBarr, TAarr = _tmpl(_TH), _tmpl(_TU), _tmpl(_TB), _tmpl(_TA)
# ── Shared print routine ──────────────────────────────────────────
pr_ptr = Signal(range(len(_ROM) + 1))
pr_end = Signal(range(len(_ROM) + 1))
pr_ret = Signal(3)
to_ctr = Signal(range(self._to_cyc + 1)) # reply timeout countdown
gap = Signal(range(self._gap_cyc + 1)) # inter-poll gap countdown
# ROM read port is addressed by the print pointer (1-cycle latency).
m.d.comb += rom_rd.addr.eq(pr_ptr)
def start_print(msg_key, ret, nxt="PRINT_PRIME"):
s, e = _OFF[msg_key]
m.d.sync += [pr_ptr.eq(s), pr_end.eq(e), pr_ret.eq(ret)]
m.next = nxt
# ── Parse / send scratch ──────────────────────────────────────────
ip_b = Array([Signal(8, name=f"ip{i}") for i in range(4)])
ip_pos = Signal(range(N + 1))
octet = Signal(9)
noct = Signal(3)
kwend = Signal(range(N + 1))
use_def = Signal()
req_r = Signal()
ech = Signal(8) # character currently being echoed
# sequential command scan
si = Signal(range(_TMAX + 2)) # scan index
mh, muni, mbc, mbca = (Signal(init=1), Signal(init=1),
Signal(init=1), Signal(init=1))
dh, dbc, dbca = Signal(), Signal(), Signal() # delimiter-ok captures
# payload streaming (prefetch: `la` runs one byte ahead of pl_byte)
pl_base = Signal(range(N + 1))
pl_len = Signal(range(N + 1))
j = Signal(range(N + 1))
pl_byte = Signal(8)
m.d.comb += [self.udp_send_req.eq(req_r),
self.udp_dst_ip.eq(Cat(ip_b[3], ip_b[2], ip_b[1], ip_b[0])),
self.udp_pl_data.eq(pl_byte)]
tx_load_i = Signal()
tx_byte_i = Signal(8)
m.d.comb += [tx_load.eq(tx_load_i), tx_byte.eq(tx_byte_i)]
# ── Shell FSM ─────────────────────────────────────────────────────
with m.FSM(name="shell"):
with m.State("BOOT"):
start_print("BANNER", _RET_PROMPT)
# One prime cycle so the block RAM read (rom_rd.data = rom[pr_ptr])
# is valid before the first byte is loaded.
with m.State("PRINT_PRIME"):
m.next = "PRINT"
# Generic ROM printer → dispatch on pr_ret. rom_rd.data holds
# rom[pr_ptr] (address is comb-driven; stable across the tx-busy
# wait, so it is valid whenever ~tx_busy lets a byte load).
with m.State("PRINT"):
with m.If(~tx_busy):
with m.If(pr_ptr == pr_end):
with m.Switch(pr_ret):
with m.Case(_RET_PROMPT):
m.next = "PROMPT"
with m.Case(_RET_INPUT):
m.next = "INPUT"
with m.Case(_RET_PARSE):
m.next = "PARSE"
with m.Case(_RET_RXWAIT):
m.next = "RX_INIT"
with m.Case(_RET_RXBODY):
m.next = "RX_BODY"
with m.Else():
m.d.comb += [tx_load_i.eq(1), tx_byte_i.eq(rom_rd.data)]
m.d.sync += pr_ptr.eq(pr_ptr + 1)
with m.State("PROMPT"):
m.d.sync += llen.eq(0)
start_print("PROMPT", _RET_INPUT)
# Read + echo a line into lbuf until CR/LF.
with m.State("INPUT"):
with m.If(rx_pending):
m.d.sync += rx_pending.eq(0)
with m.If((rx_hold == 0x0D) | (rx_hold == 0x0A)):
start_print("CRLF", _RET_PARSE)
with m.Elif((rx_hold == 0x08) | (rx_hold == 0x7F)):
with m.If(llen != 0):
m.d.sync += llen.eq(llen - 1)
start_print("BKSP", _RET_INPUT)
with m.Elif((rx_hold >= 0x20) & (rx_hold < 0x7F)):
with m.If(llen != N):
m.d.comb += lbuf_wr.en.eq(1) # store at addr=llen
m.d.sync += [llen.eq(llen + 1), ech.eq(rx_hold)]
m.next = "ECHO"
# Echo one stored character.
with m.State("ECHO"):
with m.If(~tx_busy):
m.d.comb += [tx_load_i.eq(1), tx_byte_i.eq(ech)]
m.next = "INPUT"
# Decide which command the line holds — set up the sequential scan.
with m.State("PARSE"):
with m.If(llen == 0):
m.next = "PROMPT"
with m.Else():
m.d.sync += [si.eq(0), mh.eq(1), muni.eq(1),
mbc.eq(1), mbca.eq(1), dh.eq(0),
dbc.eq(0), dbca.eq(0)]
m.next = "SCAN_RD"
# Scan the first up-to-14 bytes, comparing each against all four
# command templates in parallel (one byte/cycle from block RAM).
with m.State("SCAN_RD"):
m.d.comb += la.eq(si) # address the current byte
m.next = "SCAN_USE"
with m.State("SCAN_USE"):
m.d.comb += la.eq(si)
c = lbuf_rd.data
lc = low(c)
inb = si < llen # byte position is within the line
with m.If(si < len(_TH)):
m.d.sync += mh.eq(mh & inb & (lc == THarr[si]))
with m.If(si == len(_TH)):
m.d.sync += dh.eq((si >= llen) | (c == ord(' ')))
with m.If(si < len(_TU)):
m.d.sync += muni.eq(muni & inb & (lc == TUarr[si]))
with m.If(si < len(_TB)):
m.d.sync += mbc.eq(mbc & inb & (lc == TBarr[si]))
with m.If(si == len(_TB)):
m.d.sync += dbc.eq((si >= llen) | (c == ord(' ')))
with m.If(si < len(_TA)):
m.d.sync += mbca.eq(mbca & inb & (lc == TAarr[si]))
with m.If(si == len(_TA)):
m.d.sync += dbca.eq((si >= llen) | (c == ord(' ')))
with m.If(si == len(_TB)): # scanned enough to decide
m.next = "EVAL"
with m.Else():
m.d.sync += si.eq(si + 1)
m.next = "SCAN_RD"
# Classify from the accumulated match/delimiter bits (same priority
# and semantics as the old parallel parser).
with m.State("EVAL"):
with m.If(mh & dh):
start_print("HELP", _RET_PROMPT)
with m.Elif(muni): # "udp unicast " prefix present
m.d.sync += [ip_pos.eq(len(_TU)), octet.eq(0), noct.eq(0),
ip_b[0].eq(0), ip_b[1].eq(0),
ip_b[2].eq(0), ip_b[3].eq(0)]
m.next = "IP_RD"
with m.Elif(mbc & dbc):
m.d.sync += [kwend.eq(len(_TB)),
ip_b[0].eq(0xFF), ip_b[1].eq(0xFF),
ip_b[2].eq(0xFF), ip_b[3].eq(0xFF)]
m.next = "BC_RD"
with m.Elif(mbca & dbca):
m.d.sync += [kwend.eq(len(_TA)),
ip_b[0].eq(0xFF), ip_b[1].eq(0xFF),
ip_b[2].eq(0xFF), ip_b[3].eq(0xFF)]
m.next = "BC_RD"
with m.Else():
start_print("ERR", _RET_PROMPT)
# Sequential dotted-quad IPv4 parse from lbuf[ip_pos ...].
with m.State("IP_RD"):
m.d.comb += la.eq(ip_pos)
m.next = "IP_USE"
with m.State("IP_USE"):
m.d.comb += la.eq(ip_pos)
c = lbuf_rd.data
with m.If((ip_pos == llen) | (c == ord(' '))):
# end of address token → commit final octet + payload bounds
with m.If((noct == 3) & (octet <= 255)):
m.d.sync += ip_b[3].eq(octet)
with m.If(ip_pos == llen):
m.d.sync += use_def.eq(1) # no payload
with m.Else(): # c == ' '
m.d.sync += [pl_base.eq(ip_pos + 1),
pl_len.eq(llen - (ip_pos + 1)),
use_def.eq((ip_pos + 1) >= llen)]
m.next = "SEND_SETUP"
with m.Else():
start_print("BADIP", _RET_PROMPT)
with m.Elif(c == ord('.')):
with m.If((noct < 3) & (octet <= 255)):
m.d.sync += [ip_b[noct].eq(octet), noct.eq(noct + 1),
octet.eq(0), ip_pos.eq(ip_pos + 1)]
m.next = "IP_RD"
with m.Else():
start_print("BADIP", _RET_PROMPT)
with m.Elif((c >= ord('0')) & (c <= ord('9'))):
m.d.sync += [octet.eq(octet * 10 + (c - ord('0'))),
ip_pos.eq(ip_pos + 1)]
m.next = "IP_RD"
with m.Else():
start_print("BADIP", _RET_PROMPT)
# Broadcast payload: everything after the keyword's trailing space.
with m.State("BC_RD"):
m.d.comb += la.eq(kwend)
m.next = "BC_USE"
with m.State("BC_USE"):
m.d.comb += la.eq(kwend)
with m.If((llen > kwend) & (lbuf_rd.data == ord(' '))):
m.d.sync += [pl_base.eq(kwend + 1),
pl_len.eq(llen - (kwend + 1)),
use_def.eq((kwend + 1) >= llen)]
with m.Else():
m.d.sync += use_def.eq(1)
m.next = "SEND_SETUP"
with m.State("SEND_SETUP"):
with m.If(self.udp_test_busy):
start_print("BUSY", _RET_PROMPT)
with m.Else():
with m.If(use_def):
m.d.sync += [pl_base.eq(0), pl_len.eq(DEF_LEN)]
m.d.sync += j.eq(0)
m.next = "PL_PRIME"
# Prime the prefetch register with payload byte 0 (block RAM read
# for typed payload; the default payload is combinational).
with m.State("PL_PRIME"):
m.d.comb += la.eq(pl_base) # fetch typed byte 0
m.next = "PL_PRIME2"
with m.State("PL_PRIME2"):
m.d.comb += la.eq(pl_base)
m.d.sync += pl_byte.eq(Mux(use_def, defrom[0], lbuf_rd.data))
m.next = "SEND_REQ"
# Hold the request until the W5100 acknowledges by going busy.
with m.State("SEND_REQ"):
m.d.comb += la.eq(pl_base + j + 1) # prefetch next byte
m.d.sync += req_r.eq(1)
with m.If(self.udp_test_busy):
m.d.sync += req_r.eq(0)
m.next = "SEND_STREAM"
# Feed payload bytes as the W5100 consumes them. `pl_byte` holds
# the current byte; `la` prefetches the next so it is ready by the
# time the master pulls it (block RAM 1-cycle latency hidden).
with m.State("SEND_STREAM"):
m.d.comb += la.eq(pl_base + j + 1)
m.d.comb += [self.udp_pl_valid.eq(1),
self.udp_pl_last.eq(j + 1 == pl_len)]
with m.If(self.udp_pl_ready):
with m.If(j + 1 == pl_len):
m.next = "SEND_WAIT"
with m.Else():
m.d.sync += [pl_byte.eq(Mux(use_def, defrom[j + 1],
lbuf_rd.data)),
j.eq(j + 1)]
with m.State("SEND_WAIT"):
with m.If(~self.udp_test_busy):
# datagram is out; now wait (bounded) for a reply.
start_print("SENT", _RET_RXWAIT)
# ── Wait for a UDP reply on the socket, with a timeout ────────────
with m.State("RX_INIT"):
m.d.sync += to_ctr.eq(self._to_cyc)
m.next = "RX_POLL"
# Ask the W5100 to check the socket RX buffer.
with m.State("RX_POLL"):
m.d.comb += self.udp_rx_req.eq(1)
with m.If(self.udp_rx_busy):
m.next = "RX_WAIT"
# One of: a datagram streams (udp_rx_valid) or none (udp_rx_none).
with m.State("RX_WAIT"):
with m.If(self.udp_rx_valid):
start_print("RXPFX", _RET_RXBODY)
with m.Elif(self.udp_rx_none):
m.d.sync += gap.eq(self._gap_cyc)
m.next = "RX_GAP"
# Idle a bit between polls; count down the overall timeout.
with m.State("RX_GAP"):
with m.If(to_ctr == 0):
start_print("TIMEOUT", _RET_PROMPT)
with m.Elif(gap == 0):
m.next = "RX_POLL"
with m.Else():
m.d.sync += [gap.eq(gap - 1), to_ctr.eq(to_ctr - 1)]
# Stream the reply payload to the UART (throttled by tx_busy).
with m.State("RX_BODY"):
with m.If(self.udp_rx_valid & ~tx_busy):
m.d.comb += [tx_load_i.eq(1), tx_byte_i.eq(self.udp_rx_data),
self.udp_rx_ready.eq(1)]
with m.If(self.udp_rx_eof):
m.next = "RX_BODY_END"
with m.State("RX_BODY_END"):
start_print("CRLF", _RET_PROMPT)
return m
# ── Testbench ───────────────────────────────────────────────────────────────
if __name__ == "__main__":
import sys
from amaranth.sim import Simulator, Period
# Bit period is irrelevant to the shell logic, so use a tiny divisor to
# keep the simulation fast; the real build uses 24 MHz / 115200 (div≈208).
# Small reply timeout/gap so the timeout test finishes quickly.
CLK, BAUD = 24_000_000, 3_000_000
DIV = round(CLK / BAUD) # = 8 sim cycles per bit
dut = UARTShell(clk_freq=CLK, baud_rate=BAUD,
reply_timeout_cycles=4000, poll_gap_cycles=300)
errors = []
# ── UART line helpers (drive uart_rx, sample uart_tx) ──────────────────
async def send_byte(ctx, val):
ctx.set(dut.uart_rx, 0)
await ctx.tick().repeat(DIV)
for i in range(8):
ctx.set(dut.uart_rx, (val >> i) & 1)
await ctx.tick().repeat(DIV)
ctx.set(dut.uart_rx, 1)
await ctx.tick().repeat(DIV)
async def send_line(ctx, s):
for ch in s:
await send_byte(ctx, ord(ch))
await send_byte(ctx, 0x0D) # Enter
# Background collector: continuously samples uart_tx into a byte list.
tx_chars = bytearray()
async def tx_collector(ctx):
while True:
# wait for start bit
if ctx.get(dut.uart_tx) == 0:
await ctx.tick().repeat(DIV // 2)
if ctx.get(dut.uart_tx) != 0:
continue
b = 0
for i in range(8):
await ctx.tick().repeat(DIV)
b |= ctx.get(dut.uart_tx) << i
await ctx.tick().repeat(DIV) # stop
tx_chars.append(b)
else:
await ctx.tick()
# Background responder: models the W5100 UDP interface for BOTH directions.
# On udp_send_req it drains + records the payload (send handshake); on
# udp_rx_req it either streams a staged reply datagram or pulses udp_rx_none
# (nothing waiting) — mirroring the real master's poll-driven RX.
captured = {"payload": None, "dst": None, "count": 0}
staged = {"reply": None} # bytes to deliver on the next RX poll
async def w5100_udp_model(ctx):
while True:
await ctx.tick()
if ctx.get(dut.udp_send_req):
ctx.set(dut.udp_test_busy, 1)
dst = ctx.get(dut.udp_dst_ip)
await ctx.tick().repeat(6)
pl = bytearray(); last = False; guard = 0
while not last:
ctx.set(dut.udp_pl_ready, 1)
got = ctx.get(dut.udp_pl_valid)
if got:
pl.append(ctx.get(dut.udp_pl_data))
last = bool(ctx.get(dut.udp_pl_last))
await ctx.tick()
if got:
# Space consumes ~5 cycles apart, like the W5100's bus
# cycle — the shell's block-RAM prefetch needs ≥1 cycle
# between pulls to present the next byte.
ctx.set(dut.udp_pl_ready, 0)
await ctx.tick().repeat(4)
guard += 1
if guard > 5000: break
ctx.set(dut.udp_pl_ready, 0)
await ctx.tick().repeat(6)
captured["payload"] = bytes(pl)
captured["dst"] = [(dst >> 24) & 0xFF, (dst >> 16) & 0xFF,
(dst >> 8) & 0xFF, dst & 0xFF]
captured["count"] += 1
ctx.set(dut.udp_test_busy, 0)
elif ctx.get(dut.udp_rx_req):
ctx.set(dut.udp_rx_busy, 1)
await ctx.tick().repeat(4) # emulate RSR/RD/header latency
rep = staged["reply"]
if rep is None:
ctx.set(dut.udp_rx_none, 1)
await ctx.tick()
ctx.set(dut.udp_rx_none, 0)
else:
staged["reply"] = None
for i, b in enumerate(rep):
ctx.set(dut.udp_rx_data, b)
ctx.set(dut.udp_rx_valid, 1)
ctx.set(dut.udp_rx_eof, 1 if i == len(rep) - 1 else 0)
g = 0
while not ctx.get(dut.udp_rx_ready):
await ctx.tick(); g += 1
if g > 20000: break
await ctx.tick() # consume cycle
ctx.set(dut.udp_rx_valid, 0)
ctx.set(dut.udp_rx_eof, 0)
ctx.set(dut.udp_rx_busy, 0)
async def wait_prompt(ctx, timeout=400_000):
"""Wait until the tail of tx_chars ends with 'rebbarb> '."""
for _ in range(timeout):
if tx_chars.endswith(b"rebbarb> "):
return True
await ctx.tick()
return False
async def testbench(ctx):
ctx.set(dut.uart_rx, 1)
# udp_pl_ready / udp_test_busy are owned by w5100_udp_model — do not
# drive them here (two testbenches on one signal deadlocks the send).
# Boot banner + first prompt.
if not await wait_prompt(ctx):
errors.append("no initial prompt"); return
print(f"boot tx: {bytes(tx_chars)!r}")
# T1: 'help' lists commands.
tx_chars.clear()
await send_line(ctx, "help")
await wait_prompt(ctx)
if b"udp unicast" not in tx_chars or b"udp broadcast" not in tx_chars:
errors.append(f"T1 help missing commands: {bytes(tx_chars)!r}")
print(f"T1 help ok ({len(tx_chars)} bytes)")
# T2: unicast with payload, and a staged reply → 'rx <reply>' printed.
tx_chars.clear(); captured["count"] = 0
staged["reply"] = b"pong-A"
await send_line(ctx, "udp unicast 192.168.1.55 hello world")
await wait_prompt(ctx)
if captured["dst"] != [192, 168, 1, 55]:
errors.append(f"T2 dst {captured['dst']} != [192,168,1,55]")
if captured["payload"] != b"hello world":
errors.append(f"T2 payload {captured['payload']!r} != b'hello world'")
if b"sent" not in tx_chars:
errors.append(f"T2 no 'sent' ack: {bytes(tx_chars)!r}")
if b"rx pong-A" not in tx_chars:
errors.append(f"T2 reply not printed: {bytes(tx_chars)!r}")
print(f"T2 unicast+reply: dst={captured['dst']} "
f"payload={captured['payload']!r} tx={bytes(tx_chars)!r}")
# T3: broadcast default payload, NO reply → 'timeout' printed (no hang).
tx_chars.clear(); staged["reply"] = None
await send_line(ctx, "udp broadcast")
await wait_prompt(ctx)
if captured["dst"] != [255, 255, 255, 255]:
errors.append(f"T3 dst {captured['dst']} != broadcast")
if captured["payload"] != bytes(_DEFAULT_PAYLOAD):
errors.append(f"T3 payload {captured['payload']!r} != default")
if b"timeout" not in tx_chars:
errors.append(f"T3 no timeout on no-reply: {bytes(tx_chars)!r}")
print(f"T3 broadcast default→timeout: dst={captured['dst']} "
f"payload={captured['payload']!r}")
# T4: broadcast with payload + reply.
tx_chars.clear(); staged["reply"] = b"Hello World"
await send_line(ctx, "udp broadcast ping123")
await wait_prompt(ctx)
if captured["payload"] != b"ping123":
errors.append(f"T4 payload {captured['payload']!r} != b'ping123'")
if b"rx Hello World" not in tx_chars:
errors.append(f"T4 reply not printed: {bytes(tx_chars)!r}")
print(f"T4 broadcast+reply: payload={captured['payload']!r} "
f"tx={bytes(tx_chars)!r}")
# T5: case-insensitive command + uppercase preserved in payload.
tx_chars.clear(); staged["reply"] = b"ok"
await send_line(ctx, "UDP UNICAST 10.0.0.9 MixedCase")
await wait_prompt(ctx)
if captured["dst"] != [10, 0, 0, 9]:
errors.append(f"T5 dst {captured['dst']} != [10,0,0,9]")
if captured["payload"] != b"MixedCase":
errors.append(f"T5 payload {captured['payload']!r} != b'MixedCase'")
print(f"T5 case: dst={captured['dst']} payload={captured['payload']!r}")
# T6: bad IP → 'bad ip', no send.
tx_chars.clear(); before = captured["count"]
await send_line(ctx, "udp unicast 1.2.3 x")
await wait_prompt(ctx)
if b"bad ip" not in tx_chars:
errors.append(f"T6 no 'bad ip': {bytes(tx_chars)!r}")
if captured["count"] != before:
errors.append("T6 sent despite bad ip")
print(f"T6 bad ip ok (no send)")
# T7: unknown command → error.
tx_chars.clear()
await send_line(ctx, "frobnicate")
await wait_prompt(ctx)
if b"?" not in tx_chars:
errors.append(f"T7 no error marker: {bytes(tx_chars)!r}")
print(f"T7 unknown ok")
# T8: backspace editing (type 'helX' <bs> 'p' → 'help'). The backspace
# echo is 3 bytes ("\b \b"), longer than one incoming byte time, so a
# realistic source pauses after it (a human always does). Settle
# between edits — this is the one spot the 1-deep RX register can't
# absorb, and it never occurs with paste (no backspaces) or typing.
tx_chars.clear()
for ch in "helX":
await send_byte(ctx, ord(ch))
await ctx.tick().repeat(DIV * 15) # > 1-byte echo (10 bits)
await send_byte(ctx, 0x08) # backspace removes 'X'
await ctx.tick().repeat(DIV * 40) # > 3-byte "\b \b" echo (~30 bits)
await send_byte(ctx, ord('p'))
await ctx.tick().repeat(DIV * 15)
await send_byte(ctx, 0x0D)
await wait_prompt(ctx)
if b"udp unicast" not in tx_chars:
errors.append(f"T8 backspace edit failed: {bytes(tx_chars)!r}")
print(f"T8 backspace edit ok")
sim = Simulator(dut)
sim.add_clock(Period(MHz=24))
sim.add_testbench(testbench)
sim.add_testbench(tx_collector, background=True)
sim.add_testbench(w5100_udp_model, background=True)
sim.run()
if errors:
print("\nFAILURES:")
for e in errors:
print(" ", e)
sys.exit(1)
print("\nAll UARTShell tests passed.")
exi_bba/w5100_parallel_master.py
+523 -9
View File
@@ -45,7 +45,10 @@ __all__ = ["W5100ParallelMaster"]
# ── W5100 register addresses (indirect 16-bit address space) ──────────────── # ── W5100 register addresses (indirect 16-bit address space) ────────────────
_MR = 0x0000 # Mode register (common) _MR = 0x0000 # Mode register (common)
_GAR0 = 0x0001 # Gateway IP, 4 bytes
_SUBR0 = 0x0005 # Subnet mask, 4 bytes
_SHAR0 = 0x0009 # Source MAC, 6 bytes _SHAR0 = 0x0009 # Source MAC, 6 bytes
_SIPR0 = 0x000F # Source IP, 4 bytes
_IR = 0x0015 # Interrupt register _IR = 0x0015 # Interrupt register
_IMR = 0x0016 # Interrupt mask _IMR = 0x0016 # Interrupt mask
_RMSR = 0x001A # RX memory size (2 bits/socket) _RMSR = 0x001A # RX memory size (2 bits/socket)
@@ -60,20 +63,50 @@ _S0_TX_WR = 0x0424 # Socket 0 TX write pointer
_S0_RX_RSR = 0x0426 # Socket 0 RX received size (2 bytes) _S0_RX_RSR = 0x0426 # Socket 0 RX received size (2 bytes)
_S0_RX_RD = 0x0428 # Socket 0 RX read pointer _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) _TX_BASE = 0x4000 # Socket 0 TX buffer base (default 2 KB window)
_RX_BASE = 0x6000 # Socket 0 RX buffer base _RX_BASE = 0x6000 # Socket 0 RX buffer base
_S0_TX_MASK = 0x07FF # 2 KB ring mask _S0_TX_MASK = 0x07FF # 2 KB ring mask
_S0_RX_MASK = 0x07FF _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 bits / command / mode values
_MR_RST = 0x80 _MR_RST = 0x80
_MR_AI = 0x02 # address auto-increment (indirect mode) _MR_AI = 0x02 # address auto-increment (indirect mode)
_MR_IND = 0x01 # indirect bus interface mode _MR_IND = 0x01 # indirect bus interface mode
_S0_MR_MACRAW = 0x04 _S0_MR_MACRAW = 0x04
_S1_MR_UDP = 0x02 # socket UDP mode
_CR_OPEN = 0x01 _CR_OPEN = 0x01
_CR_SEND = 0x20 _CR_SEND = 0x20
_CR_RECV = 0x40 _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]) # Indirect-mode address selects (A[1:0])
_A_MR = 0b00 _A_MR = 0b00
_A_AR0 = 0b01 # IDM_AR high byte _A_AR0 = 0b01 # IDM_AR high byte
@@ -97,12 +130,64 @@ class W5100ParallelMaster(Elaboratable):
Init / TX / RX interfaces are identical to W5500SPIMaster. Init / TX / RX interfaces are identical to W5500SPIMaster.
""" """
def __init__(self, strobe_cycles=3, reset_cycles=24000): 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). # /RD//WR strobe width in sync cycles (≥ W5100 access time).
self._strobe = strobe_cycles self._strobe = strobe_cycles
# MR-reset settle wait; testbench overrides with a small value. # MR-reset settle wait; testbench overrides with a small value.
self._reset_cycles = reset_cycles 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 # Physical parallel bus
self.bus_addr = Signal(2) self.bus_addr = Signal(2)
self.bus_data_o = Signal(8) self.bus_data_o = Signal(8)
@@ -136,6 +221,7 @@ class W5100ParallelMaster(Elaboratable):
def elaborate(self, platform): def elaborate(self, platform):
m = Module() m = Module()
STROBE = self._strobe STROBE = self._strobe
sn = self._sn # UDP-test socket register/buffer addresses
# ── Bus access engine: one indirect-bus read or write cycle ────────── # ── Bus access engine: one indirect-bus read or write cycle ──────────
bus_go = Signal() bus_go = Signal()
@@ -209,13 +295,43 @@ class W5100ParallelMaster(Elaboratable):
s_data, s_valid, s_last, s_consume = Signal(8), Signal(), Signal(), Signal() s_data, s_valid, s_last, s_consume = Signal(8), Signal(), Signal(), Signal()
r_data, r_valid, r_first, r_last, r_ready = ( r_data, r_valid, r_first, r_last, r_ready = (
Signal(8), Signal(), Signal(), Signal(), Signal()) Signal(8), Signal(), Signal(), Signal(), Signal())
# TX stream source = external tx interface (Phase 2). # TX stream-write source mux: during a UDP-test send the payload comes
m.d.comb += [s_data.eq(self.tx_data), s_valid.eq(self.tx_valid), # from the external `udp_pl_*` stream (the shell); otherwise the normal
s_last.eq(self.tx_eof), self.tx_ready.eq(s_consume)] # MACRAW TX interface feeds it. `udp_streaming` is raised only while the
# RX stream sink = external rx interface (Phase 3). # UDP payload is being written to the socket-1 TX buffer.
m.d.comb += [self.rx_data.eq(r_data), self.rx_valid.eq(r_valid), udp_streaming = Signal()
self.rx_sof.eq(r_first), self.rx_eof.eq(r_last), if self._enable_udp:
r_ready.eq(self.rx_ready)] 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 # Socket-buffer ring wraparound. Unlike the W5500, the W5100's IDM
# address does NOT auto-wrap at the socket-buffer boundary — it just # address does NOT auto-wrap at the socket-buffer boundary — it just
@@ -391,6 +507,10 @@ class W5100ParallelMaster(Elaboratable):
rx_rsr = Signal(16) rx_rsr = Signal(16)
rx_rd = Signal(16) rx_rd = Signal(16)
pkt_len = 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): def write_reg(name, addr, payload, nxt, direct=False):
"""Emit a 2-state block that writes `payload` (a list) to `addr`.""" """Emit a 2-state block that writes `payload` (a list) to `addr`."""
@@ -409,6 +529,8 @@ class W5100ParallelMaster(Elaboratable):
m.next = nxt m.next = nxt
# ── Main control FSM (Phase 1: init only) ──────────────────────────── # ── 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.FSM(domain="sync", name="main_fsm"):
with m.State("IDLE"): with m.State("IDLE"):
m.d.sync += self.init_done.eq(0) m.d.sync += self.init_done.eq(0)
@@ -420,6 +542,11 @@ class W5100ParallelMaster(Elaboratable):
m.next = "RX_CHECK" m.next = "RX_CHECK"
with m.Elif(self.tx_valid & self.tx_sof): with m.Elif(self.tx_valid & self.tx_sof):
m.next = "TX_START" 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. # MR = 0x80 software reset (direct A=00), then settle.
write_reg("MR_RST", _MR, [_MR_RST], "MR_WAIT", direct=True) write_reg("MR_RST", _MR, [_MR_RST], "MR_WAIT", direct=True)
@@ -452,7 +579,21 @@ class W5100ParallelMaster(Elaboratable):
# Socket 0: MACRAW mode, OPEN, enable interrupt. # Socket 0: MACRAW mode, OPEN, enable interrupt.
write_reg("S0_MODE", _S0_MR, [_S0_MR_MACRAW], "S0_OPEN") write_reg("S0_MODE", _S0_MR, [_S0_MR_MACRAW], "S0_OPEN")
write_reg("S0_OPEN", _S0_CR, [_CR_OPEN], "S0_IMR") write_reg("S0_OPEN", _S0_CR, [_CR_OPEN], "S0_IMR")
write_reg("S0_IMR", _IMR, [0x01], "INIT_DONE") # enable S0 IRQ # 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"): with m.State("INIT_DONE"):
m.d.sync += self.init_done.eq(1) m.d.sync += self.init_done.eq(1)
@@ -581,6 +722,168 @@ class W5100ParallelMaster(Elaboratable):
write_reg("RX_RECV", _S0_CR, [_CR_RECV], "RX_CLR_IR") write_reg("RX_RECV", _S0_CR, [_CR_RECV], "RX_CLR_IR")
write_reg("RX_CLR_IR", _S0_IR, [0x04], "IDLE") 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 return m
@@ -831,6 +1134,217 @@ if __name__ == "__main__":
sim.run() 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: if errors:
print("\nFAILURES:") print("\nFAILURES:")
for e in errors: for e in errors: