litex/misoclib/tools/litescope/host/driver/etherbone.py

import socket
from misoclib.tools.litescope.host.driver.reg import *

from liteeth.test.model.etherbone import *


class LiteScopeEtherboneDriver:
    def __init__(self, ip_address, udp_port=20000, addrmap=None, busword=8, debug=False):
        self.ip_address = ip_address
        self.udp_port = udp_port
        self.debug = debug

        self.tx_sock = None
        self.rx_sock = None
        if addrmap is not None:
            self.regs = build_map(addrmap, busword, self.read, self.write)

    def open(self):
        self.tx_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.rx_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.rx_sock.bind(("", self.udp_port))

    def close(self):
        pass

    def read(self, addr, burst_length=None, repeats=None):
        def to_int(v):
            return 1 if v is None else v
        reads_addrs = []
        for i in range(to_int(repeats)):
            reads_addrs += [addr+4*j for j in range(to_int(burst_length))]
        reads = EtherboneReads(base_ret_addr=0x1000, addrs=reads_addrs)
        record = EtherboneRecord()
        record.writes = None
        record.reads = reads
        record.bca = 0
        record.rca = 0
        record.rff = 0
        record.cyc = 0
        record.wca = 0
        record.wff = 0
        record.byte_enable = 0xf
        record.wcount = 0
        record.rcount = len(reads_addrs)

        packet = EtherbonePacket()
        packet.records = [record]
        packet.encode()
        self.tx_sock.sendto(bytes(packet), (self.ip_address, self.udp_port))

        datas, addrs = self.rx_sock.recvfrom(8192)
        packet = EtherbonePacket(datas)
        packet.decode()
        datas = packet.records.pop().writes.get_datas()
        if self.debug:
            for i, data in enumerate(datas):
                print("RD %08X @ %08X" %(data, addr + 4*(i%to_int(burst_length))))
        return datas

    def write(self, addr, datas):
        if not isinstance(datas, list):
            datas = [datas]
        writes_datas = [d for d in datas]
        writes = EtherboneWrites(base_addr=addr, datas=writes_datas)
        record = EtherboneRecord()
        record.writes = writes
        record.reads = None
        record.bca = 0
        record.rca = 0
        record.rff = 0
        record.cyc = 0
        record.wca = 0
        record.wff = 0
        record.byte_enable = 0xf
        record.wcount = len(writes_datas)
        record.rcount = 0

        packet = EtherbonePacket()
        packet.records = [record]
        packet.encode()
        self.tx_sock.sendto(bytes(packet), (self.ip_address, self.udp_port))

        if self.debug:
            for i, data in enumerate(datas):
                print("WR %08X @ %08X" %(data, addr + 4*i))