If you've ever tried building a fitness app or working with hardware, you know the "Dirty Secret" of the industry: sensor data is incredibly messy.
Whether it's a $10 DIY pulse sensor or a mid-range wearable, Photoplethysmogram (PPG) signals are notorious for "motion artifacts." Every time you move your wrist, the sensor loses contact, creating massive spikes that look more like a mountain range than a heartbeat. In the world of Data Engineering and Signal Processing, dealing with these outliers is the difference between a high-precision health tool and a random number generator.
Today, we're going to dive into a "Learning in Public" project where we combine NumPy and SciPy to build a dual-stage cleaning pipeline using Sliding Window Median Filters and the legendary Kalman Filter.
Pro Tip: If you're looking for more production-ready examples and advanced architectural patterns for data pipelines, I highly recommend checking out the deep dives over at the WellAlly Blog. It’s where I get most of my inspiration for robust system design!
The Architecture: A Two-Stage Defense 🛡️
When dealing with raw PPG signals, a single filter usually isn't enough. We need a strategy to first "crush" the extreme spikes and then "smooth" the remaining jitter.
graph TD
A[Raw PPG Signal] --> B[Sliding Window Median Filter]
B -- Remove Extreme Outliers --> C[Kalman Filter]
C -- Predictive Smoothing --> D[Clean Heart Rate Estimation]
D --> E[Final Dashboard/Storage]
style B fill:#f96,stroke:#333
style C fill:#69f,stroke:#333
- Median Filter: This handles "Salt and Pepper" noise—those random 0s or 1000s that happen when the sensor glitches.
- Kalman Filter: This is a recursive mathematical power-house that predicts the "true state" of the signal based on previous measurements and the current estimate.
Prerequisites 🛠️
We’ll keep it lean. You just need the standard Python data science stack:
pip install numpy scipy matplotlib
Step 1: Simulating "Dirty" PPG Data
First, let's create a synthetic heart rate signal. Real heartbeats are roughly sinusoidal, but we'll add high-frequency noise and some "motion artifacts" (giant spikes).
import numpy as np
import matplotlib.pyplot as plt
from scipy.signal import medfilt
# 1. Generate a clean sine wave (simulating 70 BPM)
fs = 100 # Sampling frequency (Hz)
t = np.linspace(0, 10, 10 * fs)
clean_signal = np.sin(2 * np.pi * 1.1 * t)
# 2. Add Gaussian noise + Random Outliers (The "Motion Artifacts")
noise = np.random.normal(0, 0.2, len(t))
dirty_signal = clean_signal + noise
# Simulate 5 massive spikes (sensor losing contact)
outlier_indices = np.random.choice(len(t), 5, replace=False)
dirty_signal[outlier_indices] *= 10.0
print(f"🚀 Generated {len(t)} samples of messy PPG data.")
Step 2: The First Defense (Median Filter)
A Median Filter is brilliant because it doesn't average values (which would pull the result toward the outlier). It picks the middle value in a window, effectively deleting spikes.
# Apply a sliding window median filter
# Window size of 5 usually works well for high-frequency PPG
denoised_median = medfilt(dirty_signal, kernel_size=5)
Step 3: The Secret Sauce (The Kalman Filter)
The Kalman Filter is famous for its role in the Apollo moon landings, but it's perfect for heart rates too! It operates in a Predict -> Correct cycle.
class SimpleKalman:
def __init__(self, process_variance, measurement_variance, estimate_error, initial_value):
self.q = process_variance # How much we trust our model
self.r = measurement_variance # How much we trust the sensor
self.p = estimate_error # Initial confidence
self.x = initial_value # Initial state
def update(self, measurement):
# 1. Prediction Phase
# (State stays same in simple 1D model)
self.p = self.p + self.q
# 2. Measurement Update (Correction)
kalman_gain = self.p / (self.p + self.r)
self.x = self.x + kalman_gain * (measurement - self.x)
self.p = (1 - kalman_gain) * self.p
return self.x
# Initialize Kalman with some "tuning" parameters
kf = SimpleKalman(process_variance=1e-5, measurement_variance=0.1, estimate_error=1, initial_value=0)
kalman_output = [kf.update(val) for val in denoised_median]
Visualizing the Results 📊
Let's see how well our pipeline performed compared to the raw, noisy input.
plt.figure(figsize=(12, 6))
plt.plot(t, dirty_signal, label='Raw Signal (with Outliers)', alpha=0.3, color='red')
plt.plot(t, kalman_output, label='Filtered Signal (Kalman)', color='teal', linewidth=2)
plt.title("PPG Signal Denoising: Raw vs. Kalman Filtered")
plt.xlabel("Time (s)")
plt.ylabel("Amplitude")
plt.legend()
plt.grid(True)
plt.show()
Why This Matters for Data Engineering 🥑
In a real-world Data Engineering pipeline, you wouldn't just run this on your laptop. You would likely implement this logic inside a stream processing engine like Apache Flink or as a pre-processing step in your IoT Gateway.
By filtering at the edge (or close to it), you:
- Save Bandwidth: You don't need to send "trash" data to the cloud.
- Improve Downstream Accuracy: Your ML models for sleep tracking or stress detection won't be confused by motion artifacts.
For those interested in how to scale these signal processing algorithms across millions of devices, the team at wellally.tech/blog has some fantastic resources on building resilient data architectures that handle "noisy" inputs at scale.
Conclusion 🏁
We've successfully taken a jittery, outlier-ridden sensor signal and transformed it into a smooth, usable waveform using just a few lines of NumPy and some Kalman logic.
Next Steps:
- Try adjusting the
process_variance (q)in the Kalman filter. What happens if you set it to a very high value? (Spoiler: It trusts the sensor more and becomes "noisier"). - Try implementing this in a real-time loop with a Raspberry Pi and a PulseSensor!
What's your favorite way to handle sensor noise? Drop a comment below! 👇
Top comments (0)