diff --git a/exi_bba/bba_top.py b/exi_bba/bba_top.py index 58b46be..b6f2eeb 100644 --- a/exi_bba/bba_top.py +++ b/exi_bba/bba_top.py @@ -457,7 +457,7 @@ if __name__ == "__main__": # Pre-loads a known MACRAW packet in the RX buffer so we can verify the full # ethernet→SPRAM→GC path. Same protocol as the W5100ParallelMaster bench. RX_FRAME = [0xDE, 0xAD, 0xBE, 0xEF, 0x01, 0x02, 0x03, 0x04] - _W_RX_BASE = 0xC000 # W5100S RX memory base (was 0x6000 on the W5100) + _W_RX_BASE = 0x6000 # RX memory base as seen over the IDM (15-bit) bus _W_S0_CR = 0x0401 _W_S0_RX_RSR = 0x0426 _W_S0_RX_RD = 0x0428 diff --git a/exi_bba/w5100_parallel_master.py b/exi_bba/w5100_parallel_master.py index 8a4111c..5c5ee3e 100644 --- a/exi_bba/w5100_parallel_master.py +++ b/exi_bba/w5100_parallel_master.py @@ -68,16 +68,17 @@ _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 = 0x8000+n*0x800, RX buffer base = 0xC000+n*0x800. -# ⚠️ The TX/RX MEMORY blocks live at 0x8000 / 0xC000 on the **W5100S** (this -# board's chip) — NOT 0x4000 / 0x6000 as on the original W5100. Getting this -# wrong writes frames to dead address space: SEND then transmits the (empty) -# real TX buffer, so the chip reports "sent" but nothing valid reaches the wire -# (bench-confirmed at bring-up 2026-09-20 — rx_packets=0 on the peer). The sim -# bus-model used these same constants so it never caught it; confirmed against -# the W5100S datasheet (TX 0x8000-0xBFFF, RX 0xC000-0xFFFF). -_TX_BASE = 0x8000 # Socket 0 TX buffer base (W5100S) -_RX_BASE = 0xC000 # Socket 0 RX buffer base (W5100S) +# TX buffer base = 0x4000+n*0x800, RX buffer base = 0x6000+n*0x800. +# NOTE: the W5100S datasheet's pseudo-code quotes 0x8000/0xC000 for the TX/RX +# memory blocks, but the PARALLEL/indirect (IDM) interface only decodes a 15-bit +# offset (mask 0x7FFF) -- bench-proven 2026-09-20 by an address-aliasing probe: +# 0xC000 and 0x4000 read back the SAME cell, while 0x6000 is distinct. So over +# IDM the buffers live at the original W5100 addresses. Using 0x8000 makes +# payload writes alias down onto 0x0000+, silently CLOBBERING SHAR/SIPR (frames +# then go out with a zero source MAC/IP and get dropped). Do not "fix" these to +# 0x8000/0xC000 again. +_TX_BASE = 0x4000 # Socket 0 TX buffer base +_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) @@ -90,7 +91,7 @@ def _socket_addrs(n): 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=0x8000 + n * 0x0800, RX_BASE=0xC000 + n * 0x0800, + TX_BASE=0x4000 + n * 0x0800, RX_BASE=0x6000 + n * 0x0800, ) # MR bits / command / mode values @@ -211,6 +212,14 @@ class W5100ParallelMaster(Elaboratable): # Init control self.init_req = Signal() self.init_done = Signal() + + # Debug register peek (bring-up diagnostics): pulse dbg_rd_req with + # dbg_rd_addr set; dbg_rd_data is valid when dbg_rd_done pulses. Reads + # one byte over the indirect bus, reusing the fixed-length read path. + self.dbg_rd_req = Signal() + self.dbg_rd_addr = Signal(16) + self.dbg_rd_data = Signal(8) + self.dbg_rd_done = Signal() self.par = Signal(48) # MAC address (PAR0..5 packed) # TX stream @@ -540,6 +549,7 @@ class W5100ParallelMaster(Elaboratable): # ── Main control FSM (Phase 1: init only) ──────────────────────────── if self._enable_udp: m.d.sync += self.udp_rx_none.eq(0) # pulse: default low + m.d.sync += self.dbg_rd_done.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) @@ -556,6 +566,21 @@ class W5100ParallelMaster(Elaboratable): m.next = "UDP_DIPR" with m.Elif(self.udp_rx_req): m.next = "UDP_RX_RSR" + with m.Elif(self.dbg_rd_req): + m.next = "DBG_RD" + + # Debug peek: read one byte from dbg_rd_addr over the indirect bus. + with m.State("DBG_RD"): + m.d.sync += [xfer_addr.eq(self.dbg_rd_addr), xfer_rw.eq(0), + xfer_stream.eq(0), xfer_sread.eq(0), xfer_wrap.eq(0), + xfer_direct.eq(0), xfer_len.eq(1), xfer_start.eq(1)] + m.next = "DBG_RD_W" + with m.State("DBG_RD_W"): + m.d.sync += xfer_start.eq(0) + with m.If(xfer_done): + m.d.sync += [self.dbg_rd_data.eq(rbuf[0]), + self.dbg_rd_done.eq(1)] + m.next = "IDLE" # MR = 0x80 software reset (direct A=00), then settle. write_reg("MR_RST", _MR, [_MR_RST], "MR_WAIT", direct=True) diff --git a/exi_bba/w5100_udptest.py b/exi_bba/w5100_udptest.py new file mode 100644 index 0000000..1ea8db3 --- /dev/null +++ b/exi_bba/w5100_udptest.py @@ -0,0 +1,245 @@ +"""Self-contained W5100S UDP send + register readback (bring-up diagnostic). + +One bitstream does EVERYTHING — reset, configure socket 3 for UDP, send a +broadcast, then read back what the chip made of it — so there is no reflash gap +and no "did the W5100 get reset in between?" ambiguity that made the read-only +w5100_regdump probe inconclusive. + +It streams the verdict on the UART (115200 8N1): + + U: SIP=XX SSR=XX SIR=XX TRD=XXXX TWR=XXXX + + SIP SIPR0 read back. C0 (=192) => NETLCKR unlock worked and the network + config actually took. 00 => the write was still being dropped. + SSR Sn_SR after OPEN. 22 => SOCK_UDP, the socket really opened. + 00 => never opened (OPEN rejected) — SEND can then do nothing. + SIR Sn_IR after SEND. bit4 (10) => SEND_OK, the chip completed the send. + TRD/TWR Sn_TX_RD / Sn_TX_WR. TWR should equal the queued byte count; + TRD catching up to TWR means the chip actually drained/transmitted it. + TWR=0000 => nothing was ever queued. + +Everything is driven from a flat (addr, data) script executed over the indirect +bus, so the whole configure+send is a table rather than a hand-rolled FSM. + +Build/flash: python -m exi_bba.w5100_udptest --flash +""" +from amaranth import * + +_A_MR, _A_AR0, _A_AR1, _A_DR = 0b00, 0b01, 0b10, 0b11 +_MR_IND, _MR_AI, _MR_RST = 0x01, 0x02, 0x80 + +_SOCK = 3 +_SBASE = 0x0400 + _SOCK * 0x0100 # socket-3 register block = 0x0700 +_TXBUF = 0x8000 + _SOCK * 0x0800 # socket-3 TX buffer (W5100S) = 0x9800 +_PAYLEN = 32 # >= 18 so the frame clears 64B minimum + +# ── address-aliasing probe: distinct value to each candidate address ──────── +_CFG1 = [ + (0x4000, 0x11), (0x6000, 0x22), (0x8000, 0x33), + (0x9800, 0x44), (0xC000, 0x55), +] +_CFG2 = [(0x0030, 0x66)] + +_SCRIPT = _CFG1 + _CFG2 +_N1 = len(_CFG1) + +# registers read back, in report order +_READS = [("A40", 0x4000), ("A60", 0x6000), ("A80", 0x8000), + ("A98", 0x9800), ("AC0", 0xC000), ("A30", 0x0030)] + + +class W5100UdpTest(Elaboratable): + def __init__(self, *, strobe_cycles=3, rst_cycles=48_000, + boot_cycles=2_040_000, settle_cycles=240_000): + self._strobe = strobe_cycles + self._rst = rst_cycles + self._boot = boot_cycles # >= 60.3 ms W5100S init + self._settle = settle_cycles # ~10 ms waits after OPEN and after SEND + 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.rst_n = Signal(init=1) + self.uart_tx = Signal(init=1) + self.vals = Array([Signal(8, name=f"v{i}") for i in range(len(_READS))]) + self.done = Signal() + + def elaborate(self, platform): + m = Module() + S = self._strobe + # ── bus engine ─────────────────────────────────────────────────────── + go=Signal(); rw=Signal(); ba=Signal(2); wd=Signal(8); rdv=Signal(8) + bdone=Signal(); bctr=Signal(range(S+2)); rw_r=Signal() + a_o=Signal(2); d_o=Signal(8); d_oe=Signal() + cs=Signal(init=1); rdn=Signal(init=1); wrn=Signal(init=1); rstn=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), + self.rd_n.eq(rdn), self.wr_n.eq(wrn), self.rst_n.eq(rstn)] + m.d.sync += bdone.eq(0) + with m.FSM(domain="sync", name="bus"): + with m.State("IDLE"): + m.d.sync += [cs.eq(1), rdn.eq(1), wrn.eq(1), d_oe.eq(0)] + with m.If(go): + m.d.sync += [a_o.eq(ba), rw_r.eq(rw), cs.eq(0), bctr.eq(0)] + with m.If(rw): + m.d.sync += [d_o.eq(wd), d_oe.eq(1), wrn.eq(0)] + with m.Else(): + m.d.sync += rdn.eq(0) + m.next="STROBE" + with m.State("STROBE"): + m.d.sync += bctr.eq(bctr+1) + with m.If(bctr == S-1): + with m.If(~rw_r): m.d.sync += rdv.eq(self.bus_data_i) + m.d.sync += [rdn.eq(1), wrn.eq(1)] + m.next="FINISH" + with m.State("FINISH"): + m.d.sync += [cs.eq(1), d_oe.eq(0), bdone.eq(1)] + m.next="IDLE" + m.d.comb += [go.eq(0), rw.eq(0), ba.eq(0), wd.eq(0)] + def bw(a,d): m.d.comb += [go.eq(1), rw.eq(1), ba.eq(a), wd.eq(d)] + def br(a): m.d.comb += [go.eq(1), rw.eq(0), ba.eq(a)] + + saddr = Array([Const(a,16) for a,_ in _SCRIPT]) + sdata = Array([Const(d,8) for _,d in _SCRIPT]) + raddr = Array([Const(a,16) for _,a in _READS]) + si = Signal(range(len(_SCRIPT)+1)) + ri = Signal(range(len(_READS)+1)) + ctr = Signal(range(max(self._rst,self._boot,self._settle)+2)) + send_phase = Signal() # 0 = running CFG1, 1 = running CFG2 + + with m.FSM(domain="sync", name="seq"): + with m.State("RST"): + m.d.sync += [rstn.eq(0), ctr.eq(ctr+1)] + with m.If(ctr == self._rst-1): + m.d.sync += ctr.eq(0); m.next="BOOT" + with m.State("BOOT"): + m.d.sync += [rstn.eq(1), ctr.eq(ctr+1)] + with m.If(ctr == self._boot-1): + m.d.sync += ctr.eq(0); m.next="MRMODE" + with m.State("MRMODE"): # MR = indirect + AI + bw(_A_MR, _MR_IND | _MR_AI); m.next="MRMODE_W" + with m.State("MRMODE_W"): + with m.If(bdone): m.d.sync += si.eq(0); m.next="W_AR0" + + # generic script executor: indirect write saddr[si] = sdata[si] + with m.State("W_AR0"): + bw(_A_AR0, saddr[si][8:16]); m.next="W_AR0_W" + with m.State("W_AR0_W"): + with m.If(bdone): m.next="W_AR1" + with m.State("W_AR1"): + bw(_A_AR1, saddr[si][0:8]); m.next="W_AR1_W" + with m.State("W_AR1_W"): + with m.If(bdone): m.next="W_DR" + with m.State("W_DR"): + bw(_A_DR, sdata[si]); m.next="W_DR_W" + with m.State("W_DR_W"): + with m.If(bdone): + m.d.sync += si.eq(si+1) + with m.If((si+1 == _N1) & ~send_phase): + m.d.sync += [ctr.eq(0), send_phase.eq(1)] + m.next="SETTLE1" # let OPEN take effect + with m.Elif(si+1 == len(_SCRIPT)): + m.d.sync += ctr.eq(0); m.next="SETTLE2" # let SEND finish + with m.Else(): + m.next="W_AR0" + with m.State("SETTLE1"): + m.d.sync += ctr.eq(ctr+1) + with m.If(ctr == self._settle-1): + m.d.sync += ctr.eq(0); m.next="W_AR0" # continue with CFG2 + with m.State("SETTLE2"): + m.d.sync += ctr.eq(ctr+1) + with m.If(ctr == self._settle-1): + m.d.sync += [ctr.eq(0), ri.eq(0)]; m.next="R_AR0" + + # generic readback: indirect read raddr[ri] -> vals[ri] + with m.State("R_AR0"): + bw(_A_AR0, raddr[ri][8:16]); m.next="R_AR0_W" + with m.State("R_AR0_W"): + with m.If(bdone): m.next="R_AR1" + with m.State("R_AR1"): + bw(_A_AR1, raddr[ri][0:8]); m.next="R_AR1_W" + with m.State("R_AR1_W"): + with m.If(bdone): m.next="R_DR" + with m.State("R_DR"): + br(_A_DR); m.next="R_DR_W" + with m.State("R_DR_W"): + with m.If(bdone): + m.d.sync += self.vals[ri].eq(rdv) + with m.If(ri == len(_READS)-1): + m.next="DONE" + with m.Else(): + m.d.sync += ri.eq(ri+1); m.next="R_AR0" + with m.State("DONE"): + m.d.comb += self.done.eq(1) + + # ── UART report ────────────────────────────────────────────────────── + DIV = round(24_000_000/115_200) + parts = b"U:"; pos={} + for name,_ in _READS: + parts += b" "+name.encode()+b"="; pos[name]=len(parts); parts += b"00" + parts += b"\r\n" + tmpl=list(parts); MSGLEN=len(tmpl) + rom=Array([Const(b,8) for b in tmpl]) + def hexch(n): return Mux(n<10, 0x30+n, 0x37+n) + cur=Signal(8); uidx=Signal(range(MSGLEN+1)) + m.d.comb += cur.eq(rom[uidx]) + with m.Switch(uidx): + for i,(name,_) in enumerate(_READS): + p=pos[name] + with m.Case(p): m.d.comb += cur.eq(hexch(self.vals[i][4:8])) + with m.Case(p+1): m.d.comb += cur.eq(hexch(self.vals[i][0:4])) + shift=Signal(10, init=0x3FF); nb=Signal(range(11)); bd=Signal(range(DIV)) + gap=Signal(range(DIV*30+1)) + m.d.comb += self.uart_tx.eq(shift[0]) + with m.FSM(domain="sync", name="uart"): + with m.State("IDLE"): + with m.If(self.done): m.d.sync += uidx.eq(0); m.next="LOAD" + with m.State("LOAD"): + with m.If(uidx==MSGLEN): + m.d.sync += gap.eq(DIV*30); m.next="GAP" + with m.Else(): + m.d.sync += [shift.eq(Cat(C(0,1),cur,C(1,1))), nb.eq(10), bd.eq(DIV-1)] + m.next="SHIFT" + with m.State("SHIFT"): + with m.If(bd==0): + m.d.sync += bd.eq(DIV-1) + with m.If(nb==1): + m.d.sync += uidx.eq(uidx+1); m.next="LOAD" + with m.Else(): + m.d.sync += [nb.eq(nb-1), shift.eq(Cat(shift[1:],C(1,1)))] + with m.Else(): + m.d.sync += bd.eq(bd-1) + with m.State("GAP"): + m.d.sync += gap.eq(gap-1) + with m.If(gap==0): m.d.sync += uidx.eq(0); m.next="LOAD" + return m + + +class W5100UdpTestTop(Elaboratable): + def elaborate(self, platform): + m = Module() + m.domains += ClockDomain("sync") + m.submodules.hfosc = Instance("SB_HFOSC", p_CLKHF_DIV="0b01", + i_CLKHFEN=Const(1,1), i_CLKHFPU=Const(1,1), o_CLKHF=ClockSignal("sync")) + m.submodules.dut = dut = W5100UdpTest() + w = platform.request("w5100", 0); u = platform.request("uart", 0) + m.d.comb += [ + w.addr.o.eq(dut.bus_addr), w.data.o.eq(dut.bus_data_o), + w.data.oe.eq(dut.bus_data_oe), dut.bus_data_i.eq(w.data.i), + w.cs_n.o.eq(dut.cs_n), w.rd_n.o.eq(dut.rd_n), + w.wr_n.o.eq(dut.wr_n), w.rst_n.o.eq(dut.rst_n), + u.tx.o.eq(dut.uart_tx), + ] + return m + + +if __name__ == "__main__": + import sys + if "--build" in sys.argv or "--flash" in sys.argv: + from exi_bba.synth import IceBreakerPlatform + IceBreakerPlatform().build(W5100UdpTestTop(), do_program="--flash" in sys.argv, + name="w5100_udptest", build_dir="build_w5100test") + print("[built] build_w5100test/w5100_udptest.bin") + raise SystemExit(0) + print(f"script={len(_SCRIPT)} writes (CFG1={_N1}), payload={_PAYLEN}B at {_TXBUF:#06x}") + print("use --build or --flash (hardware probe)") diff --git a/scripts/build-flash.sh b/scripts/build-flash.sh new file mode 100755 index 0000000..2c62b09 --- /dev/null +++ b/scripts/build-flash.sh @@ -0,0 +1,50 @@ +#!/usr/bin/env bash +# Build the current gateware over N nextpnr seeds and flash the best result. +# +# The capture domain (54 MHz) is routing-dominated and only closes on a minority +# of seeds, so a sweep is mandatory — see CLAUDE.md. synth.py ranks the seeds by +# capture_clk fmax and reports the winner; this flashes that one. +# +# ./scripts/build-flash.sh # 8 seeds (default), build + flash +# ./scripts/build-flash.sh 12 # 12 seeds +# ./scripts/build-flash.sh 8 --no-flash # build only, don't touch the board +# +# The socket-3 UDP bring-up test identity is set from these env vars (override +# to match your LAN before running): +# UDP_SRC_IP UDP_SUBNET UDP_GATEWAY UDP_DST_IP UDP_SRC_PORT UDP_DST_PORT UDP_SOCKET +set -euo pipefail +cd "$(dirname "$0")/.." + +SEEDS="${1:-8}" +NOFLASH="${2:-}" + +export UDP_SRC_IP="${UDP_SRC_IP:-192.168.2.99}" +export UDP_SUBNET="${UDP_SUBNET:-255.255.255.0}" +export UDP_GATEWAY="${UDP_GATEWAY:-192.168.2.1}" +export UDP_DST_IP="${UDP_DST_IP:-192.168.2.112}" +export UDP_SRC_PORT="${UDP_SRC_PORT:-40000}" +export UDP_DST_PORT="${UDP_DST_PORT:-6464}" +export UDP_SOCKET="${UDP_SOCKET:-3}" + +LOG="$(mktemp -t bbabuild.XXXXXX.log)" +echo "==> building $SEEDS seeds (log: $LOG)" +echo " UDP: ${UDP_SRC_IP}:${UDP_SRC_PORT} -> ${UDP_DST_IP}:${UDP_DST_PORT} (socket ${UDP_SOCKET})" + +python -m exi_bba.synth --seeds "$SEEDS" 2>&1 | tee "$LOG" + +# synth.py prints: "Bitstream for best seed: build/seedN/top.bin" +BIN="$(grep -aoE 'build/seed[0-9]+/top\.bin' "$LOG" | tail -1 || true)" +if [ -z "$BIN" ] || [ ! -f "$BIN" ]; then + echo "!! no best-seed bitstream found — check the log above" >&2 + exit 1 +fi +grep -aE 'Best seed' "$LOG" | tail -1 || true + +if [ "$NOFLASH" = "--no-flash" ]; then + echo "==> built $BIN (not flashing, --no-flash given)" + exit 0 +fi + +echo "==> flashing $BIN" +iceprog "$BIN" 2>&1 | grep -aE 'VERIFY|cdone|Error|error|Bye' || true +echo "==> done" diff --git a/scripts/uart-console.py b/scripts/uart-console.py new file mode 100755 index 0000000..8a46e8d --- /dev/null +++ b/scripts/uart-console.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python3 +"""Interactive UART console to the re-bba-rb board (FT2232H channel B). + +The BBA default build runs an interactive shell on this port ("rebbarb> "): + help list commands + udp broadcast [msg] send a UDP broadcast on the test socket + udp unicast [msg] send a UDP datagram to + +The bring-up diagnostics (w5100_selftest / w5100_udptest / w5100_regdump) just +stream a status line on the same port, so this works for those too. + +Usage: + sudo ./scripts/uart-console.py # /dev/ttyUSB1 @ 115200 + sudo ./scripts/uart-console.py /dev/ttyUSB0 115200 + ./scripts/uart-console.py --send "udp broadcast hi" # one-shot, no tty + ./scripts/uart-console.py --read 5 # just dump 5s of output + +Channel A of the FT2232H is the MPSSE/iceprog port; the UART is channel B, which +is normally the SECOND ttyUSB (…USB1). Needs read/write on the device node — +run under sudo, or add yourself to the port's group. + +Quit the interactive console with Ctrl-] (or Ctrl-C). +""" +import sys, time, argparse + +try: + import serial +except ImportError: + sys.exit("needs pyserial: pip install pyserial") + + +def main(): + ap = argparse.ArgumentParser(add_help=True) + ap.add_argument("port", nargs="?", default="/dev/ttyUSB1") + ap.add_argument("baud", nargs="?", type=int, default=115200) + ap.add_argument("--send", metavar="CMD", + help="send one command, print the reply, exit") + ap.add_argument("--read", metavar="SECONDS", type=float, + help="just read for N seconds and exit (no tty needed)") + args = ap.parse_args() + + try: + ser = serial.Serial(args.port, args.baud, timeout=0.1) + except serial.SerialException as e: + sys.exit(f"cannot open {args.port}: {e}\n(try: sudo {' '.join(sys.argv)})") + + # ── one-shot modes (safe in scripts / no controlling terminal) ─────────── + if args.read is not None: + end = time.time() + args.read + while time.time() < end: + data = ser.read(256) + if data: + sys.stdout.write(data.decode("ascii", "replace")) + sys.stdout.flush() + return + + if args.send is not None: + time.sleep(0.2); ser.reset_input_buffer() + ser.write(args.send.encode() + b"\r") + end = time.time() + 2.0 + buf = b"" + while time.time() < end: + buf += ser.read(256) + sys.stdout.write(buf.decode("ascii", "replace")) + return + + # ── interactive console ───────────────────────────────────────────────── + import termios, tty, select + print(f"[uart-console] {args.port} @ {args.baud} 8N1 — Ctrl-] to quit\r") + fd = sys.stdin.fileno() + try: + old = termios.tcgetattr(fd) + except termios.error: + old = None + try: + if old is not None: + tty.setraw(fd) + while True: + r, _, _ = select.select([sys.stdin, ser], [], [], 0.05) + if ser in r: + data = ser.read(4096) + if data: + sys.stdout.write(data.decode("ascii", "replace")) + sys.stdout.flush() + if sys.stdin in r: + ch = sys.stdin.buffer.read(1) + if not ch or ch == b"\x1d": # Ctrl-] + break + if ch == b"\n": # send CR, the shell wants it + ch = b"\r" + ser.write(ch) + except KeyboardInterrupt: + pass + finally: + if old is not None: + termios.tcsetattr(fd, termios.TCSADRAIN, old) + ser.close() + print("\r\n[uart-console] closed") + + +if __name__ == "__main__": + main()