add HID delay

[Adafruit_Blinka-hackapet.git] / src / adafruit_blinka / microcontroller / mcp2221 / mcp2221.py

diff --git a/src/adafruit_blinka/microcontroller/mcp2221/mcp2221.py b/src/adafruit_blinka/microcontroller/mcp2221/mcp2221.py
index 118d27c76612349a9c68154e64e8c5beb1a331da..8247006153d35be26993ed41c030520bfd6c2487 100644 (file)
--- a/src/adafruit_blinka/microcontroller/mcp2221/mcp2221.py
+++ b/src/adafruit_blinka/microcontroller/mcp2221/mcp2221.py
@@ -1,6 +1,38 @@
+import os

 import time
 import hid
 
 import time
 import hid
 

+# Small values seem to help on some Windows setups
+MCP2221_HID_DELAY = float(os.environ.get('BLINKA_MCP2221_HID_DELAY', 0))
+
+# from the C driver
+# http://ww1.microchip.com/downloads/en/DeviceDoc/mcp2221_0_1.tar.gz
+# others (???) determined during driver developement
+# pylint: disable=bad-whitespace
+RESP_ERR_NOERR              = 0x00
+RESP_ADDR_NACK              = 0x25
+RESP_READ_ERR               = 0x7F
+RESP_READ_COMPL             = 0x55
+RESP_READ_PARTIAL           = 0x54 # ???
+RESP_I2C_IDLE               = 0x00
+RESP_I2C_START_TOUT         = 0x12
+RESP_I2C_RSTART_TOUT        = 0x17
+RESP_I2C_WRADDRL_TOUT       = 0x23
+RESP_I2C_WRADDRL_WSEND      = 0x21
+RESP_I2C_WRADDRL_NACK       = 0x25
+RESP_I2C_WRDATA_TOUT        = 0x44
+RESP_I2C_RDDATA_TOUT        = 0x52
+RESP_I2C_STOP_TOUT          = 0x62
+
+RESP_I2C_MOREDATA           = 0x43 # ???
+RESP_I2C_PARTIALDATA        = 0x41 # ???
+RESP_I2C_WRITINGNOSTOP      = 0x45 # ???
+
+MCP2221_RETRY_MAX           = 50
+MCP2221_MAX_I2C_DATA_LEN    = 60
+MASK_ADDR_NACK              = 0x40
+# pylint: enable=bad-whitespace
+

 class MCP2221:
 
     VID = 0x04D8
 class MCP2221:
 
     VID = 0x04D8


@@ -15,12 +47,15 @@ class MCP2221:
     def __init__(self):
         self._hid = hid.device()
         self._hid.open(MCP2221.VID, MCP2221.PID)
     def __init__(self):
         self._hid = hid.device()
         self._hid.open(MCP2221.VID, MCP2221.PID)

+        self._reset()
+        time.sleep(0.25)

 
     def _hid_xfer(self, report, response=True):
         # first byte is report ID, which =0 for MCP2221
         # remaing bytes = 64 byte report data
         # https://github.com/libusb/hidapi/blob/083223e77952e1ef57e6b77796536a3359c1b2a3/hidapi/hidapi.h#L185
         self._hid.write(b'\0' + report + b'\0'*(64-len(report)))
 
     def _hid_xfer(self, report, response=True):
         # first byte is report ID, which =0 for MCP2221
         # remaing bytes = 64 byte report data
         # https://github.com/libusb/hidapi/blob/083223e77952e1ef57e6b77796536a3359c1b2a3/hidapi/hidapi.h#L185
         self._hid.write(b'\0' + report + b'\0'*(64-len(report)))

+        time.sleep(MCP2221_HID_DELAY)

         if response:
             # return is 64 byte response report
             return self._hid.read(64)
         if response:
             # return is 64 byte response report
             return self._hid.read(64)


@@ -29,13 +64,13 @@ class MCP2221:
     # MISC
     #----------------------------------------------------------------
     def gp_get_mode(self, pin):
     # MISC
     #----------------------------------------------------------------
     def gp_get_mode(self, pin):

