#!/usr/bin/env python import cocotb from cocotb.triggers import * from cocotb.result import * from cocotb.queue import Queue class tx_buffer: def __init__(self, dut, prefix): self.queue = Queue() self.queue_start_ev = Event() self.queue_done_ev = Event() self.dut = dut self.prefix = prefix pass async def main(self): while True: await self.queue_start_ev.wait() self.queue_start_ev.clear() while not self.queue.empty(): data = await self.queue.get() getattr(self.dut, self.prefix + "dv").value = True for byte in data: getattr(self.dut, self.prefix + "d").value = byte await Edge(getattr(self.dut, self.prefix + "da")) getattr(self.dut, self.prefix + "dv").value = False await Edge(getattr(self.dut, self.prefix + "da")) self.queue_done_ev.set() async def start(self): await cocotb.start(self.main()) async def send(self, data, send_now=True, wait=None): if wait is None: wait = send_now self.queue.put_nowait(data) if send_now: self.queue_start_ev.set() if wait: await self.queue_done_ev.wait() self.queue_done_ev.clear() class rx_buffer: def __init__(self, dut, prefix): self.queue = Queue() self.dut = dut self.prefix = prefix async def main(self): while True: d = bytearray(b'') await RisingEdge(getattr(self.dut, self.prefix + "a")) while True: await First(Edge(getattr(self.dut, self.prefix + "dt")), Edge(getattr(self.dut, self.prefix + "a"))) if not getattr(self.dut, self.prefix + "a").value: break d.append(getattr(self.dut, self.prefix + "d").value) self.queue.put_nowait(d) async def start(self): await cocotb.start(self.main())