Skip to content

Session 3: Handling Exceptions in Concurrent Systems

Synopsis

Covers error propagation, task failure visibility, defensive coding, and recovery patterns for asynchronous embedded applications.

Session Content

Session 3: Handling Exceptions in Concurrent Systems

Duration: ~45 minutes
Topic: Handling Exceptions in Concurrent Systems
Target Audience: Python developers with basic programming knowledge learning MicroPython asynchronous programming on Raspberry Pi Pico 2 W

1) Session Overview

In concurrent MicroPython programs, multiple tasks may run “at the same time” using uasyncio. When one task fails, the error can affect the whole application if it is not handled properly. This session focuses on identifying, catching, logging, and recovering from exceptions in asynchronous programs on the Raspberry Pi Pico 2 W.

By the end of the session, learners will be able to: - Understand how exceptions behave in asynchronous code - Catch and handle errors inside uasyncio tasks - Protect hardware control loops from crashes - Implement recovery strategies such as retries and fallback states - Build a robust concurrent application with safe error handling

2) Prerequisites

Knowledge required

  • Basic Python syntax
  • Functions, loops, conditionals
  • Introductory understanding of uasyncio
  • Basic familiarity with GPIO pins on Raspberry Pi Pico 2 W

Hardware required

  • Raspberry Pi Pico 2 W
  • USB cable
  • Breadboard
  • 1 LED
  • 1 resistor (220Ω–330Ω)
  • 1 pushbutton
  • Optional: I2C sensor such as BME280 or similar

Software required

  • Thonny IDE
  • MicroPython firmware for Raspberry Pi Pico 2 W

3) Development Environment Setup

Thonny setup

  1. Install Thonny on your computer.
  2. Connect the Raspberry Pi Pico 2 W via USB.
  3. In Thonny:
  4. Go to Tools > Options > Interpreter
  5. Select MicroPython (Raspberry Pi Pico)
  6. Choose the correct serial port for the Pico
  7. Upload MicroPython firmware if needed.
  8. Save scripts directly to the Pico using Thonny.

MicroPython verification

Run this in Thonny:

import sys
print(sys.platform)

Expected output:

rp2

4) Theory: Exceptions in Concurrent Systems

What is an exception?

An exception is an error that interrupts normal execution. In MicroPython, common examples include: - ValueError - OSError - RuntimeError - KeyboardInterrupt

Why exceptions are important in concurrent code

In asynchronous programs: - One task may fail while others continue - A background task may stop silently if the exception is not caught - Hardware loops may leave devices in unsafe states - Network operations often fail due to unstable connectivity

Common failure sources in Pico projects

  • Invalid sensor data
  • I2C communication issues
  • Network connection failures
  • Incorrect pin usage
  • Unexpected user input
  • Task cancellation during shutdown

5) Key Concepts

A. Exceptions inside uasyncio tasks

If a task raises an exception and nobody catches it, the task stops. Depending on the design, the rest of the system may continue, but the failed task is lost.

B. Defensive programming

Use: - try - except - finally

This helps ensure: - errors are logged - hardware is reset safely - tasks can retry or exit cleanly

C. Recovery strategies

  • Retry after delay
  • Use fallback values
  • Restart failing task
  • Safely disable outputs
  • Report errors over serial or network

D. Cancellation handling

Tasks may be cancelled intentionally. In async systems, cancellation should be handled cleanly to: - turn off LEDs - close connections - release resources

6) Example 1: Catching Exceptions in an Async Task

This example shows a task that intentionally raises an error sometimes, then catches it and continues.

Code: safe task with exception handling

import uasyncio as asyncio

async def risky_task():
    count = 0
    while True:
        try:
            count += 1
            print("Task running:", count)

            # Simulate a failure every 5 iterations
            if count % 5 == 0:
                raise ValueError("Simulated task error")

            await asyncio.sleep(1)

        except ValueError as e:
            print("Caught error:", e)
            print("Recovering by waiting and continuing...")
            await asyncio.sleep(2)

        except Exception as e:
            print("Unexpected error:", e)
            print("Stopping task safely.")
            break

async def main():
    await risky_task()

asyncio.run(main())

Example output

Task running: 1
Task running: 2
Task running: 3
Task running: 4
Task running: 5
Caught error: Simulated task error
Recovering by waiting and continuing...
Task running: 6
Task running: 7

7) Example 2: Concurrent LED Blinking with Safe Error Handling

