src/adafruit_blinka/microcontroller/mcp2221/mcp2221.py

   1 # SPDX-FileCopyrightText: 2021 Melissa LeBlanc-Williams for Adafruit Industries
   2 #
   3 # SPDX-License-Identifier: MIT
   4 """Chip Definition for MCP2221"""
   5
   6 import os
   7 import time
   8 import atexit
   9
  10 import hid
  11
  12 # Here if you need it
  13 MCP2221_HID_DELAY = float(os.environ.get("BLINKA_MCP2221_HID_DELAY", 0))
  14 # Use to set delay between reset and device reopen. if negative, don't reset at all
  15 MCP2221_RESET_DELAY = float(os.environ.get("BLINKA_MCP2221_RESET_DELAY", 0.5))
  16
  17 # from the C driver
  18 # http://ww1.microchip.com/downloads/en/DeviceDoc/mcp2221_0_1.tar.gz
  19 # others (???) determined during driver developement
  20 RESP_ERR_NOERR = 0x00
  21 RESP_ADDR_NACK = 0x25
  22 RESP_READ_ERR = 0x7F
  23 RESP_READ_COMPL = 0x55
  24 RESP_READ_PARTIAL = 0x54  # ???
  25 RESP_I2C_IDLE = 0x00
  26 RESP_I2C_START_TOUT = 0x12
  27 RESP_I2C_RSTART_TOUT = 0x17
  28 RESP_I2C_WRADDRL_TOUT = 0x23
  29 RESP_I2C_WRADDRL_WSEND = 0x21
  30 RESP_I2C_WRADDRL_NACK = 0x25
  31 RESP_I2C_WRDATA_TOUT = 0x44
  32 RESP_I2C_RDDATA_TOUT = 0x52
  33 RESP_I2C_STOP_TOUT = 0x62
  34
  35 RESP_I2C_MOREDATA = 0x43  # ???
  36 RESP_I2C_PARTIALDATA = 0x41  # ???
  37 RESP_I2C_WRITINGNOSTOP = 0x45  # ???
  38
  39 MCP2221_RETRY_MAX = 50
  40 MCP2221_MAX_I2C_DATA_LEN = 60
  41 MASK_ADDR_NACK = 0x40
  42
  43
  44 class MCP2221:
  45     """MCP2221 Device Class Definition"""
  46
  47     VID = 0x04D8
  48     PID = 0x00DD
  49
  50     GP_GPIO = 0b000
  51     GP_DEDICATED = 0b001
  52     GP_ALT0 = 0b010
  53     GP_ALT1 = 0b011
  54     GP_ALT2 = 0b100
  55
  56     def __init__(self):
  57         self._hid = hid.device()
  58         self._hid.open(MCP2221.VID, MCP2221.PID)
  59         # make sure the device gets closed before exit
  60         atexit.register(self.close)
  61         if MCP2221_RESET_DELAY >= 0:
  62             self._reset()
  63         self._gp_config = [0x07] * 4  # "don't care" initial value
  64         for pin in range(4):
  65             self.gp_set_mode(pin, self.GP_GPIO)  # set to GPIO mode
  66             self.gpio_set_direction(pin, 1)  # set to INPUT
  67
  68     def close(self):
  69         """Close the hid device. Does nothing if the device is not open."""
  70         self._hid.close()
  71
  72     def __del__(self):
  73         # try to close the device before destroying the instance
  74         self.close()
  75
  76     def _hid_xfer(self, report, response=True):
  77         """Perform HID Transfer"""
  78         # first byte is report ID, which =0 for MCP2221
  79         # remaing bytes = 64 byte report data
  80         # https://github.com/libusb/hidapi/blob/083223e77952e1ef57e6b77796536a3359c1b2a3/hidapi/hidapi.h#L185
  81         self._hid.write(b"\0" + report + b"\0" * (64 - len(report)))
  82         time.sleep(MCP2221_HID_DELAY)
  83         if response:
  84             # return is 64 byte response report
  85             return self._hid.read(64)
  86         return None
  87
  88     # ----------------------------------------------------------------
  89     # MISC
  90     # ----------------------------------------------------------------
  91     def gp_get_mode(self, pin):
  92         """Get Current Pin Mode"""
  93         return self._hid_xfer(b"\x61")[22 + pin] & 0x07
  94
  95     def gp_set_mode(self, pin, mode):
  96         """Set Current Pin Mode"""
  97         # already set to that mode?
  98         mode &= 0x07
  99         if mode == (self._gp_config[pin] & 0x07):
 100             return
 101         # update GP mode for pin
 102         self._gp_config[pin] = mode
 103         # empty report, this is safe since 0's = no change
 104         report = bytearray(b"\x60" + b"\x00" * 63)
 105         # set the alter GP flag byte
 106         report[7] = 0xFF
 107         # add GP setttings
 108         report[8] = self._gp_config[0]
 109         report[9] = self._gp_config[1]
 110         report[10] = self._gp_config[2]
 111         report[11] = self._gp_config[3]
 112         # and make it so
 113         self._hid_xfer(report)
 114
 115     def _pretty_report(self, register):
 116         report = self._hid_xfer(register)
 117         print("     0  1  2  3  4  5  6  7  8  9")
 118         index = 0
 119         for row in range(7):
 120             print("{} : ".format(row), end="")
 121             for _ in range(10):
 122                 print("{:02x} ".format(report[index]), end="")
 123                 index += 1
 124                 if index > 63:
 125                     break
 126             print()
 127
 128     def _status_dump(self):
 129         self._pretty_report(b"\x10")
 130
 131     def _sram_dump(self):
 132         self._pretty_report(b"\x61")
 133
 134     def _reset(self):
 135         self._hid_xfer(b"\x70\xAB\xCD\xEF", response=False)
 136         self._hid.close()
 137         time.sleep(MCP2221_RESET_DELAY)
 138         start = time.monotonic()
 139         while time.monotonic() - start < 5:
 140             try:
 141                 self._hid.open(MCP2221.VID, MCP2221.PID)
 142             except OSError:
 143                 # try again
 144                 time.sleep(0.1)
 145                 continue
 146             return
 147         raise OSError("open failed")
 148
 149     # ----------------------------------------------------------------
 150     # GPIO
 151     # ----------------------------------------------------------------
 152     def gpio_set_direction(self, pin, mode):
 153         """Set Current GPIO Pin Direction"""
 154         if mode:
 155             # set bit 3 for INPUT
 156             self._gp_config[pin] |= 1 << 3
 157         else:
 158             # clear bit 3 for OUTPUT
 159             self._gp_config[pin] &= ~(1 << 3)
 160         report = bytearray(b"\x50" + b"\x00" * 63)  # empty set GPIO report
 161         offset = 4 * (pin + 1)
 162         report[offset] = 0x01  # set pin direction
 163         report[offset + 1] = mode  # to this
 164         self._hid_xfer(report)
 165
 166     def gpio_set_pin(self, pin, value):
 167         """Set Current GPIO Pin Value"""
 168         if value:
 169             # set bit 4
 170             self._gp_config[pin] |= 1 << 4
 171         else:
 172             # clear bit 4
 173             self._gp_config[pin] &= ~(1 << 4)
 174         report = bytearray(b"\x50" + b"\x00" * 63)  # empty set GPIO report
 175         offset = 2 + 4 * pin
 176         report[offset] = 0x01  # set pin value
 177         report[offset + 1] = value  # to this
 178         self._hid_xfer(report)
 179
 180     def gpio_get_pin(self, pin):
 181         """Get Current GPIO Pin Value"""
 182         resp = self._hid_xfer(b"\x51")
 183         offset = 2 + 2 * pin
 184         if resp[offset] == 0xEE:
 185             raise RuntimeError("Pin is not set for GPIO operation.")
 186         return resp[offset]
 187
 188     # ----------------------------------------------------------------
 189     # I2C
 190     # ----------------------------------------------------------------
 191     def _i2c_status(self):
 192         resp = self._hid_xfer(b"\x10")
 193         if resp[1] != 0:
 194             raise RuntimeError("Couldn't get I2C status")
 195         return resp
 196
 197     def _i2c_state(self):
 198         return self._i2c_status()[8]
 199
 200     def _i2c_cancel(self):
 201         resp = self._hid_xfer(b"\x10\x00\x10")
 202         if resp[1] != 0x00:
 203             raise RuntimeError("Couldn't cancel I2C")
 204         if resp[2] == 0x10:
 205             # bus release will need "a few hundred microseconds"
 206             time.sleep(0.001)
 207
 208     # pylint: disable=too-many-arguments,too-many-branches
 209     def _i2c_write(self, cmd, address, buffer, start=0, end=None):
 210         if self._i2c_state() != 0x00:
 211             self._i2c_cancel()
 212
 213         end = end if end else len(buffer)
 214         length = end - start
 215         retries = 0
 216
 217         while (end - start) > 0 or not buffer:
 218             chunk = min(end - start, MCP2221_MAX_I2C_DATA_LEN)
 219             # write out current chunk
 220             resp = self._hid_xfer(
 221                 bytes([cmd, length & 0xFF, (length >> 8) & 0xFF, address << 1])
 222                 + buffer[start : (start + chunk)]
 223             )
 224             # check for success
 225             if resp[1] != 0x00:
 226                 if resp[2] in (
 227                     RESP_I2C_START_TOUT,
 228                     RESP_I2C_WRADDRL_TOUT,
 229                     RESP_I2C_WRADDRL_NACK,
 230                     RESP_I2C_WRDATA_TOUT,
 231                     RESP_I2C_STOP_TOUT,
 232                 ):
 233                     raise RuntimeError("Unrecoverable I2C state failure")
 234                 retries += 1
 235                 if retries >= MCP2221_RETRY_MAX:
 236                     raise RuntimeError("I2C write error, max retries reached.")
 237                 time.sleep(0.001)
 238                 continue  # try again
 239             # yay chunk sent!
 240             while self._i2c_state() == RESP_I2C_PARTIALDATA:
 241                 time.sleep(0.001)
 242             if not buffer:
 243                 break
 244             start += chunk
 245             retries = 0
 246
 247         # check status in another loop
 248         for _ in range(MCP2221_RETRY_MAX):
 249             status = self._i2c_status()
 250             if status[20] & MASK_ADDR_NACK:
 251                 raise OSError("I2C slave address was NACK'd")
 252             usb_cmd_status = status[8]
 253             if usb_cmd_status == 0:
 254                 break
 255             if usb_cmd_status == RESP_I2C_WRITINGNOSTOP and cmd == 0x94:
 256                 break  # this is OK too!
 257             if usb_cmd_status in (
 258                 RESP_I2C_START_TOUT,
 259                 RESP_I2C_WRADDRL_TOUT,
 260                 RESP_I2C_WRADDRL_NACK,
 261                 RESP_I2C_WRDATA_TOUT,
 262                 RESP_I2C_STOP_TOUT,
 263             ):
 264                 raise RuntimeError("Unrecoverable I2C state failure")
 265             time.sleep(0.001)
 266         else:
 267             raise RuntimeError("I2C write error: max retries reached.")
 268         # whew success!
 269
 270     def _i2c_read(self, cmd, address, buffer, start=0, end=None):
 271         if self._i2c_state() not in (RESP_I2C_WRITINGNOSTOP, 0):
 272             self._i2c_cancel()
 273
 274         end = end if end else len(buffer)
 275         length = end - start
 276
 277         # tell it we want to read
 278         resp = self._hid_xfer(
 279             bytes([cmd, length & 0xFF, (length >> 8) & 0xFF, (address << 1) | 0x01])
 280         )
 281
 282         # check for success
 283         if resp[1] != 0x00:
 284             raise RuntimeError("Unrecoverable I2C read failure")
 285
 286         # and now the read part
 287         while (end - start) > 0:
 288             for _ in range(MCP2221_RETRY_MAX):
 289                 # the actual read
 290                 resp = self._hid_xfer(b"\x40")
 291                 # check for success
 292                 if resp[1] == RESP_I2C_PARTIALDATA:
 293                     time.sleep(0.001)
 294                     continue
 295                 if resp[1] != 0x00:
 296                     raise RuntimeError("Unrecoverable I2C read failure")
 297                 if resp[2] == RESP_ADDR_NACK:
 298                     raise RuntimeError("I2C NACK")
 299                 if resp[3] == 0x00 and resp[2] == 0x00:
 300                     break
 301                 if resp[3] == RESP_READ_ERR:
 302                     time.sleep(0.001)
 303                     continue
 304                 if resp[2] in (RESP_READ_COMPL, RESP_READ_PARTIAL):
 305                     break
 306             else:
 307                 raise RuntimeError("I2C read error: max retries reached.")
 308
 309             # move data into buffer
 310             chunk = min(end - start, 60)
 311             for i, k in enumerate(range(start, start + chunk)):
 312                 buffer[k] = resp[4 + i]
 313             start += chunk
 314
 315     # pylint: enable=too-many-arguments
 316
 317     def _i2c_configure(self, baudrate=100000):
 318         """Configure I2C"""
 319         self._hid_xfer(
 320             bytes(
 321                 [
 322                     0x10,  # set parameters
 323                     0x00,  # don't care
 324                     0x00,  # no effect
 325                     0x20,  # next byte is clock divider
 326                     12000000 // baudrate - 3,
 327                 ]
 328             )
 329         )
 330
 331     def i2c_writeto(self, address, buffer, *, start=0, end=None):
 332         """Write data from the buffer to an address"""
 333         self._i2c_write(0x90, address, buffer, start, end)
 334
 335     def i2c_readfrom_into(self, address, buffer, *, start=0, end=None):
 336         """Read data from an address and into the buffer"""
 337         self._i2c_read(0x91, address, buffer, start, end)
 338
 339     def i2c_writeto_then_readfrom(
 340         self,
 341         address,
 342         out_buffer,
 343         in_buffer,
 344         *,
 345         out_start=0,
 346         out_end=None,
 347         in_start=0,
 348         in_end=None,
 349     ):
 350         """Write data from buffer_out to an address and then
 351         read data from an address and into buffer_in
 352         """
 353         self._i2c_write(0x94, address, out_buffer, out_start, out_end)
 354         self._i2c_read(0x93, address, in_buffer, in_start, in_end)
 355
 356     def i2c_scan(self, *, start=0, end=0x79):
 357         """Perform an I2C Device Scan"""
 358         found = []
 359         for addr in range(start, end + 1):
 360             # try a write
 361             try:
 362                 self.i2c_writeto(addr, b"\x00")
 363             except OSError:  # no reply!
 364                 # We got a NACK, which could be correct
 365                 continue
 366             # store if success
 367             found.append(addr)
 368         return found
 369
 370     # ----------------------------------------------------------------
 371     # ADC
 372     # ----------------------------------------------------------------
 373     def adc_configure(self, vref=0):
 374         """Configure the Analog-to-Digital Converter"""
 375         report = bytearray(b"\x60" + b"\x00" * 63)
 376         report[5] = 1 << 7 | (vref & 0b111)
 377         self._hid_xfer(report)
 378
 379     def adc_read(self, pin):
 380         """Read from the Analog-to-Digital Converter"""
 381         resp = self._hid_xfer(b"\x10")
 382         return resp[49 + 2 * pin] << 8 | resp[48 + 2 * pin]
 383
 384     # ----------------------------------------------------------------
 385     # DAC
 386     # ----------------------------------------------------------------
 387     def dac_configure(self, vref=0):
 388         """Configure the Digital-to-Analog Converter"""
 389         report = bytearray(b"\x60" + b"\x00" * 63)
 390         report[3] = 1 << 7 | (vref & 0b111)
 391         self._hid_xfer(report)
 392
 393     # pylint: disable=unused-argument
 394     def dac_write(self, pin, value):
 395         """Write to the Digital-to-Analog Converter"""
 396         report = bytearray(b"\x60" + b"\x00" * 63)
 397         report[4] = 1 << 7 | (value & 0b11111)
 398         self._hid_xfer(report)
 399
 400     # pylint: enable=unused-argument
 401
 402
 403 mcp2221 = MCP2221()