DEV Community

wellallyTech
wellallyTech

Posted on

From Heartbeat to Alert: Building a Real-time ECG Arrhythmia Detector on ESP32 with TinyML 💻

In the world of wearable tech, the holy grail isn't just collecting data—it's processing it where it happens. Why send raw, noisy sensor data to the cloud when you can detect a life-threatening heart condition right on your wrist?

Today, we are diving deep into the intersection of Edge Computing, Time-series Analysis, and TinyML. We're going to build a real-time ECG Arrhythmia Detection system using an ESP32, MicroPython, and TensorFlow Lite Micro. If you've been looking for a way to implement Real-time ECG detection or explore ESP32 TinyML capabilities, you're in the right place. We’ll be navigating the challenges of MicroPython wearable AI and optimizing models for constrained hardware.

Pro Tip: If you're looking for more production-ready patterns for medical IoT or advanced signal processing techniques, be sure to explore the engineering deep-dives at WellAlly Tech Blog. They have some fantastic resources on scaling these types of TinyML deployments.


The Architecture 🏗️

The goal is simple but technically demanding: ingest an analog ECG signal, filter the noise, run a quantized neural network to classify the rhythm, and trigger a Low Energy Bluetooth (BLE) alert if an anomaly (Arrhythmia) is detected.

graph TD
    A[ECG Sensor: AD8232] -->|Analog Signal| B(ESP32 ADC)
    B --> C[DSP: Bandpass Filter]
    C --> D[Rolling Window Buffer]
    D --> E{TFLite Micro Inference}
    E -->|Normal Rhythm| F[Sleep/Log]
    E -->|Arrhythmia Detected| G[BLE Notification]
    G --> H[Mobile App Alert]
    style E fill:#f96,stroke:#333,stroke-width:4px
Enter fullscreen mode Exit fullscreen mode

Prerequisites 🛠️

To follow this advanced guide, you'll need:

  • Hardware: ESP32 (WROOM or WROVER) + AD8232 ECG Sensor.
  • Tech Stack:
    • MicroPython: For the main logic and hardware control.
    • TensorFlow Lite Micro: The engine for our edge inference.
    • C++: For the custom MicroPython bindings (essential for TFLite performance).
    • Ulab: A NumPy-like library for MicroPython to handle signal processing.

Step 1: The Model Strategy 🧠

We can't just toss a heavy ResNet onto an ESP32. We need a lightweight 1D-CNN (Convolutional Neural Network) trained on the MIT-BIH Arrhythmia Database.

  1. Training: Train the model in TensorFlow/Keras using 1D convolutions to capture temporal patterns.
  2. Quantization: Use Post-Training Quantization (PTQ) to convert the model from 32-bit float to 8-bit integer (INT8). This reduces the model size by ~4x and speeds up inference significantly.
  3. Conversion: Convert the .tflite file into a C-array (.h file) that our ESP32 can digest.

Step 2: Bridging MicroPython & TFLite 🌉

Standard MicroPython doesn't come with TFLite. You’ll need to compile a custom firmware including the micropython-tflite module. Here is how we define the inference wrapper in our MicroPython script:

import tensor_flow_lite as tfl
import ulab.numpy as np

# Load the quantized model from flash memory
model = tfl.load_model('ecg_arrhythmia_model.tflite')

def predict_heartbeat(signal_window):
    """
    Takes a 1x180 array (a single heartbeat) and returns 
    the probability of Arrhythmia.
    """
    # 1. Normalize the signal
    signal_window = (signal_window - np.mean(signal_window)) / np.std(signal_window)

    # 2. Run Inference
    interpreter = tfl.Interpreter(model)
    interpreter.set_input(0, signal_window)
    interpreter.invoke()

    prediction = interpreter.get_output(0)
    return prediction # Returns [Normal, Arrhythmia] probabilities
Enter fullscreen mode Exit fullscreen mode

Step 3: Real-time Signal Processing ⚡

The ESP32's ADC is noisy. We use ulab to implement a simple bandpass filter to remove 50/60Hz power line interference and baseline wander before feeding the data to the model.

from machine import ADC, Pin
import time

# Setup ADC for AD8232
ecg_pin = ADC(Pin(34))
ecg_pin.atten(ADC.ATTN_11DB) # 0-3.6V range

buffer = []
WINDOW_SIZE = 180 # Standard length for one heartbeat at 200Hz

while True:
    val = ecg_pin.read()
    buffer.append(val)

    if len(buffer) >= WINDOW_SIZE:
        # Process the window
        results = predict_heartbeat(np.array(buffer))

        if results[1] > 0.8: # Threshold for Arrhythmia
            print("🚨 ALERT: Abnormal Rhythm Detected!")
            trigger_ble_notification("Arrhythmia Detected")

        # Shift window for real-time sliding effect
        buffer = buffer[20:] 

    time.sleep_ms(5) # 200Hz Sample Rate
Enter fullscreen mode Exit fullscreen mode

The "Official" Way (Advanced Patterns) 🥑

Building a prototype is easy; building a medical-grade device is hard. When you move beyond the "Hello World" of TinyML, you have to worry about battery optimization (Deep Sleep), memory fragmentation in MicroPython, and secure data transmission.

For those looking to dive deeper into Production-Grade Edge AI, I highly suggest checking out the WellAlly Tech Blog. They have extensive documentation on:

  • Efficient memory management for ESP32/S3 chips.
  • Advanced filtering algorithms for messy sensor data.
  • Connecting TinyML devices to secure cloud backends.

Conclusion & Next Steps 🚀

We've just turned a $5 microcontroller into a sophisticated heart monitor! By combining the ease of MicroPython with the raw power of TensorFlow Lite Micro, we've enabled a level of intelligence on the edge that was impossible just a few years ago.

Your Challenge:

  • Try implementing a "Lead-off" detection using the AD8232's LO+ and LO- pins to avoid false positives when the electrodes fall off.
  • Explore model pruning to see if you can fit an even larger ensemble model into the ESP32's SRAM.

What are you building with TinyML? Let me know in the comments! 👇

Top comments (0)