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