MicroPython for Microcontrollers Module 13 – Ports and boards: RP2040/RP2350, ESP32 and the rest
RP2040 and RP2350 specifics: PIO, DMA and flash
Program the RP2040 and RP2350 from MicroPython 1.29: PIO state machines timed to the cycle, DMA copies and paced transfers, rp2.Flash and BOOTSEL.
What you will learn
- Count the cycles of a PIO program and predict the pulse lengths it makes
- Choose a StateMachine freq from the clock divider it becomes, and say what a fraction costs
- Configure an rp2.DMA transfer with pack_ctrl(), a count in transfers and a pace
- Use rp2.Flash(start, len) and rp2.bootsel_button() without corrupting files or stalling the board
- Name what the RP2350 adds to the rp2 port and what stays the same
Before you start
What you’ll learn · Before you start
What you will learn
- Count the cycles of a PIO program and predict the pulse lengths it makes
- Choose a StateMachine freq from the clock divider it becomes, and say what a fraction costs
- Configure an rp2.DMA transfer with pack_ctrl(), a count in transfers and a pace
- Use rp2.Flash(start, len) and rp2.bootsel_button() without corrupting files or stalling the board
- Name what the RP2350 adds to the rp2 port and what stays the same
Before you start
On this page
An infrared remote’s 38 kHz carrier flips a pin 76,000 times a second, once every 13.2 µs, for as long as a burst lasts. A Python loop on a Pico cannot keep that rhythm while it does anything else: the rp2 quick reference warns that even timer callbacks are prone to jitter from the garbage collector.
The RP2040 has hardware that keeps time on its own: eight PIO state machines that run one instruction per clock
cycle, and twelve DMA channels that move data without the processor. This lesson programs both from MicroPython
1.29: a PIO program timed to the cycle, the clock divider behind freq, DMA transfers set up with pack_ctrl() and
paced by a request or the new DMATimer, the flash behind the filesystem, the BOOTSEL button, and what the RP2350
adds. Every program runs here on a model of the rp2 module built from the port’s own source.
A state machine runs one instruction per cycle, so its timing can be counted
PIO stands for programmable I/O. The RP2040 has two PIO blocks of four
PIO state machine: One of the small processors in the programmable I/O blocks of the RP2040 and RP2350. It runs a program of up to 32 instructions, one per cycle of its own clock, to drive and read pins with exact timing. each, numbered 0 to 7 by rp2.StateMachine, and each block
holds 32 instructions that its four machines share. A state machine knows nine instructions, among them set to
drive pins, pull to take a word that Python sent, mov to copy it and jmp to loop. The RP2040 datasheet makes
the promise that matters: every instruction takes exactly one cycle unless it stalls, and up to 31 delay cycles may
follow it, written in MicroPython as [n] after the instruction. A PIO program’s timing is counted, not measured.
The program below sends bursts of a 38 kHz carrier. Python puts the number of periods minus one into the state
machine’s transmit FIFO: First in, first out: a small queue in hardware. Each PIO state machine has a 4-word FIFO for data from the processor and one for data back to it.; the program pulls it, copies it into the scratch register X and
loops: a set with 12 delay cycles keeps the pin high for 13 cycles, a set with 11 keeps it low for 12, and the
jump adds the thirteenth. Step through one burst:
The drawing shows the whole scene; its 9 steps are listed below.
Every step as text
- Python sends a count. The burst needs 24 periods, so Python puts 23 into the transmit FIFO with sm.put(24 - 1). The state machine is already waiting at pull().
- Cycle 0: pull(). pull() moves the word from the FIFO into the output shift register, OSR, in one cycle. With the FIFO empty it would stall here instead, cycle after cycle.
- Cycle 1: mov(x, osr). X becomes 23, the number of periods still to send after the first one. Two cycles have passed and the pin is still low.
- Cycles 2 to 14: high. set(pins, 1) drives GP16 high at cycle 2, and its delay of 12 holds the state machine for 12 more cycles. The pin is high for 13 cycles.
- Cycles 15 to 26: low. set(pins, 0) drives the pin low at cycle 15 and its delay of 11 adds 11 cycles. So far the pin has been low for 12 cycles.
- Cycle 27: the jump. jmp(x_dec) finds X = 23, which is not zero, so it jumps back to the label, and X becomes 22. Its cycle is the 13th low one, so a period is 26 cycles.
- Periods 2 to 24. The loop repeats. Rising edges come at cycles 28, 54 and so on, 26 cycles apart, and every jump leaves X one lower. At 988,020 Hz that is a 38,000.8 Hz carrier.
- Cycle 625: X is 0. The 24th jump finds X = 0, so it does not jump, and X still decrements, to 0xFFFFFFFF. The program wraps to pull(), which costs no cycle.
- Waiting for the next count. From cycle 626 pull() finds the FIFO empty and stalls. GP16 stays low, and the next sm.put() starts the next burst two cycles after it lands.
The model beside the program runs those instructions one cycle at a time at freq=988_000, which the next section
explains; its bench_ functions report what a board keeps to itself, so they exist only in the model. Run it, then
press Edit, change sm.put(24 - 1) to sm.put(24) and run it again:
rp2/burst.py
"""A burst of a 38 kHz infrared carrier from a PIO program, run on the model of rp2 beside this file."""
import rp2
from machine import Pin
@rp2.asm_pio(set_init=rp2.PIO.OUT_LOW)
def burst():
pull() # wait for a count from the CPU
mov(x, osr) # x = periods - 1
label("period")
set(pins, 1)[12] # 1 + 12 cycles high
set(pins, 0)[11] # 1 + 11 cycles low ...
jmp(x_dec, "period") # ... and 1 more for the jump
sm = rp2.StateMachine(0, burst, freq=988_000, set_base=Pin(16))
sm.active(1)
sm.put(24 - 1) # 24 carrier periods
whole, frac = rp2.bench_divider(sm)
ns = rp2.bench_cycle_ns(sm)
print(sm, "divider", whole, "+", frac, "/ 256: one cycle is", round(ns, 1), "ns")
trace = rp2.bench_run(sm, 700)
levels = [level for _, level in trace]
rises = [c for c in range(1, len(levels)) if levels[c] and not levels[c - 1]]
high = [levels[r:].index(0) for r in rises]
period = rises[1] - rises[0]
print("first rising edge at cycle", rises[0], "after pull and mov")
print(len(rises), "periods of", period, "cycles:", high[0], "high and", period - high[0], "low")
print("carrier:", round(1e9 / (period * ns), 1), "Hz")
stall = [what for what, _ in trace].index("stall")
print("burst: cycles", rises[0], "to", stall - 1, "=", round((stall - rises[0]) * ns / 1000, 1), "us")
print("from cycle", stall, "pull() waits for the next count; the pin stays", levels[-1])Output
StateMachine(0) divider 126 + 132 / 256: one cycle is 1012.1 ns first rising edge at cycle 2 after pull and mov 24 periods of 26 cycles: 13 high and 13 low carrier: 38000.8 Hz burst: cycles 2 to 625 = 631.6 us from cycle 626 pull() waits for the next count; the pin stays 0
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 burst.py
rp2/divider.py
"""The clock divider StateMachine(freq=...) sets at a 125 MHz system clock, on the model of rp2 beside this file."""
import rp2
@rp2.asm_pio()
def idle():
nop()
print("freq asked divider one cycle real freq system clocks a cycle")
for freq in (125_000_000, 10_000_000, 1_000_000, 988_000, 2_000, 1_908, 1_907):
try:
sm = rp2.StateMachine(0, idle, freq=freq)
except ValueError as e:
print(f"{freq:>11,} ValueError: {e}")
continue
whole, frac = rp2.bench_divider(sm)
ns = rp2.bench_cycle_ns(sm)
clocks = f"{whole}, every cycle" if frac == 0 else f"{whole} or {whole + 1}"
print(f"{freq:>11,} {whole:>5} + {frac:>3}/256 {ns:>9.1f} ns {1e9 / ns:>13,.2f} Hz {clocks}")Output
freq asked divider one cycle real freq system clocks a cycle
125,000,000 1 + 0/256 8.0 ns 125,000,000.00 Hz 1, every cycle
10,000,000 12 + 128/256 100.0 ns 10,000,000.00 Hz 12 or 13
1,000,000 125 + 0/256 1000.0 ns 1,000,000.00 Hz 125, every cycle
988,000 126 + 132/256 1012.1 ns 988,020.25 Hz 126 or 127
2,000 62500 + 0/256 500000.0 ns 2,000.00 Hz 62500, every cycle
1,908 65513 + 160/256 524109.0 ns 1,908.00 Hz 65513 or 65514
1,907 ValueError: freq out of range
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 divider.py
rp2.py (the model) · rp2/rp2.py
"""A model of MicroPython's rp2 module on a Raspberry Pi Pico (RP2040, 125 MHz), so rp2 code runs on a computer.
It follows the MicroPython 1.29.0 documentation and the rp2 port's source: a subset of the PIO assembler of
ports/rp2/modules/rp2.py (set, mov, pull, jmp, nop, label, wrap_target, wrap and [delay]), the clock divider of
StateMachine.init() in ports/rp2/rp2_pio.c, the DMA control fields and defaults of ports/rp2/rp2_dma.c with the
RP2040's bit positions, the DMATimer ratio search of the same file, and rp2.Flash(start, len) of
ports/rp2/rp2_flash.c. Functions whose names start with bench_ are not MicroPython: they run a state machine for a
number of cycles, or report what the hardware would have done. Never copy this file to a board.
"""
FAKE = True
SYS_CLK_HZ = 125_000_000 # machine.freq() after boot on a Pico (rp2 quick reference)
_MASK32 = 0xFFFFFFFF
class PIOASMError(Exception):
pass
class BenchError(Exception):
"""What the hardware would do silently, said out loud by the model."""
class PIO:
IN_LOW, IN_HIGH, OUT_LOW, OUT_HIGH = 0, 1, 2, 3
SHIFT_LEFT, SHIFT_RIGHT = 0, 1
JOIN_NONE, JOIN_TX, JOIN_RX = 0, 1, 2
class _Instr:
def __init__(self, op, args):
self.op, self.args, self.delay = op, args, 0
def __getitem__(self, cycles): # set(pins, 1) [12]
if not 0 <= cycles <= 31:
raise PIOASMError("delay too large")
self.delay = cycles
return self
class _Program:
def __init__(self, set_init):
self.set_init, self.instrs, self.labels = set_init, [], {}
self.wrap_target, self.wrap = 0, None
def _add(self, op, *args):
ins = _Instr(op, args)
self.instrs.append(ins)
return ins
def names(self):
"""What a PIO function sees as globals while asm_pio() assembles it."""
def set_(dest, data):
if not 0 <= data <= 31:
raise PIOASMError("set takes a value from 0 to 31")
return self._add("set", dest, data)
def jmp(cond, label=None):
return self._add("jmp", "always", cond) if label is None else self._add("jmp", cond, label)
def label(name):
self.labels[name] = len(self.instrs)
def wrap_target():
self.wrap_target = len(self.instrs)
def wrap():
self.wrap = len(self.instrs) - 1
return {
"set": set_, "jmp": jmp, "label": label, "wrap_target": wrap_target, "wrap": wrap,
"nop": lambda: self._add("nop"), "pull": lambda: self._add("pull"),
"mov": lambda dest, src: self._add("mov", dest, src),
"pins": "pins", "x": "x", "y": "y", "osr": "osr",
"x_dec": "x_dec", "y_dec": "y_dec", "not_x": "not_x", "not_y": "not_y",
}
def asm_pio(*, set_init=None, **settings):
"""Assemble the decorated function's instructions (as @rp2.asm_pio does, for the subset above)."""
def dec(f):
prog = _Program(set_init)
names = prog.names()
gl = f.__globals__
saved = {k: gl[k] for k in names if k in gl}
gl.update(names)
try:
f()
finally:
for k in names:
if k in saved:
gl[k] = saved[k]
else:
gl.pop(k, None)
if prog.wrap is None:
prog.wrap = len(prog.instrs) - 1
return prog
return dec
class StateMachine:
"""One of the RP2040's 8 state machines (2 PIO blocks of 4), running at a divided system clock."""
def __init__(self, id, program=None, freq=-1, *, set_base=None):
if not 0 <= id <= 7:
raise ValueError("invalid StateMachine")
self.id = id
if program is not None:
self.init(program, freq, set_base=set_base)
def init(self, program, freq=-1, *, set_base=None):
if len(program.instrs) > 32:
raise OSError(12, "ENOMEM") # a PIO block holds 32 instructions
if freq < 0:
div = 256 # the default: run at the system clock
elif freq == 0:
div = 65536 * 256 # a divider of 0 means 65536, the slowest
else:
div = SYS_CLK_HZ * 256 // freq # ports/rp2/rp2_pio.c: clkdiv in 1/256 steps
if not 256 <= div <= 65536 * 256:
raise ValueError("freq out of range")
self._div, self._prog = div, program
self._pc, self._delay, self._x, self._y, self._osr = 0, 0, 0, 0, 0
self._fifo, self._running = [], False
self.level = 1 if program.set_init == PIO.OUT_HIGH else 0
def active(self, value=None):
if value is not None:
self._running = bool(value)
return self._running
def put(self, value):
if len(self._fifo) == 4 and not self._running:
raise BenchError("put() would wait for ever: the 4-word TX FIFO is full and nothing pulls")
self._fifo.append(value & _MASK32)
def tx_fifo(self):
return len(self._fifo)
def __repr__(self):
return "StateMachine(%d)" % self.id
def _cycle(self):
"""One state machine cycle: an instruction, a delay cycle or a stall."""
if self._delay:
self._delay -= 1
return "delay"
ins = self._prog.instrs[self._pc]
nxt = self._pc + 1
if ins.op == "pull":
if not self._fifo:
return "stall" # a blocking pull waits; its delay starts only after it completes
self._osr = self._fifo.pop(0)
elif ins.op == "mov":
dest, src = ins.args
setattr(self, "_" + dest, getattr(self, "_" + src))
elif ins.op == "set":
dest, data = ins.args
if dest == "pins":
self.level = data & 1
else:
setattr(self, "_" + dest, data)
elif ins.op == "jmp":
cond, target = ins.args
take = True
if cond in ("x_dec", "y_dec"):
reg = "_" + cond[0]
take = getattr(self, reg) != 0 # tested before the decrement, which always happens
setattr(self, reg, (getattr(self, reg) - 1) & _MASK32)
elif cond in ("not_x", "not_y"):
take = getattr(self, "_" + cond[-1]) == 0
if take:
nxt = self._prog.labels[target]
if self._pc == self._prog.wrap and nxt == self._pc + 1:
nxt = self._prog.wrap_target # wrapping costs no cycle
self._pc, self._delay = nxt, ins.delay
return ins.op
def bench_divider(sm):
"""The (whole, frac) clock divider the port set: one cycle lasts whole + frac/256 system clocks."""
return sm._div // 256, sm._div % 256
def bench_cycle_ns(sm):
"""The average length of one state machine cycle in nanoseconds."""
return sm._div / 256 * 1e9 / SYS_CLK_HZ
def bench_run(sm, cycles):
"""Run sm for a number of cycles; return what each cycle did and the pin level during it."""
if not sm._running:
raise BenchError("the state machine is not active")
return [(sm._cycle(), sm.level) for _ in range(cycles)]
# ---- DMA ------------------------------------------------------------------------------------------------------
# (name, lowest bit, width) of the RP2040's CH0_CTRL_TRIG register, in the order rp2_dma.c lists them.
_CTRL_FIELDS = (
("enable", 0, 1), ("high_pri", 1, 1), ("size", 2, 2), ("inc_read", 4, 1), ("inc_write", 5, 1),
("ring_size", 6, 4), ("ring_sel", 10, 1), ("chain_to", 11, 4), ("treq_sel", 15, 6), ("irq_quiet", 21, 1),
("bswap", 22, 1), ("sniff_en", 23, 1), ("busy", 24, 1), ("write_err", 29, 1), ("read_err", 30, 1),
("ahb_err", 31, 1),
)
_READ_ONLY = ("busy", "ahb_err")
_PERMANENT = 0x3F # treq_sel: no pacing, run at full speed
_claimed_channels, _claimed_timers = set(), set()
class DMA:
"""One of the RP2040's 12 DMA channels, claimed for this object."""
def __init__(self):
free = [c for c in range(12) if c not in _claimed_channels]
if not free:
raise OSError(16, "EBUSY")
self.channel = free[0]
_claimed_channels.add(self.channel)
self._busy = False
def __repr__(self):
return "DMA(%d)" % self.channel
def pack_ctrl(self, default=None, **fields):
value = default
if value is None: # quiet, unpaced, both addresses incrementing, 4-byte words, enabled, chained to itself
value = 1 << 21 | _PERMANENT << 15 | 1 << 5 | 1 << 4 | 2 << 2 | 1 | (self.channel & 0xF) << 11
known = {name: (shift, width) for name, shift, width in _CTRL_FIELDS}
for name, v in fields.items():
if name not in known:
raise TypeError("unexpected keyword argument '%s'" % name)
if name in _READ_ONLY:
continue
shift, width = known[name]
mask = (1 << width) - 1
v = int(v)
if v & mask != v:
raise ValueError("bad field value")
value = value & ~(mask << shift) | v << shift
return value
@staticmethod
def unpack_ctrl(value):
return {name: value >> shift & (1 << width) - 1 for name, shift, width in _CTRL_FIELDS}
def config(self, read=None, write=None, count=None, ctrl=None, trigger=False):
self._read, self._write, self._count, self._ctrl = read, write, count, ctrl
if trigger:
self._run()
def _run(self):
f = DMA.unpack_ctrl(self._ctrl)
width = 1 << f["size"]
total = self._count * width
short = [(name, len(buf)) for name, buf, inc in (("read", self._read, f["inc_read"]),
("write", self._write, f["inc_write"]))
if inc and total > len(buf)]
if short:
raise BenchError(
"%d transfers of %d bytes move %d bytes, but %s: on a board the channel carries on into the memory "
"after the buffer" % (self._count, width, total,
" and ".join("the %s buffer holds %d" % s for s in short))
)
if f["inc_read"] and f["inc_write"]:
self._write[:total] = self._read[:total]
self._busy = False
def active(self, value=None):
return self._busy
def close(self):
_claimed_channels.discard(self.channel)
class DMATimer:
"""One of the 4 DMA pacing timers: it asks for a transfer X/Y times per system clock cycle."""
def __init__(self, timer_id=None, *, freq=None, ratio=None):
if timer_id is not None and (not 0 <= timer_id <= 3 or timer_id in _claimed_timers):
raise ValueError("value out of range") if not 0 <= timer_id <= 3 else OSError(16, "EBUSY")
if timer_id is None:
free = [t for t in range(4) if t not in _claimed_timers]
if not free:
raise OSError(16, "EBUSY")
timer_id = free[0]
self.timer_id = timer_id
_claimed_timers.add(timer_id)
self._x, self._y = 0, 0
if ratio is not None:
self.ratio = ratio
elif freq is not None:
self.freq = freq
@property
def ratio(self):
return (self._x, self._y)
@ratio.setter
def ratio(self, xy):
x, y = xy
if not (1 <= x <= 65535 and 1 <= y <= 65535 and x <= y):
raise ValueError("value out of range")
self._x, self._y = x, y
@property
def freq(self):
return SYS_CLK_HZ * self._x // self._y # read back as a whole number of hertz
@freq.setter
def freq(self, hz):
if hz > SYS_CLK_HZ or SYS_CLK_HZ // 65535 > hz:
raise ValueError("value out of range")
self._x, self._y = _best_ratio(hz, SYS_CLK_HZ)
def __int__(self):
return 0x3B + self.timer_id # the treq_sel value of this timer
def __repr__(self):
return "DMATimer(%d)" % self.timer_id
def close(self):
_claimed_timers.discard(self.timer_id)
def _best_ratio(p, q, limit=65536):
"""The fraction x/y closest to p/q with both terms below limit: continued fractions, as rp2_dma.c does."""
h, k, h0, k0 = 1, 0, 0, 1
while q:
a = p // q
h1, k1 = a * h + h0, a * k + k0
if h1 >= limit or k1 >= limit:
n = (limit - 1 - h0) // h if h else limit
if k:
n = min(n, (limit - 1 - k0) // k)
if n != limit and n > 0 and n >= (a + 1) // 2:
return n * h + h0, n * k + k0
break
p, q = q, p % q
h0, k0, h, k = h, k, h1, k1
return h, k
# ---- Flash and the BOOTSEL button ------------------------------------------------------------------------------
_FLASH_BYTES = 2 * 1024 * 1024 # a Pico's flash chip
_STORAGE_BYTES = 1441792 # MICROPY_HW_FLASH_STORAGE_BYTES of RPI_PICO: 1408 KiB at the top of the flash
_BLOCK = 4096 # one flash sector
class Flash:
"""The flash kept for files, or a part of it: the block device behind the filesystem."""
def __init__(self, *, start=-1, len=-1):
if start == -1:
start = 0
elif not (0 <= start < _STORAGE_BYTES and start % _BLOCK == 0):
raise ValueError
if len == -1:
len = _STORAGE_BYTES - start
elif not (0 < len and start + len <= _STORAGE_BYTES and len % _BLOCK == 0):
raise ValueError
self._offset = _FLASH_BYTES - _STORAGE_BYTES + start # where it starts in the flash chip
self._size = len
self._xip = 0x10000000 + self._offset # the address its buffer has, as uctypes.addressof() reports it
def ioctl(self, op, arg):
return {4: self._size // _BLOCK, 5: _BLOCK}.get(op, 0) # 4: block count, 5: block size
def bench_boot():
"""What the rp2 port's frozen _boot.py does at every start-up, before boot.py: mount Flash() at /, or format it."""
import vfs
bdev = Flash()
try:
fs = vfs.VfsLfs2(bdev, progsize=256)
print("boot: mounted the", bdev.ioctl(4, 0), "block filesystem at /")
except OSError as e:
print("boot: VfsLfs2(Flash()) raised OSError", e.args[0], "so the boot code formats all", bdev.ioctl(4, 0), "blocks")
vfs.VfsLfs2.mkfs(bdev, progsize=256)
fs = vfs.VfsLfs2(bdev, progsize=256)
vfs.mount(fs, "/")
def bootsel_button():
"""Reads the BOOTSEL button through the flash's chip-select line: 1 while it is pressed."""
global _bootsel_reads
_bootsel_reads += 1
return _button
_button, _bootsel_reads = 0, 0
BOOTSEL_PAUSE_US = 8 # MICROPY_HW_BOOTSEL_DELAY_US: the flash and interrupts stay off at least this long per read
def bench_press(pressed):
"""Hold the BOOTSEL button down (True) or let it go (False)."""
global _button
_button = 1 if pressed else 0rp2/machine.py
"""A stand-in for machine.Pin, enough for the rp2 examples beside it. Never copy this file to a board."""
FAKE = True
class Pin:
OUT, IN = 1, 0
def __init__(self, id, mode=-1, value=None):
self.id = id
def __repr__(self):
return "Pin(GPIO%d)" % self.id
def freq():
return 125_000_000Runs 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 period took 26 cycles, 13 high and 13 low, so the carrier is 988,020 Hz ÷ 26 = 38,000.8 Hz, and 24 periods
lasted 631.6 µs. Nothing in the program measured time. The numbers follow from the instructions alone, and they stay
the same while Python collects garbage or handles an interrupt, because the state machine runs beside the processor.
When the count runs out, the jump falls through, the program wraps to pull() without spending a cycle, and the
state machine waits there with the pin low until the next put().
Putting the number of periods into X
jmp(x_dec, "period") tests X before it decrements it, so a loop that starts with X = n runs n + 1 times:
sm.put(24) sends 25 periods. Put n − 1, as the program does, and write the reason in a comment.
Checkpoint
Check: Counting a PIO program's cycles
freq sets the length of a cycle, in steps of 1/256 of a system clock
A state machine cannot run at any frequency you name. StateMachine.init() turns freq into a
clock divider: A circuit that makes one cycle of a slower clock last a chosen number of cycles of a faster one. A PIO state machine's divider counts system clocks in steps of 1/256.: the rp2 port multiplies the system clock by 256 and divides by freq,
rounding down, which is 125_000_000 * 256 // freq on a Pico at its boot clock, and loads the result as a whole
number and a fraction in 256ths, the 16-bit integer and 8-bit fraction that the RP2040 datasheet gives every state
machine. The divisor runs from 1 to 65,536, so at the Pico’s 125 MHz the slowest state machine runs at
1,908 Hz, as the StateMachine documentation says, and freq=1_907 raises ValueError. The divider.py tab of the
program above asks for seven frequencies and prints what each becomes:
| freq asked | Divider | One cycle | Real frequency | System clocks a cycle |
|---|---|---|---|---|
| 10,000,000 | 12 + 128/256 | 100.0 ns | 10,000,000.00 Hz | 12 or 13 |
| 1,000,000 | 125 + 0/256 | 1000.0 ns | 1,000,000.00 Hz | 125, every cycle |
| 988,000 | 126 + 132/256 | 1012.1 ns | 988,020.25 Hz | 126 or 127 |
| 2,000 | 62500 + 0/256 | 500000.0 ns | 2,000.00 Hz | 62500, every cycle |
| 1,908 | 65513 + 160/256 | 524109.0 ns | 1,908.00 Hz | 65513 or 65514 |
| 1,907 | ValueError: freq out of range |
Two things follow. The frequency you get is close to the one you asked for, but often not equal: 988,000 became 988,020.25 Hz, because the divider keeps only whole 256ths. And a divider with a fraction does not make every cycle the same length. The datasheet’s divider is a first-order delta-sigma, so the state machine’s cycles last 126 or 127 system clocks, 1,008 ns or 1,016 ns, and only their average is 1,012.1 ns. A whole divider such as 125 makes every cycle exactly 1 µs.
So the carrier has a choice. At freq=988_000 its average is 38,000.8 Hz and an edge can move by one system clock,
8 ns, which is 0.06 % of the 13.2 µs half period. At freq=1_000_000 every edge is exact, but the carrier is
1,000,000 ÷ 26 = 38,461.5 Hz, 1.2 % fast on every period. Choose a whole divider when a protocol needs every cycle
the same, and a fraction when the average frequency matters more.
Key idea
A PIO program’s timing is its cycle count times the length of one cycle, and that length is the system clock’s period times the divider, a whole number of 256ths: count first, then divide.
Checkpoint
Check: The clock divider behind freq
DMA moves data by itself, and counts transfers, not bytes
The RP2040’s DMA: Direct memory access: hardware channels that copy data between memory and peripherals by themselves, so the processor is free while a transfer runs. controller has 12 channels. Each one reads from one address and writes to
another, again and again, adding to either address or to both after each transfer, while the processor does other
work. rp2.DMA() claims a free channel, config() gives it its addresses, a count and a control word, and
trigger=True starts it. Any object with the buffer protocol can stand for an address, a bytearray or a state
machine’s FIFO.
The control word holds the details, and pack_ctrl() builds it from named fields. On the RP2040 its default for
channel 0 is 0x3f8039: 4-byte transfers (size 2), both addresses incrementing, treq_sel 0x3f, which means as fast as the bus
allows, and chain_to the channel itself, which means no chaining. The program prints it field by field, copies 1,024
bytes, and then makes the slip that the documentation’s own example warns about in a comment:
rp2/dma_copy.py
"""A DMA copy on a Pico: the control word, a count in transfers, and the slip, on the model of rp2 beside this file."""
import rp2
src = bytearray(range(256)) * 4 # 1,024 bytes
dst = bytearray(1024)
d = rp2.DMA()
ctrl = d.pack_ctrl() # the default for this channel
print(d, "default ctrl:", hex(ctrl))
f = rp2.DMA.unpack_ctrl(ctrl)
print(" size", f["size"], "(4-byte words), inc_read", f["inc_read"], "inc_write", f["inc_write"])
print(" treq_sel", hex(f["treq_sel"]), "(no pacing), chain_to", f["chain_to"], "(itself), irq_quiet", f["irq_quiet"])
d.config(read=src, write=dst, count=len(src) // 4, ctrl=ctrl, trigger=True)
while d.active():
pass
print("count", len(src) // 4, "words: copied", dst == src)
# Bytes to the TX FIFO of PIO 0's state machine 0: one byte a transfer, the same address, paced by its request.
to_pio = d.pack_ctrl(size=0, inc_write=False, treq_sel=(0 << 3) + 0)
print("byte transfers to PIO0 SM0:", hex(to_pio))
try:
d.config(read=src, write=dst, count=len(src), ctrl=ctrl, trigger=True) # count in bytes: the slip
except rp2.BenchError as e:
print("BenchError:", e)
d.close()Output
DMA(0) default ctrl: 0x3f8039 size 2 (4-byte words), inc_read 1 inc_write 1 treq_sel 0x3f (no pacing), chain_to 0 (itself), irq_quiet 1 count 256 words: copied True byte transfers to PIO0 SM0: 0x200011 BenchError: 1024 transfers of 4 bytes move 4096 bytes, but the read buffer holds 1024 and the write buffer holds 1024: on a board the channel carries on into the memory after the buffer
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 dma_copy.py
rp2/dma_pace.py
"""DMATimer (MicroPython 1.29) paces transfers at X/Y of the 125 MHz system clock, on the model of rp2 beside this file."""
import rp2
timers = []
print("asked timer ratio (X, Y) freq read back treq_sel")
for want in (10_000, 44_100, 48_000, 1_000_000):
t = rp2.DMATimer(freq=want)
timers.append(t) # keep a reference, or the timer is freed and reused
print(f"{want:>9,} {str(t):<12} {str(t.ratio):<14} {t.freq:>10,} Hz {hex(int(t))}")
try:
rp2.DMATimer(freq=8_000)
except OSError as e:
print("a fifth timer: OSError", e)
for t in timers:
t.close()
try:
rp2.DMATimer(freq=1_000)
except ValueError as e:
print("1,000 Hz: ValueError:", e, "(the lowest is 125,000,000 // 65,535 = 1,907 Hz)")Output
asked timer ratio (X, Y) freq read back treq_sel 10,000 DMATimer(0) (1, 12500) 10,000 Hz 0x3b 44,100 DMATimer(1) (15, 42517) 44,100 Hz 0x3c 48,000 DMATimer(2) (6, 15625) 48,000 Hz 0x3d 1,000,000 DMATimer(3) (1, 125) 1,000,000 Hz 0x3e a fifth timer: OSError [Errno 16] EBUSY 1,000 Hz: ValueError: value out of range (the lowest is 125,000,000 // 65,535 = 1,907 Hz)
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 dma_pace.py
rp2.py (the model) · rp2/rp2.py
"""A model of MicroPython's rp2 module on a Raspberry Pi Pico (RP2040, 125 MHz), so rp2 code runs on a computer.
It follows the MicroPython 1.29.0 documentation and the rp2 port's source: a subset of the PIO assembler of
ports/rp2/modules/rp2.py (set, mov, pull, jmp, nop, label, wrap_target, wrap and [delay]), the clock divider of
StateMachine.init() in ports/rp2/rp2_pio.c, the DMA control fields and defaults of ports/rp2/rp2_dma.c with the
RP2040's bit positions, the DMATimer ratio search of the same file, and rp2.Flash(start, len) of
ports/rp2/rp2_flash.c. Functions whose names start with bench_ are not MicroPython: they run a state machine for a
number of cycles, or report what the hardware would have done. Never copy this file to a board.
"""
FAKE = True
SYS_CLK_HZ = 125_000_000 # machine.freq() after boot on a Pico (rp2 quick reference)
_MASK32 = 0xFFFFFFFF
class PIOASMError(Exception):
pass
class BenchError(Exception):
"""What the hardware would do silently, said out loud by the model."""
class PIO:
IN_LOW, IN_HIGH, OUT_LOW, OUT_HIGH = 0, 1, 2, 3
SHIFT_LEFT, SHIFT_RIGHT = 0, 1
JOIN_NONE, JOIN_TX, JOIN_RX = 0, 1, 2
class _Instr:
def __init__(self, op, args):
self.op, self.args, self.delay = op, args, 0
def __getitem__(self, cycles): # set(pins, 1) [12]
if not 0 <= cycles <= 31:
raise PIOASMError("delay too large")
self.delay = cycles
return self
class _Program:
def __init__(self, set_init):
self.set_init, self.instrs, self.labels = set_init, [], {}
self.wrap_target, self.wrap = 0, None
def _add(self, op, *args):
ins = _Instr(op, args)
self.instrs.append(ins)
return ins
def names(self):
"""What a PIO function sees as globals while asm_pio() assembles it."""
def set_(dest, data):
if not 0 <= data <= 31:
raise PIOASMError("set takes a value from 0 to 31")
return self._add("set", dest, data)
def jmp(cond, label=None):
return self._add("jmp", "always", cond) if label is None else self._add("jmp", cond, label)
def label(name):
self.labels[name] = len(self.instrs)
def wrap_target():
self.wrap_target = len(self.instrs)
def wrap():
self.wrap = len(self.instrs) - 1
return {
"set": set_, "jmp": jmp, "label": label, "wrap_target": wrap_target, "wrap": wrap,
"nop": lambda: self._add("nop"), "pull": lambda: self._add("pull"),
"mov": lambda dest, src: self._add("mov", dest, src),
"pins": "pins", "x": "x", "y": "y", "osr": "osr",
"x_dec": "x_dec", "y_dec": "y_dec", "not_x": "not_x", "not_y": "not_y",
}
def asm_pio(*, set_init=None, **settings):
"""Assemble the decorated function's instructions (as @rp2.asm_pio does, for the subset above)."""
def dec(f):
prog = _Program(set_init)
names = prog.names()
gl = f.__globals__
saved = {k: gl[k] for k in names if k in gl}
gl.update(names)
try:
f()
finally:
for k in names:
if k in saved:
gl[k] = saved[k]
else:
gl.pop(k, None)
if prog.wrap is None:
prog.wrap = len(prog.instrs) - 1
return prog
return dec
class StateMachine:
"""One of the RP2040's 8 state machines (2 PIO blocks of 4), running at a divided system clock."""
def __init__(self, id, program=None, freq=-1, *, set_base=None):
if not 0 <= id <= 7:
raise ValueError("invalid StateMachine")
self.id = id
if program is not None:
self.init(program, freq, set_base=set_base)
def init(self, program, freq=-1, *, set_base=None):
if len(program.instrs) > 32:
raise OSError(12, "ENOMEM") # a PIO block holds 32 instructions
if freq < 0:
div = 256 # the default: run at the system clock
elif freq == 0:
div = 65536 * 256 # a divider of 0 means 65536, the slowest
else:
div = SYS_CLK_HZ * 256 // freq # ports/rp2/rp2_pio.c: clkdiv in 1/256 steps
if not 256 <= div <= 65536 * 256:
raise ValueError("freq out of range")
self._div, self._prog = div, program
self._pc, self._delay, self._x, self._y, self._osr = 0, 0, 0, 0, 0
self._fifo, self._running = [], False
self.level = 1 if program.set_init == PIO.OUT_HIGH else 0
def active(self, value=None):
if value is not None:
self._running = bool(value)
return self._running
def put(self, value):
if len(self._fifo) == 4 and not self._running:
raise BenchError("put() would wait for ever: the 4-word TX FIFO is full and nothing pulls")
self._fifo.append(value & _MASK32)
def tx_fifo(self):
return len(self._fifo)
def __repr__(self):
return "StateMachine(%d)" % self.id
def _cycle(self):
"""One state machine cycle: an instruction, a delay cycle or a stall."""
if self._delay:
self._delay -= 1
return "delay"
ins = self._prog.instrs[self._pc]
nxt = self._pc + 1
if ins.op == "pull":
if not self._fifo:
return "stall" # a blocking pull waits; its delay starts only after it completes
self._osr = self._fifo.pop(0)
elif ins.op == "mov":
dest, src = ins.args
setattr(self, "_" + dest, getattr(self, "_" + src))
elif ins.op == "set":
dest, data = ins.args
if dest == "pins":
self.level = data & 1
else:
setattr(self, "_" + dest, data)
elif ins.op == "jmp":
cond, target = ins.args
take = True
if cond in ("x_dec", "y_dec"):
reg = "_" + cond[0]
take = getattr(self, reg) != 0 # tested before the decrement, which always happens
setattr(self, reg, (getattr(self, reg) - 1) & _MASK32)
elif cond in ("not_x", "not_y"):
take = getattr(self, "_" + cond[-1]) == 0
if take:
nxt = self._prog.labels[target]
if self._pc == self._prog.wrap and nxt == self._pc + 1:
nxt = self._prog.wrap_target # wrapping costs no cycle
self._pc, self._delay = nxt, ins.delay
return ins.op
def bench_divider(sm):
"""The (whole, frac) clock divider the port set: one cycle lasts whole + frac/256 system clocks."""
return sm._div // 256, sm._div % 256
def bench_cycle_ns(sm):
"""The average length of one state machine cycle in nanoseconds."""
return sm._div / 256 * 1e9 / SYS_CLK_HZ
def bench_run(sm, cycles):
"""Run sm for a number of cycles; return what each cycle did and the pin level during it."""
if not sm._running:
raise BenchError("the state machine is not active")
return [(sm._cycle(), sm.level) for _ in range(cycles)]
# ---- DMA ------------------------------------------------------------------------------------------------------
# (name, lowest bit, width) of the RP2040's CH0_CTRL_TRIG register, in the order rp2_dma.c lists them.
_CTRL_FIELDS = (
("enable", 0, 1), ("high_pri", 1, 1), ("size", 2, 2), ("inc_read", 4, 1), ("inc_write", 5, 1),
("ring_size", 6, 4), ("ring_sel", 10, 1), ("chain_to", 11, 4), ("treq_sel", 15, 6), ("irq_quiet", 21, 1),
("bswap", 22, 1), ("sniff_en", 23, 1), ("busy", 24, 1), ("write_err", 29, 1), ("read_err", 30, 1),
("ahb_err", 31, 1),
)
_READ_ONLY = ("busy", "ahb_err")
_PERMANENT = 0x3F # treq_sel: no pacing, run at full speed
_claimed_channels, _claimed_timers = set(), set()
class DMA:
"""One of the RP2040's 12 DMA channels, claimed for this object."""
def __init__(self):
free = [c for c in range(12) if c not in _claimed_channels]
if not free:
raise OSError(16, "EBUSY")
self.channel = free[0]
_claimed_channels.add(self.channel)
self._busy = False
def __repr__(self):
return "DMA(%d)" % self.channel
def pack_ctrl(self, default=None, **fields):
value = default
if value is None: # quiet, unpaced, both addresses incrementing, 4-byte words, enabled, chained to itself
value = 1 << 21 | _PERMANENT << 15 | 1 << 5 | 1 << 4 | 2 << 2 | 1 | (self.channel & 0xF) << 11
known = {name: (shift, width) for name, shift, width in _CTRL_FIELDS}
for name, v in fields.items():
if name not in known:
raise TypeError("unexpected keyword argument '%s'" % name)
if name in _READ_ONLY:
continue
shift, width = known[name]
mask = (1 << width) - 1
v = int(v)
if v & mask != v:
raise ValueError("bad field value")
value = value & ~(mask << shift) | v << shift
return value
@staticmethod
def unpack_ctrl(value):
return {name: value >> shift & (1 << width) - 1 for name, shift, width in _CTRL_FIELDS}
def config(self, read=None, write=None, count=None, ctrl=None, trigger=False):
self._read, self._write, self._count, self._ctrl = read, write, count, ctrl
if trigger:
self._run()
def _run(self):
f = DMA.unpack_ctrl(self._ctrl)
width = 1 << f["size"]
total = self._count * width
short = [(name, len(buf)) for name, buf, inc in (("read", self._read, f["inc_read"]),
("write", self._write, f["inc_write"]))
if inc and total > len(buf)]
if short:
raise BenchError(
"%d transfers of %d bytes move %d bytes, but %s: on a board the channel carries on into the memory "
"after the buffer" % (self._count, width, total,
" and ".join("the %s buffer holds %d" % s for s in short))
)
if f["inc_read"] and f["inc_write"]:
self._write[:total] = self._read[:total]
self._busy = False
def active(self, value=None):
return self._busy
def close(self):
_claimed_channels.discard(self.channel)
class DMATimer:
"""One of the 4 DMA pacing timers: it asks for a transfer X/Y times per system clock cycle."""
def __init__(self, timer_id=None, *, freq=None, ratio=None):
if timer_id is not None and (not 0 <= timer_id <= 3 or timer_id in _claimed_timers):
raise ValueError("value out of range") if not 0 <= timer_id <= 3 else OSError(16, "EBUSY")
if timer_id is None:
free = [t for t in range(4) if t not in _claimed_timers]
if not free:
raise OSError(16, "EBUSY")
timer_id = free[0]
self.timer_id = timer_id
_claimed_timers.add(timer_id)
self._x, self._y = 0, 0
if ratio is not None:
self.ratio = ratio
elif freq is not None:
self.freq = freq
@property
def ratio(self):
return (self._x, self._y)
@ratio.setter
def ratio(self, xy):
x, y = xy
if not (1 <= x <= 65535 and 1 <= y <= 65535 and x <= y):
raise ValueError("value out of range")
self._x, self._y = x, y
@property
def freq(self):
return SYS_CLK_HZ * self._x // self._y # read back as a whole number of hertz
@freq.setter
def freq(self, hz):
if hz > SYS_CLK_HZ or SYS_CLK_HZ // 65535 > hz:
raise ValueError("value out of range")
self._x, self._y = _best_ratio(hz, SYS_CLK_HZ)
def __int__(self):
return 0x3B + self.timer_id # the treq_sel value of this timer
def __repr__(self):
return "DMATimer(%d)" % self.timer_id
def close(self):
_claimed_timers.discard(self.timer_id)
def _best_ratio(p, q, limit=65536):
"""The fraction x/y closest to p/q with both terms below limit: continued fractions, as rp2_dma.c does."""
h, k, h0, k0 = 1, 0, 0, 1
while q:
a = p // q
h1, k1 = a * h + h0, a * k + k0
if h1 >= limit or k1 >= limit:
n = (limit - 1 - h0) // h if h else limit
if k:
n = min(n, (limit - 1 - k0) // k)
if n != limit and n > 0 and n >= (a + 1) // 2:
return n * h + h0, n * k + k0
break
p, q = q, p % q
h0, k0, h, k = h, k, h1, k1
return h, k
# ---- Flash and the BOOTSEL button ------------------------------------------------------------------------------
_FLASH_BYTES = 2 * 1024 * 1024 # a Pico's flash chip
_STORAGE_BYTES = 1441792 # MICROPY_HW_FLASH_STORAGE_BYTES of RPI_PICO: 1408 KiB at the top of the flash
_BLOCK = 4096 # one flash sector
class Flash:
"""The flash kept for files, or a part of it: the block device behind the filesystem."""
def __init__(self, *, start=-1, len=-1):
if start == -1:
start = 0
elif not (0 <= start < _STORAGE_BYTES and start % _BLOCK == 0):
raise ValueError
if len == -1:
len = _STORAGE_BYTES - start
elif not (0 < len and start + len <= _STORAGE_BYTES and len % _BLOCK == 0):
raise ValueError
self._offset = _FLASH_BYTES - _STORAGE_BYTES + start # where it starts in the flash chip
self._size = len
self._xip = 0x10000000 + self._offset # the address its buffer has, as uctypes.addressof() reports it
def ioctl(self, op, arg):
return {4: self._size // _BLOCK, 5: _BLOCK}.get(op, 0) # 4: block count, 5: block size
def bench_boot():
"""What the rp2 port's frozen _boot.py does at every start-up, before boot.py: mount Flash() at /, or format it."""
import vfs
bdev = Flash()
try:
fs = vfs.VfsLfs2(bdev, progsize=256)
print("boot: mounted the", bdev.ioctl(4, 0), "block filesystem at /")
except OSError as e:
print("boot: VfsLfs2(Flash()) raised OSError", e.args[0], "so the boot code formats all", bdev.ioctl(4, 0), "blocks")
vfs.VfsLfs2.mkfs(bdev, progsize=256)
fs = vfs.VfsLfs2(bdev, progsize=256)
vfs.mount(fs, "/")
def bootsel_button():
"""Reads the BOOTSEL button through the flash's chip-select line: 1 while it is pressed."""
global _bootsel_reads
_bootsel_reads += 1
return _button
_button, _bootsel_reads = 0, 0
BOOTSEL_PAUSE_US = 8 # MICROPY_HW_BOOTSEL_DELAY_US: the flash and interrupts stay off at least this long per read
def bench_press(pressed):
"""Hold the BOOTSEL button down (True) or let it go (False)."""
global _button
_button = 1 if pressed else 0Runs 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
count is a number of transfers, and with the default size of 2 each transfer moves 4 bytes, so 1,024 bytes are
256 transfers. The last call passed len(src), 1,024 transfers of 4 bytes, 4,096 bytes. The model refused, but a
board raises nothing: the channel reads past src and writes 3,072 bytes into whatever follows dst in RAM, and the
program fails later, somewhere unrelated.
Passing a length in bytes as count
d.config(read=src, write=dst, count=len(src), ctrl=d.pack_ctrl(), trigger=True) moves four times too much. Write
count=len(src) // 4 for the default word size, or count=len(src) with pack_ctrl(size=0) for byte transfers.
Checkpoint
Check: DMA transfers and their count
A request paces each transfer: a state machine’s DREQ, or a DMATimer
At treq_sel 0x3f a channel copies as fast as the bus allows, which suits memory to memory. A peripheral needs data
at its own pace, so treq_sel names who asks for each transfer: a DREQ: Data request: the signal by which a peripheral tells a DMA channel that it can take or give one more item, so the channel transfers at the peripheral's pace.. On the RP2040 the
transmit FIFO of PIO block p, state machine n, asks with DREQ (p << 3) + n, the formula of the DMA documentation.
To feed bytes to PIO 0’s state machine 0, the program’s middle step packs size=0 for one byte a transfer,
inc_write=False because the FIFO’s address never moves, and treq_sel=0, which gives 0x200011. Without the
request the channel would write faster than the program pulls, and the RP2040 datasheet says a write to a full FIFO
has no effect: those bytes would be lost. Follow a paced transfer:
The drawing shows the whole scene; its 6 steps are listed below.
Every step as text
- Configure and start. The program gives the channel src as the read address, the state machine as the write address, a count of 1,024 transfers and the control word 0x200011 (bytes, a fixed write address, treq_sel 0), and starts it.
- A request. The FIFO has room, so PIO 0's state machine 0 raises its data request, DREQ 0. Only now may the channel move one item.
- One byte moves. The channel reads one byte from src, adds 1 to its read address, and writes the byte to the FIFO's address, which stays the same. The count drops by one.
- The FIFO fills. When the FIFO holds 4 words the request drops and the channel waits. Unpaced, it would keep writing, and the datasheet says a write to a full FIFO has no effect, so those bytes would be lost.
- pull() sets the pace. Each pull() of the state machine frees a word, the request rises again and the next byte moves. The transfer runs exactly as fast as the PIO program consumes.
- Count 0. After 1,024 transfers the count is 0 and the channel stops. The CPU copied nothing; it polls active(), which now returns False, or takes the channel's interrupt.
Memory and peripherals without a request of their own can be paced by one of the DMA controller’s four pacing
timers, which MicroPython 1.29 opens as rp2.DMATimer. A timer asks for a transfer X/Y times per system clock cycle,
and DMATimer(freq=...) finds the closest fraction with both terms below 65,536. The dma_pace.py tab asked for four
rates: 44,100 Hz became (15, 42517), which reads back as 44,100 Hz because 125,000,000 × 15 ÷ 42,517 =
44,100.007, and int(timer) is the treq_sel value to pass, 0x3b to 0x3e. A fifth timer raised
OSError: [Errno 16] EBUSY, and below 1,907 Hz, 1/65,535 of the clock, ValueError.
Keep the timer object
The documentation warns that a pacing timer is released when its DMATimer object is garbage collected. Keep a
reference for as long as the transfer runs, or another caller may claim the timer and change its rate.
Checkpoint
Check: Pacing a channel with a DREQ or a DMATimer
rp2.Flash and the BOOTSEL button pause the flash, so use them rarely
rp2.Flash() is the block device: An object that stores data in numbered blocks of one size and offers readblocks(), writeblocks() and ioctl() for the block count, the block size and erasing. MicroPython's filesystems sit on one, such as rp2.Flash or an SD card. More in “Filesystems: LittleFS, FAT and vfs.mount” that the board’s start-up code mounts at / as
LittleFS: on a Pico, the 352 blocks of 4,096 bytes kept for files, as
the filesystems lesson showed. Its keyword arguments start
and len, both multiples of 4,096 inside that area, hand out a part of it. The program below takes two parts, then
tries to keep the second one free for raw records:
rp2/flash_split.py
"""rp2.Flash(start, len) on a Pico, and what the boot code does to a split, on the models of rp2, uctypes and vfs beside this file."""
import rp2
import uctypes
import vfs
whole = rp2.Flash() # what the board's boot code mounts at /
print("Flash():", whole.ioctl(4, 0), "blocks of", whole.ioctl(5, 0), "bytes, read at", hex(uctypes.addressof(whole)))
files = rp2.Flash(len=1024 * 1024) # 1 MiB for a filesystem
log = rp2.Flash(start=1024 * 1024) # the rest for raw records
print("Flash(len=1 MiB):", files.ioctl(4, 0), "blocks from", hex(uctypes.addressof(files)))
print("Flash(start=1 MiB):", log.ioctl(4, 0), "blocks from", hex(uctypes.addressof(log)))
for kwargs in ({"start": 1000}, {"len": 2 * 1024 * 1024}, {"start": 4096, "len": 6000}):
try:
rp2.Flash(**kwargs)
except ValueError:
print("Flash(%s): ValueError" % ", ".join("%s=%d" % kv for kv in kwargs.items()))
rp2.bench_boot() # the first start-up of a board with stock firmware
rp2.bench_boot() # every start-up after it
vfs.umount("/")
vfs.VfsLfs2.mkfs(files, progsize=256) # a smaller filesystem, to keep the log area free
print("mkfs on Flash(len=1 MiB): a", files.ioctl(4, 0), "block filesystem")
rp2.bench_boot() # the next start-upOutput
Flash(): 352 blocks of 4096 bytes, read at 0x100a0000 Flash(len=1 MiB): 256 blocks from 0x100a0000 Flash(start=1 MiB): 96 blocks from 0x101a0000 Flash(start=1000): ValueError Flash(len=2097152): ValueError Flash(start=4096, len=6000): ValueError boot: VfsLfs2(Flash()) raised OSError 22 so the boot code formats all 352 blocks boot: mounted the 352 block filesystem at / mkfs on Flash(len=1 MiB): a 256 block filesystem boot: VfsLfs2(Flash()) raised OSError 22 so the boot code formats all 352 blocks
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 flash_split.py
rp2.py (the model) · rp2/rp2.py
"""A model of MicroPython's rp2 module on a Raspberry Pi Pico (RP2040, 125 MHz), so rp2 code runs on a computer.
It follows the MicroPython 1.29.0 documentation and the rp2 port's source: a subset of the PIO assembler of
ports/rp2/modules/rp2.py (set, mov, pull, jmp, nop, label, wrap_target, wrap and [delay]), the clock divider of
StateMachine.init() in ports/rp2/rp2_pio.c, the DMA control fields and defaults of ports/rp2/rp2_dma.c with the
RP2040's bit positions, the DMATimer ratio search of the same file, and rp2.Flash(start, len) of
ports/rp2/rp2_flash.c. Functions whose names start with bench_ are not MicroPython: they run a state machine for a
number of cycles, or report what the hardware would have done. Never copy this file to a board.
"""
FAKE = True
SYS_CLK_HZ = 125_000_000 # machine.freq() after boot on a Pico (rp2 quick reference)
_MASK32 = 0xFFFFFFFF
class PIOASMError(Exception):
pass
class BenchError(Exception):
"""What the hardware would do silently, said out loud by the model."""
class PIO:
IN_LOW, IN_HIGH, OUT_LOW, OUT_HIGH = 0, 1, 2, 3
SHIFT_LEFT, SHIFT_RIGHT = 0, 1
JOIN_NONE, JOIN_TX, JOIN_RX = 0, 1, 2
class _Instr:
def __init__(self, op, args):
self.op, self.args, self.delay = op, args, 0
def __getitem__(self, cycles): # set(pins, 1) [12]
if not 0 <= cycles <= 31:
raise PIOASMError("delay too large")
self.delay = cycles
return self
class _Program:
def __init__(self, set_init):
self.set_init, self.instrs, self.labels = set_init, [], {}
self.wrap_target, self.wrap = 0, None
def _add(self, op, *args):
ins = _Instr(op, args)
self.instrs.append(ins)
return ins
def names(self):
"""What a PIO function sees as globals while asm_pio() assembles it."""
def set_(dest, data):
if not 0 <= data <= 31:
raise PIOASMError("set takes a value from 0 to 31")
return self._add("set", dest, data)
def jmp(cond, label=None):
return self._add("jmp", "always", cond) if label is None else self._add("jmp", cond, label)
def label(name):
self.labels[name] = len(self.instrs)
def wrap_target():
self.wrap_target = len(self.instrs)
def wrap():
self.wrap = len(self.instrs) - 1
return {
"set": set_, "jmp": jmp, "label": label, "wrap_target": wrap_target, "wrap": wrap,
"nop": lambda: self._add("nop"), "pull": lambda: self._add("pull"),
"mov": lambda dest, src: self._add("mov", dest, src),
"pins": "pins", "x": "x", "y": "y", "osr": "osr",
"x_dec": "x_dec", "y_dec": "y_dec", "not_x": "not_x", "not_y": "not_y",
}
def asm_pio(*, set_init=None, **settings):
"""Assemble the decorated function's instructions (as @rp2.asm_pio does, for the subset above)."""
def dec(f):
prog = _Program(set_init)
names = prog.names()
gl = f.__globals__
saved = {k: gl[k] for k in names if k in gl}
gl.update(names)
try:
f()
finally:
for k in names:
if k in saved:
gl[k] = saved[k]
else:
gl.pop(k, None)
if prog.wrap is None:
prog.wrap = len(prog.instrs) - 1
return prog
return dec
class StateMachine:
"""One of the RP2040's 8 state machines (2 PIO blocks of 4), running at a divided system clock."""
def __init__(self, id, program=None, freq=-1, *, set_base=None):
if not 0 <= id <= 7:
raise ValueError("invalid StateMachine")
self.id = id
if program is not None:
self.init(program, freq, set_base=set_base)
def init(self, program, freq=-1, *, set_base=None):
if len(program.instrs) > 32:
raise OSError(12, "ENOMEM") # a PIO block holds 32 instructions
if freq < 0:
div = 256 # the default: run at the system clock
elif freq == 0:
div = 65536 * 256 # a divider of 0 means 65536, the slowest
else:
div = SYS_CLK_HZ * 256 // freq # ports/rp2/rp2_pio.c: clkdiv in 1/256 steps
if not 256 <= div <= 65536 * 256:
raise ValueError("freq out of range")
self._div, self._prog = div, program
self._pc, self._delay, self._x, self._y, self._osr = 0, 0, 0, 0, 0
self._fifo, self._running = [], False
self.level = 1 if program.set_init == PIO.OUT_HIGH else 0
def active(self, value=None):
if value is not None:
self._running = bool(value)
return self._running
def put(self, value):
if len(self._fifo) == 4 and not self._running:
raise BenchError("put() would wait for ever: the 4-word TX FIFO is full and nothing pulls")
self._fifo.append(value & _MASK32)
def tx_fifo(self):
return len(self._fifo)
def __repr__(self):
return "StateMachine(%d)" % self.id
def _cycle(self):
"""One state machine cycle: an instruction, a delay cycle or a stall."""
if self._delay:
self._delay -= 1
return "delay"
ins = self._prog.instrs[self._pc]
nxt = self._pc + 1
if ins.op == "pull":
if not self._fifo:
return "stall" # a blocking pull waits; its delay starts only after it completes
self._osr = self._fifo.pop(0)
elif ins.op == "mov":
dest, src = ins.args
setattr(self, "_" + dest, getattr(self, "_" + src))
elif ins.op == "set":
dest, data = ins.args
if dest == "pins":
self.level = data & 1
else:
setattr(self, "_" + dest, data)
elif ins.op == "jmp":
cond, target = ins.args
take = True
if cond in ("x_dec", "y_dec"):
reg = "_" + cond[0]
take = getattr(self, reg) != 0 # tested before the decrement, which always happens
setattr(self, reg, (getattr(self, reg) - 1) & _MASK32)
elif cond in ("not_x", "not_y"):
take = getattr(self, "_" + cond[-1]) == 0
if take:
nxt = self._prog.labels[target]
if self._pc == self._prog.wrap and nxt == self._pc + 1:
nxt = self._prog.wrap_target # wrapping costs no cycle
self._pc, self._delay = nxt, ins.delay
return ins.op
def bench_divider(sm):
"""The (whole, frac) clock divider the port set: one cycle lasts whole + frac/256 system clocks."""
return sm._div // 256, sm._div % 256
def bench_cycle_ns(sm):
"""The average length of one state machine cycle in nanoseconds."""
return sm._div / 256 * 1e9 / SYS_CLK_HZ
def bench_run(sm, cycles):
"""Run sm for a number of cycles; return what each cycle did and the pin level during it."""
if not sm._running:
raise BenchError("the state machine is not active")
return [(sm._cycle(), sm.level) for _ in range(cycles)]
# ---- DMA ------------------------------------------------------------------------------------------------------
# (name, lowest bit, width) of the RP2040's CH0_CTRL_TRIG register, in the order rp2_dma.c lists them.
_CTRL_FIELDS = (
("enable", 0, 1), ("high_pri", 1, 1), ("size", 2, 2), ("inc_read", 4, 1), ("inc_write", 5, 1),
("ring_size", 6, 4), ("ring_sel", 10, 1), ("chain_to", 11, 4), ("treq_sel", 15, 6), ("irq_quiet", 21, 1),
("bswap", 22, 1), ("sniff_en", 23, 1), ("busy", 24, 1), ("write_err", 29, 1), ("read_err", 30, 1),
("ahb_err", 31, 1),
)
_READ_ONLY = ("busy", "ahb_err")
_PERMANENT = 0x3F # treq_sel: no pacing, run at full speed
_claimed_channels, _claimed_timers = set(), set()
class DMA:
"""One of the RP2040's 12 DMA channels, claimed for this object."""
def __init__(self):
free = [c for c in range(12) if c not in _claimed_channels]
if not free:
raise OSError(16, "EBUSY")
self.channel = free[0]
_claimed_channels.add(self.channel)
self._busy = False
def __repr__(self):
return "DMA(%d)" % self.channel
def pack_ctrl(self, default=None, **fields):
value = default
if value is None: # quiet, unpaced, both addresses incrementing, 4-byte words, enabled, chained to itself
value = 1 << 21 | _PERMANENT << 15 | 1 << 5 | 1 << 4 | 2 << 2 | 1 | (self.channel & 0xF) << 11
known = {name: (shift, width) for name, shift, width in _CTRL_FIELDS}
for name, v in fields.items():
if name not in known:
raise TypeError("unexpected keyword argument '%s'" % name)
if name in _READ_ONLY:
continue
shift, width = known[name]
mask = (1 << width) - 1
v = int(v)
if v & mask != v:
raise ValueError("bad field value")
value = value & ~(mask << shift) | v << shift
return value
@staticmethod
def unpack_ctrl(value):
return {name: value >> shift & (1 << width) - 1 for name, shift, width in _CTRL_FIELDS}
def config(self, read=None, write=None, count=None, ctrl=None, trigger=False):
self._read, self._write, self._count, self._ctrl = read, write, count, ctrl
if trigger:
self._run()
def _run(self):
f = DMA.unpack_ctrl(self._ctrl)
width = 1 << f["size"]
total = self._count * width
short = [(name, len(buf)) for name, buf, inc in (("read", self._read, f["inc_read"]),
("write", self._write, f["inc_write"]))
if inc and total > len(buf)]
if short:
raise BenchError(
"%d transfers of %d bytes move %d bytes, but %s: on a board the channel carries on into the memory "
"after the buffer" % (self._count, width, total,
" and ".join("the %s buffer holds %d" % s for s in short))
)
if f["inc_read"] and f["inc_write"]:
self._write[:total] = self._read[:total]
self._busy = False
def active(self, value=None):
return self._busy
def close(self):
_claimed_channels.discard(self.channel)
class DMATimer:
"""One of the 4 DMA pacing timers: it asks for a transfer X/Y times per system clock cycle."""
def __init__(self, timer_id=None, *, freq=None, ratio=None):
if timer_id is not None and (not 0 <= timer_id <= 3 or timer_id in _claimed_timers):
raise ValueError("value out of range") if not 0 <= timer_id <= 3 else OSError(16, "EBUSY")
if timer_id is None:
free = [t for t in range(4) if t not in _claimed_timers]
if not free:
raise OSError(16, "EBUSY")
timer_id = free[0]
self.timer_id = timer_id
_claimed_timers.add(timer_id)
self._x, self._y = 0, 0
if ratio is not None:
self.ratio = ratio
elif freq is not None:
self.freq = freq
@property
def ratio(self):
return (self._x, self._y)
@ratio.setter
def ratio(self, xy):
x, y = xy
if not (1 <= x <= 65535 and 1 <= y <= 65535 and x <= y):
raise ValueError("value out of range")
self._x, self._y = x, y
@property
def freq(self):
return SYS_CLK_HZ * self._x // self._y # read back as a whole number of hertz
@freq.setter
def freq(self, hz):
if hz > SYS_CLK_HZ or SYS_CLK_HZ // 65535 > hz:
raise ValueError("value out of range")
self._x, self._y = _best_ratio(hz, SYS_CLK_HZ)
def __int__(self):
return 0x3B + self.timer_id # the treq_sel value of this timer
def __repr__(self):
return "DMATimer(%d)" % self.timer_id
def close(self):
_claimed_timers.discard(self.timer_id)
def _best_ratio(p, q, limit=65536):
"""The fraction x/y closest to p/q with both terms below limit: continued fractions, as rp2_dma.c does."""
h, k, h0, k0 = 1, 0, 0, 1
while q:
a = p // q
h1, k1 = a * h + h0, a * k + k0
if h1 >= limit or k1 >= limit:
n = (limit - 1 - h0) // h if h else limit
if k:
n = min(n, (limit - 1 - k0) // k)
if n != limit and n > 0 and n >= (a + 1) // 2:
return n * h + h0, n * k + k0
break
p, q = q, p % q
h0, k0, h, k = h, k, h1, k1
return h, k
# ---- Flash and the BOOTSEL button ------------------------------------------------------------------------------
_FLASH_BYTES = 2 * 1024 * 1024 # a Pico's flash chip
_STORAGE_BYTES = 1441792 # MICROPY_HW_FLASH_STORAGE_BYTES of RPI_PICO: 1408 KiB at the top of the flash
_BLOCK = 4096 # one flash sector
class Flash:
"""The flash kept for files, or a part of it: the block device behind the filesystem."""
def __init__(self, *, start=-1, len=-1):
if start == -1:
start = 0
elif not (0 <= start < _STORAGE_BYTES and start % _BLOCK == 0):
raise ValueError
if len == -1:
len = _STORAGE_BYTES - start
elif not (0 < len and start + len <= _STORAGE_BYTES and len % _BLOCK == 0):
raise ValueError
self._offset = _FLASH_BYTES - _STORAGE_BYTES + start # where it starts in the flash chip
self._size = len
self._xip = 0x10000000 + self._offset # the address its buffer has, as uctypes.addressof() reports it
def ioctl(self, op, arg):
return {4: self._size // _BLOCK, 5: _BLOCK}.get(op, 0) # 4: block count, 5: block size
def bench_boot():
"""What the rp2 port's frozen _boot.py does at every start-up, before boot.py: mount Flash() at /, or format it."""
import vfs
bdev = Flash()
try:
fs = vfs.VfsLfs2(bdev, progsize=256)
print("boot: mounted the", bdev.ioctl(4, 0), "block filesystem at /")
except OSError as e:
print("boot: VfsLfs2(Flash()) raised OSError", e.args[0], "so the boot code formats all", bdev.ioctl(4, 0), "blocks")
vfs.VfsLfs2.mkfs(bdev, progsize=256)
fs = vfs.VfsLfs2(bdev, progsize=256)
vfs.mount(fs, "/")
def bootsel_button():
"""Reads the BOOTSEL button through the flash's chip-select line: 1 while it is pressed."""
global _bootsel_reads
_bootsel_reads += 1
return _button
_button, _bootsel_reads = 0, 0
BOOTSEL_PAUSE_US = 8 # MICROPY_HW_BOOTSEL_DELAY_US: the flash and interrupts stay off at least this long per read
def bench_press(pressed):
"""Hold the BOOTSEL button down (True) or let it go (False)."""
global _button
_button = 1 if pressed else 0uctypes.py (the model) · rp2/uctypes.py
"""A stand-in for uctypes.addressof(), enough for the rp2 examples beside it. Never copy this file to a board."""
FAKE = True
def addressof(obj):
"""The address of an object's buffer: for rp2.Flash, where the processor reads its first byte through XIP."""
address = getattr(obj, "_xip", None)
if address is None:
raise TypeError("object with buffer protocol required")
return addressvfs.py (the model) · rp2/vfs.py
"""A stand-in for vfs.VfsLfs2, enough to show what a mount checks. Never copy this file to a board.
LittleFS keeps the block count of the device it was made on in its superblock, at the start of the device, and its
mount refuses a superblock whose block count differs from the device's (lib/littlefs/lfs2.c, "Invalid block count").
"""
FAKE = True
_superblocks = {} # flash offset -> the block count written there by mkfs
_mounted = {}
class VfsLfs2:
def __init__(self, bdev, progsize=256):
made = _superblocks.get(bdev._offset)
if made != bdev.ioctl(4, 0):
raise OSError(22, "EINVAL") # no filesystem here, or one of another size
self.bdev = bdev
@staticmethod
def mkfs(bdev, progsize=256):
for start in [o for o in _superblocks if bdev._offset <= o < bdev._offset + bdev._size]:
del _superblocks[start] # formatting erases what was there
_superblocks[bdev._offset] = bdev.ioctl(4, 0)
def mount(fs, path):
_mounted[path] = fs
def umount(path):
_mounted.pop(path, None)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 area for files starts 640 KiB into the flash, at offset 0xa0000, and the processor reads it at address
0x100a0000, the address uctypes.addressof() printed. Flash(start=1024 * 1024) is the last 96 blocks, from
0x101a0000, Flash(len=1024 * 1024) the first 256, and a start or length that is not a whole number of blocks, or
that runs past the area, raises ValueError.
The split itself failed, and on stock firmware it always does. The rp2 port’s frozen _boot.py mounts Flash(), all
352 blocks, before boot.py runs, and formats them if the mount fails. LittleFS refuses a superblock whose block count
differs from the device’s, so the 256-block filesystem made by mkfs raised OSError 22 at the next start-up, and
the boot code formatted the whole area, files and all. A raw area beside the files needs firmware built with a
smaller MICROPY_HW_FLASH_STORAGE_BYTES or its own boot code; on stock firmware, keep raw records in a file.
Writing to the flash has a cost of its own. The program runs from that same flash, read through
XIP: Execute in place: running a program straight from external flash, read through a cache. While that flash is erased or written, no code can be read from it., and while a sector is erased or programmed the port turns interrupts off and pauses the
other core. PIO state machines, and DMA transfers that do not read the flash, keep running, but every Python callback
waits. bootsel_button() costs a pause too, because the BOOTSEL button sits on the flash’s chip-select line:
The drawing shows the whole scene; its 5 steps are listed below.
Every step as text
- Interrupts off. The BOOTSEL button shares the flash chip's select line, QSPI_SS. Code is read from that flash, so the port first turns interrupts off and holds the other core, which might be running from flash.
- The line is let go. The port stops driving QSPI_SS, so the button alone decides its level. From here until the end, nothing can be read from the flash.
- The wait. A busy loop that runs from RAM waits more than 8 µs, MICROPY_HW_BOOTSEL_DELAY_US, for the line to settle. Timer and pin interrupts that arrive now wait as well.
- The read. A pressed button pulls the line low, so the function will return 1 for low and 0 for high, as the rp2 documentation says.
- Back to normal. The port drives QSPI_SS again, ends the atomic section and returns. At 1,000 calls a second this costs at least 8 ms of every second; at a few calls a second it costs nothing you could see.
Each call holds the flash away for at least 8 µs, MICROPY_HW_BOOTSEL_DELAY_US, with interrupts off. Polled in a
loop 1,000 times a second, that is at least 8 ms of every second with no interrupt served; read a few times a
second, it is nothing.
Checkpoint
Check: rp2.Flash parts and the BOOTSEL button
The RP2350 keeps the API and adds blocks, channels and cores
The Pico 2 and Pico 2 W use the RP2350, and the same rp2 module drives it. The PIO tutorial calls basic PIO
programs fully compatible between the two chips, with StateMachine ids from 0 to 11 on the RP2350:
| Feature | RP2040: Pico, Pico W | RP2350: Pico 2, Pico 2 W |
|---|---|---|
| Processors | 2 Arm Cortex-M0+, 133 MHz | 2 Arm Cortex-M33 or 2 Hazard3 RISC-V, 150 MHz |
| SRAM | 264 KB | 520 KB |
| PIO | 2 blocks, 8 state machines | 3 blocks, 12 state machines |
| DMA | 12 channels, 4 pacing timers | 16 channels, 4 pacing timers |
| GPIO | 30 | 30 (RP2350A) or 48 (RP2350B) |
| PWM channels | 16 | 24 |
| Inline assembler | asm_thumb |
asm_thumb, or asm_rv32 in RISC-V mode |
| Firmware | RPI_PICO, RPI_PICO_W | RPI_PICO2, RPI_PICO2_W, each with a RISCV variant |
| Flash kept for files | 1,408 KiB on a Pico | 3 MiB on a Pico 2 |
The chip runs either pair of processors, chosen by the firmware: the RISCV variant of RPI_PICO2 or RPI_PICO2_W starts
the Hazard3 cores, and since MicroPython 1.25 it has the @micropython.asm_rv32 inline assembler in place of asm_thumb. The
native and viper emitters compile for whichever cores run. The tutorial also warns that the two chips reset their
GPIO pins differently, so a program that relies on a pin’s state after reset may need changes: set every pin a
program uses, as the pins lesson recommends anyway.
One codebase for both chips
Ask the hardware instead of guessing the board: sys.implementation._machine names the board and its chip, and
rp2.StateMachine(8) raises ValueError on an RP2040, which has no third PIO block.
Checkpoint
Check: What the RP2350 adds
Practise: a bit-timing table for a PIO encoder
A pixel chip that reads a 1 as a long pulse and a 0 as a short one is driven by a PIO program of fixed cycle counts
per bit, like the burst above. Before such a program goes near a board, its timing can be worked out on paper. Write
the two functions of pio_timing.py: the divider the port sets for a freq, and the high and low time of every bit
of some bytes:
Exercise · Medium · Python
Work out a PIO encoder's divider and bit timings
Write the two functions of pio_timing.py, which work out on a computer what a PIO encoder will send.
divider(freq, sys_hz=125_000_000) returns the clock divider that rp2.StateMachine(freq=freq) sets, as a pair (whole, frac). The port computes value = sys_hz * 256 // freq, refuses a value below 256 or above 65536 * 256 with ValueError, and splits it into whole = value // 256 and frac = value % 256.
bit_pulses(data, one_high, zero_high, bit_cycles, freq, sys_hz=125_000_000) returns one (high_ns, low_ns) pair for every bit of the bytes in data, most significant bit first. A 1 stays high for one_high cycles and a 0 for zero_high cycles, and every bit lasts bit_cycles cycles. One cycle lasts value / 256 system clock periods, the divider's average; round each time to the nearest nanosecond with round(). Unless 0 < zero_high < one_high < bit_cycles, raise ValueError.
Starter code · pio_timing.py
SYS_HZ = 125_000_000 # the Pico's system clock after boot
def divider(freq, sys_hz=SYS_HZ):
"""The (whole, frac) clock divider StateMachine(freq=freq) sets."""
return (0, 0)
def bit_pulses(data, one_high, zero_high, bit_cycles, freq, sys_hz=SYS_HZ):
"""One (high_ns, low_ns) pair per bit of data, most significant bit first."""
return []The sample tests · test_pio_timing.py
from pio_timing import bit_pulses, divider
def raises(fn, *args):
try:
fn(*args)
except ValueError:
return True
return False
def test_whole_divider():
"""freq=2_000 is a whole divider of 62,500"""
assert divider(2_000) == (62500, 0)
def test_fractional_divider():
"""freq=988_000 is 126 + 132/256, as in the lesson"""
assert divider(988_000) == (126, 132)
def test_divider_range():
"""refuses a freq below 1,908 Hz or above the system clock"""
assert raises(divider, 1_907)
assert raises(divider, 200_000_000)
assert divider(1_908) == (65513, 160)
def test_one_byte():
"""0x80 at 8 MHz: a long pulse, then seven short ones"""
assert bit_pulses(b"\x80", 7, 3, 10, 8_000_000) == [(875, 375)] + [(375, 875)] * 7
def test_bit_order():
"""two bytes, most significant bit first"""
pulses = bit_pulses(b"\x01\x40", 7, 3, 10, 8_000_000)
assert len(pulses) == 16
assert [p[0] for p in pulses].count(875) == 2
assert pulses[7] == (875, 375) and pulses[9] == (875, 375)
def test_fractional_cycle():
"""at 988 kHz one cycle averages 1012.125 ns"""
assert bit_pulses(b"\xff", 3, 1, 4, 988_000)[0] == (3036, 1012)
def test_bad_timings():
"""refuses pulses that do not fit the bit"""
assert raises(bit_pulses, b"\x00", 3, 3, 10, 8_000_000)
assert raises(bit_pulses, b"\x00", 10, 3, 10, 8_000_000)
assert raises(bit_pulses, b"\x00", 7, 0, 10, 8_000_000)A hint
One cycle in nanoseconds is value / 256 * 1e9 / sys_hz, with the same value that divider() computes. For each byte, take its bits from 7 down to 0 with (byte >> bit) & 1, pick the high cycle count, and turn both the high part and the rest of the bit into nanoseconds.
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 cycles, dividers, DMA, the flash and the RP2350, one question at a time:
Practice that adapts to you
Practise this lesson
12 questions about 6 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): Why can a PIO state machine keep timing that a Python loop on the same chip cannot?
A state machine is a small processor beside the CPU with its own program of up to 32 instructions. Every instruction
takes exactly one of its cycles unless it waits for data, and delays are written in cycles, so its timing is a count:
the burst in this lesson spends 13 cycles high and 13 low in every period. A Python loop shares the CPU with the
interpreter, the garbage collector and interrupt handlers, and none of them promises a cycle count, so its pulses
stretch whenever something else runs. The state machine also keeps going while Python is busy, as long as its FIFO
has data. Python’s job becomes feeding it, with put() or a DMA channel.
Two longer rp2 questions follow; the link under each leads to what it tests, the first step of a good reply and the follow-ups to expect:
Model answers open with any pass: ₹79 for 1 week ($2.99), one payment that never renews.
-
Mid to senior
A Pico sends an infrared carrier from a PIO program of 26 cycles a period at freq=1_000_000, and the 38 kHz receiver works poorly. A colleague proposes freq=988_000. What does each setting give, what does the change cost, and how else could you get closer to 38 kHz?
-
Mid to senior
A Pico logger reads a sensor in a 1 ms timer callback, polls the BOOTSEL button as fast as its main loop runs and appends a record to a file every second. Readings arrive late. After a change that wrote raw records to rp2.Flash(start=...), every file vanished. Explain both faults and fix the design.
Flashcards · 8 cards
Recall it, then check
Say the answer to yourself before you look: recalling it is what makes it stick.
-
Why can a PIO state machine keep timing that a Python loop cannot?
Each instruction takes exactly one cycle unless it stalls, plus up to 31 delay cycles written as [n], and the state machine runs beside the processor, untouched by garbage collection and interrupts.
-
How many times does a loop on jmp(x_dec, ...) run when X starts at n?
n + 1 times, because the jump tests X before it decrements it. To send n periods, put n − 1, as sm.put(24 - 1) does for 24.
-
How does a StateMachine's freq become its clock divider on a Pico?
125_000_000 * 256 // freq: a whole number and a fraction in 256ths, from 1 to 65,536. The slowest state machine runs at 1,908 Hz, and freq=1_907 raises ValueError.
-
What does a fractional PIO divider cost?
Uneven cycles: at freq=988_000 they last 126 or 127 system clocks, and only their average is right. A whole divider, such as 125 for 1 MHz, makes every cycle the same.
-
What does a DMA channel's count mean?
A number of transfers of size bytes, not a number of bytes: with the default size of 4-byte words, 1,024 bytes are count=256. A count in bytes overruns both buffers without an error.
-
How do you pace a DMA channel that feeds a PIO state machine?
Set treq_sel to the FIFO's DREQ, (p << 3) + n for PIO block p and state machine n, with size=0 for bytes and inc_write=False, so no byte is written to a full FIFO.
-
Why poll rp2.bootsel_button() only a few times a second?
The button sits on the flash's chip-select line, so each call keeps the flash away for at least 8 µs with interrupts off.
-
What does the RP2350 add for MicroPython over the RP2040?
12 PIO state machines, 16 DMA channels, 520 KB of SRAM and 24 PWM channels, under the same rp2 module; its RISCV firmware variants use asm_rv32 instead of asm_thumb.
Key takeaways
- A PIO state machine runs one instruction per cycle plus the delays you write, so a program’s pulses are counted:
the burst spends 26 cycles a period, 13 high and 13 low, and
jmp(x_dec)loops X + 1 times. freqbecomes a divider of whole 256ths of the 125 MHz clock: the slowest is 1,908 Hz, a fraction gets the average right but makes cycles 126 or 127 clocks long, and a whole divider makes every cycle the same.- A DMA channel’s
countis in transfers ofsizebytes, so 1,024 bytes are 256 word transfers, and a count in bytes overruns both buffers without an error. treq_selpaces a channel: a PIO FIFO’s DREQ is(p << 3) + n, and aDMATimerasks at X/Y of the system clock, for as long as you keep the object.rp2.Flash(start, len)hands out parts of the file area, but stock firmware formats the whole area at start-up when its filesystem has another size, so raw records go in a file; flash writes andbootsel_button()turn interrupts off, so both belong outside fast loops.- The RP2350 runs the same
rp2code with 12 state machines and 16 DMA channels, and its RISC-V mode tradesasm_thumbforasm_rv32.
Terms in this lesson
- clock divider
- A circuit that makes one cycle of a slower clock last a chosen number of cycles of a faster one. A PIO state machine's divider counts system clocks in steps of 1/256.
- DMA
- Direct memory access: hardware channels that copy data between memory and peripherals by themselves, so the processor is free while a transfer runs.
- DREQ
- Data request: the signal by which a peripheral tells a DMA channel that it can take or give one more item, so the channel transfers at the peripheral's pace.
- FIFO
- First in, first out: a small queue in hardware. Each PIO state machine has a 4-word FIFO for data from the processor and one for data back to it.
- PIO state machine
- One of the small processors in the programmable I/O blocks of the RP2040 and RP2350. It runs a program of up to 32 instructions, one per cycle of its own clock, to drive and read pins with exact timing.
- XIP
- Execute in place: running a program straight from external flash, read through a cache. While that flash is erased or written, no code can be read from it.
References
- rp2 – functionality specific to the RP2040 (PIO assembler, bootsel_button) (MicroPython project)
- class StateMachine – access to the RP2040's programmable I/O interface (MicroPython project)
- class PIO – advanced PIO usage (MicroPython project)
- class DMA and class DMATimer – access to the RP2040's DMA controller (MicroPython project)
- class Flash – access to built-in flash storage (MicroPython project)
- Programmable IO (the rp2 PIO tutorial) (MicroPython project)
- General information about the RP2xxx port (MicroPython project)
- Quick reference for the RP2 (MicroPython project)
- MicroPython v1.29.0 release notes (rp2.DMATimer) (MicroPython project)
- MicroPython v1.25.0 release notes (asm_rv32 on the RP2350, bootsel_button on the RP2350) (MicroPython project)
- StateMachine.init() and its clock divider (ports/rp2/rp2_pio.c, v1.29.0) (MicroPython project)
- DMA control fields, defaults and DMATimer ratios (ports/rp2/rp2_dma.c, v1.29.0) (MicroPython project)
- rp2.Flash, its start and len arguments and the flash lock-out (ports/rp2/rp2_flash.c, v1.29.0) (MicroPython project)
- bootsel_button() (ports/rp2/modrp2.c, v1.29.0) (MicroPython project)
- MICROPY_HW_BOOTSEL_DELAY_US (ports/rp2/mpconfigport.h, v1.29.0) (MicroPython project)
- Pico storage size (ports/rp2/boards/RPI_PICO/mpconfigboard.cmake, v1.29.0) (MicroPython project)
- Pico 2 board and its RISCV variant (ports/rp2/boards/RPI_PICO2, v1.29.0) (MicroPython project)
- Pico 2 W board and its RISCV variant (ports/rp2/boards/RPI_PICO2_W, v1.29.0) (MicroPython project)
- uctypes – access binary data in a structured way (addressof) (MicroPython project)
- RP2040 datasheet (2.5 DMA, 3.2.2 control flow, 3.5.5 clock dividers) (Raspberry Pi Ltd)
- RP2040 platform limits (pico-sdk 2.3.0, src/rp2040/hardware_regs/include/hardware/platform_defs.h) (Raspberry Pi Ltd)
- RP2350 platform limits (pico-sdk 2.3.0, src/rp2350/hardware_regs/include/hardware/platform_defs.h) (Raspberry Pi Ltd)
Related tools
Report a problem with this lesson
Kept only in this browser. Your Learn progress