litex/liteeth/test/model/mac.py

106 lines
2.4 KiB
Python
Raw Normal View History

2015-01-28 13:07:59 -05:00
import binascii
from liteeth.common import *
from liteeth.mac.common import *
from liteeth.test.common import *
2015-01-28 14:44:41 -05:00
def print_mac(s):
print_with_prefix(s, "[MAC]")
preamble = [0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0xD5]
2015-01-28 13:07:59 -05:00
def crc32(l):
crc = []
crc_bytes = binascii.crc32(bytes(l)).to_bytes(4, byteorder="little")
for byte in crc_bytes:
crc.append(int(byte))
return crc
# MAC model
class MACPacket(list):
def __init__(self, init=[]):
self.ongoing = False
self.done = False
for byte in init:
self.append(byte)
class MACRXPacket(MACPacket):
def check_remove_preamble(self):
2015-01-28 14:44:41 -05:00
if comp(self[0:8], preamble):
2015-01-28 13:07:59 -05:00
for i in range(8):
self.pop(0)
return False
else:
return True
def check_remove_crc(self):
if comp(self[-4:], crc32(self[:-4])):
for i in range(4):
self.pop()
return False
else:
return True
class MACTXPacket(MACPacket):
def insert_crc(self):
2015-01-28 14:44:41 -05:00
for d in crc32(self):
self.append(d)
2015-01-28 13:07:59 -05:00
def insert_preamble(self):
2015-01-28 14:44:41 -05:00
for d in reversed(preamble):
self.insert(0, d)
2015-01-28 13:07:59 -05:00
class MAC(Module):
2015-01-28 14:44:41 -05:00
def __init__(self, phy, debug=False, loopback=False):
2015-01-28 13:07:59 -05:00
self.phy = phy
self.debug = debug
2015-01-28 14:44:41 -05:00
self.loopback = loopback
2015-01-28 13:07:59 -05:00
self.tx_packets = []
self.tx_packet = MACTXPacket()
self.rx_packet = MACRXPacket()
self.ip_callback = None
def set_ip_callback(self, callback):
self.ip_callback = callback
def send(self, datas):
tx_packet = MACTXPacket(datas)
2015-01-28 14:44:41 -05:00
if self.debug:
r = ">>>>>>>>\n"
r += "length " + str(len(tx_packet)) + "\n"
for d in tx_packet:
r += "%02x" %d
print_mac(r)
2015-01-28 13:07:59 -05:00
tx_packet.insert_crc()
tx_packet.insert_preamble()
self.tx_packets.append(tx_packet)
def callback(self, datas):
rx_packet = MACRXPacket(datas)
preamble_error = rx_packet.check_remove_preamble()
crc_error = rx_packet.check_remove_crc()
2015-01-28 14:44:41 -05:00
if self.debug:
r = "<<<<<<<<\n"
r += "preamble_error " + str(preamble_error) + "\n"
r += "crc_error " + str(crc_error) + "\n"
r += "length " + str(len(rx_packet)) + "\n"
for d in rx_packet:
r += "%02x" %d
print_mac(r)
2015-01-28 13:07:59 -05:00
if (not preamble_error) and (not crc_error):
2015-01-28 14:44:41 -05:00
if self.loopback:
self.send(rx_packet)
elif self.ip_callback is not None:
2015-01-28 13:07:59 -05:00
self.ip_callback(rx_packet)
def gen_simulation(self, selfp):
self.tx_packet.done = True
while True:
yield from self.phy.receive()
self.callback(self.phy.packet)
# XXX add full duplex
if len(self.tx_packets) != 0:
tx_packet = self.tx_packets.pop(0)
yield from self.phy.send(tx_packet)