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