taches-cooperatives.py

Raspberry Pi Pico · File : taches-cooperatives.py · Language : Python · Use case : Ordonnanceur de tâches non bloquant pour animer plusieurs broches

Three periodic tasks (LED 2 Hz, beep 1 Hz, reading every 5 s) run on a 15-line cooperative mini scheduler. This pattern replaces nested delay() calls as soon as a project animates more than one pin.

Hardware

A Raspberry Pi Pico (RP2040 dual Cortex-M0+ @ 133 MHz) with MicroPython installed, powered over USB. 1 LED + 330 Ω on GP25, 1 passive buzzer on GP2 (optional, drop the task otherwise).

Wiring

LED on GP25 through 330 Ω to GND; buzzer between GP2 and GND. Adapt pins to your free ones.

How it works

Each task stores its deadline; the loop compares with time.ticks_diff (robust to 32-bit wraparound) and reschedules the next deadline. sleep_ms(1) avoids full CPU occupancy.

Running it

Install MicroPython if needed (hold BOOTSEL, plug USB, drop the .uf2 file), then copy the script to the board drive or via Thonny. The script auto-starts when named main.py.

Variants

Add tasks on the fly with creer(), or move to uasyncio (MicroPython asyncio) for event-driven tasks with non-blocking waits.

taches-cooperatives.py

# Ordonnanceur cooperatif : plusieurs taches sans fil ni RTOS.
# Chaque tache est appelee a sa frequence ; aucune ne bloque les autres.
from machine import Pin, ADC
import time

led = Pin(25, Pin.OUT)
buzz = Pin(2, Pin.OUT)
adc = ADC(4)

taches = []

def creer(periodo_ms, fonction):
    taches.append([time.ticks_ms() + periodo_ms, periodo_ms, fonction])

def clignoter():
    led.value(not led.value())

def bip_court():
    buzz.on()
    time.sleep_ms(20)
    buzz.off()

def mesurer():
    v = adc.read_u16() * 3.3 / 65535.0
    t = 27.0 - (v - 0.706) / 0.001721
    print("T = %.1f C" % t)

creer(500, clignoter)     # 2 Hz
creer(1000, bip_court)    # 1 Hz
creer(5000, mesurer)      # toutes les 5 s

print("Ordonnanceur demarre (%d taches)" % len(taches))
while True:
    maintenant = time.ticks_ms()
    for tache in taches:
        if time.ticks_diff(maintenant, tache[0]) >= 0:
            tache[0] = time.ticks_add(tache[0], tache[1])
            tache[2]()
    time.sleep_ms(1)   # cede la main : evite la pleine charge CPU

← Back to the board