-        return self._hid_xfer(bytes([0x61]))[22+pin] & 0x07
+        return self._hid_xfer(b'\x61')[22+pin] & 0x07

 
     def gp_set_mode(self, pin, mode):
         # get current settings
 
     def gp_set_mode(self, pin, mode):
         # get current settings

-        current = self._hid_xfer(bytes([0x61]))
+        current = self._hid_xfer(b'\x61')

         # empty report, this is safe since 0's = no change
         # empty report, this is safe since 0's = no change

-        report = bytearray([0x60]+[0]*63)
+        report = bytearray(b'\x60'+b'\x00'*63)

         # set the alter GP flag byte
         report[7] = 0xFF
         # each pin can be set individually
         # set the alter GP flag byte
         report[7] = 0xFF
         # each pin can be set individually


@@ -63,35 +98,43 @@ class MCP2221:
             print()
 
     def _status_dump(self):
             print()
 
     def _status_dump(self):

-        self._pretty_report(self._hid_xfer(bytes([0x10])))
+        self._pretty_report(self._hid_xfer(b'\x10'))

 
     def _sram_dump(self):
 
     def _sram_dump(self):

-        self._pretty_report(self._hid_xfer(bytes([0x61])))
+        self._pretty_report(self._hid_xfer(b'\x61'))

 
     def _reset(self):
         self._hid_xfer(b'\x70\xAB\xCD\xEF', response=False)
 
     def _reset(self):
         self._hid_xfer(b'\x70\xAB\xCD\xEF', response=False)

-        time.sleep(1)
-        self._hid.open(MCP2221.VID, MCP2221.PID)
+        start = time.monotonic()
+        while time.monotonic() - start < 5:
+            try:
+                self._hid.open(MCP2221.VID, MCP2221.PID)
+            except OSError:
+                # try again
+                time.sleep(0.1)
+                continue
+            return
+        raise OSError("open failed")

 
     #----------------------------------------------------------------
     # GPIO
     #----------------------------------------------------------------
     def gpio_set_direction(self, pin, mode):
 
     #----------------------------------------------------------------
     # GPIO
     #----------------------------------------------------------------
     def gpio_set_direction(self, pin, mode):

-        report = bytearray([0x50]+[0]*63)  # empty set GPIO report
+        report = bytearray(b'\x50'+b'\x00'*63)  # empty set GPIO report

         offset = 4 * (pin + 1)
         offset = 4 * (pin + 1)

-        report[offset] = 0x01              # set pin direction
-        report[offset+1] = mode            # to this
+        report[offset] = 0x01                   # set pin direction
+        report[offset+1] = mode                 # to this

         self._hid_xfer(report)
 
     def gpio_set_pin(self, pin, value):
         self._hid_xfer(report)
 
     def gpio_set_pin(self, pin, value):

-        report = bytearray([0x50]+[0]*63)  # empty set GPIO report
+        report = bytearray(b'\x50'+b'\x00'*63)  # empty set GPIO report

         offset = 2 + 4 * pin
         offset = 2 + 4 * pin

-        report[offset] = 0x01              # set pin value
-        report[offset+1] = value           # to this
+        report[offset] = 0x01                   # set pin value
+        report[offset+1] = value                # to this

         self._hid_xfer(report)
 
     def gpio_get_pin(self, pin):
         self._hid_xfer(report)
 
     def gpio_get_pin(self, pin):

-        resp = self._hid_xfer(bytes([0x51]))
+        resp = self._hid_xfer(b'\x51')

         offset = 2 + 2 * pin
         if resp[offset] == 0xEE:
             raise RuntimeError("Pin is not set for GPIO operation.")
         offset = 2 + 2 * pin
         if resp[offset] == 0xEE:
             raise RuntimeError("Pin is not set for GPIO operation.")


@@ -100,22 +143,124 @@ class MCP2221:
 
     #----------------------------------------------------------------
     # I2C
 
     #----------------------------------------------------------------
     # I2C

