src/adafruit_blinka/microcontroller/mcp2221/mcp2221.py

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