#!/usr/bin/env python import cocotb from cocotb.triggers import * from cocotb.result import * from cocotb.queue import Queue # DUT interface (internal) tb_rx_queue = Queue() tb_rx_queue_done_ev = Event() async def dut_receiver(dut): scoreboard = tb_rx_queue while True: await Edge(dut.coco_dut_rxs) byte = dut.coco_dut_rxd.value dut._log.info("[DT:RX <-] Byte %02X", byte) assert not scoreboard.empty(), "Data was provided on the RX interface even though none was expected" if not scoreboard.empty(): sb_byte = await scoreboard.get() assert byte == sb_byte, f"Received {byte} instead of {sb_byte}" async def dut_rxa(dut): scoreboard = tb_rx_queue await Edge(dut.coco_dut_rxa) # TODO: Workaround for some stupid bug somewhere while True: await Edge(dut.coco_dut_rxa) if dut.coco_dut_rxa.value: dut._log.info("[DT:RX <-] Start of frame") else: dut._log.info("[DT:RX <-] End of frame") if scoreboard.empty(): tb_rx_queue_done_ev.set() # TB / ETH interface (external) tb_tx_queue = Queue() async def tb_transmitter(dut): queue = tb_tx_queue while True: byte = await queue.get() if byte == -1: dut._log.info("[TB:TX ->] Wait IPG") await Timer(16, units="us") # IPG else: # TX byte dut._log.info("[TB:TX ->] Byte %02X", byte) dut.coco_tb_txd.value = byte dut.coco_tb_txs.value = not dut.coco_tb_txs.value await Edge(dut.coco_tb_txa); async def tb_send_packet(dut, data): queue = tb_tx_queue scoreboard = tb_rx_queue queue.put_nowait(0x55) queue.put_nowait(0x55) queue.put_nowait(0x55) queue.put_nowait(0xD5) for byte in data: scoreboard.put_nowait(byte) queue.put_nowait(byte) queue.put_nowait(-1) async def timeout(dut): await Timer(50, units="us") assert False, "Timeout" @cocotb.test() async def simple_rx(dut): """Receive a packet using trashernet.""" await cocotb.start(timeout(dut)) await cocotb.start(dut_receiver(dut)) await cocotb.start(dut_rxa(dut)) await cocotb.start(tb_transmitter(dut)) dut.rst.value = 1 await Timer(20, units="ns") dut.rst.value = 0 await Timer(20, units="ns") await tb_send_packet(dut, b'\x02\x03\x05\x07') await cocotb.start(timeout(dut)) await tb_rx_queue_done_ev.wait()