"Tutorial: Predictive Maintenance with Vibration Analysis"

Introduction

Unplanned downtime in rotating machinery costs industrial facilities millions of pounds per year. This tutorial shows you how to build a predictive maintenance pipeline on a Raspberry Pi 4 using an ADXL345 accelerometer, compiled anomaly-detection hot-paths, and built-in cost modelling. By the end you will have a native-compiled rule that detects bearing degradation before it becomes catastrophic.

Hardware Setup

ADXL345 Accelerometer

The ADXL345 is a low-power 3-axis MEMS accelerometer with I2C and SPI interfaces. For this tutorial, wire it to the Pi via I2C:

• VCC → 3.3 V (Pin 1)
• GND → Ground (Pin 6)
• SDA → GPIO 2 (Pin 3)
• SCL → GPIO 3 (Pin 5)

Baseline Vibration Profile

Before you can detect anomalies, you need a baseline. Collect RMS acceleration values during normal operation. The formula below is implemented in pure Python first; later we compile it to a native .so.

import math
from pyvorin_edge.sensors import Sensor, SensorType, SensorReading


def rms(values):
    """Root-mean-square of a sequence."""
    return math.sqrt(sum(v * v for v in values) / len(values))


# Simulate 1 kHz samples for 1 second (1000 points)
normal_samples = [0.01, -0.02, 0.015, -0.01, 0.005] * 200
baseline_rms = rms(normal_samples)
print(f"Baseline RMS: {baseline_rms:.4f} g")
  

Compiled Hot-Path

The @edge decorator is the primary compilation entry point in the Pyvorin Edge SDK. In practice, you decorate a hot-path function with @edge; the compiler intercepts the definition, parses its AST, and produces a native .so. For explicit control, you can also invoke CompilerBridge.compile_hotpath() directly, as shown here.

from pathlib import Path
from pyvorin_edge.compiler_bridge import CompilerBridge

# In production, decorate with @edge; here we compile explicitly via CompilerBridge.
source_code = '''
def bearing_anomaly(rms_value: float, threshold: float) -> bool:
    return rms_value > threshold
'''

bridge = CompilerBridge()
so_path = bridge.compile_hotpath(
    source_code=source_code,
    function_name="bearing_anomaly",
    output_path=Path("/tmp/edge_modules/bearing_anomaly.so"),
)
print(f"Compiled .so written to: {so_path}")

# Validate against the reference Python implementation
def reference_anomaly(rms_value: float, threshold: float) -> bool:
    return rms_value > threshold

report = bridge.validate_compilation(
    so_path,
    reference_anomaly,
    test_inputs=[(0.05, 0.07), (0.10, 0.07), (0.06, 0.07)],
)
print(report)
# {'passed': 3, 'failed': 0, 'mismatches': []}
  

Threshold Tuning

ISO 10816 provides velocity-based vibration severity guidelines. For small machines mounted on rigid foundations, the "warning" boundary is roughly 0.71 mm/s RMS and "critical" is 1.12 mm/s RMS. When converted to acceleration at typical bearing frequencies (~60 Hz), these map to approximately 0.07 g and 0.11 g. We use multipliers of the baseline to adapt to each individual machine.

# Adaptive thresholds derived from the baseline profile
thresholds = {
    "warning": baseline_rms * 2.0,
    "critical": baseline_rms * 3.5,
}

print(f"Warning threshold:  {thresholds['warning']:.4f} g")
print(f"Critical threshold: {thresholds['critical']:.4f} g")
  

Pipeline Integration

We wire the thresholds into a Pipeline using RuleConfig. The pipeline runs on the Edge Agent and evaluates every incoming reading against the rules.

from pyvorin_edge.pipeline import Pipeline, WindowConfig, RuleConfig
from pyvorin_edge.sensors import Sensor, SensorType, SensorReading
import time

pipeline = Pipeline(name="predictive_maintenance")
pipeline.add_sensor(
    Sensor(name="bearing_x", sensor_type=SensorType.VIBRATION, unit="g",
           normal_range=(0.0, 0.2), location="Motor_A")
)

# 60-second rolling window for RMS trend analysis
pipeline.add_window(
    WindowConfig(duration_seconds=60.0, sensor_name="bearing_x", window_type="rolling")
)

# Rules with severity-appropriate cooldowns
pipeline.add_rule(
    RuleConfig(
        name="bearing_warning",
        condition_expr="ctx.value > 0.07",
        severity="warning",
        cooldown_seconds=60.0,
    )
)
pipeline.add_rule(
    RuleConfig(
        name="bearing_critical",
        condition_expr="ctx.value > 0.11",
        severity="critical",
        cooldown_seconds=10.0,
    )
)

# Simulate a run
now = time.time()
readings = [
    SensorReading(sensor_name="bearing_x", timestamp=now, value=0.05, unit="g"),
    SensorReading(sensor_name="bearing_x", timestamp=now + 1.0, value=0.12, unit="g"),
]
result = pipeline.run(readings)
for ev in result.events:
    print(f"[{ev.severity.upper()}] {ev.rule_name} at {ev.timestamp}")
  

Cost Savings Report

Uploading raw 1 kHz vibration streams to the cloud is prohibitively expensive. By summarising at the edge—keeping only RMS, peak, and alarm counts—you reduce traffic by orders of magnitude. The CostModel and ReductionReport classes quantify this saving.

