cocotb_helpers/buffers.py

63 lines
1.6 KiB
Python

#!/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())