Automate pump and fan vibration monitoring with an ESP32 accelerometer, InduQT and Claude

// Vibration with FFT band features (ISO-style velocity RMS, 1x/2x/3x, high-frequency band) → MQTT
// Board: ESP32-DevKitC. Arduino IDE 2.x, "esp32 by Espressif Systems" 3.x.
// Libraries: PubSubClient by Nick O'Leary, arduinoFFT by Enrique Condes (2.x)
// Sensor: ADXL345 on I2C (SDA 21, SCL 22), read directly at 400 kHz for a steady 2,048 Hz sample rate.
#include <WiFi.h>
#include <PubSubClient.h>
#include <Wire.h>
#include <arduinoFFT.h>

const char* WIFI_SSID     = "factory-wifi";
const char* WIFI_PASSWORD = "wifi-password";
const char* MQTT_HOST     = "192.168.1.20";   // the Raspberry Pi running Mosquitto
const int   MQTT_PORT     = 1883;
const char* MQTT_USER     = "sensors";
const char* MQTT_PASSWORD = "sensors-password";
const char* TOPIC         = "factory/pump-3/vibration";
const double RUNNING_HZ   = 24.75;  // shaft speed in Hz (rpm / 60); 0 if unknown
const uint8_t AXIS        = 0;      // 0 = X, 1 = Y, 2 = Z: the axis that points radially at the bearing

WiFiClient net;
PubSubClient mqtt(net);

void connectAll() {
  while (WiFi.status() != WL_CONNECTED) { delay(500); }
  while (!mqtt.connected()) {
    if (!mqtt.connect("pump-3-vibration", MQTT_USER, MQTT_PASSWORD)) delay(2000);
  }
}

const int    N  = 1024;             // 0.5 s of data, 2 Hz per bin
const double FS = 2048.0;           // samples per second (Nyquist 1,024 Hz)
const double G  = 9806.65;          // mm/s² per g
double vReal[N], vImag[N];
ArduinoFFT<double> fft(vReal, vImag, N, FS);

void adxlWrite(uint8_t reg, uint8_t v) { Wire.beginTransmission(0x53); Wire.write(reg); Wire.write(v); Wire.endTransmission(); }

double readAxisG() {
  Wire.beginTransmission(0x53); Wire.write(0x32 + 2 * AXIS); Wire.endTransmission(false);
  Wire.requestFrom(0x53, 2);
  int16_t raw = Wire.read() | (Wire.read() << 8);
  return raw * 0.0039;              // full resolution: 3.9 mg per count
}

void setup() {
  Serial.begin(115200);
  Wire.begin(); Wire.setClock(400000);
  adxlWrite(0x31, 0x0B);            // DATA_FORMAT: full resolution, ±16 g
  adxlWrite(0x2C, 0x0F);            // BW_RATE: 3,200 Hz output data rate
  adxlWrite(0x2D, 0x08);            // POWER_CTL: measure
  WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
  mqtt.setServer(MQTT_HOST, MQTT_PORT);
  mqtt.setBufferSize(512);
}

// Mean-square (g²) of one bin, scaled so the bins of a band add up to that band's mean square.
double binMs(int k, double s2) { return 2.0 * vReal[k] * vReal[k] / (N * s2); }

// Peak velocity (mm/s) of a component near f: the energy of ±3 bins, so off-bin speeds still read right.
double bandPeakV(double f, double s2) {
  int k = (int)round(f / (FS / N)); double ms = 0;
  for (int j = max(1, k - 3); j <= min(N / 2 - 1, k + 3); j++) { double fj = j * FS / N; ms += binMs(j, s2) * sq(G / (2 * PI * fj)); }
  return sqrt(ms) * sqrt(2.0);
}

void loop() {
  connectAll();
  // 1. Sample at a fixed rate.
  unsigned long next = micros();
  for (int i = 0; i < N; i++) {
    while ((long)(micros() - next) < 0) {}
    vReal[i] = readAxisG(); vImag[i] = 0;
    next += 488;                    // 1e6 / 2048 µs, rounded
  }
  // 2. Remove the mean (gravity), apply a Hann window and keep its sums for scaling.
  double mean = 0; for (int i = 0; i < N; i++) mean += vReal[i]; mean /= N;
  double s2 = 0;
  for (int i = 0; i < N; i++) { double w = 0.5 - 0.5 * cos(2 * PI * i / (N - 1)); vReal[i] = (vReal[i] - mean) * w; s2 += w * w; }
  // 3. FFT → magnitudes in vReal[0..N/2].
  fft.compute(FFTDirection::Forward);
  fft.complexToMagnitude();
  // 4. Band features.
  double velMs = 0, accMs = 0, hfMs = 0, peakV = 0, peakHz = 0;
  for (int k = 1; k < N / 2; k++) {
    double f = k * FS / N, a = binMs(k, s2);
    if (f >= 10 && f <= 1000) {
      double v = a * sq(G / (2 * PI * f));
      velMs += v; accMs += a;
      if (v > peakV) { peakV = v; peakHz = f; }
    }
    if (f >= 500 && f <= 1000) hfMs += a;
  }
  char json[300];
  int n = snprintf(json, sizeof json, "{\"vel_rms_mm_s\":%.3f,\"acc_rms_g\":%.4f,\"acc_hf_rms_g\":%.4f,\"peak_hz\":%.1f",
                   sqrt(velMs), sqrt(accMs), sqrt(hfMs), peakHz);
  if (RUNNING_HZ > 0) {
    n += snprintf(json + n, sizeof json - n, ",\"v1x_mm_s\":%.3f,\"v2x_mm_s\":%.3f,\"v3x_mm_s\":%.3f",
                  bandPeakV(RUNNING_HZ, s2), bandPeakV(2 * RUNNING_HZ, s2), bandPeakV(3 * RUNNING_HZ, s2));
  }
  snprintf(json + n, sizeof json - n, "}");
  mqtt.publish(TOPIC, json);
  mqtt.loop();
  delay(4000);
}

