Session 2: Using Events, Flags, and Task Signaling
Synopsis
Introduces synchronization primitives for notifying tasks about external changes, internal milestones, and application-level state updates.
Session Content
Session 2: Using Events, Flags, and Task Signaling
Session Overview
Duration: ~45 minutes
Topic: Using Events, Flags, and Task Signaling in MicroPython
Target Device: Raspberry Pi Pico 2 W
IDE: Thonny (MicroPython interpreter)
Learning Objectives
By the end of this session, learners will be able to:
- Explain why task signaling is important in asynchronous programs
- Use
uasyncio.Eventto coordinate tasks - Use
uasyncio.ThreadSafeFlagto signal between interrupt context and async tasks - Use shared state safely with flags and events
- Build a simple responsive Pico 2 W application using multiple async tasks
Prerequisites
- Basic Python syntax
- Raspberry Pi Pico 2 W flashed with MicroPython
- Thonny installed and connected to the Pico
- Familiarity with
async/awaitfrom Session 1 - Optional hardware:
- Push button
- LED
- 220Ω resistor
- Breadboard and jumper wires
Development Setup in Thonny
- Connect the Pico 2 W to your computer via USB.
- Open Thonny.
- Go to Run > Select interpreter.
- Choose MicroPython (Raspberry Pi Pico).
- Select the correct port.
- Confirm the REPL opens and shows the MicroPython prompt.
- Save scripts to the Pico using File > Save as... > Raspberry Pi Pico.
Session Outline
- Theory: Why signaling matters in async systems — 10 min
- Events for task coordination — 10 min
- ThreadSafeFlag for interrupt-to-task signaling — 10 min
- Hands-on exercise: Button-controlled LED using signaling — 15 min
1) Theory: Why Signaling Matters in Async Systems
In asynchronous programming, tasks run cooperatively. Each task yields control at await points. When multiple tasks need to coordinate, they often need a way to tell each other when something has happened.
Common examples: - A sensor reading becomes available - A button is pressed - A network message arrives - A timeout expires - An interrupt occurs
Instead of constantly checking conditions in a loop, tasks can wait efficiently for a signal.
Common signaling tools in MicroPython
- Shared boolean flags: simple state sharing
uasyncio.Event: task-to-task synchronizationuasyncio.ThreadSafeFlag: safe signaling from interrupt handlers or other non-async contexts
2) Using uasyncio.Event
An Event is a synchronization primitive used to notify one or more tasks that something has happened.
Key behavior
- Tasks can wait on an event using
await event.wait() - Another task can notify waiting tasks using
event.set() - The event can be cleared using
event.clear()
Typical use cases
- Start all tasks when setup is complete
- Notify a task that new data is ready
- Pause/resume workflows
Example: Simple event-based task signaling
# event_demo.py
import uasyncio as asyncio
event = asyncio.Event()
async def waiter():
print("Waiter: waiting for event...")
await event.wait()
print("Waiter: event received!")
async def setter():
print("Setter: preparing to signal...")
await asyncio.sleep(2)
print("Setter: setting event")
event.set()
async def main():
asyncio.create_task(waiter())
asyncio.create_task(setter())
await asyncio.sleep(3)
asyncio.run(main())
Example output
Waiter: waiting for event...
Setter: preparing to signal...
Setter: setting event
Waiter: event received!
Important notes
- If an event is already set,
await event.wait()returns immediately - If multiple tasks are waiting, all of them wake up when the event is set
- If you need one-time notification to a single task, use a different pattern such as a queue or shared state
3) Using uasyncio.ThreadSafeFlag
ThreadSafeFlag is designed for signaling from an interrupt context to an async task.
This is important because: - Interrupt handlers must be very short - You should not perform blocking operations in an interrupt - You should not call many normal async functions directly from interrupts
Instead, the interrupt handler sets a flag, and an async task handles the work later.
Typical use cases
- Button interrupt presses
- Hardware sensor interrupts
- Timing events from timer callbacks
Pattern
- Interrupt handler triggers
- Interrupt handler sets a
ThreadSafeFlag - Async task waits on
await flag.wait() - Task handles the event safely
4) Hands-on Exercise: Button-Controlled LED Using Signaling
In this exercise, you will:
- Use a push button as an input
- Use an LED as an output
- Use a hardware interrupt to detect the button press
- Use ThreadSafeFlag to notify an async task
- Toggle the LED each time the button is pressed
Wiring
Components
- 1x LED
- 1x 220Ω resistor
- 1x push button
- Breadboard and jumper wires
Connections
LED - LED anode (+) → GPIO 15 through 220Ω resistor - LED cathode (−) → GND
Button - One side of button → GPIO 14 - Other side of button → GND
We will use the internal pull-up resistor on GPIO 14, so the button reads:
- 1 when not pressed
- 0 when pressed
Code: Button interrupt + ThreadSafeFlag + async LED control
# main.py
from machine import Pin
import uasyncio as asyncio
# Hardware setup
led = Pin(15, Pin.OUT)
button = Pin(14, Pin.IN, Pin.PULL_UP)
# Thread-safe flag for interrupt-to-async signaling
button_flag = asyncio.ThreadSafeFlag()
# Shared state
led_state = False
def button_handler(pin):
"""
Interrupt handler for the push button.
Keep interrupt handlers short:
- Do not print
- Do not sleep
- Do not perform blocking work
- Only signal the async task
"""
button_flag.set()
button.irq(trigger=Pin.IRQ_FALLING, handler=button_handler)
async def blink_on_press():
"""
Wait for button presses and toggle the LED each time.
"""
global led_state
print("Ready. Press the button to toggle the LED.")
while True:
await button_flag.wait()
# Toggle LED state
led_state = not led_state
led.value(led_state)
print("Button pressed -> LED is", "ON" if led_state else "OFF")
async def heartbeat():
"""
Simple background task to show the system is alive.
"""
while True:
await asyncio.sleep(1)
print("Heartbeat: running")
async def main():
asyncio.create_task(heartbeat())
await blink_on_press()
try:
asyncio.run(main())
finally:
asyncio.new_event_loop()
Expected behavior
- When you press the button, the LED toggles state
- The REPL prints a message each time the button is pressed
- A heartbeat message appears every second
Example output
Ready. Press the button to toggle the LED.
Heartbeat: running
Heartbeat: running
Button pressed -> LED is ON
Heartbeat: running
Button pressed -> LED is OFF
Heartbeat: running
Code walkthrough
1. Hardware objects
led = Pin(15, Pin.OUT)
button = Pin(14, Pin.IN, Pin.PULL_UP)
- LED is configured as an output
- Button is configured as an input with pull-up resistor enabled
2. Flag creation
button_flag = asyncio.ThreadSafeFlag()
This creates a signaling object that can be safely triggered from the interrupt handler.
3. Interrupt handler
def button_handler(pin):
button_flag.set()
The interrupt handler does only one thing: it signals the async task.
4. Waiting for the signal
await button_flag.wait()
The async task sleeps efficiently until the button is pressed.
5. LED toggling
led_state = not led_state
led.value(led_state)
Each button press toggles the LED.
5) Hands-on Exercise Extension: Using Event for Start/Stop Control
Now modify the program so that:
- A button press signals a ThreadSafeFlag
- The async task toggles the LED only when a separate Event is set
- A second task can enable or disable blinking
This demonstrates combining signaling tools.
Example: Event-controlled LED task
# event_control_demo.py
from machine import Pin
import uasyncio as asyncio
led = Pin(15, Pin.OUT)
button = Pin(14, Pin.IN, Pin.PULL_UP)
press_flag = asyncio.ThreadSafeFlag()
run_event = asyncio.Event()
run_event.set() # Start in running mode
def button_handler(pin):
press_flag.set()
button.irq(trigger=Pin.IRQ_FALLING, handler=button_handler)
async def toggle_task():
state = False
print("Toggle task started")
while True:
await press_flag.wait()
if run_event.is_set():
state = not state
led.value(state)
print("LED:", "ON" if state else "OFF")
else:
print("Press ignored: system paused")
async def mode_task():
while True:
await asyncio.sleep(5)
if run_event.is_set():
run_event.clear()
print("System paused")
else:
run_event.set()
print("System running")
async def main():
asyncio.create_task(mode_task())
await toggle_task()
try:
asyncio.run(main())
finally:
asyncio.new_event_loop()
Expected behavior
- The system alternates between running and paused every 5 seconds
- Button presses only toggle the LED when running
- While paused, button presses are ignored
Example output
Toggle task started
LED: ON
LED: OFF
System paused
Press ignored: system paused
Press ignored: system paused
System running
LED: ON
6) Practice Tasks
Complete one or more of the following:
Task A: Add debounce
Button presses can trigger multiple interrupts due to switch bounce. Add a simple debounce delay in the async task:
- Ignore presses that happen within 200 ms of the previous one
Hint:
- Use asyncio.ticks_ms() or time.ticks_ms() if available
- Store the timestamp of the last valid press
Task B: Add two LEDs
- Use one LED for button press indication
- Use another LED as a status indicator
- Make the status LED blink only when the system is running
Task C: Add a second button
- Button 1 toggles the LED
- Button 2 pauses/resumes the system using an
Event
7) Common Mistakes and Troubleshooting
Mistake: Doing too much in the interrupt handler
Bad:
def handler(pin):
print("Button pressed")
led.toggle()
Why this is bad: - Interrupt handlers should be short - Printing and logic can cause instability
Better: - Set a flag in the interrupt - Handle the work in an async task
Mistake: Forgetting the pull-up resistor
If the button input floats, you may see random presses.
Fix:
button = Pin(14, Pin.IN, Pin.PULL_UP)
Mistake: Not clearing or managing event state
If an Event remains set, tasks may not block as expected.
Use:
run_event.clear()
run_event.set()
to control state explicitly.
Mistake: Confusing Event and ThreadSafeFlag
- Use
Eventfor task-to-task coordination - Use
ThreadSafeFlagfor interrupt-to-task signaling
8) Session Summary
In this session, you learned:
- Why signaling is necessary in async embedded systems
- How to use
uasyncio.Eventfor task coordination - How to use
uasyncio.ThreadSafeFlagsafely with interrupts - How to build a responsive Pico 2 W application using a button and LED
- How to combine multiple async tasks for better responsiveness
9) Quick Reference
uasyncio.Event
event = asyncio.Event()
event.set()
event.clear()
await event.wait()
event.is_set()
uasyncio.ThreadSafeFlag
flag = asyncio.ThreadSafeFlag()
flag.set()
await flag.wait()
Pin setup
from machine import Pin
led = Pin(15, Pin.OUT)
button = Pin(14, Pin.IN, Pin.PULL_UP)
10) Suggested Homework
Build a small dashboard with:
- One button to cycle between 3 modes
- One LED for each mode
- A status heartbeat task
- ThreadSafeFlag for button interrupts
- Event to enable/disable mode updates
Back to Chapter | Back to Master Plan | Previous Session | Next Session