Practical insights on processing high frequency sensor streams, Fast Fourier Transform edge processing, and predictive AI inference for industrial assets
tags: iot, python, architecture, machinelearning
Handling time-series data in Industrial IoT (IIoT) is not the same as standard API metrics. When you're monitoring high rpm industrial turbines or hydraulic pumps, accelerometers can have thousands of samples per second (1 kHz to 20 kHz)
Trying to directly stream these arrays of vibration data over cellular or local WAN to a central cloud database is a quick way to burn through bandwidth.
Modern IIoT platforms implement Edge-to-Cloud processing pipelines that decouple processing at the edge from predictive ML in the cloud. In this article, we'll explore how edge gateways perform signal processing and how these clean telemetry streams get ingested to predictive machine learning models.
Sensor Layer: High frequency tri-axial accelerometers and thermal sensors capture physical motion
Edge Processing Node: Converts raw time-domain waveforms into frequency-domain metrics (FFT) to extract features like RMS velocity and peak frequency
Cloud & AI Layer: Scores extracted features against anomaly detection models to provide real-time machine health scores
Step 1: Converting Raw Waveforms with Edge FFT
Instead of transmitting 10,000 raw amplitude points, the edge device transforms time-domain signals into spectral frequency metrics:
Here is a simple Python code snippet showing how edge gateways process raw accelerometer streams using numpy:
Python
import numpy as np
def extract_vibration_features(raw_samples: np.ndarray, sampling_rate: int) -> dict:
"""
Transforms raw time-series vibration data into frequency-domain features.
"""
RMS (Root Mean Square) Velocity
rms_velocity = np.sqrt(np.mean(raw_samples2))-
Fast Fourier Transform (FFT)
fft_values = np.fft.rfft(raw_samples)
frequencies = np.fft.rfftfreq(len(raw_samples), 1 / sampling_rate)
spectral_amplitudes = np.abs(fft_values)- Peak Frequency Component peak_freq_idx = np.argmax(spectral_amplitudes) dominant_frequency = frequencies[peak_freq_idx]
return {
"rms_velocity": float(rms_velocity),
"dominant_frequency_hz": float(dominant_frequency),
"peak_amplitude": float(spectral_amplitudes[peak_freq_idx])
}
Example: 1,000 samples collected over 1 second at 1 kHz
simulated_signal = np.sin(2 np.pi 60 np.linspace(0, 1, 1000)) + np.random.normal(0, 0.1, 1000)
features = extract_vibration_features(simulated_signal, sampling_rate=1000)
print("Extracted Edge Features:", features)
You can see how the gateway reduces a large array of 1,000+ raw floats into a tiny structured payload of just a few health metrics (∼100 bytes).
Step 2: MQTT Telemetry Streaming
Once features are extracted at the edge, we publish them to MQTT with a simple JSON payload:
JSON
{
"asset_id": "pump_station_04",
"component": "bearing_casing_a",
"timestamp": 1718293021,
"metrics": {
"rms_velocity": 4.12,
"dominant_freq_hz": 59.8,
"casing_temp_c": 68.4
}
}
This allows edge gateways to stream updates even across low bandwidth industrial cellular or mesh networks.
Step 3: Anomaly Detection and Predictive Inference
In the cloud, these feature payloads get ingested into time-series anomaly detection algorithms (e.g., Isolation Forests or Autoencoders). When a bearing begins to degrade, its dominant frequency shifts and its RMS velocity spikes above baseline thresholds.
Engineering teams can study predictive maintenance architectures, like those used in PlantLog AI, to see how pre-trained models ingest time-series telemetry and predictively alert before structural failures occur.
Key Takeaways for Developers
Filter at the Edge: Don't stream raw 10 kHz data streams over WAN. Always first perform FFT feature extraction at the gateway
Standardize Data Formats: Use standardized lightweight JSON/Protobuf schemas across edge nodes
Decouple Ingestion: Use lightweight MQTT brokers (Mosquitto or EMQX) to handle reconnects for hardware devices
Top comments (0)