feat: add I2C and SPI

[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 39b1c0715970f1501a414c4b44ad17cde594ef1e..90ed49e50c046ba10d48259a1e4e96a1b6a6b6a3 100644 (file)
--- a/src/adafruit_blinka/microcontroller/mcp2221/mcp2221.py
+++ b/src/adafruit_blinka/microcontroller/mcp2221/mcp2221.py
@@ -1,7 +1,48 @@
+# SPDX-FileCopyrightText: 2021 Melissa LeBlanc-Williams for Adafruit Industries
+#
+# SPDX-License-Identifier: MIT
+"""Chip Definition for MCP2221"""
+
+import os

 import time
 import time

+import atexit
+

 import hid
 
 import hid
 

+# Here if you need it
+MCP2221_HID_DELAY = float(os.environ.get("BLINKA_MCP2221_HID_DELAY", 0))
+# Use to set delay between reset and device reopen. if negative, don't reset at all
+MCP2221_RESET_DELAY = float(os.environ.get("BLINKA_MCP2221_RESET_DELAY", 0.5))
+
+# from the C driver
+# http://ww1.microchip.com/downloads/en/DeviceDoc/mcp2221_0_1.tar.gz
+# others (???) determined during driver developement
+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
+
+

 class MCP2221:
 class MCP2221:

+    """MCP2221 Device Class Definition"""

 
     VID = 0x04D8
     PID = 0x00DD
 
     VID = 0x04D8
     PID = 0x00DD


@@ -15,62 +56,85 @@ 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()
+        # make sure the device gets closed before exit
+        atexit.register(self.close)
+        if MCP2221_RESET_DELAY >= 0:
+            self._reset()
+        self._gp_config = [0x07] * 4  # "don't care" initial value
+        for pin in range(4):
+            self.gp_set_mode(pin, self.GP_GPIO)  # set to GPIO mode
+            self.gpio_set_direction(pin, 1)  # set to INPUT
+
+    def close(self):
+        """Close the hid device. Does nothing if the device is not open."""
+        self._hid.close()
+
+    def __del__(self):
+        # try to close the device before destroying the instance
+        self.close()

 
     def _hid_xfer(self, report, response=True):
 
     def _hid_xfer(self, report, response=True):

+        """Perform HID Transfer"""

         # 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
         # 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)))
+        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)

+        return None

 
 

-    #----------------------------------------------------------------
+    # ----------------------------------------------------------------

     # MISC
     # MISC

-    #----------------------------------------------------------------
+    # ----------------------------------------------------------------

     def gp_get_mode(self, pin):
     def gp_get_mode(self, pin):

-        return self._hid_xfer(b'\x61')[22+pin] & 0x07
+        """Get Current Pin Mode"""
+        return self._hid_xfer(b"\x61")[22 + pin] & 0x07

 
     def gp_set_mode(self, pin, mode):
 
     def gp_set_mode(self, pin, mode):

-        # get current settings
-        current = self._hid_xfer(b'\x61')
+        """Set Current Pin Mode"""
+        # already set to that mode?
+        mode &= 0x07
+        if mode == (self._gp_config[pin] & 0x07):
+            return
+        # update GP mode for pin
+        self._gp_config[pin] = mode

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

-        report = bytearray(b'\x60'+b'\x00'*63)
+        report = bytearray(b"\x60" + b"\x00" * 63)

         # set the alter GP flag byte
         report[7] = 0xFF
         # set the alter GP flag byte
         report[7] = 0xFF

-        # each pin can be set individually
-        # but all 4 get set at once, so we need to
-        # transpose current settings
-        report[8]  = current[22]  # GP0
-        report[9]  = current[23]  # GP1
-        report[10] = current[24]  # GP2
-        report[11] = current[25]  # GP3
-        # then change only the one
-        report[8+pin] = mode & 0x07
+        # add GP setttings
+        report[8] = self._gp_config[0]
+        report[9] = self._gp_config[1]
+        report[10] = self._gp_config[2]
+        report[11] = self._gp_config[3]

         # and make it so
         self._hid_xfer(report)
 
         # and make it so
         self._hid_xfer(report)
 

-    def _pretty_report(self, report):
+    def _pretty_report(self, register):
+        report = self._hid_xfer(register)

         print("     0  1  2  3  4  5  6  7  8  9")
         index = 0
         for row in range(7):
         print("     0  1  2  3  4  5  6  7  8  9")
         index = 0
         for row in range(7):

