Automate air compressor monitoring with a Shelly Pro 3EM, InduQT and Claude

#!/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