Skip to content

Session 1: Designing Task-Oriented Application Architectures

Synopsis

Presents ways to divide an embedded application into cooperating services, workers, and supervisors with clear ownership of resources and responsibilities.

Session Content

Session 1: Designing Task-Oriented Application Architectures

Session Overview

Duration: ~45 minutes
Audience: Python developers with basic programming knowledge
Platform: Raspberry Pi Pico 2 W
Firmware: MicroPython
IDE: Thonny

Learning Outcomes

By the end of this session, learners will be able to: - Explain what a task-oriented application architecture is. - Identify application tasks and split them into cooperative units. - Design simple event-driven flows for embedded systems. - Use uasyncio to structure concurrent tasks in MicroPython. - Build a basic Pico 2 W application with parallel task behavior.

1) Theory: Why Task-Oriented Architecture Matters

1.1 Embedded systems are often task-driven

On the Pico 2 W, applications often need to: - Read sensors periodically - Control LEDs or motors - Respond to buttons or interrupts - Maintain Wi-Fi connectivity - Communicate over the network - Avoid blocking other activities

A task-oriented architecture organizes these responsibilities as separate, cooperative tasks rather than one long blocking program.

1.2 Benefits

  • Responsiveness: The device can react quickly to inputs.
  • Maintainability: Each task has a clear purpose.
  • Scalability: New features can be added without rewriting everything.
  • Predictability: Timing and periodic work are easier to manage.

1.3 Common architectural styles for MicroPython IoT

  • Polling loop: Simple but can become blocking.
  • State machine: Good for sequential device behavior.
  • Event-driven: Reacts to signals, timers, or interrupts.
  • Task-oriented with uasyncio: Best for multiple cooperative activities.

1.4 What tasks look like in Pico projects

Examples: - blink_task() for an LED status indicator - read_sensor_task() for periodic readings - button_task() for button monitoring - wifi_task() for connectivity management - publish_task() for sending data to a cloud service

2) Core Concepts

2.1 Blocking vs non-blocking behavior

A blocking delay like time.sleep(2) pauses the entire program.

With asynchronous programming: - Tasks yield control using await asyncio.sleep(...) - Other tasks continue running while one task waits

2.2 Cooperative multitasking

uasyncio does not preempt tasks. Each task must: - Do a small amount of work - Yield control regularly - Avoid long blocking calls

2.3 Task communication

Tasks can communicate through: - Shared variables - uasyncio.Event - uasyncio.Queue

For this session, we'll start with: - Shared state - Simple task separation

3) Development Environment Setup

3.1 Install Thonny

  1. Install Thonny IDE on your computer.
  2. Connect the Raspberry Pi Pico 2 W via USB.
  3. Open Thonny.
  4. In the bottom-right interpreter selector, choose:
  5. MicroPython (Raspberry Pi Pico)

3.2 Flash MicroPython firmware

  1. Hold the BOOTSEL button on the Pico 2 W.
  2. Plug it into USB.
  3. It appears as a USB storage device.
  4. Copy the latest Pico MicroPython UF2 firmware to the device.
  5. The board reboots automatically.

3.3 Verify connection in Thonny

Open the Shell and run:

import sys
print(sys.platform)

Expected output:

rp2

For this session, create: - main.py — application entry point - lib/ — optional folder for helper modules in future sessions

4) Architecture Design: A Simple Task Model

4.1 Example application

We will design a small system with: - A blinking onboard LED as a heartbeat - A simulated sensor task - A status task that reports system activity

4.2 Task responsibilities

  • LED task: Shows the board is alive
  • Sensor task: Produces periodic readings
  • Logger task: Prints summary data

4.3 Design principle

Each task should: - Have one job - Run independently - Yield often - Not assume other tasks are synchronous

5) Hands-On Exercise 1: Cooperative LED and Status Tasks

5.1 Goal

Build a MicroPython program that runs two asynchronous tasks: - Blink the onboard LED every second - Print a status message every 3 seconds

5.2 Wiring

No external wiring required.
Use the onboard LED on the Pico 2 W.

5.3 Code: main.py

# main.py
# Session 1: Task-Oriented Application Architectures
# Raspberry Pi Pico 2 W + MicroPython + uasyncio

import uasyncio as asyncio
from machine import Pin

# Onboard LED on most Pico boards
led = Pin("LED", Pin.OUT)

async def blink_task():
    """Blink the onboard LED once per second."""
    while True:
        led.toggle()
        print("LED:", "ON" if led.value() else "OFF")
        await asyncio.sleep(1)

async def status_task():
    """Print a periodic status message."""
    counter = 0
    while True:
        counter += 1
        print("Status task running. Tick:", counter)
        await asyncio.sleep(3)

async def main():
    """Run all application tasks concurrently."""
    print("Starting task-oriented application...")
    await asyncio.gather(
        blink_task(),
        status_task(),
    )

try:
    asyncio.run(main())
finally:
    asyncio.new_event_loop()

5.4 Expected output

Example Shell output:

Starting task-oriented application...
LED: ON
Status task running. Tick: 1
LED: OFF
LED: ON
LED: OFF
Status task running. Tick: 2
LED: ON

5.5 Discussion

  • blink_task() and status_task() run concurrently.
  • Neither task blocks the other.
  • await asyncio.sleep(...) gives control back to the scheduler.

6) Hands-On Exercise 2: Adding a Simulated Sensor Task

6.1 Goal

