MicroPython for Microcontrollers Module 4 – Buses and device drivers: I2C, SPI and UART
I2C, SPI and UART from MicroPython
Create I2C, SPI and UART buses in MicroPython 1.29 with explicit pins and speeds, read device registers into buffers, and read serial lines with timeouts.
What you will learn
- Create I2C, SPI and UART objects with explicit pins and speeds, and read the real speed back
- Read and write device registers with readfrom_mem_into, writeto_mem and struct
- Use write_readinto() inside a chip select to exchange bytes with an SPI device
- Read serial data into preallocated buffers with a timeout and assemble whole lines
- Predict the error or silence each bus gives when a device does not answer
Before you start
What you’ll learn · Before you start
What you will learn
- Create I2C, SPI and UART objects with explicit pins and speeds, and read the real speed back
- Read and write device registers with readfrom_mem_into, writeto_mem and struct
- Use write_readinto() inside a chip select to exchange bytes with an SPI device
- Read serial data into preallocated buffers with a timeout and assemble whole lines
- Predict the error or silence each bus gives when a device does not answer
Before you start
On this page
A temperature sensor, a memory chip and a GPS receiver can all hang off one Pico, because each speaks a bus: I2C on two wires shared by up to 112 addresses, SPI on four wires at several megahertz, and UART on two wires with no clock at all. The code is short, and three slips cause most of the failures: a pin the bus cannot reach, a value read as unsigned, and a serial read taken for a whole line.
This lesson creates the three buses with explicit pins and speeds, reads and writes the registers of an I2C sensor,
exchanges bytes with an SPI memory inside a chip select, and assembles whole lines from a UART into buffers made
once. The devices are made up for this track, MSC-ENV1 and MSC-MEM1, and every program runs here on a model of
machine built from MicroPython 1.29’s documentation and the rp2 port’s source.
Each bus has its own pins, its own speed and its own way to find a device
The three classes of machine look alike, but the buses under them work differently:
| Feature | I2C | SPI | UART |
|---|---|---|---|
| Wires | SCL clock and SDA data, shared by every device | SCK, MOSI and MISO, plus one chip select a device | TX and RX, crossed between the two ends |
| Who sets the speed | the controller’s clock, freq |
the controller’s clock, baudrate |
nobody: both ends agree on baudrate |
| How a device is chosen | its 7-bit address, sent in every transfer | its chip select driven low | one device a port |
| Does a device say it is there? | yes, it acknowledges every byte | no | no |
| Classes | I2C, SoftI2C |
SPI, SoftSPI |
UART |
I2C: A two-wire bus, clock SCL and data SDA, on which a controller picks one of up to 112 devices by a 7-bit address. Every byte is acknowledged, so a missing device is noticed. and SPI: A bus of a clock, a line out (MOSI) and a line in (MISO), plus one chip-select line per device. Every clock moves one bit each way, and nothing acknowledges. each come twice: the hardware class uses the chip’s own bus
block, and the Soft class drives any pins from software with the same methods, more slowly. A hardware bus reaches
only some pins. On a Pico, I2C 0 can use GP4 for SDA and GP5 for SCL, and the constructor refuses a pin its bus
cannot reach. Since 1.25 the rp2 port lets you leave out the bus number of I2C and SPI, and the esp32 port that
of I2C, to use the board’s default, but writing the number and every pin keeps the code portable and easy to
review. Create the three buses; machine.bench_wire(), which only the model has, puts MSC-ENV1 on the bus:
bus/bus_setup.py
"""Three buses on a Pico with explicit ids, pins and speeds, on the model of machine beside this file."""
from machine import I2C, SPI, UART, Pin
import machine
machine.bench_wire("ENV1", address=0x41) # a made-up sensor on the I2C bus
i2c = I2C(0, scl=Pin(5), sda=Pin(4), freq=400_000)
spi = SPI(0, baudrate=1_000_000, sck=Pin(18), mosi=Pin(19), miso=Pin(16))
cs = Pin(17, Pin.OUT, value=1) # SPI chip select: a plain output you drive
uart = UART(1, baudrate=9600, tx=Pin(8), rx=Pin(9))
print(i2c)
print("scan:", [hex(a) for a in i2c.scan()])
print(uart)
for freq in (100_000, 1_000_000):
print("asked", freq, "->", I2C(0, scl=Pin(5), sda=Pin(4), freq=freq)) # printing shows the real rate
try:
I2C(1, scl=Pin(5), sda=Pin(4))
except ValueError as e:
print("I2C(1) on GP5 and GP4: ValueError:", e) Output
I2C(0, freq=399361, scl=5, sda=4, timeout=50000) scan: ['0x41'] UART(1, baudrate=9600, bits=8, parity=None, stop=1, tx=8, rx=9, txbuf=256, rxbuf=256, timeout=0, timeout_char=2, invert=None, irq=0) asked 100000 -> I2C(0, freq=100000, scl=5, sda=4, timeout=50000) asked 1000000 -> I2C(0, freq=1000000, scl=5, sda=4, timeout=50000) I2C(1) on GP5 and GP4: ValueError: bad SCL pin
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 bus_setup.py
machine.py (the model) · bus/machine.py
"""A model of machine.I2C, SoftI2C, SPI and UART on a Raspberry Pi Pico (125 MHz), with made-up devices on the buses,
so bus code runs on a computer.
It follows the MicroPython 1.29.0 documentation and the rp2 port's source: the pin rules and the print formats of
ports/rp2/machine_i2c.c, machine_i2c.h and machine_uart.c, the speed the pico-sdk's i2c_set_baudrate() really sets
(a period of whole system clocks), OSError(5) EIO from the hardware I2C when nobody acknowledges and OSError(19)
ENODEV from SoftI2C, and the UART defaults (256-byte buffers, timeout 0, timeout_char at least 13000 // baudrate + 1).
The devices are made up for this track: MSC-ENV1, a temperature sensor with 8-bit registers, and MSC-MEM1, a memory
that answers SPI command 0x9F with three ID bytes. Functions whose names start with bench_ are not MicroPython: they
wire the devices and feed the UART. Never copy this file to a board.
"""
FAKE = True
SYS_CLK_HZ = 125_000_000
EIO, ENODEV = 5, 19
class Pin:
IN, OUT = 0, 1
def __init__(self, id, mode=-1, value=None):
self.id, self.level = id, value
def __call__(self, v=None):
return self.value(v)
def value(self, v=None):
if v is None:
return self.level
self.level = 1 if v else 0
for dev in _spi_devices:
dev.select(self.id, self.level)
def __repr__(self):
return "Pin(GPIO%d)" % self.id
class _Env1:
"""MSC-ENV1 (made up): 0x00 ID 0xE1, 0x01 CTRL, 0x02 STATUS, 0x03-0x04 TEMP (signed, big-endian, 1/128 °C)."""
def __init__(self):
self.regs = bytearray(8)
self.regs[0] = 0xE1
self.set_temp(2752) # 21.5 °C
def set_temp(self, raw):
self.regs[3:5] = (raw & 0xFFFF).to_bytes(2, "big")
_i2c_devices, _spi_devices = {}, []
class I2C:
def __init__(self, id=0, *, scl=None, sda=None, freq=400_000, timeout=50_000):
default = {0: (5, 4), 1: (7, 6)}[id] # (SCL, SDA): GP5 and GP4 on a Pico for bus 0
self.id = id
self.scl = scl.id if scl is not None else default[0]
self.sda = sda.id if sda is not None else default[1]
if not (self.scl & 1 == 1 and (self.scl & 2) >> 1 == id):
raise ValueError("bad SCL pin")
if not (self.sda & 1 == 0 and (self.sda & 2) >> 1 == id):
raise ValueError("bad SDA pin")
period = (SYS_CLK_HZ + freq // 2) // freq # whole system clocks per SCL period (pico-sdk)
self.freq, self.timeout = SYS_CLK_HZ // period, timeout
def __repr__(self):
return "I2C(%d, freq=%d, scl=%d, sda=%d, timeout=%d)" % (self.id, self.freq, self.scl, self.sda, self.timeout)
_missing = EIO # the rp2 hardware I2C reports a missing acknowledgement as EIO
def _device(self, addr):
dev = _i2c_devices.get(addr)
if dev is None:
raise OSError(self._missing)
return dev
def scan(self):
return sorted(_i2c_devices)
def readfrom_mem(self, addr, memaddr, nbytes, *, addrsize=8):
dev = self._device(addr)
return bytes(dev.regs[memaddr:memaddr + nbytes])
def readfrom_mem_into(self, addr, memaddr, buf, *, addrsize=8):
dev = self._device(addr)
buf[:] = dev.regs[memaddr:memaddr + len(buf)]
def writeto_mem(self, addr, memaddr, buf, *, addrsize=8):
dev = self._device(addr)
dev.regs[memaddr:memaddr + len(buf)] = buf
class SoftI2C(I2C):
_missing = ENODEV # the bit-banged I2C sees the missing ACK after the address
def __init__(self, scl, sda, *, freq=400_000, timeout=50_000):
self.id, self.scl, self.sda, self.freq, self.timeout = -1, scl.id, sda.id, freq, timeout
def __repr__(self):
return "SoftI2C(scl=%d, sda=%d, freq=%d)" % (self.scl, self.sda, self.freq)
class _Mem1:
"""MSC-MEM1 (made up): after chip select falls, command 0x9F returns the ID bytes 0x6D 0x53 0x01, then 0xFF."""
def __init__(self, cs):
self.cs, self.selected, self.out = cs, False, []
def select(self, pin, level):
if pin == self.cs:
if not level and not self.selected:
self.out = None # a command starts with the next byte
self.selected = not level
def exchange(self, byte):
if not self.selected:
return 0xFF
if self.out is None: # this byte is the command
self.out = [0x6D, 0x53, 0x01] if byte == 0x9F else []
return 0xFF
return self.out.pop(0) if self.out else 0xFF
class SPI:
def __init__(self, id, baudrate=1_000_000, *, polarity=0, phase=0, bits=8, sck=None, mosi=None, miso=None):
for name, pin, low in (("SCK", sck, 2), ("MOSI", mosi, 3), ("MISO", miso, 0)):
if pin is not None and not (pin.id & 3 == low and (pin.id & 8) >> 3 == id):
raise ValueError("bad %s pin" % name)
self.id = id
def write_readinto(self, write_buf, read_buf):
if len(write_buf) != len(read_buf):
raise ValueError("buffers must be the same length")
for i, b in enumerate(write_buf):
read_buf[i] = _spi_devices[0].exchange(b) if _spi_devices else 0xFF
def write(self, buf):
self.write_readinto(buf, bytearray(len(buf)))
def read(self, nbytes, write=0x00):
buf = bytearray(nbytes)
self.write_readinto(bytes([write]) * nbytes, buf)
return bytes(buf)
class UART:
def __init__(self, id, baudrate=115_200, bits=8, parity=None, stop=1, *, tx=None, rx=None,
timeout=0, timeout_char=0, txbuf=256, rxbuf=256):
self.id, self.baudrate, self.bits, self.parity, self.stop = id, baudrate, bits, parity, stop
self.tx, self.rx = tx.id, rx.id
for name, pin, low in (("TX", self.tx, 0), ("RX", self.rx, 1)):
if not (pin & 3 == low and ((pin + 4) & 8) >> 3 == id):
raise ValueError("bad %s pin" % name)
self.txbuf, self.rxbuf, self.timeout = txbuf, rxbuf, timeout
self.timeout_char = max(timeout_char, 13000 // baudrate + 1) # at least one character's time
self._arrived, self._coming = bytearray(), []
def __repr__(self):
return ("UART(%d, baudrate=%d, bits=%d, parity=%s, stop=%d, tx=%d, rx=%d, txbuf=%d, rxbuf=%d, timeout=%d, "
"timeout_char=%d, invert=None, irq=0)" % (self.id, self.baudrate, self.bits, self.parity, self.stop,
self.tx, self.rx, self.txbuf, self.rxbuf, self.timeout,
self.timeout_char))
def _wait(self):
if self._coming: # each read finds the next chunk on the wire
self._arrived += self._coming.pop(0)
def any(self):
return len(self._arrived)
def read(self, nbytes=-1):
self._wait()
if not self._arrived:
return None # nothing within the timeout
n = len(self._arrived) if nbytes < 0 else nbytes
out = bytes(self._arrived[:n])
del self._arrived[:n]
return out
def readinto(self, buf, nbytes=None):
self._wait()
if not self._arrived:
return None
n = min(len(buf) if nbytes is None else nbytes, len(self._arrived))
buf[:n] = self._arrived[:n]
del self._arrived[:n]
return n
def bench_wire(device, *, address=None, cs=None):
"""Connect a made-up device: "ENV1" to the I2C bus at an address, "MEM1" to the SPI bus with a chip select."""
if device == "ENV1":
_i2c_devices[address] = _Env1()
return _i2c_devices[address]
_spi_devices.append(_Mem1(cs))
return _spi_devices[-1]
def bench_feed(uart, *chunks):
"""Bytes that arrive on the UART's RX pin, one chunk between one read and the next."""
uart._coming.extend(chunks) Runs on this device, in your browser. The first run downloads Python (about 13.5 MB), which is kept for the next runs.
Your run, in this browser
The I2C bus asked for 400,000 Hz and runs at 399,361. The pico-sdk makes each SCL period a whole number of 125 MHz
system clocks, and 125,000,000 ÷ 400,000 = 312.5 rounds to 313, so the rate is 125,000,000 ÷ 313. The documentation
warns that the real rate may be lower than the one requested, and printing the object is how you see it. The
UART: A serial port with a transmit and a receive line and no clock. Both ends use the same baud rate, and bytes arrive whenever the other end sends them. line shows the rp2 port’s defaults: 256-byte buffers, timeout=0, so a read returns at
once, and timeout_char=2, the port’s minimum of 13,000 ÷ 9,600 + 1 ms between characters. I2C(1) on GP5 failed
with bad SCL pin: GP5 belongs to bus 0.
Pull-ups on the I2C lines
I2C needs pull-up resistors on SDA and SCL, 1 to 10 kΩ by the I2C documentation. The rp2 port switches on the pins’ own pull-ups, 50 to 80 kΩ by the RP2040 datasheet, far weaker than that, so use a board or breakout that has real resistors.
Checkpoint
Check: Bus pins, speeds and finding a device
An I2C register read is a write, a repeated start and a read
Most I2C sensors hold their readings in numbered registers. readfrom_mem(addr, memaddr, nbytes) writes the register
number to the device, then reads the bytes back without letting go of the bus, through a
repeated start: An I2C start condition sent without a stop before it, so the controller keeps the bus while it turns from writing a register number to reading the register.; writeto_mem(addr, memaddr, buf) writes the number and then the
data. MSC-ENV1 answers at address 0x41, keeps its ID, 0xE1, in register 0x00, and its temperature in 0x03 and 0x04 as
a signed big-endian number in 128ths of a degree. Step through the read of those two bytes:
The drawing shows the whole scene; its 6 steps are listed below.
Every step as text
- Start and address. The Pico pulls SDA low while SCL is high, a start condition, then sends 0x41 with the write bit. The sensor knows its address and pulls SDA low for one clock, an acknowledgement.
- The register number. Next comes the register to read, 0x03, the high byte of the temperature. The sensor acknowledges it and moves its register pointer there.
- Repeated start. Without a stop, the Pico sends a new start and 0x41 again, now with the read bit. Keeping the bus means no other controller can slip in between the two halves.
- First byte. The sensor sends 0x0A from register 0x03 and moves on to 0x04. The Pico acknowledges, which asks for one more byte.
- Last byte. The sensor sends 0xC0. The Pico does not acknowledge this one, which tells the sensor to stop, and then ends the transfer with a stop condition.
- Into the buffer. readfrom_mem_into() stores the two bytes in buf and returns None. struct.unpack_from(">h", buf) reads them as 2,752 signed, which is 21.5 °C at 128ths of a degree.
The program reads the ID, writes the control register, reads the temperature twice into one buffer, and then calls an address where nothing answers:
bus/i2c_regs.py
"""Registers of a made-up I2C sensor, MSC-ENV1 at 0x41, read and written on the model of machine beside this file."""
import struct
from machine import I2C, Pin, SoftI2C
import machine
sensor = machine.bench_wire("ENV1", address=0x41)
i2c = I2C(0, scl=Pin(5), sda=Pin(4), freq=400_000)
ENV1, REG_ID, REG_CTRL, REG_TEMP = 0x41, 0x00, 0x01, 0x03
print("ID:", hex(i2c.readfrom_mem(ENV1, REG_ID, 1)[0]))
i2c.writeto_mem(ENV1, REG_CTRL, b"\x01") # one register, one byte
print("CTRL now:", i2c.readfrom_mem(ENV1, REG_CTRL, 1))
buf = bytearray(2) # made once, filled by every reading
for raw_temp in (2752, -672):
sensor.set_temp(raw_temp) # the bench: the room warms or cools
i2c.readfrom_mem_into(ENV1, REG_TEMP, buf)
raw = struct.unpack_from(">h", buf)[0] # signed, big-endian
wrong = int.from_bytes(buf, "big") # unsigned: the slip
print("TEMP bytes", [hex(b) for b in buf], "raw", raw, "=", raw / 128, "C; read unsigned:", wrong / 128)
for bus in (i2c, SoftI2C(scl=Pin(5), sda=Pin(4))):
try:
bus.readfrom_mem(0x42, REG_ID, 1) # nothing answers at 0x42
except OSError as e:
print(type(bus).__name__, "at 0x42: OSError", e.args[0]) Output
ID: 0xe1 CTRL now: b'\x01' TEMP bytes ['0xa', '0xc0'] raw 2752 = 21.5 C; read unsigned: 21.5 TEMP bytes ['0xfd', '0x60'] raw -672 = -5.25 C; read unsigned: 506.75 I2C at 0x42: OSError 5 SoftI2C at 0x42: OSError 19
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 i2c_regs.py
machine.py (the model) · bus/machine.py
"""A model of machine.I2C, SoftI2C, SPI and UART on a Raspberry Pi Pico (125 MHz), with made-up devices on the buses,
so bus code runs on a computer.
It follows the MicroPython 1.29.0 documentation and the rp2 port's source: the pin rules and the print formats of
ports/rp2/machine_i2c.c, machine_i2c.h and machine_uart.c, the speed the pico-sdk's i2c_set_baudrate() really sets
(a period of whole system clocks), OSError(5) EIO from the hardware I2C when nobody acknowledges and OSError(19)
ENODEV from SoftI2C, and the UART defaults (256-byte buffers, timeout 0, timeout_char at least 13000 // baudrate + 1).
The devices are made up for this track: MSC-ENV1, a temperature sensor with 8-bit registers, and MSC-MEM1, a memory
that answers SPI command 0x9F with three ID bytes. Functions whose names start with bench_ are not MicroPython: they
wire the devices and feed the UART. Never copy this file to a board.
"""
FAKE = True
SYS_CLK_HZ = 125_000_000
EIO, ENODEV = 5, 19
class Pin:
IN, OUT = 0, 1
def __init__(self, id, mode=-1, value=None):
self.id, self.level = id, value
def __call__(self, v=None):
return self.value(v)
def value(self, v=None):
if v is None:
return self.level
self.level = 1 if v else 0
for dev in _spi_devices:
dev.select(self.id, self.level)
def __repr__(self):
return "Pin(GPIO%d)" % self.id
class _Env1:
"""MSC-ENV1 (made up): 0x00 ID 0xE1, 0x01 CTRL, 0x02 STATUS, 0x03-0x04 TEMP (signed, big-endian, 1/128 °C)."""
def __init__(self):
self.regs = bytearray(8)
self.regs[0] = 0xE1
self.set_temp(2752) # 21.5 °C
def set_temp(self, raw):
self.regs[3:5] = (raw & 0xFFFF).to_bytes(2, "big")
_i2c_devices, _spi_devices = {}, []
class I2C:
def __init__(self, id=0, *, scl=None, sda=None, freq=400_000, timeout=50_000):
default = {0: (5, 4), 1: (7, 6)}[id] # (SCL, SDA): GP5 and GP4 on a Pico for bus 0
self.id = id
self.scl = scl.id if scl is not None else default[0]
self.sda = sda.id if sda is not None else default[1]
if not (self.scl & 1 == 1 and (self.scl & 2) >> 1 == id):
raise ValueError("bad SCL pin")
if not (self.sda & 1 == 0 and (self.sda & 2) >> 1 == id):
raise ValueError("bad SDA pin")
period = (SYS_CLK_HZ + freq // 2) // freq # whole system clocks per SCL period (pico-sdk)
self.freq, self.timeout = SYS_CLK_HZ // period, timeout
def __repr__(self):
return "I2C(%d, freq=%d, scl=%d, sda=%d, timeout=%d)" % (self.id, self.freq, self.scl, self.sda, self.timeout)
_missing = EIO # the rp2 hardware I2C reports a missing acknowledgement as EIO
def _device(self, addr):
dev = _i2c_devices.get(addr)
if dev is None:
raise OSError(self._missing)
return dev
def scan(self):
return sorted(_i2c_devices)
def readfrom_mem(self, addr, memaddr, nbytes, *, addrsize=8):
dev = self._device(addr)
return bytes(dev.regs[memaddr:memaddr + nbytes])
def readfrom_mem_into(self, addr, memaddr, buf, *, addrsize=8):
dev = self._device(addr)
buf[:] = dev.regs[memaddr:memaddr + len(buf)]
def writeto_mem(self, addr, memaddr, buf, *, addrsize=8):
dev = self._device(addr)
dev.regs[memaddr:memaddr + len(buf)] = buf
class SoftI2C(I2C):
_missing = ENODEV # the bit-banged I2C sees the missing ACK after the address
def __init__(self, scl, sda, *, freq=400_000, timeout=50_000):
self.id, self.scl, self.sda, self.freq, self.timeout = -1, scl.id, sda.id, freq, timeout
def __repr__(self):
return "SoftI2C(scl=%d, sda=%d, freq=%d)" % (self.scl, self.sda, self.freq)
class _Mem1:
"""MSC-MEM1 (made up): after chip select falls, command 0x9F returns the ID bytes 0x6D 0x53 0x01, then 0xFF."""
def __init__(self, cs):
self.cs, self.selected, self.out = cs, False, []
def select(self, pin, level):
if pin == self.cs:
if not level and not self.selected:
self.out = None # a command starts with the next byte
self.selected = not level
def exchange(self, byte):
if not self.selected:
return 0xFF
if self.out is None: # this byte is the command
self.out = [0x6D, 0x53, 0x01] if byte == 0x9F else []
return 0xFF
return self.out.pop(0) if self.out else 0xFF
class SPI:
def __init__(self, id, baudrate=1_000_000, *, polarity=0, phase=0, bits=8, sck=None, mosi=None, miso=None):
for name, pin, low in (("SCK", sck, 2), ("MOSI", mosi, 3), ("MISO", miso, 0)):
if pin is not None and not (pin.id & 3 == low and (pin.id & 8) >> 3 == id):
raise ValueError("bad %s pin" % name)
self.id = id
def write_readinto(self, write_buf, read_buf):
if len(write_buf) != len(read_buf):
raise ValueError("buffers must be the same length")
for i, b in enumerate(write_buf):
read_buf[i] = _spi_devices[0].exchange(b) if _spi_devices else 0xFF
def write(self, buf):
self.write_readinto(buf, bytearray(len(buf)))
def read(self, nbytes, write=0x00):
buf = bytearray(nbytes)
self.write_readinto(bytes([write]) * nbytes, buf)
return bytes(buf)
class UART:
def __init__(self, id, baudrate=115_200, bits=8, parity=None, stop=1, *, tx=None, rx=None,
timeout=0, timeout_char=0, txbuf=256, rxbuf=256):
self.id, self.baudrate, self.bits, self.parity, self.stop = id, baudrate, bits, parity, stop
self.tx, self.rx = tx.id, rx.id
for name, pin, low in (("TX", self.tx, 0), ("RX", self.rx, 1)):
if not (pin & 3 == low and ((pin + 4) & 8) >> 3 == id):
raise ValueError("bad %s pin" % name)
self.txbuf, self.rxbuf, self.timeout = txbuf, rxbuf, timeout
self.timeout_char = max(timeout_char, 13000 // baudrate + 1) # at least one character's time
self._arrived, self._coming = bytearray(), []
def __repr__(self):
return ("UART(%d, baudrate=%d, bits=%d, parity=%s, stop=%d, tx=%d, rx=%d, txbuf=%d, rxbuf=%d, timeout=%d, "
"timeout_char=%d, invert=None, irq=0)" % (self.id, self.baudrate, self.bits, self.parity, self.stop,
self.tx, self.rx, self.txbuf, self.rxbuf, self.timeout,
self.timeout_char))
def _wait(self):
if self._coming: # each read finds the next chunk on the wire
self._arrived += self._coming.pop(0)
def any(self):
return len(self._arrived)
def read(self, nbytes=-1):
self._wait()
if not self._arrived:
return None # nothing within the timeout
n = len(self._arrived) if nbytes < 0 else nbytes
out = bytes(self._arrived[:n])
del self._arrived[:n]
return out
def readinto(self, buf, nbytes=None):
self._wait()
if not self._arrived:
return None
n = min(len(buf) if nbytes is None else nbytes, len(self._arrived))
buf[:n] = self._arrived[:n]
del self._arrived[:n]
return n
def bench_wire(device, *, address=None, cs=None):
"""Connect a made-up device: "ENV1" to the I2C bus at an address, "MEM1" to the SPI bus with a chip select."""
if device == "ENV1":
_i2c_devices[address] = _Env1()
return _i2c_devices[address]
_spi_devices.append(_Mem1(cs))
return _spi_devices[-1]
def bench_feed(uart, *chunks):
"""Bytes that arrive on the UART's RX pin, one chunk between one read and the next."""
uart._coming.extend(chunks) Runs on this device, in your browser. The first run downloads Python (about 13.5 MB), which is kept for the next runs.
Your run, in this browser
readfrom_mem_into() fills a buffer made once, where readfrom_mem() returns a new bytes object at every call,
which matters in a loop that reads ten times a second. The bytes 0x0A 0xC0 are 2,752, 21.5 °C. At −5.25 °C the
sensor sends 0xFD 0x60, and only a signed read gets it right: struct.unpack_from(">h", buf) gives −672, while
int.from_bytes(buf, "big") gives 64,864, 506.75 °C. MicroPython’s int.from_bytes() has no signed argument, so
struct is the way to read signed registers.
Nothing answered at 0x42, and the two classes said so differently: the Pico’s hardware I2C raised OSError 5,
EIO, and SoftI2C, which sees the missing acknowledgement right after the address, raised 19, ENODEV. A device
that holds the clock low for longer than timeout, 50,000 µs by default, raises 110, ETIMEDOUT.
Reading a signed register as unsigned
int.from_bytes(buf, "big") / 128 is right above 0 °C and absurd below it. Read signed registers with
struct.unpack_from(">h", buf)[0], and check the datasheet for each register’s sign and byte order.
The roles can also swap. Since 1.26, machine.I2CTarget(0, addr=0x41, mem=bytearray(8)) makes a board answer as an
I2C device, with a bytearray as its registers, on the alif, esp32, mimxrt, rp2, samd, stm32 and zephyr ports: a second
board can then play MSC-ENV1 for your driver’s tests.
Checkpoint
Check: I2C register reads and writes
SPI exchanges bytes both ways inside a chip select you drive
On SPI every clock pulse moves one bit out on MOSI and one bit in on MISO, so sending and receiving happen together.
write_readinto(tx, rx) sends tx and fills rx at the same time, and both must be the same length; to read n
bytes you send n bytes, which is what read(n, write) does for you. The device knows a command has started because
its chip select: The line that tells one SPI device the next bytes are for it: driven low before a command and high after it. MicroPython leaves it to your code, as a Pin. went low, and that it has ended because it went high again. MicroPython
leaves that pin to you, a plain Pin output, which the documentation drives in try and finally.
MSC-MEM1 answers command 0x9F with three ID bytes. The command byte goes out first, and the ID comes back on the next three clocks, while the Pico sends filler bytes:
The drawing shows the whole scene; its 5 steps are listed below.
Every step as text
- Chip select low. cs(0) pulls the memory's chip select low. The memory now listens, and it will take the next byte as the start of a command.
- The command. Eight clocks send 0x9F, read ID, on MOSI. On the same clocks MISO carries 0xFF, because the memory has nothing to say yet, and that byte lands in rx[0].
- First ID byte. The Pico sends a filler byte, 0xFF, only to make the clock run, and the memory answers with the first ID byte, 0x6D, on MISO.
- The rest of the ID. Two more filler bytes bring 0x53 and 0x01. write_readinto() has now filled all four bytes of rx, the same length as tx.
- Chip select high. cs(1) in the finally block ends the command. Only a new falling edge lets the memory take another command, so every command gets its own select.
The program reads the ID inside one select, then sends two commands without raising the chip select between them:
bus/spi_id.py
"""The ID of a made-up SPI memory, MSC-MEM1, with the chip select driven by hand, on the model of machine beside this file."""
from machine import SPI, Pin
import machine
machine.bench_wire("MEM1", cs=17)
spi = SPI(0, baudrate=1_000_000, sck=Pin(18), mosi=Pin(19), miso=Pin(16))
cs = Pin(17, Pin.OUT, value=1)
tx = bytearray(b"\x9f\xff\xff\xff") # command 0x9F, then three bytes that clock the answer in
rx = bytearray(4)
cs(0)
try:
spi.write_readinto(tx, rx) # sends and receives at the same time
finally:
cs(1) # always deselect, even after an error
print("rx:", [hex(b) for b in rx], "ID:", [hex(b) for b in rx[1:]])
cs(0) # the slip: two commands inside one select
spi.write_readinto(tx, rx)
spi.write_readinto(tx, rx)
cs(1)
print("second command, same select:", [hex(b) for b in rx[1:]]) Output
rx: ['0xff', '0x6d', '0x53', '0x1'] ID: ['0x6d', '0x53', '0x1'] second command, same select: ['0xff', '0xff', '0xff']
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 spi_id.py
machine.py (the model) · bus/machine.py
"""A model of machine.I2C, SoftI2C, SPI and UART on a Raspberry Pi Pico (125 MHz), with made-up devices on the buses,
so bus code runs on a computer.
It follows the MicroPython 1.29.0 documentation and the rp2 port's source: the pin rules and the print formats of
ports/rp2/machine_i2c.c, machine_i2c.h and machine_uart.c, the speed the pico-sdk's i2c_set_baudrate() really sets
(a period of whole system clocks), OSError(5) EIO from the hardware I2C when nobody acknowledges and OSError(19)
ENODEV from SoftI2C, and the UART defaults (256-byte buffers, timeout 0, timeout_char at least 13000 // baudrate + 1).
The devices are made up for this track: MSC-ENV1, a temperature sensor with 8-bit registers, and MSC-MEM1, a memory
that answers SPI command 0x9F with three ID bytes. Functions whose names start with bench_ are not MicroPython: they
wire the devices and feed the UART. Never copy this file to a board.
"""
FAKE = True
SYS_CLK_HZ = 125_000_000
EIO, ENODEV = 5, 19
class Pin:
IN, OUT = 0, 1
def __init__(self, id, mode=-1, value=None):
self.id, self.level = id, value
def __call__(self, v=None):
return self.value(v)
def value(self, v=None):
if v is None:
return self.level
self.level = 1 if v else 0
for dev in _spi_devices:
dev.select(self.id, self.level)
def __repr__(self):
return "Pin(GPIO%d)" % self.id
class _Env1:
"""MSC-ENV1 (made up): 0x00 ID 0xE1, 0x01 CTRL, 0x02 STATUS, 0x03-0x04 TEMP (signed, big-endian, 1/128 °C)."""
def __init__(self):
self.regs = bytearray(8)
self.regs[0] = 0xE1
self.set_temp(2752) # 21.5 °C
def set_temp(self, raw):
self.regs[3:5] = (raw & 0xFFFF).to_bytes(2, "big")
_i2c_devices, _spi_devices = {}, []
class I2C:
def __init__(self, id=0, *, scl=None, sda=None, freq=400_000, timeout=50_000):
default = {0: (5, 4), 1: (7, 6)}[id] # (SCL, SDA): GP5 and GP4 on a Pico for bus 0
self.id = id
self.scl = scl.id if scl is not None else default[0]
self.sda = sda.id if sda is not None else default[1]
if not (self.scl & 1 == 1 and (self.scl & 2) >> 1 == id):
raise ValueError("bad SCL pin")
if not (self.sda & 1 == 0 and (self.sda & 2) >> 1 == id):
raise ValueError("bad SDA pin")
period = (SYS_CLK_HZ + freq // 2) // freq # whole system clocks per SCL period (pico-sdk)
self.freq, self.timeout = SYS_CLK_HZ // period, timeout
def __repr__(self):
return "I2C(%d, freq=%d, scl=%d, sda=%d, timeout=%d)" % (self.id, self.freq, self.scl, self.sda, self.timeout)
_missing = EIO # the rp2 hardware I2C reports a missing acknowledgement as EIO
def _device(self, addr):
dev = _i2c_devices.get(addr)
if dev is None:
raise OSError(self._missing)
return dev
def scan(self):
return sorted(_i2c_devices)
def readfrom_mem(self, addr, memaddr, nbytes, *, addrsize=8):
dev = self._device(addr)
return bytes(dev.regs[memaddr:memaddr + nbytes])
def readfrom_mem_into(self, addr, memaddr, buf, *, addrsize=8):
dev = self._device(addr)
buf[:] = dev.regs[memaddr:memaddr + len(buf)]
def writeto_mem(self, addr, memaddr, buf, *, addrsize=8):
dev = self._device(addr)
dev.regs[memaddr:memaddr + len(buf)] = buf
class SoftI2C(I2C):
_missing = ENODEV # the bit-banged I2C sees the missing ACK after the address
def __init__(self, scl, sda, *, freq=400_000, timeout=50_000):
self.id, self.scl, self.sda, self.freq, self.timeout = -1, scl.id, sda.id, freq, timeout
def __repr__(self):
return "SoftI2C(scl=%d, sda=%d, freq=%d)" % (self.scl, self.sda, self.freq)
class _Mem1:
"""MSC-MEM1 (made up): after chip select falls, command 0x9F returns the ID bytes 0x6D 0x53 0x01, then 0xFF."""
def __init__(self, cs):
self.cs, self.selected, self.out = cs, False, []
def select(self, pin, level):
if pin == self.cs:
if not level and not self.selected:
self.out = None # a command starts with the next byte
self.selected = not level
def exchange(self, byte):
if not self.selected:
return 0xFF
if self.out is None: # this byte is the command
self.out = [0x6D, 0x53, 0x01] if byte == 0x9F else []
return 0xFF
return self.out.pop(0) if self.out else 0xFF
class SPI:
def __init__(self, id, baudrate=1_000_000, *, polarity=0, phase=0, bits=8, sck=None, mosi=None, miso=None):
for name, pin, low in (("SCK", sck, 2), ("MOSI", mosi, 3), ("MISO", miso, 0)):
if pin is not None and not (pin.id & 3 == low and (pin.id & 8) >> 3 == id):
raise ValueError("bad %s pin" % name)
self.id = id
def write_readinto(self, write_buf, read_buf):
if len(write_buf) != len(read_buf):
raise ValueError("buffers must be the same length")
for i, b in enumerate(write_buf):
read_buf[i] = _spi_devices[0].exchange(b) if _spi_devices else 0xFF
def write(self, buf):
self.write_readinto(buf, bytearray(len(buf)))
def read(self, nbytes, write=0x00):
buf = bytearray(nbytes)
self.write_readinto(bytes([write]) * nbytes, buf)
return bytes(buf)
class UART:
def __init__(self, id, baudrate=115_200, bits=8, parity=None, stop=1, *, tx=None, rx=None,
timeout=0, timeout_char=0, txbuf=256, rxbuf=256):
self.id, self.baudrate, self.bits, self.parity, self.stop = id, baudrate, bits, parity, stop
self.tx, self.rx = tx.id, rx.id
for name, pin, low in (("TX", self.tx, 0), ("RX", self.rx, 1)):
if not (pin & 3 == low and ((pin + 4) & 8) >> 3 == id):
raise ValueError("bad %s pin" % name)
self.txbuf, self.rxbuf, self.timeout = txbuf, rxbuf, timeout
self.timeout_char = max(timeout_char, 13000 // baudrate + 1) # at least one character's time
self._arrived, self._coming = bytearray(), []
def __repr__(self):
return ("UART(%d, baudrate=%d, bits=%d, parity=%s, stop=%d, tx=%d, rx=%d, txbuf=%d, rxbuf=%d, timeout=%d, "
"timeout_char=%d, invert=None, irq=0)" % (self.id, self.baudrate, self.bits, self.parity, self.stop,
self.tx, self.rx, self.txbuf, self.rxbuf, self.timeout,
self.timeout_char))
def _wait(self):
if self._coming: # each read finds the next chunk on the wire
self._arrived += self._coming.pop(0)
def any(self):
return len(self._arrived)
def read(self, nbytes=-1):
self._wait()
if not self._arrived:
return None # nothing within the timeout
n = len(self._arrived) if nbytes < 0 else nbytes
out = bytes(self._arrived[:n])
del self._arrived[:n]
return out
def readinto(self, buf, nbytes=None):
self._wait()
if not self._arrived:
return None
n = min(len(buf) if nbytes is None else nbytes, len(self._arrived))
buf[:n] = self._arrived[:n]
del self._arrived[:n]
return n
def bench_wire(device, *, address=None, cs=None):
"""Connect a made-up device: "ENV1" to the I2C bus at an address, "MEM1" to the SPI bus with a chip select."""
if device == "ENV1":
_i2c_devices[address] = _Env1()
return _i2c_devices[address]
_spi_devices.append(_Mem1(cs))
return _spi_devices[-1]
def bench_feed(uart, *chunks):
"""Bytes that arrive on the UART's RX pin, one chunk between one read and the next."""
uart._coming.extend(chunks) Runs on this device, in your browser. The first run downloads Python (about 13.5 MB), which is kept for the next runs.
Your run, in this browser
The first byte received is 0xFF, what MISO carried while the command went out, so the ID is rx[1:]: 0x6D, 0x53,
0x01. The second exchange read 0xFF three times: without a new falling edge on the chip select, the memory took
0x9F as more of the first command, whose answer was used up. SPI also has no acknowledgement. A missing or dead
device reads as 0xFF or 0x00 and raises nothing, so a driver checks an ID at start-up.
One chip select for two commands
Code that pulls CS low once at start-up and leaves it there works for one command and then reads nonsense. Drive it
low before each command and high after it, with the high in a finally, so an error cannot leave the device
selected.
Checkpoint
Check: SPI exchanges and the chip select
A UART read returns whatever has arrived, so build lines yourself
A UART carries bytes and nothing else. A sender writes T=21.5\n, and the receiver’s read() returns what has
arrived when it is called: part of a line, two lines, or None when nothing came before the timeout. A program that
treats each read as a line breaks at the first chunk boundary. Step through one, fed three chunks as a UART might
deliver them:
Program · bus/uart_bug.py
chunks = [b"T=21", b".5\nH=4", b"8\n"]
for chunk in chunks:
for line in chunk.split(b"\n"):
if line:
print(line) Variables
- just taken by the loop
- changed in this step
- passed by the loop
- read next
- where the loop stopped
Printed so far
Nothing printed yetb'T=21' b'.5' b'H=4' b'8'
Every step was recorded from a run with Python 3.14.8 on macOS 26 arm64.
The drawing shows the variables when the program ends, and the program's 23 steps are listed below.
Change the program and run it here. Runs on this device, in your browser. The first run downloads Python (about 13.5 MB), which is kept for the next runs.
Your run, in this browser
Every step as text
- Start. No line has run yet, so there are no variables. Python starts with line 1.
- Line 1.
chunks = [b"T=21", b".5\nH=4", b"8\n"]makeschunksa list of 3 items. - Line 2.
for chunk in chunks:takes the item at position 0 ofchunks:chunkisb'T=21'. - Line 3.
for line in chunk.split(b"\n"):takes the item at position 0 ofchunk.split(b"\n"):lineisb'T=21'. - Line 4.
lineis true (lineisb'T=21'), so the block underif line:runs. - Line 5.
print(line)printsb'T=21'. The bug This prints T=21, but the sensor sent T=21.5: the rest of the reading has not arrived yet, and nothing in the chunk says so. - Line 3.
for line in chunk.split(b"\n"):finds no item at position 1 ofchunk.split(b"\n")(it has 1 item), so the loop ends. - Line 2.
for chunk in chunks:takes the item at position 1 ofchunks:chunkisb'.5\nH=4'. - Line 3.
for line in chunk.split(b"\n"):takes the item at position 0 ofchunk.split(b"\n"):lineisb'.5'. - Line 4.
lineis true (lineisb'.5'), so the block underif line:runs. - Line 5.
print(line)printsb'.5'. The bug The .5 comes out as a line of its own, because the second chunk starts in the middle of the first message. - Line 3.
for line in chunk.split(b"\n"):takes the item at position 1 ofchunk.split(b"\n"):lineisb'H=4'. - Line 4.
lineis true (lineisb'H=4'), so the block underif line:runs. - Line 5.
print(line)printsb'H=4'. - Line 3.
for line in chunk.split(b"\n"):finds no item at position 2 ofchunk.split(b"\n")(it has 2 items), so the loop ends. - Line 2.
for chunk in chunks:takes the item at position 2 ofchunks:chunkisb'8\n'. - Line 3.
for line in chunk.split(b"\n"):takes the item at position 0 ofchunk.split(b"\n"):lineisb'8'. - Line 4.
lineis true (lineisb'8'), so the block underif line:runs. - Line 5.
print(line)printsb'8'. - Line 3.
for line in chunk.split(b"\n"):takes the item at position 1 ofchunk.split(b"\n"):lineisb''. - Line 4.
lineis false (lineisb''), so Python skips the block underif line:. - Line 3.
for line in chunk.split(b"\n"):finds no item at position 2 ofchunk.split(b"\n")(it has 2 items), so the loop ends. - Line 2.
for chunk in chunks:finds no item at position 3 ofchunks(it has 3 items), so the loop ends. That was the last line, so the program ends. Four pieces for two readings, both cut at chunk boundaries. A reader has to keep the unfinished part until its newline arrives.
It printed four pieces for two readings. The fix keeps the unfinished line in a bytearray until its newline
arrives, and reads with readinto() into a second buffer made once, so nothing is allocated for each byte. The model
of machine delivers one chunk between one read and the next:
bus/uart_lines.py
"""Whole lines from a UART that delivers bytes in chunks, with buffers made once, on the model of machine beside this file."""
from machine import UART, Pin
import machine
uart = UART(1, baudrate=9600, tx=Pin(8), rx=Pin(9), timeout=0)
machine.bench_feed(uart, b"T=21", b".5\nH=4", b"8\nT=2", b"1.6\n")
line = bytearray(32) # the line being built
chunk = bytearray(16) # what one readinto() delivers
view = memoryview(chunk)
n = 0
while True:
got = uart.readinto(chunk)
if got is None: # nothing arrived within the timeout
break
for b in view[:got]:
if b == 0x0A: # the end of a line
print(bytes(line[:n]))
n = 0
elif n < len(line):
line[n] = b
n += 1 Output
b'T=21.5' b'H=48' b'T=21.6'
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 uart_lines.py
machine.py (the model) · bus/machine.py
"""A model of machine.I2C, SoftI2C, SPI and UART on a Raspberry Pi Pico (125 MHz), with made-up devices on the buses,
so bus code runs on a computer.
It follows the MicroPython 1.29.0 documentation and the rp2 port's source: the pin rules and the print formats of
ports/rp2/machine_i2c.c, machine_i2c.h and machine_uart.c, the speed the pico-sdk's i2c_set_baudrate() really sets
(a period of whole system clocks), OSError(5) EIO from the hardware I2C when nobody acknowledges and OSError(19)
ENODEV from SoftI2C, and the UART defaults (256-byte buffers, timeout 0, timeout_char at least 13000 // baudrate + 1).
The devices are made up for this track: MSC-ENV1, a temperature sensor with 8-bit registers, and MSC-MEM1, a memory
that answers SPI command 0x9F with three ID bytes. Functions whose names start with bench_ are not MicroPython: they
wire the devices and feed the UART. Never copy this file to a board.
"""
FAKE = True
SYS_CLK_HZ = 125_000_000
EIO, ENODEV = 5, 19
class Pin:
IN, OUT = 0, 1
def __init__(self, id, mode=-1, value=None):
self.id, self.level = id, value
def __call__(self, v=None):
return self.value(v)
def value(self, v=None):
if v is None:
return self.level
self.level = 1 if v else 0
for dev in _spi_devices:
dev.select(self.id, self.level)
def __repr__(self):
return "Pin(GPIO%d)" % self.id
class _Env1:
"""MSC-ENV1 (made up): 0x00 ID 0xE1, 0x01 CTRL, 0x02 STATUS, 0x03-0x04 TEMP (signed, big-endian, 1/128 °C)."""
def __init__(self):
self.regs = bytearray(8)
self.regs[0] = 0xE1
self.set_temp(2752) # 21.5 °C
def set_temp(self, raw):
self.regs[3:5] = (raw & 0xFFFF).to_bytes(2, "big")
_i2c_devices, _spi_devices = {}, []
class I2C:
def __init__(self, id=0, *, scl=None, sda=None, freq=400_000, timeout=50_000):
default = {0: (5, 4), 1: (7, 6)}[id] # (SCL, SDA): GP5 and GP4 on a Pico for bus 0
self.id = id
self.scl = scl.id if scl is not None else default[0]
self.sda = sda.id if sda is not None else default[1]
if not (self.scl & 1 == 1 and (self.scl & 2) >> 1 == id):
raise ValueError("bad SCL pin")
if not (self.sda & 1 == 0 and (self.sda & 2) >> 1 == id):
raise ValueError("bad SDA pin")
period = (SYS_CLK_HZ + freq // 2) // freq # whole system clocks per SCL period (pico-sdk)
self.freq, self.timeout = SYS_CLK_HZ // period, timeout
def __repr__(self):
return "I2C(%d, freq=%d, scl=%d, sda=%d, timeout=%d)" % (self.id, self.freq, self.scl, self.sda, self.timeout)
_missing = EIO # the rp2 hardware I2C reports a missing acknowledgement as EIO
def _device(self, addr):
dev = _i2c_devices.get(addr)
if dev is None:
raise OSError(self._missing)
return dev
def scan(self):
return sorted(_i2c_devices)
def readfrom_mem(self, addr, memaddr, nbytes, *, addrsize=8):
dev = self._device(addr)
return bytes(dev.regs[memaddr:memaddr + nbytes])
def readfrom_mem_into(self, addr, memaddr, buf, *, addrsize=8):
dev = self._device(addr)
buf[:] = dev.regs[memaddr:memaddr + len(buf)]
def writeto_mem(self, addr, memaddr, buf, *, addrsize=8):
dev = self._device(addr)
dev.regs[memaddr:memaddr + len(buf)] = buf
class SoftI2C(I2C):
_missing = ENODEV # the bit-banged I2C sees the missing ACK after the address
def __init__(self, scl, sda, *, freq=400_000, timeout=50_000):
self.id, self.scl, self.sda, self.freq, self.timeout = -1, scl.id, sda.id, freq, timeout
def __repr__(self):
return "SoftI2C(scl=%d, sda=%d, freq=%d)" % (self.scl, self.sda, self.freq)
class _Mem1:
"""MSC-MEM1 (made up): after chip select falls, command 0x9F returns the ID bytes 0x6D 0x53 0x01, then 0xFF."""
def __init__(self, cs):
self.cs, self.selected, self.out = cs, False, []
def select(self, pin, level):
if pin == self.cs:
if not level and not self.selected:
self.out = None # a command starts with the next byte
self.selected = not level
def exchange(self, byte):
if not self.selected:
return 0xFF
if self.out is None: # this byte is the command
self.out = [0x6D, 0x53, 0x01] if byte == 0x9F else []
return 0xFF
return self.out.pop(0) if self.out else 0xFF
class SPI:
def __init__(self, id, baudrate=1_000_000, *, polarity=0, phase=0, bits=8, sck=None, mosi=None, miso=None):
for name, pin, low in (("SCK", sck, 2), ("MOSI", mosi, 3), ("MISO", miso, 0)):
if pin is not None and not (pin.id & 3 == low and (pin.id & 8) >> 3 == id):
raise ValueError("bad %s pin" % name)
self.id = id
def write_readinto(self, write_buf, read_buf):
if len(write_buf) != len(read_buf):
raise ValueError("buffers must be the same length")
for i, b in enumerate(write_buf):
read_buf[i] = _spi_devices[0].exchange(b) if _spi_devices else 0xFF
def write(self, buf):
self.write_readinto(buf, bytearray(len(buf)))
def read(self, nbytes, write=0x00):
buf = bytearray(nbytes)
self.write_readinto(bytes([write]) * nbytes, buf)
return bytes(buf)
class UART:
def __init__(self, id, baudrate=115_200, bits=8, parity=None, stop=1, *, tx=None, rx=None,
timeout=0, timeout_char=0, txbuf=256, rxbuf=256):
self.id, self.baudrate, self.bits, self.parity, self.stop = id, baudrate, bits, parity, stop
self.tx, self.rx = tx.id, rx.id
for name, pin, low in (("TX", self.tx, 0), ("RX", self.rx, 1)):
if not (pin & 3 == low and ((pin + 4) & 8) >> 3 == id):
raise ValueError("bad %s pin" % name)
self.txbuf, self.rxbuf, self.timeout = txbuf, rxbuf, timeout
self.timeout_char = max(timeout_char, 13000 // baudrate + 1) # at least one character's time
self._arrived, self._coming = bytearray(), []
def __repr__(self):
return ("UART(%d, baudrate=%d, bits=%d, parity=%s, stop=%d, tx=%d, rx=%d, txbuf=%d, rxbuf=%d, timeout=%d, "
"timeout_char=%d, invert=None, irq=0)" % (self.id, self.baudrate, self.bits, self.parity, self.stop,
self.tx, self.rx, self.txbuf, self.rxbuf, self.timeout,
self.timeout_char))
def _wait(self):
if self._coming: # each read finds the next chunk on the wire
self._arrived += self._coming.pop(0)
def any(self):
return len(self._arrived)
def read(self, nbytes=-1):
self._wait()
if not self._arrived:
return None # nothing within the timeout
n = len(self._arrived) if nbytes < 0 else nbytes
out = bytes(self._arrived[:n])
del self._arrived[:n]
return out
def readinto(self, buf, nbytes=None):
self._wait()
if not self._arrived:
return None
n = min(len(buf) if nbytes is None else nbytes, len(self._arrived))
buf[:n] = self._arrived[:n]
del self._arrived[:n]
return n
def bench_wire(device, *, address=None, cs=None):
"""Connect a made-up device: "ENV1" to the I2C bus at an address, "MEM1" to the SPI bus with a chip select."""
if device == "ENV1":
_i2c_devices[address] = _Env1()
return _i2c_devices[address]
_spi_devices.append(_Mem1(cs))
return _spi_devices[-1]
def bench_feed(uart, *chunks):
"""Bytes that arrive on the UART's RX pin, one chunk between one read and the next."""
uart._coming.extend(chunks) Runs on this device, in your browser. The first run downloads Python (about 13.5 MB), which is kept for the next runs.
Your run, in this browser
readinto() returned the number of bytes it stored, then None when nothing more arrived, which ended the loop.
Iterating over a memoryview slice reads the chunk in place; the only new objects are one view for each read and one
bytes for each finished line. A line longer than its 32-byte buffer is cut rather than written past the end.
Two settings decide when a read gives up: timeout, the milliseconds to wait for the first character, and
timeout_char, the wait between characters. readline() stops at the timeout too, so with a short timeout it can
return half a line. For message-shaped data, UART.irq(trigger=UART.IRQ_RXIDLE) calls a handler when the line goes
quiet after a burst of bytes; MicroPython 1.24 made the UART interrupts behave alike on the esp32, mimxrt, nrf,
renesas-ra, rp2 and samd ports.
Key idea
A serial read hands you the bytes that have arrived, never a message: keep the unfinished part in a buffer of your own and split on the line end yourself.
Checkpoint
Check: UART reads, timeouts and lines
Practise: a UART line reader that keeps its buffers
Write LineReader, the reader of the last program as a class a whole application can use: it keeps an unfinished
line between calls, ignores carriage returns, skips empty lines, and drops lines too long for its buffer while
counting them. The sample tests feed it chunks through a fake UART that only offers readinto():
Exercise · Medium · Python
A UART line reader that keeps its buffers
Write the LineReader class in lines.py. It collects whole lines from a UART whose reads return any number of bytes, as a real one does.
LineReader(uart, size=64, chunk=16) makes its buffers once: a bytearray(size) for the line being built and a bytearray(chunk) for one read. poll() calls uart.readinto(...) with that same chunk buffer until it returns None or 0, and returns a list of the lines it finished, as bytes without the line end.
A line ends at b"\n". Ignore every b"\r", skip empty lines, and keep an unfinished line for the next poll(). A line with more than size bytes is dropped when its b"\n" arrives, and self.dropped counts such lines; the line after it is read normally. The sample tests use a fake UART that offers only readinto().
Starter code · lines.py
class LineReader:
"""Whole lines from a UART, with buffers made once."""
def __init__(self, uart, size=64, chunk=16):
self.uart = uart
self.dropped = 0
def poll(self):
"""Read what has arrived; return the finished lines as bytes, without the line end."""
return [] The sample tests · test_lines.py
from lines import LineReader
class FakeUART:
"""Hands out one chunk per readinto() call, then None, like a UART with timeout=0."""
def __init__(self, *chunks):
self.chunks = list(chunks)
self.buffers = []
def readinto(self, buf, nbytes=None):
self.buffers.append(buf)
if not self.chunks:
return None
data = self.chunks.pop(0)
assert len(data) <= len(buf), "the test's chunk must fit the buffer"
buf[: len(data)] = data
return len(data)
def feed(self, *chunks):
self.chunks.extend(chunks)
def test_one_line():
"""returns one line from one chunk"""
assert LineReader(FakeUART(b"OK\n")).poll() == [b"OK"]
def test_line_across_chunks():
"""joins a line that arrives in pieces"""
assert LineReader(FakeUART(b"T=21", b".5\nH=4", b"8\n")).poll() == [b"T=21.5", b"H=48"]
def test_keeps_unfinished_line():
"""keeps an unfinished line until the next poll"""
uart = FakeUART(b"P=10")
reader = LineReader(uart)
assert reader.poll() == []
uart.feed(b"13\n")
assert reader.poll() == [b"P=1013"]
def test_crlf_and_empty_lines():
"""ignores carriage returns and skips empty lines"""
assert LineReader(FakeUART(b"A\r\n\r\nB\r\n")).poll() == [b"A", b"B"]
def test_long_line_dropped():
"""drops a line longer than size, counts it, and reads the next one"""
reader = LineReader(FakeUART(b"ABCDEFGHIJ\nOK\n"), size=8)
assert reader.poll() == [b"OK"]
assert reader.dropped == 1
def test_line_of_exactly_size():
"""keeps a line of exactly size bytes"""
assert LineReader(FakeUART(b"ABCDEFGH\n"), size=8).poll() == [b"ABCDEFGH"]
def test_same_buffer_every_read():
"""reads into the same chunk buffer every time"""
uart = FakeUART(b"x=1\n", b"y=2\n")
LineReader(uart).poll()
assert len(uart.buffers) == 3 and all(b is uart.buffers[0] for b in uart.buffers) A hint
Keep the number of bytes in the line buffer, self.n, and a flag for a line that has grown too long. For every byte of view[:got], where view = memoryview(self.chunk): on 0x0A finish or drop the line and reset both; skip 0x0D; otherwise store the byte if self.n < size, or set the flag.
Results of the sample tests
| Test | Result | Details |
|---|
What your code printed
The sample tests run on this device, in your browser (Pyodide): nothing is sent to mysmartcopilot.com. The first run downloads Python (about 13.5 MB), which is kept for the next runs. A check in your browser is feedback for you, not proof that the code is right for every input.
Then check yourself on pins, registers, chip selects and serial reads:
Practice that adapts to you
Practise this lesson
8 questions about 4 ideas of this lesson. Each answer is explained at once, and what you get right is kept in this browser.
Interview questions
Warm-up (fresher): How do I2C, SPI and UART differ?
I2C uses two shared wires, a clock and data, and the controller picks a device by sending its 7-bit address; every device acknowledges each byte, so a missing one shows up as an error. SPI uses a clock, a line out and a line in, and a separate chip-select line for each device; it is faster and full duplex, but nothing acknowledges, so a missing device just reads as 0xFF or 0x00. A UART has a transmit and a receive line and no clock at all: both ends must agree on the baud rate, and bytes arrive whenever the other side sends them, so the receiver has to find message boundaries itself.
Warm-up (fresher): Why does an I2C bus need pull-up resistors?
Every device on an I2C bus drives SDA and SCL only low, as open-drain outputs, so that several devices can share the wires without a short circuit. Something has to bring the lines back high when nobody pulls them down, and that is the pull-up resistors’ job, 1 to 10 kΩ by MicroPython’s I2C documentation. Without them the lines never rise properly, and transfers fail, hang or return wrong values. The internal pull-ups the rp2 port switches on are 50 to 80 kΩ, too weak for a reliable bus.
Two harder bus questions come next. Each row leads to what the question probes and to a way into it:
Model answers open with any pass: ₹79 for 1 week ($2.99), one payment that never renews.
-
Mid to senior
A sensor answers every read on the bench, but mounted at the end of a 1 m cable it raises OSError 5 every few minutes on a Pico's hardware I2C at 400 kHz. How do you find the cause, and what do you change in the hardware and in the code?
-
Fresher to mid
A module sends text lines ending in \r\n at 9,600 baud. The parser does line = uart.read() and sometimes sees half lines, and over days the board starts raising MemoryError. What is wrong, and how do you build the reader?
Flashcards · 8 cards
Recall it, then check
Say the answer to yourself before you look: recalling it is what makes it stick.
-
How do I2C, SPI and UART each pick the device they talk to?
I2C sends a 7-bit address in every transfer and each byte is acknowledged; SPI drives that device's chip-select line low; a UART has one device a port and no clock.
-
Why print an I2C object after creating it?
The printed object shows the real clock rate: asked for 400,000 Hz, a Pico runs at 399,361 Hz, since every SCL period is a whole number of 125 MHz system clocks.
-
Why read registers with readfrom_mem_into() in a loop?
It fills a buffer made once, while readfrom_mem() returns a new bytes object at every call, which matters in a loop that reads ten times a second.
-
How do you read a signed big-endian 16-bit register?
struct.unpack_from(">h", buf)[0]. MicroPython's int.from_bytes() has no signed mode, so 0xFD 0x60 would read as 64,864 instead of −672.
-
What does a missing I2C device raise on a Pico?
OSError 5, EIO, from the hardware I2C, and 19, ENODEV, from SoftI2C. A device that holds the clock low past the timeout raises 110, ETIMEDOUT.
-
How does a missing SPI device show itself?
It does not: reads return 0xFF or 0x00 and nothing raises, since SPI has no acknowledgement. A driver checks an ID at start-up.
-
Why drive the SPI chip select high after every command?
The device starts a command on a falling chip select and ends it on a rising one; without a new falling edge it takes the next bytes as more of the first command. Put the high in a finally.
-
What does a UART read return?
Whatever has arrived: part of a line, two lines, or None when nothing came before the timeout. Keep the unfinished line in a buffer of your own and split on the line end yourself.
Key takeaways
- I2C picks a device by address and gets an acknowledgement for every byte, SPI by a chip select with no acknowledgement, and a UART has one device a port and no clock; each hardware bus reaches only some pins.
- Ask for a speed, then print the bus: a Pico’s I2C at 400,000 Hz really runs at 399,361 Hz, and the rp2 UART starts
with
timeout=0. readfrom_mem_into()reads registers into a buffer made once, andstruct.unpack_from(">h", buf)reads a signed register thatint.from_bytes()would read as unsigned.- A missing I2C device raises
OSError5 on the Pico’s hardware bus and 19 onSoftI2C; a missing SPI device raises nothing, so check its ID. - SPI exchanges bytes both ways in buffers of equal length, inside a chip select you drive low before each command and high after it.
- A UART read returns whatever has arrived, so keep the unfinished line in your own buffer, read with
readinto(), and split on the line end yourself.
Terms in this lesson
- chip select
- The line that tells one SPI device the next bytes are for it: driven low before a command and high after it. MicroPython leaves it to your code, as a Pin.
- I2C
- A two-wire bus, clock SCL and data SDA, on which a controller picks one of up to 112 devices by a 7-bit address. Every byte is acknowledged, so a missing device is noticed.
- repeated start
- An I2C start condition sent without a stop before it, so the controller keeps the bus while it turns from writing a register number to reading the register.
- SPI
- A bus of a clock, a line out (MOSI) and a line in (MISO), plus one chip-select line per device. Every clock moves one bit each way, and nothing acknowledges.
- UART
- A serial port with a transmit and a receive line and no clock. Both ends use the same baud rate, and bytes arrive whenever the other end sends them.
References
- class I2C – a two-wire serial protocol (MicroPython project)
- class SPI – a Serial Peripheral Interface bus protocol (controller side) (MicroPython project)
- class UART – duplex serial communication bus (MicroPython project)
- class I2CTarget – an I2C target device (MicroPython project)
- Quick reference for the RP2 (UART, SPI and I2C pins) (MicroPython project)
- struct – pack and unpack primitive data types (MicroPython project)
- Builtin functions and exceptions (int.from_bytes without signed) (MicroPython project)
- MicroPython v1.24.0 release notes (UART IRQs made consistent across ports) (MicroPython project)
- MicroPython v1.25.0 release notes (default I2C and SPI bus ids on rp2 and esp32) (MicroPython project)
- MicroPython v1.26.0 release notes (machine.I2CTarget) (MicroPython project)
- rp2 hardware I2C, its pull-ups and its errors (ports/rp2/machine_i2c.c, v1.29.0) (MicroPython project)
- rp2 I2C pin rules and default pins (ports/rp2/machine_i2c.h, v1.29.0) (MicroPython project)
- Software I2C and its ENODEV (extmod/machine_i2c.c, v1.29.0) (MicroPython project)
- rp2 UART defaults, timeouts and print (ports/rp2/machine_uart.c, v1.29.0) (MicroPython project)
- i2c_set_baudrate() (pico-sdk 2.3.0, src/rp2_common/hardware_i2c/i2c.c) (Raspberry Pi Ltd)
- RP2040 datasheet (5.5.3 pad pull-up resistance) (Raspberry Pi Ltd)
Related tools
Report a problem with this lesson
Kept only in this browser. Your Learn progress