This example uses two tasks: - one blinks an LED - one simulates a fault and handles it without crashing the whole program

Wiring

  • LED anode → GPIO 15 through a resistor
  • LED cathode → GND

Code: concurrent tasks with exception protection

import uasyncio as asyncio
from machine import Pin

led = Pin(15, Pin.OUT)

async def blink_led():
    while True:
        try:
            led.toggle()
            print("LED state:", led.value())
            await asyncio.sleep(0.5)
        except Exception as e:
            print("LED task error:", e)
            led.off()
            await asyncio.sleep(1)

async def monitor_system():
    counter = 0
    while True:
        try:
            counter += 1
            print("Monitor tick:", counter)

            # Simulate a fault
            if counter == 8:
                raise RuntimeError("Simulated monitor failure")

            await asyncio.sleep(1)

        except RuntimeError as e:
            print("Monitor recovered from:", e)
            await asyncio.sleep(2)

        except Exception as e:
            print("Unexpected monitor error:", e)
            break

async def main():
    print("Starting concurrent tasks...")
    t1 = asyncio.create_task(blink_led())
    t2 = asyncio.create_task(monitor_system())

    await asyncio.gather(t1, t2)

asyncio.run(main())

Example output

Starting concurrent tasks...
LED state: 1
Monitor tick: 1
LED state: 0
Monitor tick: 2
LED state: 1
Monitor tick: 3
Monitor tick: 4
Monitor tick: 5
Monitor tick: 6
Monitor tick: 7
Monitor tick: 8
Monitor recovered from: Simulated monitor failure
LED state: 0

8) Example 3: Handling Hardware Exceptions from an I2C Sensor

A very common source of exceptions is sensor communication. If a sensor is disconnected or returns invalid data, your code should recover gracefully.

Code: safe I2C read pattern

import uasyncio as asyncio
from machine import Pin, I2C

# I2C setup for Raspberry Pi Pico 2 W
# SDA = GP0, SCL = GP1 are common choices
i2c = I2C(0, sda=Pin(0), scl=Pin(1), freq=400000)

async def read_sensor():
    while True:
        try:
            devices = i2c.scan()
            print("I2C devices found:", devices)

            if not devices:
                raise OSError("No I2C device detected")

            # Simulate sensor reading logic
            # Replace this with actual sensor driver code
            print("Reading sensor data...")
            await asyncio.sleep(2)

        except OSError as e:
            print("Sensor/I2C error:", e)
            print("Retrying in 3 seconds...")
            await asyncio.sleep(3)

        except Exception as e:
            print("Unexpected sensor error:", e)
            break

asyncio.run(read_sensor())

Example output

I2C devices found: [118]
Reading sensor data...
I2C devices found: [118]
Reading sensor data...

If no device is connected:

I2C devices found: []
Sensor/I2C error: No I2C device detected
Retrying in 3 seconds...

9) Hands-on Exercise 1: Build a Fault-Tolerant LED Task

Goal

Create a blinking LED task that: - blinks every 0.5 seconds - simulates an error every 10 blinks - catches the error and continues blinking after a short pause

Instructions

  1. Connect an LED to GPIO 15 through a resistor.
  2. Open Thonny and create a new script.
  3. Enter the code below.
  4. Run it and observe the LED and serial output.

Starter code

import uasyncio as asyncio
from machine import Pin

led = Pin(15, Pin.OUT)

async def blink_with_errors():
    blink_count = 0

    while True:
        try:
            led.toggle()
            blink_count += 1
            print("Blink count:", blink_count)

            if blink_count % 10 == 0:
                raise ValueError("Simulated blink failure")

            await asyncio.sleep(0.5)

        except ValueError as e:
            print("Handled blinking error:", e)
            led.off()
            await asyncio.sleep(2)

asyncio.run(blink_with_errors())

Checkpoint

  • Does the LED pause briefly after the simulated error?
  • Does the program continue running after the error?

Expected result

The LED blinks normally, pauses when the error occurs, then resumes.

10) Hands-on Exercise 2: Concurrent Button Monitor with Error Recovery

Goal

Create two concurrent tasks: - one blinks an LED - one checks a pushbutton - if the button is not responding, the task logs an error and retries

Wiring

  • Button one side → GPIO 14
  • Button other side → GND
  • Use internal pull-up resistor

Code

import uasyncio as asyncio
from machine import Pin

