From e2c72ef61d7742fbcb7e048bc928446ed4b56a82 Mon Sep 17 00:00:00 2001 From: Markus Koch Date: Mon, 18 Apr 2022 17:02:34 +0200 Subject: [PATCH] buffers: Add TX/RX buffer helpers --- buffers.py | 62 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 62 insertions(+) create mode 100644 buffers.py diff --git a/buffers.py b/buffers.py new file mode 100644 index 0000000..cb49b46 --- /dev/null +++ b/buffers.py @@ -0,0 +1,62 @@ +#!/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())