Real-Time Systems for IoT

IoT firmware must respond to hardware events within precise time bounds — a sensor overflows, a safety limit is exceeded, a button is pressed. Real-time programming patterns let you build firmware that is both responsive and stable.

What is Real-Time?

Real-Time ≠ Fast

Real-time means: "responds within a GUARANTEED deadline"

Hard real-time:   Deadline missed = system failure
                  (airbag deployment, pacemaker, industrial safety)

Soft real-time:   Deadline missed = degraded performance
                  (most IoT sensors, dashboards, logging)

Best-effort:      No deadline
                  (background tasks, data upload, OTA updates)

Most IoT projects are SOFT real-time:
  ✓ Read temperature every 5 seconds (± 100 ms is fine)
  ✓ Update OLED display at 10 fps (± 10 ms is fine)
  ✗ Hard real-time motor control (needs µs precision, use RTOS)

Hardware Timers

Hardware timers fire independently of your Python code — they're the most reliable way to run periodic tasks.

python

# MicroPython — hardware timer for periodic tasks
from machine import Timer, Pin
import utime

led = Pin(2, Pin.OUT)
reading_count = 0
sensor_buffer = []

def read_sensor_isr(timer):
    """Called by hardware timer — keep it short."""
    global reading_count
    reading_count += 1
    # In a real ISR, set a flag and do work in main loop
    # Don't allocate memory or do I/O here

def blink_isr(timer):
    led.toggle()

# Timer 0: blink LED every 500 ms
blink_timer = Timer(0)
blink_timer.init(period=500, mode=Timer.PERIODIC, callback=blink_isr)

# Timer 1: trigger sensor read every 2 seconds
sensor_timer = Timer(1)
sensor_trigger = False

def trigger_sensor(t):
    global sensor_trigger
    sensor_trigger = True

sensor_timer.init(period=2000, mode=Timer.PERIODIC, callback=trigger_sensor)

# Main loop responds to timer flags
start = utime.time()
while utime.time() - start < 30:
    if sensor_trigger:
        sensor_trigger = False
        # Do actual sensor work here (safe — not in ISR)
        reading_count += 1
        print(f"Reading #{reading_count}: T=24.5°C")

# Cleanup
blink_timer.deinit()
sensor_timer.deinit()

uasyncio — Cooperative Multitasking

uasyncio lets multiple coroutines run concurrently by cooperating — each yields control when waiting.

python

import uasyncio
from machine import Pin, ADC
import dht

# Shared state
sensor_data = {
    "temperature": None,
    "humidity":    None,
    "light":       None,
    "last_update": 0
}

async def task_read_dht(pin_num, interval_s=2):
    """Read DHT22 every 2 seconds."""
    sensor = dht.DHT22(Pin(pin_num))
    while True:
        try:
            sensor.measure()
            sensor_data["temperature"] = sensor.temperature()
            sensor_data["humidity"]    = sensor.humidity()
        except OSError:
            print("[DHT] Read failed")
        await uasyncio.sleep(interval_s)

async def task_read_light(pin_num, interval_ms=200):
    """Read LDR 5 times per second."""
    adc = ADC(Pin(pin_num))
    adc.atten(ADC.ATTN_11DB)
    while True:
        raw = adc.read()
        sensor_data["light"] = 100 - (raw / 4095 * 100)
        await uasyncio.sleep_ms(interval_ms)

async def task_print_data(interval_s=5):
    """Print sensor data every 5 seconds."""
    while True:
        if sensor_data["temperature"] is not None:
            print(f"T={sensor_data['temperature']:.1f}°C "
                  f"H={sensor_data['humidity']:.1f}% "
                  f"Light={sensor_data['light']:.1f}%")
        await uasyncio.sleep(interval_s)

async def task_wifi_upload(interval_s=30):
    """Upload to cloud every 30 seconds."""
    import urequests
    while True:
        if sensor_data["temperature"] is not None:
            try:
                # In a real app: post to MQTT or HTTP API
                print(f"[Upload] Sending to cloud...")
                # urequests.post(url, json=sensor_data)
            except Exception as e:
                print(f"[Upload] Failed: {e}")
        await uasyncio.sleep(interval_s)

async def main():
    """Launch all tasks concurrently."""
    await uasyncio.gather(
        task_read_dht(pin_num=4),
        task_read_light(pin_num=34),
        task_print_data(),
        task_wifi_upload(),
    )

uasyncio.run(main())

FreeRTOS on ESP32 (Arduino Framework)

The ESP32 runs FreeRTOS natively. From the Arduino framework, you can create true multi-core tasks.

cpp

// Arduino (ESP32) — FreeRTOS tasks on dual cores
#include <Arduino.h>

// Shared data (protect with mutex)
SemaphoreHandle_t dataMutex;
float temperature = 0.0;
float humidity    = 0.0;

// Task 1: Read sensor on Core 1 (APP CPU)
void sensorTask(void* parameter) {
    for (;;) {
        float t = 24.5 + random(-10, 10) / 10.0f;  // Simulated
        float h = 60.0 + random(-10, 10) / 10.0f;

        if (xSemaphoreTake(dataMutex, portMAX_DELAY) == pdTRUE) {
            temperature = t;
            humidity    = h;
            xSemaphoreGive(dataMutex);
        }
        vTaskDelay(pdMS_TO_TICKS(2000));  // Yield for 2 seconds
    }
}