led = Pin(15, Pin.OUT)
button = Pin(14, Pin.IN, Pin.PULL_UP)

async def blink_led():
    while True:
        led.toggle()
        print("LED:", led.value())
        await asyncio.sleep(0.5)

async def monitor_button():
    while True:
        try:
            if button.value() == 0:
                print("Button pressed")
                await asyncio.sleep(0.2)  # debounce delay

            await asyncio.sleep(0.1)

        except Exception as e:
            print("Button monitor error:", e)
            await asyncio.sleep(1)

async def main():
    await asyncio.gather(
        blink_led(),
        monitor_button()
    )

asyncio.run(main())

Challenge

Modify the code so that: - pressing the button turns the LED on for 1 second - any exception causes the LED to turn off safely

11) Best Practices for Exception Handling in Concurrent Systems

Do

  • Catch exceptions close to where they occur
  • Keep tasks small and focused
  • Log errors clearly
  • Use finally for cleanup
  • Recover from temporary hardware/network errors
  • Turn off actuators on failure

Don’t

  • Hide all exceptions with a broad except and no logging
  • Let background tasks fail silently
  • Ignore cleanup when cancelling tasks
  • Retry infinitely without delay
  • Leave outputs in unsafe states

12) Advanced Pattern: Cleanup with finally

Use finally to ensure cleanup runs even when errors happen.

import uasyncio as asyncio
from machine import Pin

led = Pin(15, Pin.OUT)

async def controlled_blink():
    try:
        while True:
            led.toggle()
            print("Blinking...")
            await asyncio.sleep(1)

    except Exception as e:
        print("Task error:", e)

    finally:
        led.off()
        print("LED turned off safely.")

asyncio.run(controlled_blink())

13) Common Mistakes

Mistake 1: No exception handling in a task

A task crashes and disappears.

Mistake 2: Catching exceptions too broadly without logs

You lose visibility into what went wrong.

Mistake 3: Forgetting cleanup

Hardware may remain active unexpectedly.

Mistake 4: Blocking the event loop with long delays

Use await asyncio.sleep() instead of busy loops.

14) Mini Project: Resilient Sensor Monitor

Goal

Design a program that: - checks an I2C device repeatedly - blinks an LED as a status indicator - retries after communication failures - turns off the LED if the sensor is unavailable for too long

Suggested behavior

  • Green/OK indication using LED blinking
  • Failure indication using slower blink or LED off
  • Retry count tracking
  • Optional reset after repeated failures

Starter template

import uasyncio as asyncio
from machine import Pin, I2C

led = Pin(15, Pin.OUT)
i2c = I2C(0, sda=Pin(0), scl=Pin(1), freq=400000)

async def status_led():
    while True:
        led.toggle()
        await asyncio.sleep(0.5)

async def sensor_watchdog():
    failures = 0

    while True:
        try:
            devices = i2c.scan()
            print("Devices:", devices)

            if not devices:
                raise OSError("Sensor not found")

            failures = 0
            print("Sensor OK")
            await asyncio.sleep(2)

        except OSError as e:
            failures += 1
            print("Error:", e, "Failures:", failures)

            if failures >= 3:
                print("Too many failures, stopping LED and waiting...")
                led.off()
                await asyncio.sleep(5)
                failures = 0
            else:
                await asyncio.sleep(1)

async def main():
    await asyncio.gather(
        status_led(),
        sensor_watchdog()
    )

asyncio.run(main())

15) Session Summary

In this session, you learned how to: - handle exceptions in MicroPython asynchronous tasks - protect concurrent programs from crashing - recover from hardware and communication failures - use try, except, and finally effectively - design safer concurrent applications on the Pico 2 W

16) Review Questions

  1. What happens if an exception is not handled inside an async task?
  2. Why is cleanup important in concurrent hardware systems?
  3. When should you use finally?
  4. What is a good strategy for temporary I2C failures?
  5. Why should tasks log errors instead of ignoring them?

17) Further Reading

  • MicroPython Wiki: https://github.com/micropython/micropython/wiki
  • MicroPython Quick Reference: https://docs.micropython.org/en/latest/rp2/quickref.html
  • Raspberry Pi MicroPython Documentation: https://www.raspberrypi.com/documentation/microcontrollers/micropython.html
  • Raspberry Pi Pico Series Documentation: https://www.raspberrypi.com/documentation/microcontrollers/pico-series.html#pico2

Back to Chapter | Back to Master Plan | Previous Session | Next Session