litex/liteeth/test/model/etherbone.py

332 lines
7.7 KiB
Python
Raw Normal View History

2015-02-10 15:29:14 -05:00
import math, copy
from liteeth.common import *
from liteeth.test.common import *
from liteeth.test.model import udp
def print_etherbone(s):
print_with_prefix(s, "[ETHERBONE]")
# Etherbone model
class EtherboneWrite:
def __init__(self, data):
self.data = data
def __repr__(self):
2015-02-11 05:11:54 -05:00
return "WR32 0x{:08x}".format(self.data)
2015-02-10 15:29:14 -05:00
class EtherboneRead:
def __init__(self, addr):
self.addr = addr
def __repr__(self):
2015-02-11 05:11:54 -05:00
return "RD32 @ 0x{:08x}".format(self.addr)
2015-02-10 15:29:14 -05:00
class EtherboneWrites(Packet):
def __init__(self, init=[], base_addr=0):
Packet.__init__(self, init)
self.base_addr = base_addr
self.writes = []
2015-02-11 05:11:54 -05:00
self.encoded = init != []
2015-02-10 15:29:14 -05:00
def add(self, write):
self.writes.append(write)
def encode(self):
2015-02-11 05:11:54 -05:00
if self.encoded:
raise ValueError
2015-02-10 15:29:14 -05:00
for byte in split_bytes(self.base_addr, 4):
self.append(byte)
for write in self.writes:
for byte in split_bytes(write.data, 4):
self.append(byte)
2015-02-11 05:11:54 -05:00
self.encoded = True
2015-02-10 15:29:14 -05:00
def decode(self):
2015-02-11 05:11:54 -05:00
if not self.encoded:
raise ValueError
2015-02-10 15:29:14 -05:00
base_addr = []
for i in range(4):
base_addr.append(self.pop(0))
self.base_addr = merge_bytes(base_addr)
self.writes = []
while len(self) != 0:
write = []
for i in range(4):
write.append(self.pop(0))
self.writes.append(EtherboneWrite(merge_bytes(write)))
2015-02-11 05:11:54 -05:00
self.encoded = False
def __repr__(self):
r = "Writes\n"
r += "--------\n"
r += "BaseAddr @ 0x{:08x}\n".format(self.base_addr)
for write in self.writes:
r += write.__repr__() + "\n"
return r
2015-02-10 15:29:14 -05:00
class EtherboneReads(Packet):
def __init__(self, init=[], base_ret_addr=0):
Packet.__init__(self, init)
self.base_ret_addr = base_ret_addr
self.reads = []
2015-02-11 05:11:54 -05:00
self.encoded = init != []
2015-02-10 15:29:14 -05:00
def add(self, read):
self.reads.append(read)
def encode(self):
2015-02-11 05:11:54 -05:00
if self.encoded:
raise ValueError
2015-02-10 15:29:14 -05:00
for byte in split_bytes(self.base_ret_addr, 4):
self.append(byte)
for read in self.reads:
for byte in split_bytes(read.addr, 4):
self.append(byte)
2015-02-11 05:11:54 -05:00
self.encoded = True
2015-02-10 15:29:14 -05:00
def decode(self):
2015-02-11 05:11:54 -05:00
if not self.encoded:
raise ValueError
2015-02-10 15:29:14 -05:00
base_ret_addr = []
for i in range(4):
base_ret_addr.append(self.pop(0))
self.base_ret_addr = merge_bytes(base_ret_addr)
self.reads = []
while len(self) != 0:
read = []
for i in range(4):
read.append(self.pop(0))
self.reads.append(EtherboneRead(merge_bytes(read)))
2015-02-11 05:11:54 -05:00
self.encoded = False
def __repr__(self):
r = "Reads\n"
r += "--------\n"
r += "BaseRetAddr @ 0x{:08x}\n".format(self.base_ret_addr)
for read in self.reads:
r += read.__repr__() + "\n"
return r
2015-02-10 15:29:14 -05:00
class EtherboneRecord(Packet):
def __init__(self, init=[]):
Packet.__init__(self, init)
self.writes = None
self.reads = None
2015-02-11 05:11:54 -05:00
self.encoded = init != []
2015-02-10 15:29:14 -05:00
def get_writes(self):
if self.wcount == 0:
return None
else:
writes = []
for i in range((self.wcount+1)*4):
writes.append(self.pop(0))
return EtherboneWrites(writes)
def get_reads(self):
if self.rcount == 0:
return None
else:
reads = []
for i in range((self.rcount+1)*4):
reads.append(self.pop(0))
return EtherboneReads(reads)
def decode(self):
2015-02-11 05:11:54 -05:00
if not self.encoded:
raise ValueError
2015-02-10 15:29:14 -05:00
header = []
for byte in self[:etherbone_record_header_len]:
header.append(self.pop(0))
for k, v in sorted(etherbone_record_header.items()):
setattr(self, k, get_field_data(v, header))
2015-02-11 05:11:54 -05:00
self.writes = self.get_writes()
self.writes.decode()
self.reads = self.get_reads()
self.reads.decode()
self.encoded = False
2015-02-10 15:29:14 -05:00
def set_writes(self, writes):
2015-02-11 05:11:54 -05:00
self.wcount = len(writes.writes)
2015-02-10 15:29:14 -05:00
writes.encode()
for byte in writes:
self.append(byte)
def set_reads(self, reads):
2015-02-11 05:11:54 -05:00
self.rcount = len(reads.reads)
2015-02-10 15:29:14 -05:00
reads.encode()
for byte in reads:
self.append(byte)
def encode(self):
2015-02-11 05:11:54 -05:00
if self.encoded:
raise ValueError
self.set_writes(self.writes)
self.set_reads(self.reads)
2015-02-10 15:29:14 -05:00
header = 0
for k, v in sorted(etherbone_record_header.items()):
value = merge_bytes(split_bytes(getattr(self, k), math.ceil(v.width/8)), "little")
header += (value << v.offset+(v.byte*8))
for d in split_bytes(header, etherbone_record_header_len):
self.insert(0, d)
2015-02-11 05:11:54 -05:00
self.encoded = True
def __repr__(self, n=0):
r = "Record {}\n".format(n)
r += "--------\n"
if self.encoded:
for d in self:
r += "{:02x}".format(d)
else:
for k in sorted(etherbone_record_header.keys()):
r += k + " : 0x{:0x}\n".format(getattr(self,k))
if self.wcount != 0:
r += self.writes.__repr__()
if self.rcount != 0:
r += self.reads.__repr__()
2015-02-10 15:29:14 -05:00
return r
class EtherbonePacket(Packet):
def __init__(self, init=[]):
Packet.__init__(self, init)
2015-02-11 05:11:54 -05:00
self.encoded = init != []
self.records = []
2015-02-10 15:29:14 -05:00
self.magic = etherbone_magic
self.version = etherbone_version
self.addr_size = 32//8
self.port_size = 32//8
self.nr = 0
self.pr = 0
self.pf = 0
2015-02-10 15:29:14 -05:00
def get_records(self):
records = []
done = False
payload = self
while len(payload) != 0:
record = EtherboneRecord(payload)
record.decode()
records.append(copy.deepcopy(record))
payload = record
return records
def decode(self):
2015-02-11 05:11:54 -05:00
if not self.encoded:
raise ValueError
2015-02-10 15:29:14 -05:00
header = []
for byte in self[:etherbone_packet_header_len]:
2015-02-10 15:29:14 -05:00
header.append(self.pop(0))
for k, v in sorted(etherbone_packet_header.items()):
2015-02-10 15:29:14 -05:00
setattr(self, k, get_field_data(v, header))
2015-02-11 05:11:54 -05:00
self.records = self.get_records()
self.encoded = False
2015-02-10 15:29:14 -05:00
def set_records(self, records):
for record in records:
2015-02-11 05:11:54 -05:00
record.encode()
2015-02-10 15:29:14 -05:00
for byte in record:
self.append(byte)
def encode(self):
2015-02-11 05:11:54 -05:00
if self.encoded:
raise ValueError
self.set_records(self.records)
2015-02-10 15:29:14 -05:00
header = 0
for k, v in sorted(etherbone_packet_header.items()):
2015-02-10 15:29:14 -05:00
value = merge_bytes(split_bytes(getattr(self, k), math.ceil(v.width/8)), "little")
header += (value << v.offset+(v.byte*8))
for d in split_bytes(header, etherbone_packet_header_len):
2015-02-10 15:29:14 -05:00
self.insert(0, d)
2015-02-11 05:11:54 -05:00
self.encoded = True
2015-02-10 15:29:14 -05:00
def __repr__(self):
2015-02-11 05:11:54 -05:00
r = "Packet\n"
r += "--------\n"
if self.encoded:
for d in self:
r += "{:02x}".format(d)
else:
for k in sorted(etherbone_packet_header.keys()):
2015-02-11 05:11:54 -05:00
r += k + " : 0x{:0x}\n".format(getattr(self,k))
for i, record in enumerate(self.records):
r += record.__repr__(i)
2015-02-10 15:29:14 -05:00
return r
class Etherbone(Module):
def __init__(self, udp, debug=False):
self.udp = udp
self.debug = debug
self.tx_packets = []
self.tx_packet = EtherbonePacket()
self.rx_packet = EtherbonePacket()
udp.set_etherbone_callback(self.callback)
2015-02-10 15:29:14 -05:00
def send(self, packet):
packet.encode()
if self.debug:
print_etherbone(">>>>>>>>")
print_etherbone(packet)
udp_packet = udp.UDPPacket(packet)
udp_packet.src_port = 0x1234 # XXX
udp_packet.dst_port = 20000 # XXX
2015-02-10 15:29:14 -05:00
udp_packet.length = len(packet)
udp_packet.checksum = 0
self.udp.send(udp_packet)
def callback(self, packet):
packet = EtherbonePacket(packet)
2015-02-10 15:29:14 -05:00
packet.decode()
if self.debug:
print_etherbone("<<<<<<<<")
print_etherbone(packet)
self.process(packet)
def process(self, packet):
pass
if __name__ == "__main__":
# Writes/Reads
writes = EtherboneWrites(base_addr=0x1000)
for i in range(16):
writes.add(EtherboneWrite(i))
reads = EtherboneReads(base_ret_addr=0x2000)
for i in range(16):
reads.add(EtherboneRead(i))
# Record
record = EtherboneRecord()
2015-02-11 05:11:54 -05:00
record.writes = writes
record.reads = reads
2015-02-10 15:29:14 -05:00
record.bca = 0
record.rca = 0
record.rff = 0
record.cyc = 0
record.wca = 0
record.wff = 0
record.byte_enable = 0
record.wcount = 16
record.rcount = 16
# Packet
packet = EtherbonePacket()
2015-02-11 05:11:54 -05:00
packet.records = [copy.deepcopy(record) for i in range(8)]
2015-02-10 15:29:14 -05:00
packet.nr = 0
packet.pr = 0
packet.pf = 0
#print(packet)
packet.encode()
2015-02-11 05:11:54 -05:00
#print(packet)
2015-02-10 15:29:14 -05:00
# Send packet over UDP to check against Wireshark dissector
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.sendto(bytes(packet), ("192.168.1.1", 20000))
2015-02-10 15:29:14 -05:00
packet = EtherbonePacket(packet)
packet.decode()
2015-02-11 05:11:54 -05:00
print(packet)