#!/usr/bin/env python3 import cocotb from cocotb.triggers import * from cocotb.result import * from cocotb.queue import Queue from cocotb_helpers import buffers import sys import threading import socket import time import os import asyncio import fcntl import zlib ''' # Set up virtual device using sudo ip link add dev veth1 type veth peer name veth2 sudo ip link set up veth1 sudo ip link set up veth2 sudo ip addr add 192.168.2.1/24 dev veth1 sudo ip link set promisc on dev veth1 ''' class MacDevReceiver(): def __init__(self, dut, eth_tx, eth_rx, dev): self.dut = dut self.eth_tx = eth_tx self.eth_rx = eth_rx self.dev = dev self.mac_rx_ev = Event() ETH_P_ALL=3 self.macdev=socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) self.macdev.bind((dev, 0)) fcntl.fcntl(self.macdev, fcntl.F_SETFL, os.O_NONBLOCK) # Not the best way to poll, but I couldn't get asyncio to play nicely with threading... def eth_fcs(self, data): crc = zlib.crc32(data) & 0xFFFF_FFFF return crc.to_bytes(4, byteorder='little') async def main(self): ETH_HEAD = b'\x55\x55\x55\x55\xD5' while True: try: r = self.macdev.recv(2000) if len(r) < 60: r += b'\x00' * (60 - len(r)) r += self.eth_fcs(r) self.dut._log.debug("DUT RX: " + str(r)) await self.eth_tx.send(ETH_HEAD + r); except: await Timer(1, "us") pass async def main_rx(self): while True: frame = await self.eth_rx.queue.get() try: while (frame[0] != 0xD5): frame = frame[1:] frame = frame[1:] except: self.dut._log.debug("NLP / Invalid frame") continue self.dut._log.debug("DUT TX: " + str(frame)) self.macdev.send(frame) async def start(self): await cocotb.start(self.main()) await cocotb.start(self.main_rx()) @cocotb.test() async def hwitl(dut): """Real-Ethernet-hardware in the loop test""" # Start verification components eth_tx = buffers.tx_buffer(dut, "cocovc_eth_inst.cocotb_tx_") await eth_tx.start() eth_rx = buffers.rx_buffer(dut, "cocovc_eth_inst.cocotb_rx_") await eth_rx.start() # Start local monitors macdev_receiver = MacDevReceiver(dut, eth_tx, eth_rx, "veth2") # Wait for VHDL part to be ready await Edge(dut.bench_ready) print("beep") await macdev_receiver.start() print("Press Ctrl+C to stop the test.") await Timer(100, "sec")