diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 5b42d04..6e279a2 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -57,3 +57,17 @@ jobs: dockerfile: .devcontainer/Dockerfile tag-prefix: "" secrets: inherit + + svlint: + needs: select_ci_image + if: always() # dont skip if build_ci_image is skipped + runs-on: ubuntu-latest + container: + image: ${{ needs.select_ci_image.outputs.image-tag }} + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: lint uart + run: cd uart && make svlint diff --git a/.svlint.toml b/.svlint.toml new file mode 100644 index 0000000..98b2eac --- /dev/null +++ b/.svlint.toml @@ -0,0 +1,84 @@ + +# Style ruleset begin + +option.textwidth = 100 +option.indent = 2 +option.exclude_paths = [ + "test/utils.svh", +] # this file contains a complex macro that svlint chokes on + +# textrules.style_textwidth = true +textrules.style_semicolon = true +syntaxrules.tab_character = true +syntaxrules.style_indent = true +syntaxrules.multiline_if_begin = false +syntaxrules.multiline_for_begin = true +syntaxrules.style_trailingwhitespace = true +textrules.style_directives = true +syntaxrules.style_operator_arithmetic = true +syntaxrules.style_operator_boolean = false +syntaxrules.style_operator_integer = true +syntaxrules.style_operator_unary = true +syntaxrules.style_operator_arithmetic_leading_space = true +syntaxrules.style_operator_boolean_leading_space = true +syntaxrules.style_operator_integer_leading_space = true + +syntaxrules.style_keyword_0or1space = true +syntaxrules.style_keyword_0space = true +syntaxrules.style_keyword_1or2space = true +syntaxrules.style_keyword_1space = true +syntaxrules.style_keyword_construct = false +syntaxrules.style_keyword_datatype = false # overly restrictive. +syntaxrules.style_keyword_end = true +syntaxrules.style_keyword_maybelabel = true +syntaxrules.style_keyword_new = true +syntaxrules.style_keyword_newline = true +syntaxrules.style_commaleading = true +syntaxrules.eventlist_or = true + +# Style ruleset end + +# from designintents.toml +syntaxrules.blocking_assignment_in_always_ff = true +syntaxrules.blocking_assignment_in_always_latch = true +syntaxrules.non_blocking_assignment_in_always_comb = true +syntaxrules.case_default = true +syntaxrules.enum_with_type = true +syntaxrules.function_with_automatic = true +syntaxrules.keyword_forbidden_priority = true +syntaxrules.keyword_forbidden_unique = true +syntaxrules.keyword_forbidden_unique0 = true +# TODO syntaxrules.operator_case_equality = true +syntaxrules.procedural_continuous_assignment = true +syntaxrules.action_block_with_side_effect = true +# TODO syntaxrules.default_nettype_none = true +syntaxrules.function_same_as_system_function = true +# TODO syntaxrules.keyword_forbidden_always = true +# TODO syntaxrules.keyword_forbidden_wire_reg = true +syntaxrules.module_nonansi_forbidden = true +syntaxrules.generate_case_with_label = true +syntaxrules.generate_for_with_label = true +syntaxrules.generate_if_with_label = true +syntaxrules.localparam_type_twostate = true +syntaxrules.parameter_type_twostate = true +syntaxrules.localparam_explicit_type = true +# TODO syntaxrules.parameter_explicit_type = true +syntaxrules.parameter_default_value = true +syntaxrules.parameter_in_generate = true +syntaxrules.parameter_in_package = true +syntaxrules.genvar_declaration_in_loop = true +syntaxrules.genvar_declaration_out_loop = false +syntaxrules.keyword_forbidden_generate = true +syntaxrules.keyword_required_generate = false +# TODO syntaxrules.explicit_case_default = true +# TODO syntaxrules.explicit_if_else = true +syntaxrules.loop_statement_in_always_comb = true +syntaxrules.loop_statement_in_always_ff = true +syntaxrules.loop_statement_in_always_latch = true +syntaxrules.sequential_block_in_always_comb = true +syntaxrules.sequential_block_in_always_ff = true +syntaxrules.sequential_block_in_always_latch = true +syntaxrules.inout_with_tri = true +# TODO syntaxrules.input_with_var = true +# TODO syntaxrules.output_with_var = true +syntaxrules.interface_port_with_modport = true diff --git a/uart/.gitignore b/uart/.gitignore new file mode 100644 index 0000000..642d621 --- /dev/null +++ b/uart/.gitignore @@ -0,0 +1,7 @@ +dv/__pycache__ +sim_build +sw/__pycache__ +results.xml + +*.vcd +*.fst diff --git a/uart/Makefile b/uart/Makefile new file mode 100644 index 0000000..1929d3a --- /dev/null +++ b/uart/Makefile @@ -0,0 +1,35 @@ +SIM = verilator +TOPLEVEL_LANG = verilog +VERILOG_SOURCES = \ + $(CURDIR)/rtl/uart_rx.sv \ + $(CURDIR)/rtl/uart_tx.sv \ + $(CURDIR)/rtl/uart.sv + +SVLINT_SOURCES = $(wildcard $(CURDIR)/rtl/*.sv) + +TOPLEVEL = uart +MODULE = test_uart_pyuvm + +WAVES = 1 +EXTRA_ARGS += --timing --trace --trace-structs -Wno-PINMISSING -Wno-WIDTHEXPAND +PLUSARGS += --trace --trace-file dv/dump.vcd + +PYTHONPATH := $(CURDIR)/dv:$(CURDIR)/sw:$(PYTHONPATH) +export PYTHONPATH + +include $(shell cocotb-config --makefiles)/Makefile.sim + +# Post-simulation VCD to FST conversion +post_waves: sim + @if [ -f dv/dump.vcd ]; then \ + vcd2fst dv/dump.vcd dv/dump.fst; \ + echo "Generated dv/dump.fst ($$(du -h dv/dump.fst | cut -f1))"; \ + fi + +# =========================== +# Linting +# =========================== +svlint: + bash -o pipefail -c 'svlint $(if $(CI),--github-actions) $(SVLINT_SOURCES) $(if $(CI),| sed "s/::error/::warning/g")' + +.PHONY: svlint post_waves diff --git a/uart/README.md b/uart/README.md new file mode 100644 index 0000000..89d6adc --- /dev/null +++ b/uart/README.md @@ -0,0 +1,26 @@ +# How to test uart + +## Requirements: + +Python 3.9+ +cocotb — pip install cocotb +cocotbext-uart — pip install cocotbext-uart +Icarus Verilog — brew install icarus-verilog on Mac + +In the uart/ folder: + +## to run all tests: + +make SIM=icarus + +## to run a single test: + +make SIM=icarus COCOTB_TEST_FILTER=test_halt + +## What each test does: + +test_halt — sends HALT, verifies the DUT asserts hold_core +test_wr32 — sends HALT then writes a word, verifies correct address and data appear on the bus +test_load_and_run — full CLI flow: halt, load 20 program words, run. Mirrors cli option 1 in uart_cli.py. Takes ~30 seconds due to real baud-rate timing. +test_rdreg — reads back a register value through the debug interface +test_bad_checksum — verifies the DUT rejects a corrupted packet with STATUS_CHK diff --git a/uart/dv/test_uart_pyuvm.py b/uart/dv/test_uart_pyuvm.py new file mode 100644 index 0000000..a839b15 --- /dev/null +++ b/uart/dv/test_uart_pyuvm.py @@ -0,0 +1,217 @@ +import cocotb +from cocotb.triggers import RisingEdge, FallingEdge, Timer +from cocotb.clock import Clock +import pyuvm +from pyuvm import * +import random + + +# ----------------------------------------------------------------------------- +# 1. Sequence Item +# ----------------------------------------------------------------------------- +class UartItem(uvm_sequence_item): + def __init__(self, name="UartItem", data=0): + super().__init__(name) + self.data = data + + def __str__(self): + return f"UartItem(data=0x{self.data:02X})" + + +# ----------------------------------------------------------------------------- +# 2. Bus Functional Model (BFM) +# ----------------------------------------------------------------------------- +class UartBFM: + def __init__(self, dut): + self.dut = dut + self.rx_listeners = [] + + async def reset(self): + self.dut.rst.value = 1 + self.dut.i_valid_s.value = 0 + self.dut.i_data_s.value = 0 + self.dut.i_ready_m.value = 1 + self.dut.i_rxd.value = 1 + await Timer(100, units="ns") + await RisingEdge(self.dut.clk) + self.dut.rst.value = 0 + await RisingEdge(self.dut.clk) + + async def send_tx_byte(self, byte_val): + while not self.dut.o_ready_s.value: + await RisingEdge(self.dut.clk) + + self.dut.i_data_s.value = byte_val + self.dut.i_valid_s.value = 1 + await RisingEdge(self.dut.clk) + self.dut.i_valid_s.value = 0 + + while self.dut.o_ready_s.value: + await RisingEdge(self.dut.clk) + + async def monitor_rx(self): + while True: + await RisingEdge(self.dut.clk) + if self.dut.o_valid_m.value and self.dut.i_ready_m.value: + received_byte = int(self.dut.o_data_m.value) + for listener in self.rx_listeners: + listener(received_byte) + + async def loopback_wire(self): + """Connect o_txd directly to i_rxd on falling clock edges for stability.""" + while True: + await FallingEdge(self.dut.clk) + self.dut.i_rxd.value = self.dut.o_txd.value + + +# ----------------------------------------------------------------------------- +# 3. Driver +# ----------------------------------------------------------------------------- +class UartDriver(uvm_driver): + def build_phase(self): + self.ap = uvm_analysis_port("ap", self) + self.bfm = ConfigDB().get(self, "", "BFM") + + async def run_phase(self): + while True: + item = await self.seq_item_port.get_next_item() + self.logger.info(f"[TX DRIVER] Transmitting byte: 0x{item.data:02X}") + await self.bfm.send_tx_byte(item.data) + self.ap.write(item) # Publish EXPECTED item via TLM Analysis Port + self.seq_item_port.item_done() + + +# ----------------------------------------------------------------------------- +# 4. Monitor +# ----------------------------------------------------------------------------- +class UartMonitor(uvm_monitor): + def build_phase(self): + self.ap = uvm_analysis_port("ap", self) + self.bfm = ConfigDB().get(self, "", "BFM") + + def connect_phase(self): + self.bfm.rx_listeners.append(self.on_rx_byte) + + def on_rx_byte(self, byte_val): + self.logger.info(f"[RX MONITOR] Sampled byte: 0x{byte_val:02X}") + item = UartItem("rx_item", data=byte_val) + self.ap.write(item) # Publish ACTUAL item via TLM Analysis Port + + +# ----------------------------------------------------------------------------- +# 5. Scoreboard (Canonical Dual TLM FIFO) +# ----------------------------------------------------------------------------- +class UartScoreboard(uvm_scoreboard): + def build_phase(self): + self.expected_fifo = uvm_tlm_analysis_fifo("expected_fifo", self) + self.actual_fifo = uvm_tlm_analysis_fifo("actual_fifo", self) + + self.expected_export = self.expected_fifo.analysis_export + self.actual_export = self.actual_fifo.analysis_export + + self.passed_count = 0 + self.failed_count = 0 + + def check_phase(self): + while self.expected_fifo.can_get() and self.actual_fifo.can_get(): + _, exp_item = self.expected_fifo.try_get() + _, act_item = self.actual_fifo.try_get() + + if exp_item.data == act_item.data: + self.logger.info(f"[SCOREBOARD PASS] Expected=0x{exp_item.data:02X}, Received=0x{act_item.data:02X}") + self.passed_count += 1 + else: + self.logger.error(f"[SCOREBOARD FAIL] Expected=0x{exp_item.data:02X}, Got=0x{act_item.data:02X}") + self.failed_count += 1 + + if self.expected_fifo.can_get(): + self.logger.error(f"[SCOREBOARD FAIL] Unmatched expected bytes remaining in FIFO!") + self.failed_count += 1 + + if self.actual_fifo.can_get(): + self.logger.error(f"[SCOREBOARD FAIL] Unmatched actual bytes remaining in FIFO!") + self.failed_count += 1 + + if self.failed_count == 0: + self.logger.info(f"[SCOREBOARD SUMMARY] All {self.passed_count} items matched successfully!") + + +# ----------------------------------------------------------------------------- +# 6. Agent +# ----------------------------------------------------------------------------- +class UartAgent(uvm_agent): + def build_phase(self): + self.driver = UartDriver("driver", self) + self.sequencer = uvm_sequencer("sequencer", self) + self.monitor = UartMonitor("monitor", self) + + def connect_phase(self): + self.driver.seq_item_port.connect(self.sequencer.seq_item_export) + + +# ----------------------------------------------------------------------------- +# 7. Environment +# ----------------------------------------------------------------------------- +class UartEnv(uvm_env): + def build_phase(self): + self.agent = UartAgent("agent", self) + self.scoreboard = UartScoreboard("scoreboard", self) + + def connect_phase(self): + # Canonical UVM Connections: + # 1. Driver ap -> Scoreboard expected_export + self.agent.driver.ap.connect(self.scoreboard.expected_export) + # 2. Monitor ap -> Scoreboard actual_export + self.agent.monitor.ap.connect(self.scoreboard.actual_export) + + +# ----------------------------------------------------------------------------- +# 8. Sequence +# ----------------------------------------------------------------------------- +class UartRandomSequence(uvm_sequence): + def __init__(self, name="UartRandomSequence", num_items=10): + super().__init__(name) + self.num_items = num_items + + async def body(self): + for _ in range(self.num_items): + val = random.randint(0, 255) + item = UartItem("item", data=val) + await self.start_item(item) + await self.finish_item(item) + + +# ----------------------------------------------------------------------------- +# 9. Test +# ----------------------------------------------------------------------------- +@pyuvm.test() +class UartLoopbackTest(uvm_test): + def build_phase(self): + self.env = UartEnv("env", self) + self.bfm = UartBFM(cocotb.top) + ConfigDB().set(self, "*", "BFM", self.bfm) + + async def run_phase(self): + self.raise_objection() + + # Start clock + cocotb.start_soon(Clock(cocotb.top.clk, 20, units="ns").start()) + + # Start BFM background monitors (loopback wiring & RX monitor) + cocotb.start_soon(self.bfm.loopback_wire()) + cocotb.start_soon(self.bfm.monitor_rx()) + + # Reset DUT + await self.bfm.reset() + + # Run Sequence + seq = UartRandomSequence("random_seq", num_items=5) + await seq.start(self.agent_sequencer()) + + # Wait for last byte to transmit & receive over serial link + await Timer(600, units="us") + + self.drop_objection() + + def agent_sequencer(self): + return self.env.agent.sequencer diff --git a/uart/ed/Makefile b/uart/ed/Makefile new file mode 100644 index 0000000..adab05b --- /dev/null +++ b/uart/ed/Makefile @@ -0,0 +1,25 @@ +SIM = verilator +TOPLEVEL_LANG = verilog +VERILOG_SOURCES = \ + $(CURDIR)/rtl/uart_rx.sv \ + $(CURDIR)/rtl/uart_tx.sv \ + $(CURDIR)/rtl/uart.sv \ + $(CURDIR)/rtl/uart_bus_master.sv \ + $(CURDIR)/rtl/uart_top.sv +TOPLEVEL = uart_top +MODULE = test_uart_master + +EXTRA_ARGS += --timing -Wno-PINMISSING -Wno-WIDTHEXPAND + +PYTHONPATH := $(CURDIR)/dv:$(CURDIR)/sw:$(PYTHONPATH) +export PYTHONPATH + +include $(shell cocotb-config --makefiles)/Makefile.sim + +# =========================== +# Linting +# =========================== +svlint: + bash -o pipefail -c 'svlint $(if $(CI),--github-actions) $(VERILOG_SOURCES) $(if $(CI),| sed "s/::error/::warning/g")' + +.PHONY: svlint diff --git a/uart/ed/test_uart_master.py b/uart/ed/test_uart_master.py new file mode 100644 index 0000000..9d4d144 --- /dev/null +++ b/uart/ed/test_uart_master.py @@ -0,0 +1,208 @@ +import sys +import os +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +import cocotb +from cocotb.clock import Clock +from cocotb.triggers import RisingEdge, Timer +from cocotbext.uart import UartSource, UartSink + +from uart_cli import xor_chk, hx, expect_ack + +BAUD = 115200 + +# changes made: +# - in uart_bus_master: declared state before assigned rx_ready, just something icarus verilator needs +# - created a new top level wrapper with all uart modules to connect to the cocotb uart module +# - makefile + + +async def recv(sink, n: int) -> bytes: + """Wait until n bytes have arrived in the UartSink queue, then return them. + + sink.read() in cocotbext-uart v0.1.4 is synchronous — it raises QueueEmpty + if the bytes haven't arrived yet. sink.wait() is async and suspends until + at least one more byte appears, so we loop until we've collected enough. + """ + data = bytearray() + while len(data) < n: + await sink.wait() + data.extend(sink.read_nowait()) + return bytes(data[:n]) + + +# --------------------------------------------------------------------------- +# Async command helpers +# +# These mirror the cmd_* functions in uart_cli.py exactly, but use +# UartSource/UartSink instead of a pyserial Serial object. The logic — +# packet layout, checksums, response parsing — is identical to what runs +# on real hardware. +# --------------------------------------------------------------------------- + +async def cmd_halt(source, sink): + source.write_nowait(bytes([0xA5, 0x13, 0x13])) + resp = bytes(await recv(sink, 4)) + return expect_ack(resp, "HALT") + + +async def cmd_run(source, sink): + source.write_nowait(bytes([0xA5, 0x12, 0x12])) + resp = bytes(await recv(sink, 4)) + return expect_ack(resp, "RUN") + + +async def cmd_wr32(source, sink, addr: int, data: int): + pkt = bytearray([0xA5, 0x10]) + pkt += addr.to_bytes(4, "little") + pkt += data.to_bytes(4, "little") + pkt += bytes([xor_chk(pkt[1:])]) + source.write_nowait(bytes(pkt)) + resp = bytes(await recv(sink, 4)) + st = expect_ack(resp, "WR32") + if st != 0: + raise RuntimeError(f"WR32 status={st:02x} addr=0x{addr:08x} resp={hx(resp)}") + return st + + +async def cmd_rdreg(source, sink, reg_idx: int) -> int: + reg_idx &= 0x1F + cmd = 0x14 + source.write_nowait(bytes([0xA5, cmd, reg_idx, cmd ^ reg_idx])) + resp = bytes(await recv(sink, 7)) # 5A 92 d0 d1 d2 d3 chk + if resp[0] != 0x5A or resp[1] != 0x92: + raise RuntimeError(f"RDREG bad header: {hx(resp)}") + exp_chk = (0x92 ^ resp[2] ^ resp[3] ^ resp[4] ^ resp[5]) & 0xFF + if resp[6] != exp_chk: + raise RuntimeError(f"RDREG bad chk got {resp[6]:02x} expect {exp_chk:02x} frame={hx(resp)}") + return int.from_bytes(resp[2:6], "little") + + +# --------------------------------------------------------------------------- +# Setup +# --------------------------------------------------------------------------- + +async def reset_dut(dut): + dut.rst.value = 1 + dut.bus_read_data.value = 0 + for i in range(32): + dut.dbg_regs[i].value = 0 + dut.dbg_pc.value = 0 + await Timer(200, units="ns") + dut.rst.value = 0 + await Timer(100, units="ns") + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +@cocotb.test() +async def test_halt(dut): + """HALT travels through uart_rx → uart_bus_master → uart_tx and back.""" + cocotb.start_soon(Clock(dut.clk, 20, units="ns").start()) + source = UartSource(dut.i_rxd, baud=BAUD, bits=8) + sink = UartSink (dut.o_txd, baud=BAUD, bits=8) + await reset_dut(dut) + + await cmd_halt(source, sink) + + assert int(dut.hold_core.value) == 1, "hold_core should be 1 after HALT" + + +@cocotb.test() +async def test_wr32(dut): + """HALT then write 0xDEADBEEF to address 0 and verify the bus outputs. + + bus_write_enable is a one-cycle strobe inside uart_bus_master. A background + watcher catches it the cycle it fires; by the time the ACK arrives over + the serial link the strobe is long gone. + """ + cocotb.start_soon(Clock(dut.clk, 20, units="ns").start()) + source = UartSource(dut.i_rxd, baud=BAUD, bits=8) + sink = UartSink (dut.o_txd, baud=BAUD, bits=8) + await reset_dut(dut) + + await cmd_halt(source, sink) + + captured = {} + async def watch_strobe(): + while True: + await RisingEdge(dut.clk) + if int(dut.bus_write_enable.value) == 0b1111: + captured["addr"] = int(dut.bus_addr.value) + captured["data"] = int(dut.bus_write_data.value) + break + cocotb.start_soon(watch_strobe()) + + await cmd_wr32(source, sink, 0x00000000, 0xDEADBEEF) + + assert captured.get("addr") == 0x00000000, "wrong bus address" + assert captured.get("data") == 0xDEADBEEF, "wrong write data" + + +@cocotb.test() +async def test_load_and_run(dut): + """Full CLI flow: halt, load a program word-by-word, run. + + This replicates exactly what uart_cli.py does in menu option 1 + run, + using the same WORDS from test_uart.py. The entire sequence goes through + uart_rx and uart_tx at real 115200 baud timing. + """ + cocotb.start_soon(Clock(dut.clk, 20, units="ns").start()) + source = UartSource(dut.i_rxd, baud=BAUD, bits=8) + sink = UartSink (dut.o_txd, baud=BAUD, bits=8) + await reset_dut(dut) + + WORDS = [ + 0xFE010113, 0x00112E23, 0x00812C23, 0x00912A23, + 0x02010413, 0x00000493, 0x00148793, 0x3FF7F493, + 0x100007B7, 0x0097A023, 0xFE042623, 0x0100006F, + 0xFEC42783, 0x00178793, 0xFEF42623, 0xFEC42703, + 0x000F47B7, 0x23F78793, 0xFEE7D4E3, 0xFCDFF06F, + ] + + await cmd_halt(source, sink) + + base = 0x00000000 + for i, w in enumerate(WORDS): + addr = base + 4 * i + await cmd_wr32(source, sink, addr, w) + dut._log.info(f" wrote 0x{w:08x} → 0x{addr:08x}") + + await cmd_run(source, sink) + + assert int(dut.hold_core.value) == 0, "hold_core should be 0 after RUN" + + +@cocotb.test() +async def test_rdreg(dut): + """Drive dbg_regs[9] to a known value and verify RDREG echoes it back.""" + cocotb.start_soon(Clock(dut.clk, 20, units="ns").start()) + source = UartSource(dut.i_rxd, baud=BAUD, bits=8) + sink = UartSink (dut.o_txd, baud=BAUD, bits=8) + await reset_dut(dut) + + dut.dbg_regs[9].value = 0xCAFEBABE + + val = await cmd_rdreg(source, sink, 9) + + assert val == 0xCAFEBABE, f"expected 0xCAFEBABE, got {val:#010x}" + + +@cocotb.test() +async def test_bad_checksum(dut): + """Corrupt a packet checksum and verify STATUS_CHK (0x01) is returned.""" + cocotb.start_soon(Clock(dut.clk, 20, units="ns").start()) + source = UartSource(dut.i_rxd, baud=BAUD, bits=8) + sink = UartSink (dut.o_txd, baud=BAUD, bits=8) + await reset_dut(dut) + + good = bytes([0xA5, 0x13, 0x13]) + bad = good[:-1] + bytes([good[-1] ^ 0xFF]) # flip all bits in CHK byte + source.write_nowait(bad) + resp = bytes(await recv(sink, 4)) + + dut._log.info(f"bad-chk resp: {hx(resp)}") + assert resp[0] == 0x5A and resp[1] == 0x90, f"expected ACK frame, got {hx(resp)}" + assert resp[2] == 0x01, f"expected STATUS_CHK 0x01, got {resp[2]:02x}" diff --git a/uart/ed/uart_bus_master.sv b/uart/ed/uart_bus_master.sv new file mode 100644 index 0000000..9ab92e9 --- /dev/null +++ b/uart/ed/uart_bus_master.sv @@ -0,0 +1,293 @@ +//protocol: A: address D: data +//SOF: A5 +//command number(10-13) +//WRITE32: A5 10 A0 A1 A2 A3 D0 D1 D2 D3 CHK +//READ32: A5 11 A0 A1 A2 A3 CHK +//RUN: A5 12 CHK (CHK=0x12) +//HALT: A5 13 CHK (CHK=0x12) +//R_ACK:90 R_RD:91 +module uart_bus_master ( + input wire clk + , input wire rst + , input wire [7:0] rx_data + , input wire rx_valid + , output wire rx_ready + , output logic [7:0] tx_data + , output logic tx_valid + , input wire tx_ready + , output logic [31:0] bus_addr + , output logic [31:0] bus_write_data + , output logic [3:0] bus_write_enable + , input wire [31:0] bus_read_data + , output logic hold_core + , input logic [31:0] dbg_regs [0:31] + , input logic [31:0] dbg_pc +); + + localparam byte SOF = 8'hA5; + localparam byte RSOF = 8'h5A; + + localparam byte CMD_WR32 = 8'h10; + localparam byte CMD_RD32 = 8'h11; + localparam byte CMD_RUN = 8'h12; + localparam byte CMD_HALT = 8'h13; + localparam byte CMD_RDREG = 8'h14; + localparam byte R_ACK = 8'h90; + localparam byte R_RD = 8'h91; + localparam byte R_RDREG = 8'h92; + + localparam byte STATUS_OK = 8'h00; + localparam byte STATUS_CHK = 8'h01; + localparam byte STATUS_BUSY = 8'h02; + localparam byte STATUS_CMD = 8'h03; + + typedef enum logic [4:0] { + STATE_WAIT_SOF = 5'd0 + , STATE_CMD = 5'd1 + , STATE_A0 = 5'd2 + , STATE_A1 = 5'd3 + , STATE_A2 = 5'd4 + , STATE_A3 = 5'd5 + , STATE_D0 = 5'd6 + , STATE_D1 = 5'd7 + , STATE_D2 = 5'd8 + , STATE_D3 = 5'd9 + , STATE_CHK = 5'd10 + , STATE_DO_WR = 5'd11 + , STATE_DO_RD0 = 5'd12 + , STATE_DO_RD1 = 5'd13 + , STATE_DO_RD2 = 5'd14 + , STATE_SEND = 5'd15 + , STATE_REG = 5'd16 + } state_t; + + state_t state; + assign rx_ready = (state != STATE_SEND); + + logic [4:0] reg_idx; + logic [7:0] cmd; + logic [31:0] addr; + logic [31:0] wdata; + logic [7:0] chk_calc; + + logic [7:0] resp [0:6]; + logic [2:0] resp_len; + logic [2:0] resp_idx; + + task automatic prepare_ack(input byte status); + begin + // 5A 90 STATUS CHK + resp[0] = RSOF; + resp[1] = R_ACK; + resp[2] = status; + resp[3] = (R_ACK ^ status); + resp_len <= 3'd4; + resp_idx <= 3'd0; + state <= STATE_SEND; + end + endtask + + task automatic prepare_rd(input logic [31:0] d); + begin + // 5A 91 d0 d1 d2 d3 chk + resp[0] = RSOF; + resp[1] = R_RD; + resp[2] = d[7:0]; + resp[3] = d[15:8]; + resp[4] = d[23:16]; + resp[5] = d[31:24]; + resp[6] = (R_RD ^ d[7:0] ^ d[15:8] ^ d[23:16] ^ d[31:24]); + resp_len <= 3'd7; + resp_idx <= 3'd0; + state <= STATE_SEND; + end + endtask + + task automatic prepare_rdreg(input logic [31:0] d); + begin + // 5A 92 d0 d1 d2 d3 chk + resp[0] = RSOF; + resp[1] = R_RDREG; + resp[2] = d[7:0]; + resp[3] = d[15:8]; + resp[4] = d[23:16]; + resp[5] = d[31:24]; + resp[6] = (R_RDREG ^ d[7:0] ^ d[15:8] ^ d[23:16] ^ d[31:24]); + resp_len <= 3'd7; + resp_idx <= 3'd0; + state <= STATE_SEND; + end + endtask + + always @(posedge clk) begin + if (rst) begin + bus_addr <= 32'd0; + bus_write_data <= 32'd0; + bus_write_enable <= 4'b0000; + + tx_data <= 8'd0; + tx_valid <= 1'b0; + + state <= STATE_WAIT_SOF; + cmd <= 8'd0; + addr <= 32'd0; + wdata <= 32'd0; + chk_calc <= 8'd0; + + resp_len <= 3'd0; + resp_idx <= 3'd0; + + hold_core <= 1'b1; + end else begin + bus_write_enable <= 4'b0000; + if (tx_valid && tx_ready) tx_valid <= 1'b0; + + if (state == STATE_SEND) begin + if (!tx_valid && tx_ready) begin + tx_data <= resp[resp_idx]; + tx_valid <= 1'b1; + if (resp_idx == resp_len - 1) begin + state <= STATE_WAIT_SOF; + resp_idx <= 3'd0; + end else begin + resp_idx <= resp_idx + 3'd1; + end + end + + end else begin + case (state) + STATE_WAIT_SOF: begin + if (rx_valid & rx_ready && rx_data == SOF) begin + state <= STATE_CMD; + end + end + + STATE_CMD: if (rx_valid & rx_ready) begin + cmd <= rx_data; + chk_calc <= rx_data; + addr <= 32'd0; + wdata <= 32'd0; + + if (rx_data == CMD_RUN || rx_data == CMD_HALT) begin + state <= STATE_CHK; + end else if (rx_data == CMD_RDREG) begin + state <= STATE_REG; + end else begin + state <= STATE_A0; + end + end + + STATE_A0: if (rx_valid & rx_ready) begin + addr[7:0] <= rx_data; + chk_calc <= chk_calc ^ rx_data; + state <= STATE_A1; + end + + STATE_A1: if (rx_valid & rx_ready) begin + addr[15:8] <= rx_data; + chk_calc <= chk_calc ^ rx_data; + state <= STATE_A2; + end + + STATE_A2: if (rx_valid & rx_ready) begin + addr[23:16] <= rx_data; + chk_calc <= chk_calc ^ rx_data; + state <= STATE_A3; + end + + STATE_A3: if (rx_valid & rx_ready) begin + addr[31:24] <= rx_data; + chk_calc <= chk_calc ^ rx_data; + if (cmd == CMD_WR32) state <= STATE_D0; + else state <= STATE_CHK; // RD32 no data packet + end + + STATE_D0: if (rx_valid & rx_ready) begin + wdata[7:0] <= rx_data; + chk_calc <= chk_calc ^ rx_data; + state <= STATE_D1; + end + + STATE_D1: if (rx_valid & rx_ready) begin + wdata[15:8] <= rx_data; + chk_calc <= chk_calc ^ rx_data; + state <= STATE_D2; + end + + STATE_D2: if (rx_valid & rx_ready) begin + wdata[23:16] <= rx_data; + chk_calc <= chk_calc ^ rx_data; + state <= STATE_D3; + end + + STATE_D3: if (rx_valid & rx_ready) begin + wdata[31:24] <= rx_data; + chk_calc <= chk_calc ^ rx_data; + state <= STATE_CHK; + end + + STATE_REG: if (rx_valid & rx_ready) begin + reg_idx <= rx_data[4:0]; + chk_calc <= chk_calc ^ rx_data; + state <= STATE_CHK; + end + + STATE_CHK: if (rx_valid & rx_ready) begin + if (rx_data != chk_calc) begin + prepare_ack(STATUS_CHK); + end else begin + case (cmd) + CMD_HALT: begin + hold_core <= 1'b1; + prepare_ack(STATUS_OK); + end + CMD_RUN: begin + hold_core <= 1'b0; + prepare_ack(STATUS_OK); + end + CMD_WR32: begin + if (!hold_core) prepare_ack(STATUS_BUSY); + else state <= STATE_DO_WR; + end + CMD_RD32: begin + if (!hold_core) prepare_ack(STATUS_BUSY); + else state <= STATE_DO_RD0; + end + CMD_RDREG: begin + prepare_rdreg(dbg_regs[reg_idx]); + end + default: begin + prepare_ack(STATUS_CMD); + end + endcase + end + end + + STATE_DO_WR: begin + bus_addr <= addr; + bus_write_data <= wdata; + bus_write_enable <= 4'b1111; + prepare_ack(STATUS_OK); + end + + STATE_DO_RD0: begin + bus_addr <= addr; + state <= STATE_DO_RD1; + end + + STATE_DO_RD1: begin + state <= STATE_DO_RD2; + end + + STATE_DO_RD2: begin + prepare_rd(bus_read_data); + end + + default: state <= STATE_WAIT_SOF; + endcase + end + end + end + +endmodule + diff --git a/uart/ed/uart_top.sv b/uart/ed/uart_top.sv new file mode 100644 index 0000000..a4cd14a --- /dev/null +++ b/uart/ed/uart_top.sv @@ -0,0 +1,77 @@ +`timescale 1ns / 1ps + +// Top-level wrapper for cocotb full-chain testing. +// Connects uart.sv (bit-level RX/TX) to uart_bus_master.sv (protocol FSM) +// so the testbench only needs to drive i_rxd / o_txd serial lines. +module uart_top + #(parameter DATA_WIDTH = 8 + , parameter CLK_HZ = 50000000 + , parameter BAUD = 115200 + ) + ( input wire clk + , input wire rst + + , input wire i_rxd + , output wire o_txd + + , output wire [31:0] bus_addr + , output wire [31:0] bus_write_data + , output wire [3:0] bus_write_enable + , input wire [31:0] bus_read_data + , output wire hold_core + + , input logic [31:0] dbg_regs [0:31] + , input logic [31:0] dbg_pc + ); + + // Internal byte-level bus between uart and uart_bus_master + wire [7:0] rx_data; + wire rx_valid; + wire rx_ready; + wire [7:0] tx_data; + wire tx_valid; + wire tx_ready; + + uart + #(.DATA_WIDTH(DATA_WIDTH) + , .CLK_HZ(CLK_HZ) + , .BAUD(BAUD) + ) + uart_inst + ( .clk(clk) + , .rst(rst) + + // TX: bus_master → uart_tx → serial + , .i_data_s(tx_data) + , .i_valid_s(tx_valid) + , .o_ready_s(tx_ready) + + // RX: serial → uart_rx → bus_master + , .o_data_m(rx_data) + , .o_valid_m(rx_valid) + , .i_ready_m(rx_ready) + + // Serial pins + , .i_rxd(i_rxd) + , .o_txd(o_txd) + ); + + uart_bus_master bus_master_inst + ( .clk(clk) + , .rst(rst) + , .rx_data(rx_data) + , .rx_valid(rx_valid) + , .rx_ready(rx_ready) + , .tx_data(tx_data) + , .tx_valid(tx_valid) + , .tx_ready(tx_ready) + , .bus_addr(bus_addr) + , .bus_write_data(bus_write_data) + , .bus_write_enable(bus_write_enable) + , .bus_read_data(bus_read_data) + , .hold_core(hold_core) + , .dbg_regs(dbg_regs) + , .dbg_pc(dbg_pc) + ); + +endmodule diff --git a/uart/rtl/uart.sv b/uart/rtl/uart.sv new file mode 100644 index 0000000..0f29fc0 --- /dev/null +++ b/uart/rtl/uart.sv @@ -0,0 +1,60 @@ +`timescale 1ns / 1ps + +module uart + #(parameter DATA_WIDTH = 8 + , parameter CLK_HZ = 50000000 + , parameter BAUD = 115200 + ) + ( input wire clk + , input wire rst + , input wire [DATA_WIDTH - 1:0] i_data_s + , input wire i_valid_s + , output wire o_ready_s + , output wire [DATA_WIDTH - 1:0] o_data_m + , output wire o_valid_m + , input wire i_ready_m + , input wire i_rxd + , output wire o_txd + , output wire o_tx_busy + , output wire o_rx_busy + , output wire o_rx_overrun_error + , output wire o_rx_frame_error + ); + + // clocks per bit + //localparam integer CLK_HZ = 50000000; + //localparam integer BAUD = 115200; + localparam int DIV = (CLK_HZ / BAUD); // 50e6/115200 ≈ 434 + + uart_tx + #(.DATA_WIDTH(DATA_WIDTH) + , .DIV(DIV) + ) + uart_tx_inst + ( .clk(clk) + , .rst(rst) + , .i_data(i_data_s) + , .i_valid(i_valid_s) + , .o_ready(o_ready_s) + , .o_txd(o_txd) + , .o_busy(o_tx_busy) + ); + + uart_rx + #(.DATA_WIDTH(DATA_WIDTH) + , .DIV(DIV) + ) + uart_rx_inst + ( .clk(clk) + , .rst(rst) + , .o_data(o_data_m) + , .o_valid(o_valid_m) + , .i_ready(i_ready_m) + , .i_rxd(i_rxd) + , .o_busy(o_rx_busy) + , .o_overrun_error(o_rx_overrun_error) + , .o_frame_error(o_rx_frame_error) + ); + + +endmodule diff --git a/uart/rtl/uart_rx.sv b/uart/rtl/uart_rx.sv new file mode 100644 index 0000000..cda86ce --- /dev/null +++ b/uart/rtl/uart_rx.sv @@ -0,0 +1,137 @@ +`timescale 1ns / 1ps + +module uart_rx + #(parameter DATA_WIDTH = 8 + , parameter DIV = 434 // 50e6/115200 ≈ 434 + ) + ( input wire clk + , input wire rst + , output logic [DATA_WIDTH - 1:0] o_data + , output logic o_valid + , input wire i_ready + , input wire i_rxd + , output logic o_busy + , output logic o_overrun_error + , output logic o_frame_error + ); + + reg rxd_q0; + reg rxd_q1; + + // detect falling edge + reg rxd_q1_d = 1'b1; + wire start_fall = (rxd_q1_d == 1'b1) && (rxd_q1 == 1'b0); + + typedef enum logic [1:0] { + STATE_IDLE = 2'd0 + , STATE_START = 2'd1 + , STATE_DATA = 2'd2 + , STATE_STOP = 2'd3 + } uart_state_t; + uart_state_t state = STATE_IDLE; + + localparam int TIMER_W = (DIV <= 1) ? 1 : $clog2(DIV); + localparam int BIT_IDX_W = (DATA_WIDTH <= 1) ? 1 : $clog2(DATA_WIDTH); + + reg [TIMER_W -1:0] timer; + reg [BIT_IDX_W -1:0] bit_idx; + reg [DATA_WIDTH - 1:0] data_reg; + + always @(posedge clk) begin + if (rst) begin + o_data <= 0; + o_valid <= 0; + o_busy <= 0; + o_overrun_error <= 0; + o_frame_error <= 0; + rxd_q0 <= 1'b1; + rxd_q1 <= 1'b1; + rxd_q1_d <= 1'b1; + state <= STATE_IDLE; + timer <= {TIMER_W{1'b0}}; + bit_idx <= {BIT_IDX_W{1'b0}}; + data_reg <= 0; + end else begin + // 2FF synchronizer because aynchronous input + rxd_q0 <= i_rxd; + rxd_q1 <= rxd_q0; + rxd_q1_d <= rxd_q1; + o_overrun_error <= 1'b0; + o_frame_error <= 1'b0; + + if (o_valid && i_ready) begin + o_valid <= 1'b0; + end + + case (state) + STATE_IDLE: begin + o_busy <= 1'b0; + bit_idx <= 0; + timer <= 0; + + if (start_fall) begin + o_busy <= 1'b1; + state <= STATE_START; + timer <= (DIV / 2) - 1; // center + end + end + + STATE_START: begin + o_busy <= 1'b1; + if (timer != 0) begin + timer <= timer - 1'b1; + end else begin + if (rxd_q1 == 1'b0) begin + state <= STATE_DATA; + bit_idx <= {BIT_IDX_W{1'b0}}; + data_reg <= 0; + timer <= DIV - 1; + end else begin + ///starting bit not 0, error + state <= STATE_IDLE; + o_frame_error <= 1'b1; + end + end + end + + STATE_DATA: begin + o_busy <= 1'b1; + if (timer != 0) begin + timer <= timer - 1'b1; + end else begin + data_reg[bit_idx] <= rxd_q1; // LSB first + + if (bit_idx == DATA_WIDTH - 1) begin + state <= STATE_STOP; + timer <= DIV - 1; + end else begin + bit_idx <= bit_idx + 1'b1; + timer <= DIV - 1; + end + end + end + + STATE_STOP: begin + o_busy <= 1'b1; + if (timer != 0) begin + timer <= timer - 1'b1; + end else begin + // stop bit should be 1 + if (rxd_q1 == 1'b1) begin + o_data <= data_reg; + // previous data is still there + o_overrun_error <= o_valid; + o_valid <= 1'b1; + end else begin + o_frame_error <= 1'b1; + end + state <= STATE_IDLE; + end + end + + default: state <= STATE_IDLE; + endcase + end + end + +endmodule diff --git a/uart/rtl/uart_tx.sv b/uart/rtl/uart_tx.sv new file mode 100644 index 0000000..d472d8e --- /dev/null +++ b/uart/rtl/uart_tx.sv @@ -0,0 +1,78 @@ +`timescale 1ns / 1ps + +module uart_tx + #(parameter DATA_WIDTH = 8 + , parameter DIV = 434 // 50e6/115200 ≈ 434 + ) + ( input wire clk + , input wire rst + , input wire [DATA_WIDTH - 1:0] i_data + , input wire i_valid + , output logic o_ready + , output logic o_txd + , output logic o_busy + ); + + // 1 start + DATA_WIDTH data + 1 stop + localparam int FRAME_BITS = DATA_WIDTH + 2; + + localparam int BIT_IDX_W = (FRAME_BITS <= 1) ? 1 : $clog2(FRAME_BITS); + localparam int TIMER_W = (DIV <= 1) ? 1 : $clog2(DIV); + + reg [FRAME_BITS - 1:0] data_reg = {FRAME_BITS{1'b1}}; + reg [BIT_IDX_W -1:0] bit_idx; + reg [TIMER_W -1:0] timer; + + typedef enum logic {STATE_IDLE, STATE_SEND} uart_tx_state_t; + uart_tx_state_t state; + + always @(posedge clk) begin + if (rst) begin + o_ready <= 1'b1; + o_txd <= 1'b1; + o_busy <= 1'b0; + data_reg <= {FRAME_BITS{1'b1}}; + bit_idx <= {BIT_IDX_W{1'b0}}; + timer <= {TIMER_W{1'b0}}; + state <= STATE_IDLE; + end else begin + case (state) + STATE_IDLE: begin + o_txd <= 1'b1; + o_busy <= 1'b0; + o_ready <= 1'b1; + + if (i_valid) begin + data_reg <= {1'b1, i_data, 1'b0}; + bit_idx <= {BIT_IDX_W{1'b0}}; + o_txd <= 1'b0; + timer <= DIV - 1; + o_busy <= 1'b1; + o_ready <= 1'b0; + state <= STATE_SEND; + end + end + + STATE_SEND: begin + o_busy <= 1'b1; + o_ready <= 1'b0; + if (timer != 0) begin + timer <= timer - 1'b1; + end else begin + bit_idx <= bit_idx + 1'b1; + // shift 1 bit so that LSB is the data out + data_reg <= {1'b1, data_reg[FRAME_BITS - 1:1]}; + o_txd <= data_reg[1]; + timer <= DIV - 1; + if (bit_idx == FRAME_BITS - 1) begin + state <= STATE_IDLE; + end + end + end + + default: state <= STATE_IDLE; + endcase + end + end + +endmodule diff --git a/uart/sw/test_uart.py b/uart/sw/test_uart.py new file mode 100644 index 0000000..e74a3f7 --- /dev/null +++ b/uart/sw/test_uart.py @@ -0,0 +1,129 @@ +import serial +import time + +COM = "COM8" # change to your device's serial port +BAUD = 115200 +TIMEOUT = 1.0 + +WORDS = [ + 0xFE010113, 0x00112E23, 0x00812C23, 0x00912A23, + 0x02010413, 0x00000493, 0x00148793, 0x3FF7F493, + 0x100007B7, 0x0097A023, 0xFE042623, 0x0100006F, + 0xFEC42783, 0x00178793, 0xFEF42623, 0xFEC42703, + 0x000F47B7, 0x23F78793, 0xFEE7D4E3, 0xFCDFF06F, +] + +def hx(b: bytes) -> str: + return " ".join(f"{x:02x}" for x in b) + +def xor_chk(bs: bytes) -> int: + c = 0 + for b in bs: + c ^= b + return c & 0xFF + +def read_exact(ser, n) -> bytes: + d = ser.read(n) + if len(d) != n: + raise RuntimeError(f"need {n}, got {len(d)}: {hx(d)}") + return d + +def expect_ack(resp: bytes, label: str) -> int: + + if len(resp) != 4 or resp[0] != 0x5A or resp[1] != 0x90: + raise RuntimeError(f"{label}: not ACK: {hx(resp)}") + status = resp[2] + chk = resp[3] + exp = (0x90 ^ status) & 0xFF + if chk != exp: + raise RuntimeError(f"{label}: bad ACK chk got {chk:02x} expect {exp:02x} frame={hx(resp)}") + return status + +def cmd_halt(ser): + ser.write(bytes([0xA5, 0x13, 0x13])) # CHK=0x13 + resp = read_exact(ser, 4) + st = expect_ack(resp, "HALT") + print("HALT resp:", hx(resp), "status=", hex(st)) + return st + +def cmd_run(ser): + ser.write(bytes([0xA5, 0x12, 0x12])) # CHK=0x12 + resp = read_exact(ser, 4) + st = expect_ack(resp, "RUN") + print("RUN resp:", hx(resp), "status=", hex(st)) + return st + +def cmd_wr32(ser, addr: int, data: int): + pkt = bytearray([0xA5, 0x10]) + pkt += addr.to_bytes(4, "little") + pkt += data.to_bytes(4, "little") + pkt += bytes([xor_chk(pkt[1:])]) + ser.write(pkt) + resp = read_exact(ser, 4) + st = expect_ack(resp, "WR32") + if st != 0: + raise RuntimeError(f"WR32 status={st:02x} addr=0x{addr:08x} resp={hx(resp)}") + return st + +def cmd_rdreg(ser, reg_idx: int) -> int: + reg_idx &= 0x1F + cmd = 0x14 + chk = cmd ^ reg_idx + ser.write(bytes([0xA5, cmd, reg_idx, chk])) + + hdr = read_exact(ser, 2) + + if hdr == bytes([0x5A, 0x90]): + tail = read_exact(ser, 2) + st = expect_ack(hdr + tail, "RDREG(ACK)") + raise RuntimeError(f"RDREG returned ACK status=0x{st:02x} frame={hx(hdr+tail)}") + + if hdr != bytes([0x5A, 0x92]): + rest = ser.read(16) + raise RuntimeError(f"RDREG bad header: {hx(hdr)} rest={hx(rest)}") + + rest = read_exact(ser, 5) # d0 d1 d2 d3 chk + d0, d1, d2, d3, rcv_chk = rest + exp_chk = (0x92 ^ d0 ^ d1 ^ d2 ^ d3) & 0xFF + if rcv_chk != exp_chk: + raise RuntimeError(f"RDREG bad chk got {rcv_chk:02x} expect {exp_chk:02x} frame={hx(hdr+rest)}") + + return int.from_bytes(bytes([d0, d1, d2, d3]), "little") + +def main(): + ser = serial.Serial(COM, BAUD, timeout=TIMEOUT) + time.sleep(0.2) + ser.reset_input_buffer() + ser.reset_output_buffer() + + cmd_halt(ser) + + print("Loading program...") + base = 0x00000000 + for i, w in enumerate(WORDS): + addr = base + 4*i + cmd_wr32(ser, addr, w) + if (i % 4) == 3: + print(f" wrote up to 0x{addr:08x}") + + cmd_run(ser) + print("Core running. LEDR should be counting now.") + + print("Reading cnt from x9 (s1).") + last = None + try: + while True: + v = cmd_rdreg(ser, 9) # x9 = cnt + if last is None: + print(f"cnt = {v:4d} (0x{v:08x})") + else: + print(f"cnt = {v:4d} (0x{v:08x}) ") + last = v + time.sleep(0.5) + except KeyboardInterrupt: + print("\nStopped.") + finally: + ser.close() + +if __name__ == "__main__": + main() diff --git a/uart/sw/uart_cli.py b/uart/sw/uart_cli.py new file mode 100644 index 0000000..bcc08ca --- /dev/null +++ b/uart/sw/uart_cli.py @@ -0,0 +1,193 @@ +import serial +import time + + +def hx(b: bytes) -> str: + return " ".join(f"{x:02x}" for x in b) + +def xor_chk(bs: bytes) -> int: + c = 0 + for b in bs: + c ^= b + return c & 0xFF + +def read_exact(ser, n) -> bytes: + d = ser.read(n) + if len(d) != n: + raise RuntimeError(f"need {n}, got {len(d)}: {hx(d)}") + return d + +def expect_ack(resp: bytes, label: str) -> int: + + if len(resp) != 4 or resp[0] != 0x5A or resp[1] != 0x90: + raise RuntimeError(f"{label}: not ACK: {hx(resp)}") + status = resp[2] + chk = resp[3] + exp = (0x90 ^ status) & 0xFF + if chk != exp: + raise RuntimeError(f"{label}: bad ACK chk got {chk:02x} expect {exp:02x} frame={hx(resp)}") + return status + +def cmd_halt(ser): + ser.write(bytes([0xA5, 0x13, 0x13])) # CHK=0x13 + resp = read_exact(ser, 4) + st = expect_ack(resp, "HALT") + print("HALT resp:", hx(resp), "status=", hex(st)) + return st + +def cmd_run(ser): + ser.write(bytes([0xA5, 0x12, 0x12])) # CHK=0x12 + resp = read_exact(ser, 4) + st = expect_ack(resp, "RUN") + print("RUN resp:", hx(resp), "status=", hex(st)) + return st + +def cmd_wr32(ser, addr: int, data: int): + pkt = bytearray([0xA5, 0x10]) + pkt += addr.to_bytes(4, "little") + pkt += data.to_bytes(4, "little") + pkt += bytes([xor_chk(pkt[1:])]) + ser.write(pkt) + resp = read_exact(ser, 4) + st = expect_ack(resp, "WR32") + if st != 0: + raise RuntimeError(f"WR32 status={st:02x} addr=0x{addr:08x} resp={hx(resp)}") + return st + +def cmd_rdreg(ser, reg_idx: int) -> int: + reg_idx &= 0x1F + cmd = 0x14 + chk = cmd ^ reg_idx + ser.write(bytes([0xA5, cmd, reg_idx, chk])) + + hdr = read_exact(ser, 2) + + if hdr == bytes([0x5A, 0x90]): + tail = read_exact(ser, 2) + st = expect_ack(hdr + tail, "RDREG(ACK)") + raise RuntimeError(f"RDREG returned ACK status=0x{st:02x} frame={hx(hdr+tail)}") + + if hdr != bytes([0x5A, 0x92]): + rest = ser.read(16) + raise RuntimeError(f"RDREG bad header: {hx(hdr)} rest={hx(rest)}") + + rest = read_exact(ser, 5) # d0 d1 d2 d3 chk + d0, d1, d2, d3, rcv_chk = rest + exp_chk = (0x92 ^ d0 ^ d1 ^ d2 ^ d3) & 0xFF + if rcv_chk != exp_chk: + raise RuntimeError(f"RDREG bad chk got {rcv_chk:02x} expect {exp_chk:02x} frame={hx(hdr+rest)}") + + return int.from_bytes(bytes([d0, d1, d2, d3]), "little") + + +def parse_hex_file(filename): + try: + with open(filename, 'r') as f: + return [int(line.strip(), 16) for line in f if line.strip()] + except FileNotFoundError: + print(f"Wow there! Error: {filename} not found.") + return [] + except ValueError: + print(f"Hold it right there. Error: {filename} contains invalid hex data.") + return [] + + +def main_menu(ser): + print(" ___ _______________ ____") + print(" / _ \\/ _/ __/ ___/ / / _/") + print(" / , _// /_\\ \\/ /__/ /___/ / ") + print("/_/|_/___/___/\\___/____/___/ ") + print() + print("Pick an option (1-4)") + print("1. Load program in memory") + print("2. Run program in memory") + print("3. Instruction console") + print("4. Exit") + + try: + command = int(input("?> ")) + except ValueError: + print("Invalid selection. Please enter a number 1-4.") + return 0 + + if command == 1: # upload program + + filename = input("Input filename > ").strip() + WORDS = parse_hex_file(filename) + if not WORDS: + print("No data loaded; aborting upload.") + return 0 + + print("Loading program...") # TODO: Add proper RV31I assembly instruction support + base = 0x00000000 + for i, w in enumerate(WORDS): + addr = base + 4*i + cmd_wr32(ser, addr, w) + if (i % 4) == 3: + print(f"Wrote up to 0x{addr:08x}") + print("Program loaded!") + + run = input("Run program? (y/n) > ").lower() + if run == "y": + cmd_run(ser) + print("Program execution completed.") + return 0 + + if command == 2: + run = input("Run program? (y/n) > ").lower() + if run == "y": + cmd_run(ser) + print("Program execution completed.") + return 0 + + if command == 3: # TODO: Add ability to send instructions sequentially in console interface + print("Instruction console not implemented yet.") + return 0 + + if command == 4: + return 1 + + print("Invalid selection. Please choose 1-4.") + return 0 + + +def main(): + + user_ready = False + COM = "" + BAUD = 0 + + while (not user_ready): + COM = input("Input serial port (e.g. COM8) > ") # TODO: Add dynamic serial port detection and selection instead of manual input + BAUD = int(input("Input baud rate (e.g. 115200) > ")) + ready = input(f"Confirm serial port: {COM}, baud rate: {BAUD}, (y/n) > ").lower() + if ready == "y": + user_ready = True + + TIMEOUT = 1.0 + WORDS = [] + + try: + ser = serial.Serial(COM, BAUD, timeout=TIMEOUT) + time.sleep(0.2) + print(f"{COM} opened successfully!") + except serial.serialutil.SerialException as e: + print(f"Oops! An error occurred with the serial connection: {e}") + return + except Exception as e: + print(f"Oh no! An unexpected error occurred: {e}") + return + + ser.reset_input_buffer() + ser.reset_output_buffer() + + cmd_halt(ser) + + while (main_menu(ser) != 1): + pass + ser.close() + print("Serial port closed. Goodbye.") + + +if __name__ == "__main__": + main()