src/adafruit_blinka/microcontroller/mcp2221/mcp2221.py

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