litex/misoclib/com/liteusb/software/ftdi/__init__.py

353 lines
9.0 KiB
Python

import platform
import ctypes
import os
import time
import queue
import threading
_lpath = (os.path.dirname(__file__))
if _lpath == '':
_lpath = '.'
if platform.system() == "Windows":
libftdicom = ctypes.cdll.LoadLibrary(_lpath + "/libftdicom.dll")
else:
libftdicom = ctypes.cdll.LoadLibrary(_lpath + "/libftdicom.so")
class FTDI_Device(ctypes.Structure):
_fields_ = [
('_1', ctypes.c_void_p),
('_2', ctypes.c_void_p),
]
pFTDI_Device = ctypes.POINTER(FTDI_Device)
# FTDIDevice_Open
FTDIDevice_Open = libftdicom.FTDIDevice_Open
FTDIDevice_Open.argtypes = [
pFTDI_Device, # Dev
ctypes.c_int # Interface
]
FTDIDevice_Open.restype = ctypes.c_int
# FTDIDevice_Close
FTDIDevice_Close = libftdicom.FTDIDevice_Close
FTDIDevice_Close.argtypes = [pFTDI_Device]
FTDIDevice_SetMode = libftdicom.FTDIDevice_SetMode
FTDIDevice_SetMode.argtypes = [
pFTDI_Device, # Dev
ctypes.c_int, # Interface
ctypes.c_int, # Mode
ctypes.c_char, # PinDirection
ctypes.c_char, # baudrate
]
FTDIDevice_Write = libftdicom.FTDIDevice_Write
FTDIDevice_Write.argtypes = [
pFTDI_Device, # Dev
ctypes.c_int, # Interface
ctypes.c_char_p, # Buf
ctypes.c_size_t, # N
ctypes.c_bool, # async
]
FTDIDevice_Write.restype = ctypes.c_int
p_cb_StreamCallback = ctypes.CFUNCTYPE(
ctypes.c_int, # retval
ctypes.POINTER(ctypes.c_uint8), # buf
ctypes.c_int, # length
ctypes.c_void_p, # progress
ctypes.c_void_p) # userdata
FTDIDevice_ReadStream = libftdicom.FTDIDevice_ReadStream
FTDIDevice_ReadStream.argtypes = [
pFTDI_Device, # dev
ctypes.c_int, # interface
p_cb_StreamCallback, # callback
ctypes.c_void_p, # userdata
ctypes.c_int, # packetsPerTransfer
ctypes.c_int, # numTransfers
]
FTDIDevice_ReadStream.restype = ctypes.c_int
FTDI_INTERFACE_A = 1
FTDI_INTERFACE_B = 2
FTDI_BITMODE_SYNC_FIFO = (1 << 6)
class FTDIDevice:
def __init__(self, interface, mode):
self.__is_open = False
self._dev = FTDI_Device()
self.interface = interface
self.mode = mode
def __del__(self):
if self.__is_open:
self.__is_open = False
FTDIDevice_Close(self._dev)
def open(self):
err = FTDIDevice_Open(self._dev, self.interface)
if err:
return err
else:
self.__is_open = True
if self.mode == "synchronous":
err = FTDIDevice_SetMode(self._dev, interface, FTDI_BITMODE_SYNC_FIFO, 0xFF, 0)
return err
def write(self, intf, buf, async=False):
if not isinstance(buf, bytes):
raise TypeError("buf must be bytes")
return FTDIDevice_Write(self._dev, intf, buf, len(buf), async)
def read(self, intf, n):
buf = []
def callback(b, prog):
buf.extend(b)
return int(len(buf) >= n)
self.read_async(intf, callback, 4, 4)
return buf
def read_async(self, intf, callback, packetsPerTransfer, numTransfers):
def callback_wrapper(buf, ll, prog, user):
if ll:
b = ctypes.string_at(buf, ll)
else:
b = b''
return callback(b, prog)
cb = p_cb_StreamCallback(callback_wrapper)
return FTDIDevice_ReadStream(self._dev, intf, cb,
None, packetsPerTransfer, numTransfers)
class ProtocolError(Exception):
pass
class TimeoutError(Exception):
pass
INCOMPLETE = -1
UNMATCHED = 0
class BaseService:
def match_identifier(self, byt):
r = True
r = r and (byt[0] == 0x5A)
r = r and (byt[1] == 0xA5)
r = r and (byt[2] == 0x5A)
r = r and (byt[3] == 0xA5)
r = r and (byt[4] == self.tag)
return r
def get_needed_size_for_identifier(self):
return self.NEEDED_FOR_SIZE
def present_bytes(self, b):
if len(b) < self.get_needed_size_for_identifier():
return INCOMPLETE
if not self.match_identifier(b):
return UNMATCHED
size = self.get_packet_size(b)
if len(b) < size:
return INCOMPLETE
self.consume(b[:size])
return size
class UART:
class __UARTService(BaseService):
NEEDED_FOR_SIZE = 9
def __init__(self, tag):
self.tag = tag
self.q = queue.Queue()
def get_packet_size(self, buf):
return 10
def consume(self, buf):
self.q.put(buf[9])
def __init__(self, tag):
self.tag = tag
self.service = UART.__UARTService(self.tag)
def do_read(self, timeout=None):
try:
resp = self.service.q.get(True, timeout)
except queue.Empty:
return -1
return resp
def do_write(self, value):
msg = [0x5A, 0xA5, 0x5A, 0xA5, self.tag, 0x00, 0x00, 0x00, 1, value&0xFF]
self.service.write(bytes(msg))
class DMA:
class __DMAService(BaseService):
NEEDED_FOR_SIZE = 9
def __init__(self, tag):
self.tag = tag
self.q = queue.Queue()
def get_packet_size(self, buf):
payload_size = buf[5] << 24
payload_size |= buf[6] << 16
payload_size |= buf[7] << 8
payload_size |= buf[8] << 0
return 9 + payload_size
def consume(self, buf):
self.q.put(buf[9:])
def __init__(self, tag):
self.tag = tag
self.service = DMA.__DMAService(self.tag)
def do_read(self, timeout=None):
try:
resp = list(self.service.q.get(True, timeout))
except queue.Empty:
raise TimeoutError("DMA read timed out")
return resp
def do_write(self, data):
length = len(data)
msg = [0x5A, 0xA5, 0x5A, 0xA5, self.tag,
(length & 0xff000000) >> 24,
(length & 0x00ff0000) >> 16,
(length & 0x0000ff00) >> 8,
(length & 0x000000ff) >> 0]
msg += data
self.service.write(bytes(msg))
class FTDIComDevice:
def __init__(self, interface, mode, uart_tag=0, dma_tag=1, verbose=False):
self.__is_open = False
self.interface = interface
self.mode = mode
self.dev = FTDIDevice(interface, mode)
self.verbose = verbose
self.uart = UART(uart_tag)
self.dma = DMA(dma_tag)
self.__services = [self.uart.service, self.dma.service]
# Inject a write function into the services
for service in self.__services:
def write(msg):
if self.verbose:
print("< %s" % " ".join("%02x" % i for i in msg))
self.dev.write(self.interface, msg, async=False)
service.write = write
def __comms(self):
self.__buf = b""
def callback(b, prog):
try:
if self.verbose and b:
print("> %s" % " ".join("%02x" % i for i in b))
self.__buf += b
incomplete = False
while self.__buf and not incomplete:
for service in self.__services:
code = service.present_bytes(self.__buf)
if code == INCOMPLETE:
incomplete = True
break
elif code:
self.__buf = self.__buf[code:]
break
else:
self.__buf = self.__buf[1:]
return int(self.__comm_term)
except Exception as e:
self.__comm_term = True
self.__comm_exc = e
return 1
while not self.__comm_term:
self.dev.read_async(self.interface, callback, 8, 16)
if self.__comm_exc:
raise self.__comm_exc
def __del__(self):
if self.__is_open:
self.close()
def open(self):
if self.__is_open:
raise ValueError("FTDICOMDevice doubly opened")
stat = self.dev.open()
if stat:
print("USB: Error opening device\n")
return stat
self.commthread = threading.Thread(target=self.__comms, daemon=True)
self.__comm_term = False
self.__comm_exc = None
self.commthread.start()
self.__comm_term = False
self.__is_open = True
def close(self):
if not self.__is_open:
raise ValueError("FTDICOMDevice doubly closed")
self.__comm_term = True
self.commthread.join()
self.__is_open = False
def uartflush(self, timeout=0.25):
while (self.uartread(timeout) != -1):
pass
def uartread(self, timeout=None):
return self.uart.do_read(timeout)
def uartwrite(self, value):
return self.uart.do_write(value)
def dmaread(self):
return self.dma.do_read()
def dmawrite(self, data):
return self.dma.do_write(data)