-    #
-    # cribbed from the C driver:
-    #   define RESP_ERR_NOERR          0x00
-    #   define RESP_ADDR_NACK          0x25
-    #   define RESP_READ_ERR           0x7F
-    #   define RESP_READ_COMPL         0x55
-    #   define RESP_I2C_IDLE           0x00
-    #   define RESP_I2C_START_TOUT     0x12
-    #   define RESP_I2C_RSTART_TOUT    0x17
-    #   define RESP_I2C_WRADDRL_TOUT   0x23
-    #   define RESP_I2C_WRADDRL_WSEND  0x21
-    #   define RESP_I2C_WRADDRL_NACK   0x25
-    #   define RESP_I2C_WRDATA_TOUT    0x44
-    #   define RESP_I2C_RDDATA_TOUT    0x52
-    #   define RESP_I2C_STOP_TOUT      0x62

     #----------------------------------------------------------------
     #----------------------------------------------------------------

+    def _i2c_status(self):
+        resp = self._hid_xfer(b'\x10')
+        if resp[1] != 0:
+            raise RuntimeError("Couldn't get I2C status")
+        return resp
+
+    def _i2c_state(self):
+        return self._i2c_status()[8]
+
+    def _i2c_cancel(self):
+        resp = self._hid_xfer(b'\x10\x00\x10')
+        if resp[1] != 0x00:
+            raise RuntimeError("Couldn't cancel I2C")
+        if resp[2] == 0x10:
+            # bus release will need "a few hundred microseconds"
+            time.sleep(0.001)
+
+    def _i2c_write(self, cmd, address, buffer, start=0, end=None):
+        if self._i2c_state() != 0x00:
+            self._i2c_cancel()
+
+        end = end if end else len(buffer)
+        length = end - start
+        retries = 0
+
+        while (end - start) > 0:
+            chunk = min(end - start, MCP2221_MAX_I2C_DATA_LEN)
+            # write out current chunk
+            resp = self._hid_xfer(bytes([cmd,
+                                         length & 0xFF,
+                                         (length >> 8) & 0xFF,
+                                         address << 1]) +
+                                         buffer[start:(start+chunk)])
+            # check for success
+            if resp[1] != 0x00:
+                if resp[2] in (RESP_I2C_START_TOUT,
+                               RESP_I2C_WRADDRL_TOUT,
+                               RESP_I2C_WRADDRL_NACK,
+                               RESP_I2C_WRDATA_TOUT,
+                               RESP_I2C_STOP_TOUT):
+                    raise RuntimeError("Unrecoverable I2C state failure")
+                retries += 1
+                if retries >= MCP2221_RETRY_MAX:
+                    raise RuntimeError("I2C write error, max retries reached.")
+                time.sleep(0.001)
+                continue # try again
+            # yay chunk sent!
+            while self._i2c_state() == RESP_I2C_PARTIALDATA:
+                time.sleep(0.001)
+            start += chunk
+            retries = 0
+
+        # check status in another loop
+        for _ in range(MCP2221_RETRY_MAX):
+            status = self._i2c_status()
+            if status[20] & MASK_ADDR_NACK:
+                raise RuntimeError("I2C slave address was NACK'd")
+            usb_cmd_status = status[8]
+            if usb_cmd_status == 0:
+                break
+            if usb_cmd_status == RESP_I2C_WRITINGNOSTOP and cmd == 0x94:
+                break   # this is OK too!
+            if usb_cmd_status in (RESP_I2C_START_TOUT,
+                                  RESP_I2C_WRADDRL_TOUT,
+                                  RESP_I2C_WRADDRL_NACK,
+                                  RESP_I2C_WRDATA_TOUT,
+                                  RESP_I2C_STOP_TOUT):
+                raise RuntimeError("Unrecoverable I2C state failure")
+            time.sleep(0.001)
+        else:
+            raise RuntimeError("I2C write error: max retries reached.")
+        # whew success!
+
+    def _i2c_read(self, cmd, address, buffer, start=0, end=None):
+        if self._i2c_state() not in (RESP_I2C_WRITINGNOSTOP, 0):
+            self._i2c_cancel()
+
+        end = end if end else len(buffer)
+        length = end - start
+
+        # tell it we want to read
+        resp = self._hid_xfer(bytes([cmd,
+                                     length & 0xFF,
+                                     (length >> 8) & 0xFF,
+                                     (address << 1) | 0x01]))
+
+        # check for success
+        if resp[1] != 0x00:
+            raise RuntimeError("Unrecoverable I2C read failure")
+
+        # and now the read part
+        while (end - start) > 0:
+            for retry in range(MCP2221_RETRY_MAX):
+                # the actual read
+                resp = self._hid_xfer(b'\x40')
+                # check for success
+                if resp[1] == RESP_I2C_PARTIALDATA:
+                    time.sleep(0.001)
+                    continue
+                if resp[1] != 0x00:
+                    raise RuntimeError("Unrecoverable I2C read failure")
+                if resp[2] == RESP_ADDR_NACK:
+                    raise RuntimeError("I2C NACK")
+                if resp[3] == 0x00 and resp[2] == 0x00:
+                    break
+                if resp[3] == RESP_READ_ERR:
+                    time.sleep(0.001)
+                    continue
+                if resp[2] in (RESP_READ_COMPL, RESP_READ_PARTIAL):
+                    break
+
+            # move data into buffer
+            chunk = min(end - start, 60)
+            for i, k in enumerate(range(start, start+chunk)):
+                buffer[k] = resp[4 + i]
+            start += chunk
+

     def i2c_configure(self, baudrate=100000):
         self._hid_xfer(bytes([0x10,  # set parameters
                               0x00,  # don't care
     def i2c_configure(self, baudrate=100000):
         self._hid_xfer(bytes([0x10,  # set parameters
                               0x00,  # don't care


@@ -124,103 +269,51 @@ class MCP2221:
                               12000000 // baudrate - 3]))
 
     def i2c_writeto(self, address, buffer, *, start=0, end=None):
                               12000000 // baudrate - 3]))
 
     def i2c_writeto(self, address, buffer, *, start=0, end=None):

-        end = end if end else len(buffer)
-        self._hid_xfer(bytes([0x90,                    # i2c write data
-                              end - start & 0xFF,      # xfer length lo byte
-                              end - start >> 8 & 0xFF, # xfer length hi byte
-                              address << 1]) +         # i2c slave address
-                              buffer[start:end])       # user data to be sent
+        self._i2c_write(0x90, address, buffer, start, end)

 
     def i2c_readfrom_into(self, address, buffer, *, start=0, end=None):
 
     def i2c_readfrom_into(self, address, buffer, *, start=0, end=None):

-        end = end if end else len(buffer)
-        retries = 0
-        while retries < 5:
-            #
-            # why does this require two xfers?
-            #
-            # 1. tell it we want to read
-            self._hid_xfer(bytes([0x91,                     # i2c read data
-                                  end - start & 0xFF,       # xfer length lo byte
-                                  end - start >> 8 & 0xFF,  # xfer length hi byte
-                                  address << 1 | 0x01]))    # i2c slave address
-            # 2. and then actually read
-            response = self._hid_xfer(bytes([0x40]))
-            # check for success
-            if response[1] == 0x00:
-                break
-            retries += 1
-        if retries >= 5:
-            raise RuntimeError("I2C read error, max retries reached.")
-        # move data into buffer
-        for i in range(end - start):
-            buffer[start + i] = response[4 + i]
+        self._i2c_read(0x91, address, buffer, start, end)

 
     def i2c_writeto_then_readfrom(self, address, out_buffer, in_buffer, *,
                                   out_start=0, out_end=None,
                                   in_start=0, in_end=None):
 
     def i2c_writeto_then_readfrom(self, address, out_buffer, in_buffer, *,
                                   out_start=0, out_end=None,
                                   in_start=0, in_end=None):

-        out_end = out_end if out_end else len(buffer_out)
-        in_end = in_end if in_end else len(buffer_in)
-        self._hid_xfer(bytes([0x94,                            # i2c write data no stop
-                              out_end - out_start & 0xFF,      # xfer length lo byte
-                              out_end - out_start >> 8 & 0xFF, # xfer length hi byte
-                              address << 1]) +                 # i2c slave address
-                              out_buffer[out_start:out_end])   # user data to be sent
-        retries = 5
-        while retries < 5:
-            #
-            # why does this require two xfers?
-            #
-            # 1. tell it we want to read
-            self._hid_xfer(bytes([0x93,                           # i2c read data repeated start
-                                  in_end - in_start & 0xFF,       # xfer length lo byte
-                                  in_end - in_start >> 8 & 0xFF,  # xfer length hi byte
-                                  address << 1 | 0x01]))          # i2c slave address
-            # 2. and then actually read
-            response = self._hid_xfer(bytes([0x40]))
-            # check for success
-            if response[1] == 0x00:
-                break
-            retries += 1
-        if retries >= 5:
-            raise RuntimeError("I2C read error, max retries reached.")
-        # move data into buffer
-        for i in range(in_end - in_start):
-            in_buffer[in_start + i] = response[4 + i]
+        self._i2c_write(0x94, address, out_buffer, out_start, out_end)
+        self._i2c_read(0x93, address, in_buffer, in_start, in_end)

 
     def i2c_scan(self, *, start=0, end=0x79):
         found = []
         for addr in range(start, end+1):
             # try a write
 
     def i2c_scan(self, *, start=0, end=0x79):
         found = []
         for addr in range(start, end+1):
             # try a write

-            self.i2c_writeto(addr, b'\x00')
+            try:
+                self.i2c_writeto(addr, b'\x00')
+            except RuntimeError: # no reply!
+                continue

             # store if success
             # store if success

-            if self._hid_xfer(b'\x10')[8] == 0x00:
-                found.append(addr)
-            # cancel and continue
-            self._hid_xfer(b'\x10\x00\x10')
+            found.append(addr)

         return found
 
     #----------------------------------------------------------------
     # ADC
     #----------------------------------------------------------------
     def adc_configure(self, vref=0):
         return found
 
     #----------------------------------------------------------------
     # ADC
     #----------------------------------------------------------------
     def adc_configure(self, vref=0):

-        report = bytearray([0x60]+[0]*63)
+        report = bytearray(b'\x60'+b'\x00'*63)

         report[5] = 1 << 7 | (vref & 0b111)
         self._hid_xfer(report)
 
     def adc_read(self, pin):
         report[5] = 1 << 7 | (vref & 0b111)
         self._hid_xfer(report)
 
     def adc_read(self, pin):

-        resp = self._hid_xfer(bytes([0x10]))
+        resp = self._hid_xfer(b'\x10')

         return resp[49 + 2 * pin] << 8 | resp[48 + 2 * pin]
 
     #----------------------------------------------------------------
     # DAC
     #----------------------------------------------------------------
     def dac_configure(self, vref=0):
         return resp[49 + 2 * pin] << 8 | resp[48 + 2 * pin]
 
     #----------------------------------------------------------------
     # DAC
     #----------------------------------------------------------------
     def dac_configure(self, vref=0):

-        report = bytearray([0x60]+[0]*63)
+        report = bytearray(b'\x60'+b'\x00'*63)

         report[3] = 1 << 7 | (vref & 0b111)
         self._hid_xfer(report)
 
     def dac_write(self, pin, value):
         report[3] = 1 << 7 | (vref & 0b111)
         self._hid_xfer(report)
 
     def dac_write(self, pin, value):

-        report = bytearray([0x60]+[0]*63)
+        report = bytearray(b'\x60'+b'\x00'*63)

         report[4] = 1 << 7 | (value & 0b11111)
         self._hid_xfer(report)
 
         report[4] = 1 << 7 | (value & 0b11111)
         self._hid_xfer(report)