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