91 lines
2.2 KiB
Python
91 lines
2.2 KiB
Python
#!/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()
|