|
| 1 | +# amaranth: UnusedElaboratable=no |
| 2 | + |
| 3 | +import unittest |
| 4 | +from amaranth import * |
| 5 | +from amaranth.sim import * |
| 6 | + |
| 7 | +from amaranth_orchard.io import UARTPeripheral |
| 8 | + |
| 9 | +class PeripheralTestCase(unittest.TestCase): |
| 10 | + |
| 11 | + async def _csr_access(self, ctx, dut, addr, r_stb=0, w_stb=0, w_data=0, r_data=0): |
| 12 | + ctx.set(dut.bus.addr, addr) |
| 13 | + ctx.set(dut.bus.r_stb, r_stb) |
| 14 | + ctx.set(dut.bus.w_stb, w_stb) |
| 15 | + ctx.set(dut.bus.w_data, w_data) |
| 16 | + await ctx.tick() |
| 17 | + if r_stb: |
| 18 | + self.assertEqual(ctx.get(dut.bus.r_data), r_data) |
| 19 | + ctx.set(dut.bus.r_stb, 0) |
| 20 | + ctx.set(dut.bus.w_stb, 0) |
| 21 | + |
| 22 | + def test_smoke_test(self): |
| 23 | + """ |
| 24 | + Smoke test GPIO. We assume that amaranth-soc GPIO tests are fully testing functionality. |
| 25 | + """ |
| 26 | + rx_addr = 0x00 |
| 27 | + tx_addr = 0x10 |
| 28 | + config_addr = 0x00 |
| 29 | + phy_config_addr = 0x04 |
| 30 | + status_addr = 0x08 |
| 31 | + data_addr = 0x09 |
| 32 | + |
| 33 | + sim_divisor = 8 |
| 34 | + |
| 35 | + dut = UARTPeripheral(init_divisor=sim_divisor, addr_width=5) |
| 36 | + |
| 37 | + # simulated UART receiver |
| 38 | + last_sent_byte = None |
| 39 | + async def uart_rx_proc(ctx): |
| 40 | + nonlocal last_sent_byte |
| 41 | + counter = 0 |
| 42 | + tx_last = 0 |
| 43 | + sr = 0 |
| 44 | + async for clk_edge, rst, tx in ctx.tick().sample(dut.pins.tx.o): |
| 45 | + if rst: |
| 46 | + pass |
| 47 | + elif clk_edge: |
| 48 | + if counter == 0: # waiting for start transition |
| 49 | + if tx_last and not tx: # falling edge |
| 50 | + counter = 1 |
| 51 | + else: |
| 52 | + counter += 1 |
| 53 | + if (counter > (sim_divisor // 2)) and ((counter - (sim_divisor // 2)) % sim_divisor) == 0: # in middle of bit |
| 54 | + bit = ((counter - (sim_divisor // 2)) // sim_divisor) |
| 55 | + if bit >= 1 and bit <= 8: |
| 56 | + sr = (tx << 7) | (sr >> 1) |
| 57 | + if bit == 8: |
| 58 | + last_sent_byte = sr |
| 59 | + elif bit == 9: # stop |
| 60 | + counter = 0 |
| 61 | + tx_last = tx |
| 62 | + |
| 63 | + to_send_byte = None |
| 64 | + async def uart_tx_proc(ctx): |
| 65 | + nonlocal to_send_byte |
| 66 | + counter = 0 |
| 67 | + ctx.set(dut.pins.rx.i, 1) |
| 68 | + async for clk_edge, rst in ctx.tick().sample(): |
| 69 | + if rst: |
| 70 | + pass |
| 71 | + elif clk_edge: |
| 72 | + if to_send_byte is not None: |
| 73 | + bit = counter // sim_divisor |
| 74 | + if bit == 0: |
| 75 | + ctx.set(dut.pins.rx.i, 0) # start |
| 76 | + elif bit >= 1 and bit <= 8: |
| 77 | + ctx.set(dut.pins.rx.i, (to_send_byte >> (bit - 1)) & 0x1) |
| 78 | + if bit > 8: |
| 79 | + ctx.set(dut.pins.rx.i, 1) # stop |
| 80 | + to_send_byte = None |
| 81 | + counter = 0 |
| 82 | + else: |
| 83 | + counter += 1 |
| 84 | + |
| 85 | + async def testbench(ctx): |
| 86 | + nonlocal to_send_byte |
| 87 | + # enable tx |
| 88 | + await self._csr_access(ctx, dut, tx_addr|config_addr, w_stb=1, w_data=1) |
| 89 | + # check tx ready |
| 90 | + await self._csr_access(ctx, dut, tx_addr|status_addr, r_stb=1, r_data=1) |
| 91 | + # write byte |
| 92 | + await self._csr_access(ctx, dut, tx_addr|data_addr, w_stb=1, w_data=0xA5) |
| 93 | + await ctx.tick() |
| 94 | + # check tx not ready |
| 95 | + await self._csr_access(ctx, dut, tx_addr|status_addr, r_stb=1, r_data=0) |
| 96 | + # wait for UART to do its thing |
| 97 | + for i in range(sim_divisor * 12): |
| 98 | + await ctx.tick() |
| 99 | + # check tx ready |
| 100 | + await self._csr_access(ctx, dut, tx_addr|status_addr, r_stb=1, r_data=1) |
| 101 | + # check the byte was sent correctly |
| 102 | + self.assertEqual(last_sent_byte, 0xA5) |
| 103 | + |
| 104 | + # enable rx |
| 105 | + await self._csr_access(ctx, dut, rx_addr|config_addr, w_stb=1, w_data=1) |
| 106 | + # check rx not ready |
| 107 | + await self._csr_access(ctx, dut, rx_addr|status_addr, r_stb=1, r_data=0) |
| 108 | + for i in range(sim_divisor): |
| 109 | + await ctx.tick() |
| 110 | + # send a byte to the UART |
| 111 | + to_send_byte = 0x73 |
| 112 | + for i in range(sim_divisor * 12): |
| 113 | + await ctx.tick() |
| 114 | + # check rx ready |
| 115 | + await self._csr_access(ctx, dut, rx_addr|status_addr, r_stb=1, r_data=1) |
| 116 | + # check the byte was received correctly |
| 117 | + await self._csr_access(ctx, dut, rx_addr|data_addr, r_stb=1, r_data=0x73) |
| 118 | + |
| 119 | + sim = Simulator(dut) |
| 120 | + sim.add_clock(1e-6) |
| 121 | + sim.add_process(uart_rx_proc) |
| 122 | + sim.add_process(uart_tx_proc) |
| 123 | + sim.add_testbench(testbench) |
| 124 | + with sim.write_vcd(vcd_file="test_uart.vcd"): |
| 125 | + sim.run() |
0 commit comments