#!/usr/bin/env python3
"""
induqt_bridge.py — forward MQTT sensor messages to InduQT.

Subscribes to your local broker, turns each message into numbers, and posts them
to the InduQT ingest API in batches. Works on every InduQT plan, including the free one,
and your broker never has to be reachable from the internet.

  factory/<machine>/<anything>        ->  machine = <machine> (matched to the machine's externalId)
  factory/pump-3/vibration {"vib_rms_x_g": 0.12}  ->  signal vib_rms_x_g
  factory/comp-1/status/em:0 {"a_current": 12.4}  ->  signal em_0.a_current
      (two or more levels after the machine: the last level names the group, so
       two probes or two channels never overwrite each other)
  {"temperature": {"tC": 41.2}}       ->  signal temperature.tC
  factory/furnace-1/temp  21.7        ->  signal temp (a bare number takes the last level)

pip install "paho-mqtt>=2.0" requests
"""
import json
import os
import re
import threading
import time

import paho.mqtt.client as mqtt
import requests

MQTT_HOST = os.environ.get("MQTT_HOST", "localhost")
MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883"))
MQTT_USER = os.environ.get("MQTT_USER")
MQTT_PASSWORD = os.environ.get("MQTT_PASSWORD")
TOPICS = [t.strip() for t in os.environ.get("MQTT_TOPICS", "factory/#").split(",") if t.strip()]
MACHINE_LEVEL = int(os.environ.get("MACHINE_LEVEL", "1"))  # factory/<machine>/... is level 1
SEND_EVERY = float(os.environ.get("SEND_EVERY", "10"))     # seconds between posts
IGNORE = set(os.environ.get("IGNORE_FIELDS", "id,ts,time,timestamp,unixtime,Time").split(","))
INGEST_URL = os.environ.get("INDUQT_URL", "https://api.induqt.ai/v1/ingest")
API_KEY = os.environ["INDUQT_API_KEY"]  # a site key from InduQT (Settings -> API keys)

BAD = re.compile(r"[^A-Za-z0-9_.\-]")
buffer, lock = [], threading.Lock()


def key(name):
    return BAD.sub("_", name)[:64]


def numbers(prefix, value, out):
    """Collect every number in a JSON value; true/false become 1/0."""
    if isinstance(value, bool):
        out[key(prefix)] = 1 if value else 0
    elif isinstance(value, (int, float)):
        out[key(prefix)] = value
    elif isinstance(value, str):
        text = value.strip().lower()
        if text in ("on", "true", "online"):
            out[key(prefix)] = 1
        elif text in ("off", "false", "offline"):
            out[key(prefix)] = 0
        else:
            try:
                out[key(prefix)] = float(text)
            except ValueError:
                pass
    elif isinstance(value, dict):
        for k, v in value.items():
            if k not in IGNORE:
                numbers(f"{prefix}.{k}" if prefix else k, v, out)


def on_connect(client, userdata, flags, reason_code, properties):
    print(f"MQTT connected ({reason_code}); subscribing to {TOPICS}")
    for t in TOPICS:
        client.subscribe(t, qos=1)


def on_message(client, userdata, msg):
    levels = msg.topic.split("/")
    if len(levels) <= MACHINE_LEVEL:
        return
    machine = levels[MACHINE_LEVEL]
    try:
        payload = json.loads(msg.payload.decode("utf-8", "replace"))
    except ValueError:
        payload = msg.payload.decode("utf-8", "replace")
    values = {}
    group = levels[-1] if len(levels) - MACHINE_LEVEL > 2 else ""
    if isinstance(payload, dict):
        numbers(group, payload, values)
    else:
        numbers(levels[-1], payload, values)
    if values:
        with lock:
            buffer.append({"machine": machine, "ts": int(time.time() * 1000), "values": values})


def sender():
    while True:
        time.sleep(SEND_EVERY)
        with lock:
            batch = buffer[:500]
        if not batch:
            continue
        try:
            r = requests.post(
                INGEST_URL,
                headers={"Authorization": f"Bearer {API_KEY}"},
                json={"readings": batch, "autoCreate": True},
                timeout=15,
            )
            if r.status_code < 300:
                with lock:
                    del buffer[: len(batch)]
                print(f"sent {len(batch)} readings: {r.json()}")
            elif r.status_code in (400, 401, 403):
                print(f"InduQT refused the batch ({r.status_code}): {r.text}")
                with lock:
                    del buffer[: len(batch)]  # fix the key or mapping; retrying will not help
            else:
                print(f"InduQT returned {r.status_code}; will retry")
        except requests.RequestException as e:
            print(f"network error, will retry: {e}")
        with lock:
            del buffer[:-5000]  # keep at most ~5,000 readings while offline


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="induqt-bridge")
if MQTT_USER:
    client.username_pw_set(MQTT_USER, MQTT_PASSWORD)
client.on_connect = on_connect
client.on_message = on_message
threading.Thread(target=sender, daemon=True).start()
client.connect(MQTT_HOST, MQTT_PORT, keepalive=60)
client.loop_forever()

# /etc/systemd/system/induqt-bridge.service  — keeps the bridge running after a reboot
[Unit]
Description=MQTT to InduQT bridge
After=network-online.target mosquitto.service

[Service]
User=pi
EnvironmentFile=/home/pi/induqt-bridge.env
ExecStart=/home/pi/induqt/bin/python3 /home/pi/induqt_bridge.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

# then:
# sudo systemctl daemon-reload && sudo systemctl enable --now induqt-bridge
# journalctl -u induqt-bridge -f