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