trashernet/cocotb/hw_itl.py

95 lines
2.3 KiB
Python
Raw Permalink Normal View History

2022-05-13 18:19:35 +02:00
#!/usr/bin/env python3
2022-04-29 20:53:20 +02:00
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
2022-04-29 20:53:20 +02:00
2022-05-13 18:19:35 +02:00
'''
# 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
2022-05-13 18:19:35 +02:00
'''
2022-04-29 20:53:20 +02:00
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))
2022-05-13 18:19:35 +02:00
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...
2022-04-29 20:53:20 +02:00
def eth_fcs(self, data):
crc = zlib.crc32(data) & 0xFFFF_FFFF
return crc.to_bytes(4, byteorder='little')
2022-04-29 20:53:20 +02:00
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))
2022-04-29 20:53:20 +02:00
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))
2022-04-29 20:53:20 +02:00
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")
2022-04-29 20:53:20 +02:00
# 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")