Extend the architecture with a third task that simulates a sensor reading.

6.2 Code: main.py

# main.py
# Task-oriented architecture with three cooperative tasks

import uasyncio as asyncio
from machine import Pin
import random

led = Pin("LED", Pin.OUT)

async def blink_task():
    """Blink the onboard LED every second."""
    while True:
        led.toggle()
        print("LED:", "ON" if led.value() else "OFF")
        await asyncio.sleep(1)

async def status_task():
    """Print application health every 3 seconds."""
    counter = 0
    while True:
        counter += 1
        print("Status task tick:", counter)
        await asyncio.sleep(3)

async def sensor_task():
    """Simulate a periodic sensor reading."""
    while True:
        reading = random.randint(20, 35)
        print("Sensor reading:", reading, "°C")
        await asyncio.sleep(2)

async def main():
    """Run all tasks concurrently."""
    print("Starting application with sensor simulation...")
    await asyncio.gather(
        blink_task(),
        status_task(),
        sensor_task(),
    )

try:
    asyncio.run(main())
finally:
    asyncio.new_event_loop()

6.3 Expected output

Starting application with sensor simulation...
LED: ON
Sensor reading: 27 °C
Status task tick: 1
LED: OFF
Sensor reading: 29 °C
LED: ON
LED: OFF
Status task tick: 2
Sensor reading: 24 °C

6.4 What this demonstrates

  • Independent timing for each task
  • Shared execution without threads
  • A foundation for real sensor integration

7) Hands-On Exercise 3: Replace Simulation with a Real Button Task

7.1 Goal

Add a button to control application behavior through a task.

7.2 Required hardware

  • 1 push button
  • 1 breadboard
  • 2 jumper wires

7.3 Wiring

Connect: - One side of the button to GND - The other side to GP15

We will use the internal pull-up resistor.

7.4 Code: main.py

# main.py
# Task-oriented architecture with a button input

import uasyncio as asyncio
from machine import Pin

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

async def blink_task():
    """Blink the onboard LED every second."""
    while True:
        led.toggle()
        await asyncio.sleep(1)

async def button_task():
    """Monitor the button and report press/release events."""
    last_state = button.value()

    while True:
        current_state = button.value()

        if current_state != last_state:
            if current_state == 0:
                print("Button pressed")
            else:
                print("Button released")
            last_state = current_state

        await asyncio.sleep(0.05)

async def main():
    """Run the LED and button tasks together."""
    print("Starting button monitoring...")
    await asyncio.gather(
        blink_task(),
        button_task(),
    )

try:
    asyncio.run(main())
finally:
    asyncio.new_event_loop()

7.5 Expected output

Starting button monitoring...
Button pressed
Button released
Button pressed

7.6 Learning points

  • Polling can still be cooperative if the loop yields often.
  • Task architecture makes it easy to add input handling.
  • Debouncing can be added later if needed.

8) Design Exercise: Architect a Weather Station Task Set

8.1 Scenario

Design a Pico 2 W weather station that: - Reads temperature every 10 seconds - Updates an OLED display every 1 second - Sends data to the cloud every 60 seconds - Blinks an LED while Wi-Fi is connected

8.2 Suggested task breakdown

  • read_sensor_task()
  • display_task()
  • network_task()
  • publish_task()
  • heartbeat_task()

8.3 Questions to answer

  • Which task should run most frequently?
  • Which task can tolerate delay?
  • Which task must never block the others?
  • What shared state will tasks need?

8.4 Example task responsibilities

  • Sensor task writes latest value to shared data
  • Display task reads the latest value and refreshes screen
  • Publish task sends stored readings
  • Heartbeat task indicates network status

9) Best Practices for Task-Oriented MicroPython Apps

9.1 Keep tasks small

Each task should do one thing well.

9.2 Avoid blocking calls

Avoid long sleep() calls in tasks. Prefer: - await asyncio.sleep(...) - Short periodic checks

9.3 Use clear naming

Examples: - wifi_connect_task - sensor_read_task - mqtt_publish_task

9.4 Define shared state carefully

Use simple data structures like dictionaries:

app_state = {
    "temperature": None,
    "wifi_connected": False,
}

9.5 Always handle cleanup

Use:

try:
    asyncio.run(main())
finally:
    asyncio.new_event_loop()

10) Quick Knowledge Check

10.1 Questions

  1. Why is time.sleep() problematic in multi-task embedded applications?
  2. What does await asyncio.sleep() do?
  3. What is cooperative multitasking?
  4. Why should each task have a single responsibility?
  5. What kinds of tasks would a Pico 2 W IoT device commonly need?

10.2 Model answers

  1. It blocks the entire program.
  2. It pauses the current task and lets others run.
  3. Tasks share CPU time by voluntarily yielding.
  4. It improves maintainability and clarity.
  5. Sensor reading, connectivity, UI updates, publishing, input handling.

11) Session Summary

In this session, you learned how to: - Think in tasks instead of a single blocking loop - Structure embedded applications around responsibilities - Use uasyncio to run concurrent behavior on the Pico 2 W - Build a foundation for responsive IoT applications

12) Stretch Goal

Create a program with: - One task reading a sensor - One task blinking an LED based on sensor threshold - One task printing an alert when a value exceeds a limit

Example logic: - Temperature above 30°C → LED blinks faster - Temperature below 25°C → LED blinks slowly - Alert message printed when threshold is crossed

13) Reference

  • 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 | Next Session