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         """Configure I2C."""
 163         if self._i2c_index is None:
 164             raise RuntimeError("I2C bus not initialized.")
 165
 166         resp = self._hid_xfer(
 167             bytes(
 168                 [
 169                     self.I2C0_INIT if self._i2c_index == 0 else self.I2C1_INIT,
 170                     0x00 if not pullup else 0x01,
 171                 ]
 172             )
 173             + baudrate.to_bytes(4, byteorder="little"),
 174             True,
 175         )
 176         if resp[1] != self.RESP_OK:
 177             raise RuntimeError("I2C init error.")
 178
 179     def i2c_set_port(self, index):
 180         """Set I2C port."""
 181         if index not in (0, 1):
 182             raise ValueError("I2C index must be 0 or 1.")
 183         self._i2c_index = index
 184
 185     def _i2c_write(self, address, buffer, start=0, end=None, stop=True):
 186         """Write data from the buffer to an address"""
 187         if self._i2c_index is None:
 188             raise RuntimeError("I2C bus not initialized.")
 189
 190         end = end if end else len(buffer)
 191
 192         write_cmd = self.I2C0_WRITE if self._i2c_index == 0 else self.I2C1_WRITE
 193         stop_flag = 0x01 if stop else 0x00
 194
 195         while (end - start) > 0:
 196             remain_bytes = end - start
 197             chunk = min(remain_bytes, 64 - 7)
 198             resp = self._hid_xfer(
 199                 bytes([write_cmd, address, stop_flag])
 200                 + remain_bytes.to_bytes(4, byteorder="little")
 201                 + buffer[start : (start + chunk)],
 202                 True,
 203             )
 204             if resp[1] != self.RESP_OK:
 205                 raise RuntimeError("I2C write error")
 206             start += chunk
 207
 208     def _i2c_read(self, address, buffer, start=0, end=None):
 209         """Read data from an address and into the buffer"""
 210         # TODO: support chunkified reads
 211         if self._i2c_index is None:
 212             raise RuntimeError("I2C bus not initialized.")
 213
 214         end = end if end else len(buffer)
 215
 216         read_cmd = self.I2C0_READ if self._i2c_index == 0 else self.I2C1_READ
 217         stop_flag = 0x01  # always stop
 218         read_size = end - start
 219
 220         resp = self._hid_xfer(bytes([read_cmd, address, stop_flag, read_size]), True)
 221         if resp[1] != self.RESP_OK:
 222             raise RuntimeError("I2C write error")
 223         # move into buffer
 224         for i in range(read_size):
 225             buffer[start + i] = resp[i + 2]
 226
 227     def i2c_writeto(self, address, buffer, *, start=0, end=None):
 228         """Write data from the buffer to an address"""
 229         self._i2c_write(address, buffer, start, end)
 230
 231     def i2c_readfrom_into(self, address, buffer, *, start=0, end=None):
 232         """Read data from an address and into the buffer"""
 233         self._i2c_read(address, buffer, start, end)
 234
 235     def i2c_writeto_then_readfrom(
 236         self,
 237         address,
 238         out_buffer,
 239         in_buffer,
 240         *,
 241         out_start=0,
 242         out_end=None,
 243         in_start=0,
 244         in_end=None
 245     ):
 246         """Write data from buffer_out to an address and then
 247         read data from an address and into buffer_in
 248         """
 249         self._i2c_write(address, out_buffer, out_start, out_end, False)
 250         self._i2c_read(address, in_buffer, in_start, in_end)
 251
 252     def i2c_scan(self, *, start=0, end=0x79):
 253         """Perform an I2C Device Scan"""
 254         if self._i2c_index is None:
 255             raise RuntimeError("I2C bus not initialized.")
 256         found = []
 257         for addr in range(start, end + 1):
 258             # try a write
 259             try:
 260                 self.i2c_writeto(addr, b"\x00\x00\x00")
 261             except RuntimeError:  # no reply!
 262                 continue
 263             # store if success
 264             found.append(addr)
 265         return found
 266
 267     # ----------------------------------------------------------------
 268     # SPI
 269     # ----------------------------------------------------------------
 270     def spi_configure(self, baudrate):
 271         """Configure SPI."""
 272         if self._spi_index is None:
 273             raise RuntimeError("SPI bus not initialized.")
 274
 275         resp = self._hid_xfer(
 276             bytes(
 277                 [
 278                     self.SPI0_INIT if self._spi_index == 0 else self.SPI1_INIT,
 279                     0x00,  # mode, not yet implemented
 280                 ]
 281             )
 282             + baudrate.to_bytes(4, byteorder="little"),
 283             True,
 284         )
 285         if resp[1] != self.RESP_OK:
 286             raise RuntimeError("SPI init error.")
 287
 288     def spi_set_port(self, index):
 289         """Set SPI port."""
 290         if index not in (0, 1):
 291             raise ValueError("SPI index must be 0 or 1.")
 292         self._spi_index = index
 293
 294     def spi_write(self, buffer, *, start=0, end=None):
 295         """SPI write."""
 296         if self._spi_index is None:
 297             raise RuntimeError("SPI bus not initialized.")
 298
 299         end = end if end else len(buffer)
 300
 301         write_cmd = self.SPI0_WRITE if self._spi_index == 0 else self.SPI1_WRITE
 302
 303         while (end - start) > 0:
 304             remain_bytes = end - start
 305             chunk = min(remain_bytes, 64 - 3)
 306             resp = self._hid_xfer(
 307                 bytes([write_cmd, chunk]) + buffer[start : (start + chunk)], True
 308             )
 309             if resp[1] != self.RESP_OK:
 310                 raise RuntimeError("SPI write error")
 311             start += chunk
 312
 313     def spi_readinto(self, buffer, *, start=0, end=None, write_value=0):
 314         """SPI readinto."""
 315         if self._spi_index is None:
 316             raise RuntimeError("SPI bus not initialized.")
 317
 318         end = end if end else len(buffer)
 319         read_cmd = self.SPI0_READ if self._spi_index == 0 else self.SPI1_READ
 320         read_size = end - start
 321
 322         resp = self._hid_xfer(bytes([read_cmd, write_value, read_size]), True)
 323         if resp[1] != self.RESP_OK:
 324             raise RuntimeError("SPI write error")
 325         # move into buffer
 326         for i in range(read_size):
 327             buffer[start + i] = resp[i + 2]
 328
 329     def spi_write_readinto(
 330         self,
 331         buffer_out,
 332         buffer_in,
 333         *,
 334         out_start=0,
 335         out_end=None,
 336         in_start=0,
 337         in_end=None
 338     ):
 339         """SPI write and readinto."""
 340         raise NotImplementedError("SPI write_readinto Not implemented")
 341
 342     # ----------------------------------------------------------------
 343     # NEOPIXEL
 344     # ----------------------------------------------------------------
 345     def neopixel_write(self, gpio, buf):
 346         """NeoPixel write."""
 347         # open serial (data is sent over this)
 348         if self._serial is None:
 349             import serial
 350             import serial.tools.list_ports
 351
 352             ports = serial.tools.list_ports.comports()
 353             for port in ports:
 354                 if port.vid == self.VID and port.pid == self.PID:
 355                     self._serial = serial.Serial(port.device)
 356                     break
 357         if self._serial is None:
 358             raise RuntimeError("Could not find Pico com port.")
 359
 360         # init
 361         if not self._neopixel_initialized:
 362             # deinit any current setup
 363             self._hid_xfer(bytes([self.WS2812B_DEINIT]))
 364             resp = self._hid_xfer(
 365                 bytes(
 366                     [
 367                         self.WS2812B_INIT,
 368                         gpio._pin.id,
 369                     ]
 370                 ),
 371                 True,
 372             )
 373             if resp[1] != self.RESP_OK:
 374                 raise RuntimeError("Neopixel init error")
 375             self._neopixel_initialized = True
 376
 377         # write
 378         # command is done over HID
 379         remain_bytes = len(buf)
 380         resp = self._hid_xfer(
 381             bytes([self.WS2812B_WRITE]) + remain_bytes.to_bytes(4, byteorder="little"),
 382             True,
 383         )
 384         if resp[1] != self.RESP_OK:
 385             if resp[2] == 0x01:
 386                 raise RuntimeError(
 387                     "Neopixel write error : too many pixel for the firmware."
 388                 )
 389             elif resp[2] == 0x02:
 390                 raise RuntimeError(
 391                     "Neopixel write error : transfer already in progress."
 392                 )
 393             else:
 394                 raise RuntimeError("Neopixel write error")
 395         # buffer is sent over serial
 396         self._serial.write(buf)
 397         self._serial.flush()
 398
 399     # ----------------------------------------------------------------
 400     # PWM
 401     # ----------------------------------------------------------------
 402     def pwm_configure(self, pin, frequency=500, duty_cycle=0, variable_frequency=False):
 403         """Configure PWM."""
 404         self.pwm_deinit(pin)
 405         resp = self._hid_xfer(bytes([self.PWM_INIT_PIN, pin.id]), True)
 406         if resp[1] != self.RESP_OK:
 407             raise RuntimeError("PWM init error.")
 408
 409         self.pwm_set_frequency(pin, frequency)
 410         self.pwm_set_duty_cycle(pin, duty_cycle)
 411
 412     def pwm_deinit(self, pin):
 413         """Deinit PWM."""
 414         self._hid_xfer(bytes([self.PWM_DEINIT_PIN, pin.id]))
 415
 416     def pwm_get_frequency(self, pin):
 417         """PWM get freq."""
 418         resp = self._hid_xfer(
 419             bytes([self.PWM_GET_FREQ, pin.id])
 420             + frequency.to_bytes(4, byteorder="little"),
 421             True,
 422         )
 423         if resp[1] != self.RESP_OK:
 424             raise RuntimeError("PWM get frequency error.")
 425         return int.from_bytes(resp[3 : 3 + 4], byteorder="little")
 426
 427     def pwm_set_frequency(self, pin, frequency):
 428         """PWM set freq."""
 429         resp = self._hid_xfer(
 430             bytes([self.PWM_SET_FREQ, pin.id])
 431             + frequency.to_bytes(4, byteorder="little"),
 432             True,
 433         )
 434         if resp[1] != self.RESP_OK:
 435             if resp[3] == 0x01:
 436                 raise RuntimeError("PWM different frequency on same slice.")
 437             elif resp[3] == 0x02:
 438                 raise RuntimeError("PWM frequency too low.")
 439             elif resp[3] == 0x03:
 440                 raise RuntimeError("PWM frequency too high.")
 441             else:
 442                 raise RuntimeError("PWM frequency error.")
 443
 444     def pwm_get_duty_cycle(self, pin):
 445         """PWM get duty cycle."""
 446         resp = self._hid_xfer(bytes([self.PWM_GET_DUTY_U16, pin.id]), True)
 447         if resp[1] != self.RESP_OK:
 448             raise RuntimeError("PWM get duty cycle error.")
 449         return int.from_bytes(resp[3 : 3 + 4], byteorder="little")
 450
 451     def pwm_set_duty_cycle(self, pin, duty_cycle):
 452         """PWM set duty cycle."""
 453         resp = self._hid_xfer(
 454             bytes([self.PWM_SET_DUTY_U16, pin.id])
 455             + duty_cycle.to_bytes(2, byteorder="little"),
 456             True,
 457         )
 458         if resp[1] != self.RESP_OK:
 459             raise RuntimeError("PWM set duty cycle error.")
 460
 461
 462 pico_u2if = Pico_u2if()