-            print("{} : ".format(row), end='')
+            print("{} : ".format(row), end="")

             for _ in range(10):
             for _ in range(10):

-                print("{:02x} ".format(report[index]), end='')
+                print("{:02x} ".format(report[index]), end="")

                 index += 1
                 if index > 63:
                     break
             print()
 
     def _status_dump(self):
                 index += 1
                 if index > 63:
                     break
             print()
 
     def _status_dump(self):

-        self._pretty_report(self._hid_xfer(b'\x10'))
+        self._pretty_report(b"\x10")

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

-        self._pretty_report(self._hid_xfer(b'\x61'))
+        self._pretty_report(b"\x61")

 
     def _reset(self):
 
     def _reset(self):

-        self._hid_xfer(b'\x70\xAB\xCD\xEF', response=False)
+        self._hid_xfer(b"\x70\xAB\xCD\xEF", response=False)
+        self._hid.close()
+        time.sleep(MCP2221_RESET_DELAY)

         start = time.monotonic()
         while time.monotonic() - start < 5:
             try:
         start = time.monotonic()
         while time.monotonic() - start < 5:
             try:


@@ -82,160 +146,257 @@ class MCP2221:
             return
         raise OSError("open failed")
 
             return
         raise OSError("open failed")
 

-    #----------------------------------------------------------------
+    # ----------------------------------------------------------------

     # GPIO
     # GPIO

-    #----------------------------------------------------------------
+    # ----------------------------------------------------------------

     def gpio_set_direction(self, pin, mode):
     def gpio_set_direction(self, pin, mode):

-        report = bytearray(b'\x50'+b'\x00'*63)  # empty set GPIO report
+        """Set Current GPIO Pin Direction"""
+        if mode:
+            # set bit 3 for INPUT
+            self._gp_config[pin] |= 1 << 3
+        else:
+            # clear bit 3 for OUTPUT
+            self._gp_config[pin] &= ~(1 << 3)
+        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(b'\x50'+b'\x00'*63)  # empty set GPIO report
+        """Set Current GPIO Pin Value"""
+        if value:
+            # set bit 4
+            self._gp_config[pin] |= 1 << 4
+        else:
+            # clear bit 4
+            self._gp_config[pin] &= ~(1 << 4)
+        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(b'\x51')
+        """Get Current GPIO Pin Value"""
+        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.")

-        else:
-            return resp[offset]
+        return resp[offset]

 
 

-    #----------------------------------------------------------------
+    # ----------------------------------------------------------------

     # I2C
     # I2C

-    #
-    # cribbed from the C driver
-    # http://ww1.microchip.com/downloads/en/DeviceDoc/mcp2221_0_1.tar.gz
-    #   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):
     def _i2c_status(self):

-        return self._hid_xfer(b'\x10')[8]
+        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):
 
     def _i2c_cancel(self):

-        resp = self._hid_xfer(b'\x10\x00\x10')
+        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)
 
         if resp[2] == 0x10:
             # bus release will need "a few hundred microseconds"
             time.sleep(0.001)
 

+    # pylint: disable=too-many-arguments,too-many-branches

     def _i2c_write(self, cmd, address, buffer, start=0, end=None):
     def _i2c_write(self, cmd, address, buffer, start=0, end=None):

-        if self._i2c_status():
+        if self._i2c_state() != 0x00:

             self._i2c_cancel()
             self._i2c_cancel()

+

         end = end if end else len(buffer)
         length = end - start
         retries = 0
         end = end if end else len(buffer)
         length = end - start
         retries = 0

-        while (end - start) > 0:
-            chunk = min(end - start, 60)
+
+        while (end - start) > 0 or not buffer:
+            chunk = min(end - start, MCP2221_MAX_I2C_DATA_LEN)

             # write out current chunk
             # write out current chunk

