litex/soc/tools/remote/comm_udp.py

   1 import socket
   2
   3 from litex.soc.tools.remote.etherbone import EtherbonePacket, EtherboneRecord
   4 from litex.soc.tools.remote.etherbone import EtherboneReads, EtherboneWrites
   5
   6
   7 class CommUDP:
   8     def __init__(self, server="192.168.1.50", port=1234, debug=False):
   9         self.server = server
  10         self.port = port
  11         self.debug = debug
  12
  13     def open(self):
  14         if hasattr(self, "tx_socket"):
  15             return
  16         self.tx_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  17         self.rx_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  18         self.rx_socket.bind(("", self.port))
  19
  20     def close(self):
  21         if not hasattr(self, "tx_socket"):
  22             return
  23         self.tx_socket.close()
  24         del self.tx_socket
  25         self.rx_socket.close()
  26         del self.rx_socket
  27
  28     def read(self, addr, length=None):
  29         length_int = 1 if length is None else length
  30         record = EtherboneRecord()
  31         record.reads = EtherboneReads(addrs=[addr+4*j for j in range(length_int)])
  32         record.rcount = len(record.reads)
  33
  34         packet = EtherbonePacket()
  35         packet.records = [record]
  36         packet.encode()
  37         self.tx_socket.sendto(bytes(packet), (self.server, self.port))
  38
  39         datas, dummy = self.rx_socket.recvfrom(8192)
  40         packet = EtherbonePacket(datas)
  41         packet.decode()
  42         datas = packet.records.pop().writes.get_datas()
  43         if self.debug:
  44             for i, value in enumerate(datas):
  45                 print("read {:08x} @ {:08x}".format(value, addr + 4*i))
  46         return datas[0] if length is None else datas
  47
  48     def write(self, addr, datas):
  49         datas = datas if isinstance(datas, list) else [datas]
  50         length = len(datas)
  51         record = EtherboneRecord()
  52         record.writes = EtherboneWrites(base_addr=addr, datas=iter(datas))
  53         record.wcount = len(record.writes)
  54
  55         packet = EtherbonePacket()
  56         packet.records = [record]
  57         packet.encode()
  58         self.tx_socket.sendto(bytes(packet), (self.server, self.port))
  59
  60         if self.debug:
  61             for i, value in enumerate(datas):
  62                 print("write {:08x} @ {:08x}".format(value, addr + 4*i))