80 lines
1.8 KiB
Python
80 lines
1.8 KiB
Python
#!/usr/bin/env python
|
|
|
|
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
|
|
|
|
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)
|
|
|
|
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")
|