src/adafruit_blinka/microcontroller/mcp2221/mcp2221.py

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