src/adafruit_blinka/microcontroller/pico_u2if/pico_u2if.py

   1 """Chip Definition for Pico with u2if firmware"""
   2 # https://github.com/execuc/u2if
   3
   4 import hid
   5
   6 # pylint: disable=import-outside-toplevel,too-many-branches,too-many-statements
   7 # pylint: disable=too-many-arguments,too-many-function-args, too-many-public-methods
   8
   9
  10 class Pico_u2if:
  11     """MCP2221 Device Class Definition"""
  12
  13     VID = 0xCAFE
  14     PID = 0x4005
  15
  16     # MISC
  17     RESP_OK = 0x01
  18
  19     # GPIO
  20     GPIO_INIT_PIN = 0x20
  21     GPIO_SET_VALUE = 0x21
  22     GPIO_GET_VALUE = 0x22
  23
  24     # ADC
  25     ADC_INIT_PIN = 0x40
  26     ADC_GET_VALUE = 0x41
  27
  28     # I2C
  29     I2C0_INIT = 0x80
  30     I2C0_DEINIT = 0x81
  31     I2C0_WRITE = 0x82
  32     I2C0_READ = 0x83
  33     I2C0_WRITE_FROM_UART = 0x84
  34     I2C1_INIT = I2C0_INIT + 0x10
  35     I2C1_DEINIT = I2C0_DEINIT + 0x10
  36     I2C1_WRITE = I2C0_WRITE + 0x10
  37     I2C1_READ = I2C0_READ + 0x10
  38     I2C1_WRITE_FROM_UART = I2C0_WRITE_FROM_UART + 0x10
  39
  40     # SPI
  41     SPI0_INIT = 0x60
  42     SPI0_DEINIT = 0x61
  43     SPI0_WRITE = 0x62
  44     SPI0_READ = 0x63
  45     SPI0_WRITE_FROM_UART = 0x64
  46     SPI1_INIT = SPI0_INIT + 0x10
  47     SPI1_DEINIT = SPI0_DEINIT + 0x10
  48     SPI1_WRITE = SPI0_WRITE + 0x10
  49     SPI1_READ = SPI0_READ + 0x10
  50     SPI1_WRITE_FROM_UART = SPI0_WRITE_FROM_UART + 0x10
  51
  52     # WS2812B (LED)
  53     WS2812B_INIT = 0xA0
  54     WS2812B_DEINIT = 0xA1
  55     WS2812B_WRITE = 0xA2
  56
  57     # PWM
  58     PWM_INIT_PIN = 0x30
  59     PWM_DEINIT_PIN = 0x31
  60     PWM_SET_FREQ = 0x32
  61     PWM_GET_FREQ = 0x33
  62     PWM_SET_DUTY_U16 = 0x34
  63     PWM_GET_DUTY_U16 = 0x35
  64     PWM_SET_DUTY_NS = 0x36
  65     PWM_GET_DUTY_NS = 0x37
  66
  67     # UART
  68     UART0_INIT = 0x50
  69     UART0_DEINIT = 0x51
  70     UART0_WRITE = 0x52
  71     UART0_READ = 0x53
  72
  73     def __init__(self):
  74         self._hid = hid.device()
  75         self._hid.open(Pico_u2if.VID, Pico_u2if.PID)
  76         self._i2c_index = None
  77         self._spi_index = None
  78         self._serial = None
  79         self._neopixel_initialized = False
  80         self._uart_rx_buffer = None
  81
  82     def _hid_xfer(self, report, response=True):
  83         """Perform HID Transfer"""
  84         # first byte is report ID, which =0
  85         # remaing bytes = 64 byte report data
  86         # https://github.com/libusb/hidapi/blob/083223e77952e1ef57e6b77796536a3359c1b2a3/hidapi/hidapi.h#L185
  87         self._hid.write(b"\0" + report + b"\0" * (64 - len(report)))
  88         if response:
  89             # return is 64 byte response report
  90             return self._hid.read(64)
  91         return None
  92
  93     # ----------------------------------------------------------------
  94     # GPIO
  95     # ----------------------------------------------------------------
  96     def gpio_init_pin(self, pin_id, direction, pull):
  97         """Configure GPIO Pin."""
  98         self._hid_xfer(
  99             bytes(
 100                 [
 101                     self.GPIO_INIT_PIN,
 102                     pin_id,
 103                     direction,
 104                     pull,
 105                 ]
 106             )
 107         )
 108
 109     def gpio_set_pin(self, pin_id, value):
 110         """Set Current GPIO Pin Value"""
 111         self._hid_xfer(
 112             bytes(
 113                 [
 114                     self.GPIO_SET_VALUE,
 115                     pin_id,
 116                     int(value),
 117                 ]
 118             )
 119         )
 120
 121     def gpio_get_pin(self, pin_id):
 122         """Get Current GPIO Pin Value"""
 123         resp = self._hid_xfer(
 124             bytes(
 125                 [
 126                     self.GPIO_GET_VALUE,
 127                     pin_id,
 128                 ]
 129             ),
 130             True,
 131         )
 132         return resp[3] != 0x00
 133
 134     # ----------------------------------------------------------------
 135     # ADC
 136     # ----------------------------------------------------------------
 137     def adc_init_pin(self, pin_id):
 138         """Configure ADC Pin."""
 139         self._hid_xfer(
 140             bytes(
 141                 [
 142                     self.ADC_INIT_PIN,
 143                     pin_id,
 144                 ]
 145             )
 146         )
 147
 148     def adc_get_value(self, pin_id):
 149         """Get ADC value for pin."""
 150         resp = self._hid_xfer(
 151             bytes(
 152                 [
 153                     self.ADC_GET_VALUE,
 154                     pin_id,
 155                 ]
 156             ),
 157             True,
 158         )
 159         return int.from_bytes(resp[3 : 3 + 2], byteorder="little")
 160
 161     # ----------------------------------------------------------------
 162     # I2C
 163     # ----------------------------------------------------------------
 164     def i2c_configure(self, baudrate, pullup=False):
 165         """Configure I2C."""
 166         if self._i2c_index is None:
 167             raise RuntimeError("I2C bus not initialized.")
 168
 169         resp = self._hid_xfer(
 170             bytes(
 171                 [
 172                     self.I2C0_INIT if self._i2c_index == 0 else self.I2C1_INIT,
 173                     0x00 if not pullup else 0x01,
 174                 ]
 175             )
 176             + baudrate.to_bytes(4, byteorder="little"),
 177             True,
 178         )
 179         if resp[1] != self.RESP_OK:
 180             raise RuntimeError("I2C init error.")
 181
 182     def i2c_set_port(self, index):
 183         """Set I2C port."""
 184         if index not in (0, 1):
 185             raise ValueError("I2C index must be 0 or 1.")
 186         self._i2c_index = index
 187
 188     def _i2c_write(self, address, buffer, start=0, end=None, stop=True):
 189         """Write data from the buffer to an address"""
 190         if self._i2c_index is None:
 191             raise RuntimeError("I2C bus not initialized.")
 192
 193         end = end if end else len(buffer)
 194
 195         write_cmd = self.I2C0_WRITE if self._i2c_index == 0 else self.I2C1_WRITE
 196         stop_flag = 0x01 if stop else 0x00
 197
 198         while (end - start) > 0:
 199             remain_bytes = end - start
 200             chunk = min(remain_bytes, 64 - 7)
 201             resp = self._hid_xfer(
 202                 bytes([write_cmd, address, stop_flag])
 203                 + remain_bytes.to_bytes(4, byteorder="little")
 204                 + buffer[start : (start + chunk)],
 205                 True,
 206             )
 207             if resp[1] != self.RESP_OK:
 208                 raise RuntimeError("I2C write error")
 209             start += chunk
 210
 211     def _i2c_read(self, address, buffer, start=0, end=None):
 212         """Read data from an address and into the buffer"""
 213         # TODO: support chunkified reads
 214         if self._i2c_index is None:
 215             raise RuntimeError("I2C bus not initialized.")
 216
 217         end = end if end else len(buffer)
 218
 219         read_cmd = self.I2C0_READ if self._i2c_index == 0 else self.I2C1_READ
 220         stop_flag = 0x01  # always stop
 221         read_size = end - start
 222
 223         resp = self._hid_xfer(bytes([read_cmd, address, stop_flag, read_size]), True)
 224         if resp[1] != self.RESP_OK:
 225             raise RuntimeError("I2C write error")
 226         # move into buffer
 227         for i in range(read_size):
 228             buffer[start + i] = resp[i + 2]
 229
 230     def i2c_writeto(self, address, buffer, *, start=0, end=None):
 231         """Write data from the buffer to an address"""
 232         self._i2c_write(address, buffer, start, end)
 233
 234     def i2c_readfrom_into(self, address, buffer, *, start=0, end=None):
 235         """Read data from an address and into the buffer"""
 236         self._i2c_read(address, buffer, start, end)
 237
 238     def i2c_writeto_then_readfrom(
 239         self,
 240         address,
 241         out_buffer,
 242         in_buffer,
 243         *,
 244         out_start=0,
 245         out_end=None,
 246         in_start=0,
 247         in_end=None
 248     ):
 249         """Write data from buffer_out to an address and then
 250         read data from an address and into buffer_in
 251         """
 252         self._i2c_write(address, out_buffer, out_start, out_end, False)
 253         self._i2c_read(address, in_buffer, in_start, in_end)
 254
 255     def i2c_scan(self, *, start=0, end=0x79):
 256         """Perform an I2C Device Scan"""
 257         if self._i2c_index is None:
 258             raise RuntimeError("I2C bus not initialized.")
 259         found = []
 260         for addr in range(start, end + 1):
 261             # try a write
 262             try:
 263                 self.i2c_writeto(addr, b"\x00\x00\x00")
 264             except RuntimeError:  # no reply!
 265                 continue
 266             # store if success
 267             found.append(addr)
 268         return found
 269
 270     # ----------------------------------------------------------------
 271     # SPI
 272     # ----------------------------------------------------------------
 273     def spi_configure(self, baudrate):
 274         """Configure SPI."""
 275         if self._spi_index is None:
 276             raise RuntimeError("SPI bus not initialized.")
 277
 278         resp = self._hid_xfer(
 279             bytes(
 280                 [
 281                     self.SPI0_INIT if self._spi_index == 0 else self.SPI1_INIT,
 282                     0x00,  # mode, not yet implemented
 283                 ]
 284             )
 285             + baudrate.to_bytes(4, byteorder="little"),
 286             True,
 287         )
 288         if resp[1] != self.RESP_OK:
 289             raise RuntimeError("SPI init error.")
 290
 291     def spi_set_port(self, index):
 292         """Set SPI port."""
 293         if index not in (0, 1):
 294             raise ValueError("SPI index must be 0 or 1.")
 295         self._spi_index = index
 296
 297     def spi_write(self, buffer, *, start=0, end=None):
 298         """SPI write."""
 299         if self._spi_index is None:
 300             raise RuntimeError("SPI bus not initialized.")
 301
 302         end = end if end else len(buffer)
 303
 304         write_cmd = self.SPI0_WRITE if self._spi_index == 0 else self.SPI1_WRITE
 305
 306         while (end - start) > 0:
 307             remain_bytes = end - start
 308             chunk = min(remain_bytes, 64 - 3)
 309             resp = self._hid_xfer(
 310                 bytes([write_cmd, chunk]) + buffer[start : (start + chunk)], True
 311             )
 312             if resp[1] != self.RESP_OK:
 313                 raise RuntimeError("SPI write error")
 314             start += chunk
 315
 316     def spi_readinto(self, buffer, *, start=0, end=None, write_value=0):
 317         """SPI readinto."""
 318         if self._spi_index is None:
 319             raise RuntimeError("SPI bus not initialized.")
 320
 321         end = end if end else len(buffer)
 322         read_cmd = self.SPI0_READ if self._spi_index == 0 else self.SPI1_READ
 323         read_size = end - start
 324
 325         resp = self._hid_xfer(bytes([read_cmd, write_value, read_size]), True)
 326         if resp[1] != self.RESP_OK:
 327             raise RuntimeError("SPI write error")
 328         # move into buffer
 329         for i in range(read_size):
 330             buffer[start + i] = resp[i + 2]
 331
 332     def spi_write_readinto(
 333         self,
 334         buffer_out,
 335         buffer_in,
 336         *,
 337         out_start=0,
 338         out_end=None,
 339         in_start=0,
 340         in_end=None
 341     ):
 342         """SPI write and readinto."""
 343         raise NotImplementedError("SPI write_readinto Not implemented")
 344
 345     # ----------------------------------------------------------------
 346     # NEOPIXEL
 347     # ----------------------------------------------------------------
 348     def neopixel_write(self, gpio, buf):
 349         """NeoPixel write."""
 350         # open serial (data is sent over this)
 351         if self._serial is None:
 352             import serial
 353             import serial.tools.list_ports
 354
 355             ports = serial.tools.list_ports.comports()
 356             for port in ports:
 357                 if port.vid == self.VID and port.pid == self.PID:
 358                     self._serial = serial.Serial(port.device)
 359                     break
 360         if self._serial is None:
 361             raise RuntimeError("Could not find Pico com port.")
 362
 363         # init
 364         if not self._neopixel_initialized:
 365             # deinit any current setup
 366             self._hid_xfer(bytes([self.WS2812B_DEINIT]))
 367             resp = self._hid_xfer(
 368                 bytes(
 369                     [
 370                         self.WS2812B_INIT,
 371                         gpio._pin.id,
 372                     ]
 373                 ),
 374                 True,
 375             )
 376             if resp[1] != self.RESP_OK:
 377                 raise RuntimeError("Neopixel init error")
 378             self._neopixel_initialized = True
 379
 380         # write
 381         # command is done over HID
 382         remain_bytes = len(buf)
 383         resp = self._hid_xfer(
 384             bytes([self.WS2812B_WRITE]) + remain_bytes.to_bytes(4, byteorder="little"),
 385             True,
 386         )
 387         if resp[1] != self.RESP_OK:
 388             # pylint: disable=no-else-raise
 389             if resp[2] == 0x01:
 390                 raise RuntimeError(
 391                     "Neopixel write error : too many pixel for the firmware."
 392                 )
 393             elif resp[2] == 0x02:
 394                 raise RuntimeError(
 395                     "Neopixel write error : transfer already in progress."
 396                 )
 397             else:
 398                 raise RuntimeError("Neopixel write error")
 399         # buffer is sent over serial
 400         self._serial.write(buf)
 401         self._serial.flush()
 402
 403     # ----------------------------------------------------------------
 404     # PWM
 405     # ----------------------------------------------------------------
 406     # pylint: disable=unused_argument
 407     def pwm_configure(self, pin, frequency=500, duty_cycle=0, variable_frequency=False):
 408         """Configure PWM."""
 409         self.pwm_deinit(pin)
 410         resp = self._hid_xfer(bytes([self.PWM_INIT_PIN, pin.id]), True)
 411         if resp[1] != self.RESP_OK:
 412             raise RuntimeError("PWM init error.")
 413
 414         self.pwm_set_frequency(pin, frequency)
 415         self.pwm_set_duty_cycle(pin, duty_cycle)
 416
 417     def pwm_deinit(self, pin):
 418         """Deinit PWM."""
 419         self._hid_xfer(bytes([self.PWM_DEINIT_PIN, pin.id]))
 420
 421     def pwm_get_frequency(self, pin):
 422         """PWM get freq."""
 423         resp = self._hid_xfer(bytes([self.PWM_GET_FREQ, pin.id]), True)
 424         if resp[1] != self.RESP_OK:
 425             raise RuntimeError("PWM get frequency error.")
 426         return int.from_bytes(resp[3 : 3 + 4], byteorder="little")
 427
 428     def pwm_set_frequency(self, pin, frequency):
 429         """PWM set freq."""
 430         resp = self._hid_xfer(
 431             bytes([self.PWM_SET_FREQ, pin.id])
 432             + frequency.to_bytes(4, byteorder="little"),
 433             True,
 434         )
 435         if resp[1] != self.RESP_OK:
 436             # pylint: disable=no-else-raise
 437             if resp[3] == 0x01:
 438                 raise RuntimeError("PWM different frequency on same slice.")
 439             elif resp[3] == 0x02:
 440                 raise RuntimeError("PWM frequency too low.")
 441             elif resp[3] == 0x03:
 442                 raise RuntimeError("PWM frequency too high.")
 443             else:
 444                 raise RuntimeError("PWM frequency error.")
 445
 446     def pwm_get_duty_cycle(self, pin):
 447         """PWM get duty cycle."""
 448         resp = self._hid_xfer(bytes([self.PWM_GET_DUTY_U16, pin.id]), True)
 449         if resp[1] != self.RESP_OK:
 450             raise RuntimeError("PWM get duty cycle error.")
 451         return int.from_bytes(resp[3 : 3 + 4], byteorder="little")
 452
 453     def pwm_set_duty_cycle(self, pin, duty_cycle):
 454         """PWM set duty cycle."""
 455         resp = self._hid_xfer(
 456             bytes([self.PWM_SET_DUTY_U16, pin.id])
 457             + duty_cycle.to_bytes(2, byteorder="little"),
 458             True,
 459         )
 460         if resp[1] != self.RESP_OK:
 461             raise RuntimeError("PWM set duty cycle error.")
 462
 463
 464 pico_u2if = Pico_u2if()