from pyvorin_edge.cost_model import CostModel, TrafficModel
from pyv_edge_agent.reports.reduction_report import ReductionReport, DataStream

# Traffic model: one sensor streaming 1 kHz raw vs one summary per minute
traffic = TrafficModel(
    properties=1,
    sensors_per_property=1,
    readings_per_sensor_per_day=86_400_000,  # 1 kHz
    raw_payload_bytes=64,
    edge_summaries_per_sensor_per_day=1440,   # one per minute
    edge_payload_bytes=256,
)

model = CostModel(traffic)
raw_cost = model.total_monthly_cost_raw()
edge_cost = model.total_monthly_cost_edge()

print(f"Raw monthly cost:  £{raw_cost:.2f}")
print(f"Edge monthly cost: £{edge_cost:.2f}")
print(f"Savings:           £{model.cost_savings():.2f}")

raw_stream = DataStream(
    byte_counts=[traffic.raw_bytes_per_month()],
    message_counts=[traffic.raw_messages_per_month()],
)
edge_stream = DataStream(
    byte_counts=[traffic.edge_bytes_per_month()],
    message_counts=[traffic.edge_messages_per_month()],
)

report = ReductionReport(
    raw_stream=raw_stream,
    edge_stream=edge_stream,
    raw_cost=raw_cost,
    edge_cost=edge_cost,
)
print(report.to_markdown())
  

Complete Working Script

Save the following as predictive_maintenance.py. It combines baseline profiling, compiled hot-path generation, pipeline execution, and cost reporting in one file.

#!/usr/bin/env python3
"""Predictive Maintenance — complete working example."""

import math
import time
from pathlib import Path
from pyvorin_edge.pipeline import Pipeline, WindowConfig, RuleConfig
from pyvorin_edge.sensors import Sensor, SensorType, SensorReading
from pyvorin_edge.compiler_bridge import CompilerBridge
from pyvorin_edge.cost_model import CostModel, TrafficModel
from pyv_edge_agent.reports.reduction_report import ReductionReport, DataStream


def rms(values):
    return math.sqrt(sum(v * v for v in values) / len(values))


def main():
    # 1. Baseline
    normal_samples = [0.01, -0.02, 0.015, -0.01, 0.005] * 200
    baseline_rms = rms(normal_samples)
    print(f"Baseline RMS: {baseline_rms:.4f} g")

    # 2. Compile hot-path
    source = '''
def bearing_anomaly(rms_value: float, threshold: float) -> bool:
    return rms_value > threshold
'''
    bridge = CompilerBridge()
    so_path = bridge.compile_hotpath(source, "bearing_anomaly",
                                      Path("/tmp/edge_modules/bearing_anomaly.so"))
    print(f"Compiled module: {so_path}")

    # 3. Pipeline
    pipeline = Pipeline(name="predictive_maintenance")
    pipeline.add_sensor(
        Sensor(name="bearing_x", sensor_type=SensorType.VIBRATION, unit="g",
               normal_range=(0.0, 0.2), location="Motor_A")
    )
    pipeline.add_window(
        WindowConfig(duration_seconds=60.0, sensor_name="bearing_x", window_type="rolling")
    )
    pipeline.add_rule(
        RuleConfig(name="bearing_warning", condition_expr="ctx.value > 0.07",
                   severity="warning", cooldown_seconds=60.0)
    )
    pipeline.add_rule(
        RuleConfig(name="bearing_critical", condition_expr="ctx.value > 0.11",
                   severity="critical", cooldown_seconds=10.0)
    )

    now = time.time()
    readings = [
        SensorReading(sensor_name="bearing_x", timestamp=now, value=0.05, unit="g"),
        SensorReading(sensor_name="bearing_x", timestamp=now + 1.0, value=0.12, unit="g"),
    ]
    result = pipeline.run(readings)
    print(f"Events: {len(result.events)}")
    for ev in result.events:
        print(f"  [{ev.severity}] {ev.rule_name}")

    # 4. Cost report
    traffic = TrafficModel(
        properties=1, sensors_per_property=1,
        readings_per_sensor_per_day=86_400_000,
        raw_payload_bytes=64,
        edge_summaries_per_sensor_per_day=1440,
        edge_payload_bytes=256,
    )
    model = CostModel(traffic)
    raw_stream = DataStream(
        byte_counts=[traffic.raw_bytes_per_month()],
        message_counts=[traffic.raw_messages_per_month()],
    )
    edge_stream = DataStream(
        byte_counts=[traffic.edge_bytes_per_month()],
        message_counts=[traffic.edge_messages_per_month()],
    )
    report = ReductionReport(
        raw_stream=raw_stream, edge_stream=edge_stream,
        raw_cost=model.total_monthly_cost_raw(),
        edge_cost=model.total_monthly_cost_edge(),
    )
    print(report.to_markdown())


if __name__ == "__main__":
    main()
  

Summary

You now have a predictive-maintenance pipeline that profiles vibration baselines, compiles anomaly detection to native code, evaluates rules with cooldown semantics, and quantifies cloud cost savings. In production, swap the synthetic normal_samples for real ADXL345 readings via an I2C adapter and stream alerts to your CMMS over MQTT.