#!/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 ''' # Set up virtual device using sudo ip link add virt0 type dummy sudo ip link set up virt0 sudo ip addr add 192.168.2.10/24 dev virt0 ''' 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() print("Setting IF to promisc mode...") os.system("ip link set promisc on dev {}".format(dev)) 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... async def main(self): ETH_HEAD = b'\x55\x55\x55\x55\xD5' while True: try: r = self.macdev.recv(2000) 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.info("RX Frame: " + 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, "virt0") # 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")