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:
 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             start += chunk
 227             retries = 0
 228
 229         # check status in another loop
 230         for _ in range(MCP2221_RETRY_MAX):
 231             status = self._i2c_status()
 232             if status[20] & MASK_ADDR_NACK:
 233                 raise RuntimeError("I2C slave address was NACK'd")
 234             usb_cmd_status = status[8]
 235             if usb_cmd_status == 0:
 236                 break
 237             if usb_cmd_status == RESP_I2C_WRITINGNOSTOP and cmd == 0x94:
 238                 break  # this is OK too!
 239             if usb_cmd_status in (
 240                 RESP_I2C_START_TOUT,
 241                 RESP_I2C_WRADDRL_TOUT,
 242                 RESP_I2C_WRADDRL_NACK,
 243                 RESP_I2C_WRDATA_TOUT,
 244                 RESP_I2C_STOP_TOUT,
 245             ):
 246                 raise RuntimeError("Unrecoverable I2C state failure")
 247             time.sleep(0.001)
 248         else:
 249             raise RuntimeError("I2C write error: max retries reached.")
 250         # whew success!
 251
 252     def _i2c_read(self, cmd, address, buffer, start=0, end=None):
 253         if self._i2c_state() not in (RESP_I2C_WRITINGNOSTOP, 0):
 254             self._i2c_cancel()
 255
 256         end = end if end else len(buffer)
 257         length = end - start
 258
 259         # tell it we want to read
 260         resp = self._hid_xfer(
 261             bytes([cmd, length & 0xFF, (length >> 8) & 0xFF, (address << 1) | 0x01])
 262         )
 263
 264         # check for success
 265         if resp[1] != 0x00:
 266             raise RuntimeError("Unrecoverable I2C read failure")
 267
 268         # and now the read part
 269         while (end - start) > 0:
 270             for _ in range(MCP2221_RETRY_MAX):
 271                 # the actual read
 272                 resp = self._hid_xfer(b"\x40")
 273                 # check for success
 274                 if resp[1] == RESP_I2C_PARTIALDATA:
 275                     time.sleep(0.001)
 276                     continue
 277                 if resp[1] != 0x00:
 278                     raise RuntimeError("Unrecoverable I2C read failure")
 279                 if resp[2] == RESP_ADDR_NACK:
 280                     raise RuntimeError("I2C NACK")
 281                 if resp[3] == 0x00 and resp[2] == 0x00:
 282                     break
 283                 if resp[3] == RESP_READ_ERR:
 284                     time.sleep(0.001)
 285                     continue
 286                 if resp[2] in (RESP_READ_COMPL, RESP_READ_PARTIAL):
 287                     break
 288             else:
 289                 raise RuntimeError("I2C read error: max retries reached.")
 290
 291             # move data into buffer
 292             chunk = min(end - start, 60)
 293             for i, k in enumerate(range(start, start + chunk)):
 294                 buffer[k] = resp[4 + i]
 295             start += chunk
 296
 297     # pylint: enable=too-many-arguments
 298
 299     def i2c_configure(self, baudrate=100000):
 300         """Configure I2C"""
 301         self._hid_xfer(
 302             bytes(
 303                 [
 304                     0x10,  # set parameters
 305                     0x00,  # don't care
 306                     0x00,  # no effect
 307                     0x20,  # next byte is clock divider
 308                     12000000 // baudrate - 3,
 309                 ]
 310             )
 311         )
 312
 313     def i2c_writeto(self, address, buffer, *, start=0, end=None):
 314         """Write data from the buffer to an address"""
 315         self._i2c_write(0x90, address, buffer, start, end)
 316
 317     def i2c_readfrom_into(self, address, buffer, *, start=0, end=None):
 318         """Read data from an address and into the buffer"""
 319         self._i2c_read(0x91, address, buffer, start, end)
 320
 321     def i2c_writeto_then_readfrom(
 322         self,
 323         address,
 324         out_buffer,
 325         in_buffer,
 326         *,
 327         out_start=0,
 328         out_end=None,
 329         in_start=0,
 330         in_end=None
 331     ):
 332         """Write data from buffer_out to an address and then
 333         read data from an address and into buffer_in
 334         """
 335         self._i2c_write(0x94, address, out_buffer, out_start, out_end)
 336         self._i2c_read(0x93, address, in_buffer, in_start, in_end)
 337
 338     def i2c_scan(self, *, start=0, end=0x79):
 339         """Perform an I2C Device Scan"""
 340         found = []
 341         for addr in range(start, end + 1):
 342             # try a write
 343             try:
 344                 self.i2c_writeto(addr, b"\x00")
 345             except RuntimeError:  # no reply!
 346                 continue
 347             # store if success
 348             found.append(addr)
 349         return found
 350
 351     # ----------------------------------------------------------------
 352     # ADC
 353     # ----------------------------------------------------------------
 354     def adc_configure(self, vref=0):
 355         """Configure the Analog-to-Digital Converter"""
 356         report = bytearray(b"\x60" + b"\x00" * 63)
 357         report[5] = 1 << 7 | (vref & 0b111)
 358         self._hid_xfer(report)
 359
 360     def adc_read(self, pin):
 361         """Read from the Analog-to-Digital Converter"""
 362         resp = self._hid_xfer(b"\x10")
 363         return resp[49 + 2 * pin] << 8 | resp[48 + 2 * pin]
 364
 365     # ----------------------------------------------------------------
 366     # DAC
 367     # ----------------------------------------------------------------
 368     def dac_configure(self, vref=0):
 369         """Configure the Digital-to-Analog Converter"""
 370         report = bytearray(b"\x60" + b"\x00" * 63)
 371         report[3] = 1 << 7 | (vref & 0b111)
 372         self._hid_xfer(report)
 373
 374     # pylint: disable=unused-argument
 375     def dac_write(self, pin, value):
 376         """Write to the Digital-to-Analog Converter"""
 377         report = bytearray(b"\x60" + b"\x00" * 63)
 378         report[4] = 1 << 7 | (value & 0b11111)
 379         self._hid_xfer(report)
 380
 381     # pylint: enable=unused-argument
 382
 383
 384 mcp2221 = MCP2221()