trashernet/cocotb/hw_itl.py

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