// Task 2: Upload data on Core 0 (PRO CPU — same as WiFi/BT)
void uploadTask(void* parameter) {
    for (;;) {
        float t, h;
        if (xSemaphoreTake(dataMutex, pdMS_TO_TICKS(100)) == pdTRUE) {
            t = temperature;
            h = humidity;
            xSemaphoreGive(dataMutex);
        }
        Serial.printf("[Core %d] Upload: T=%.1f H=%.1f\n",
            xPortGetCoreID(), t, h);
        vTaskDelay(pdMS_TO_TICKS(10000));  // Upload every 10 seconds
    }
}

void setup() {
    Serial.begin(115200);
    dataMutex = xSemaphoreCreateMutex();

    // Pin tasks to specific cores
    xTaskCreatePinnedToCore(sensorTask, "SensorTask", 4096, NULL, 1, NULL, 1); // Core 1
    xTaskCreatePinnedToCore(uploadTask, "UploadTask", 8192, NULL, 1, NULL, 0); // Core 0

    Serial.println("FreeRTOS tasks started");
}

void loop() {
    // Main loop is itself a FreeRTOS task on Core 1
    // We're not using it — can leave empty or use for UI
    vTaskDelay(pdMS_TO_TICKS(1000));
}

Watchdog Timer

A watchdog timer resets the device if it hangs — essential for unattended field deployments.

python

# MicroPython — Software Watchdog Timer
from machine import WDT
import utime

# 8-second hardware watchdog (maximum on ESP32)
wdt = WDT(timeout=8000)

def do_sensor_reading():
    utime.sleep_ms(200)  # Simulated work
    return 24.5

def do_mqtt_publish(data):
    utime.sleep_ms(500)  # Simulated network

def main_loop():
    while True:
        try:
            # Critical operations
            data = do_sensor_reading()
            do_mqtt_publish(data)

            # Feed the watchdog ONLY if all critical work succeeded
            wdt.feed()
            print(f"Tick — WDT fed. Data: {data}")

        except Exception as e:
            print(f"Error: {e}")
            # Don't feed WDT on error — let it reset the device
            # This self-heals hangs and stuck states

        utime.sleep(5)

main_loop()

State Machine Pattern

Complex IoT firmware is best structured as a state machine.

python

from enum import Enum

class DeviceState(Enum):
    BOOT        = "BOOT"
    CONNECTING  = "CONNECTING"
    MEASURING   = "MEASURING"
    UPLOADING   = "UPLOADING"
    SLEEPING    = "SLEEPING"
    ERROR       = "ERROR"

class IoTStateMachine:
    def __init__(self):
        self.state       = DeviceState.BOOT
        self.error_count = 0
        self.data        = None

    def transition(self, new_state):
        print(f"State: {self.state.value} → {new_state.value}")
        self.state = new_state

    def run(self):
        while True:
            if self.state == DeviceState.BOOT:
                self._handle_boot()
            elif self.state == DeviceState.CONNECTING:
                self._handle_connecting()
            elif self.state == DeviceState.MEASURING:
                self._handle_measuring()
            elif self.state == DeviceState.UPLOADING:
                self._handle_uploading()
            elif self.state == DeviceState.SLEEPING:
                self._handle_sleeping()
            elif self.state == DeviceState.ERROR:
                self._handle_error()

    def _handle_boot(self):
        print("Initializing hardware...")
        import utime; utime.sleep_ms(100)
        self.transition(DeviceState.CONNECTING)

    def _handle_connecting(self):
        success = True  # connect_wifi(...)
        if success:
            self.error_count = 0
            self.transition(DeviceState.MEASURING)
        else:
            self.transition(DeviceState.ERROR)

    def _handle_measuring(self):
        self.data = {"temperature": 24.5, "humidity": 61.2}
        self.transition(DeviceState.UPLOADING)

    def _handle_uploading(self):
        success = True  # mqtt.publish(...)
        if success:
            self.transition(DeviceState.SLEEPING)
        else:
            self.error_count += 1
            self.transition(DeviceState.ERROR if self.error_count > 3 else DeviceState.MEASURING)

    def _handle_sleeping(self):
        print("Sleeping 5 minutes...")
        import machine
        machine.deepsleep(5 * 60 * 1000)

    def _handle_error(self):
        print(f"Error #{self.error_count}. Restarting in 30s...")
        import utime, machine
        utime.sleep(30)
        machine.reset()

device = IoTStateMachine()
device.run()

Summary

Hardware timers fire independently of your main code — most reliable for periodic tasks
uasyncio enables cooperative multitasking: multiple coroutines run concurrently by yielding at await points
FreeRTOS (ESP32 Arduino) provides true multi-core task scheduling with priorities and semaphores
Watchdog timers (WDT) automatically reboot the device if it hangs — essential for unattended devices
State machines make complex IoT firmware manageable by making all transitions explicit
vTaskDelay / await uasyncio.sleep() yield CPU time — never use blocking delay() or time.sleep() in concurrent firmware

Next, we'll explore IoT cloud platforms — where your device data gets stored, analyzed, and acted upon.