-            resp = self._hid_xfer(bytes([cmd,
-                                         length & 0xFF,
-                                         (length >> 8) & 0xFF,
-                                         address << 1]) +
-                                         buffer[start:(start+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:
             # 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
                 retries += 1

-                if retries >= 5:
+                if retries >= MCP2221_RETRY_MAX:

                     raise RuntimeError("I2C write error, max retries reached.")
                 time.sleep(0.001)
                     raise RuntimeError("I2C write error, max retries reached.")
                 time.sleep(0.001)

-                continue # try again
+                continue  # try again
+            # yay chunk sent!
+            while self._i2c_state() == RESP_I2C_PARTIALDATA:
+                time.sleep(0.001)
+            if not buffer:
+                break

             start += chunk
             retries = 0
 
             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):
     def _i2c_read(self, cmd, address, buffer, start=0, end=None):

-        if self._i2c_status():
+        if self._i2c_state() not in (RESP_I2C_WRITINGNOSTOP, 0):

             self._i2c_cancel()
             self._i2c_cancel()

+

         end = end if end else len(buffer)
         length = end - start
         end = end if end else len(buffer)
         length = end - start

-        retries = 0
+
+        # 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:
         while (end - start) > 0:

-            # tell it we want to read
-            resp = self._hid_xfer(bytes([cmd,
-                                         length & 0xFF,
-                                         (length >> 8) & 0xFF,
-                                         (address << 1) | 0x01]))
-            # and then actually read
-            resp = self._hid_xfer(b'\x40')
-            # check for success
-            if resp[1] != 0x00:
-                retries += 1
-                if retries >= 5:
-                    raise RuntimeError("I2C write error, max retries reached.")
-                time.sleep(0.001)
-                continue
+            for _ 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
+            else:
+                raise RuntimeError("I2C read error: max retries reached.")
+

             # move data into buffer
             chunk = min(end - start, 60)
             # move data into buffer
             chunk = min(end - start, 60)

-            for i, k in enumerate(range(start, start+chunk)):
+            for i, k in enumerate(range(start, start + chunk)):

                 buffer[k] = resp[4 + i]
             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
-                              0x00,  # no effect
-                              0x20,  # next byte is clock divider
-                              12000000 // baudrate - 3]))
+    # pylint: enable=too-many-arguments
+
+    def _i2c_configure(self, baudrate=100000):
+        """Configure I2C"""
+        self._hid_xfer(
+            bytes(
+                [
+                    0x10,  # set parameters
+                    0x00,  # don't care
+                    0x00,  # no effect
+                    0x20,  # next byte is clock divider
+                    12000000 // baudrate - 3,
+                ]
+            )
+        )

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

+        """Write data from the buffer to an address"""

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

+        """Read data from an address and into the buffer"""

         self._i2c_read(0x91, address, buffer, start, end)
 
         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,
+    ):
+        """Write data from buffer_out to an address and then
+        read data from an address and into buffer_in
+        """

         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):
         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):

+        """Perform an I2C Device Scan"""

         found = []
         found = []

-        for addr in range(start, end+1):
+        for addr in range(start, end + 1):

             # try a write
             # 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._i2c_status() == 0x00:
-                found.append(addr)
-            # cancel and continue
-            self._i2c_cancel()
+            found.append(addr)

         return found
 
         return found
 

-    #----------------------------------------------------------------
+    # ----------------------------------------------------------------

     # ADC
     # ADC

-    #----------------------------------------------------------------
+    # ----------------------------------------------------------------

     def adc_configure(self, vref=0):
     def adc_configure(self, vref=0):

-        report = bytearray(b'\x60'+b'\x00'*63)
+        """Configure the Analog-to-Digital Converter"""
+        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(b'\x10')
+        """Read from the Analog-to-Digital Converter"""
+        resp = self._hid_xfer(b"\x10")

         return resp[49 + 2 * pin] << 8 | resp[48 + 2 * pin]
 
         return resp[49 + 2 * pin] << 8 | resp[48 + 2 * pin]
 

-    #----------------------------------------------------------------
+    # ----------------------------------------------------------------

     # DAC
     # DAC

-    #----------------------------------------------------------------
+    # ----------------------------------------------------------------

     def dac_configure(self, vref=0):
     def dac_configure(self, vref=0):

-        report = bytearray(b'\x60'+b'\x00'*63)
+        """Configure the Digital-to-Analog Converter"""
+        report = bytearray(b"\x60" + b"\x00" * 63)

         report[3] = 1 << 7 | (vref & 0b111)
         self._hid_xfer(report)
 
         report[3] = 1 << 7 | (vref & 0b111)
         self._hid_xfer(report)
 

+    # pylint: disable=unused-argument

     def dac_write(self, pin, value):
     def dac_write(self, pin, value):

-        report = bytearray(b'\x60'+b'\x00'*63)
+        """Write to the Digital-to-Analog Converter"""
+        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)
 

+    # pylint: enable=unused-argument
+
+

 mcp2221 = MCP2221()
 mcp2221 = MCP2221()