trashernet/cocotb/cocotb_trashernet_phy.py

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()