Build a factory MQTT gateway on a Raspberry Pi

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

[
  {
    "id": "iqb_tab",
    "type": "tab",
    "label": "MQTT → InduQT",
    "disabled": false,
    "info": "Forwards factory/<machine>/... messages to the InduQT ingest API every 10 s. Set INDUQT_API_KEY in the Node-RED environment and the broker user in the \"Factory broker\" node."
  },
  {
    "id": "iqb_broker",
    "type": "mqtt-broker",
    "name": "Factory broker",
    "broker": "localhost",
    "port": "1883",
    "clientid": "induqt-node-red",
    "autoConnect": true,
    "usetls": false,
    "protocolVersion": "4",
    "keepalive": "60",
    "cleansession": true
  },
  {
    "id": "iqb_in",
    "type": "mqtt in",
    "z": "iqb_tab",
    "name": "factory/#",
    "topic": "factory/#",
    "qos": "1",
    "datatype": "auto-detect",
    "broker": "iqb_broker",
    "nl": false,
    "rap": true,
    "rh": 0,
    "inputs": 0,
    "x": 120,
    "y": 80,
    "wires": [
      [
        "iqb_collect"
      ]
    ]
  },
  {
    "id": "iqb_collect",
    "type": "function",
    "z": "iqb_tab",
    "name": "Collect readings",
    "func": "// Turn each MQTT message into an InduQT reading and keep it until the next send.\nconst levels = msg.topic.split('/');\nconst machine = levels[1];                       // factory/<machine>/...\nif (!machine) return null;\nconst IGNORE = ['id', 'ts', 'time', 'timestamp', 'unixtime', 'Time'];\n// Two or more levels after the machine (factory/comp-1/status/em:0): the last level names the group.\nconst values = {};\nconst key = (k) => k.replace(/[^A-Za-z0-9_.\\-]/g, '_').slice(0, 64);\n(function walk(prefix, v) {\n  if (typeof v === 'boolean') values[key(prefix)] = v ? 1 : 0;\n  else if (typeof v === 'number' && isFinite(v)) values[key(prefix)] = v;\n  else if (typeof v === 'string') {\n    const s = v.trim().toLowerCase();\n    if (s === 'on' || s === 'true') values[key(prefix)] = 1;\n    else if (s === 'off' || s === 'false') values[key(prefix)] = 0;\n    else if (s !== '' && !isNaN(Number(s))) values[key(prefix)] = Number(s);\n  } else if (v && typeof v === 'object' && !Array.isArray(v)) {\n    for (const [k, x] of Object.entries(v)) if (!IGNORE.includes(k)) walk(prefix ? prefix + '.' + k : k, x);\n  }\n})(typeof msg.payload === 'object' ? (levels.length > 3 ? levels[levels.length - 1] : '') : levels[levels.length - 1], msg.payload);\nif (!Object.keys(values).length) return null;\nconst buf = flow.get('induqt') || [];\nbuf.push({ machine, ts: Date.now(), values });\nflow.set('induqt', buf.slice(-5000));\nreturn null;",
    "outputs": 1,
    "timeout": 0,
    "noerr": 0,
    "initialize": "",
    "finalize": "",
    "libs": [],
    "x": 320,
    "y": 80,
    "wires": [
      []
    ]
  },
  {
    "id": "iqb_tick",
    "type": "inject",
    "z": "iqb_tab",
    "name": "Every 10 s",
    "props": [
      {
        "p": "payload"
      }
    ],
    "repeat": "10",
    "crontab": "",
    "once": false,
    "onceDelay": "1",
    "topic": "",
    "payload": "",
    "payloadType": "date",
    "x": 130,
    "y": 160,
    "wires": [
      [
        "iqb_flush"
      ]
    ]
  },
  {
    "id": "iqb_flush",
    "type": "function",
    "z": "iqb_tab",
    "name": "Build batch",
    "func": "// Every 10 s, send what has arrived as one batch.\nconst buf = flow.get('induqt') || [];\nif (!buf.length) return null;\nconst batch = buf.slice(0, 500);\nflow.set('induqt', buf.slice(batch.length));\nmsg.headers = {\n  'Authorization': 'Bearer ' + env.get('INDUQT_API_KEY'),\n  'Content-Type': 'application/json'\n};\nmsg.payload = { readings: batch, autoCreate: true };\nreturn msg;",
    "outputs": 1,
    "timeout": 0,
    "noerr": 0,
    "initialize": "",
    "finalize": "",
    "libs": [],
    "x": 320,
    "y": 160,
    "wires": [
      [
        "iqb_post"
      ]
    ]
  },
  {
    "id": "iqb_post",
    "type": "http request",
    "z": "iqb_tab",
    "name": "POST /v1/ingest",
    "method": "POST",
    "ret": "obj",
    "paytoqs": "ignore",
    "url": "https://api.induqt.ai/v1/ingest",
    "tls": "",
    "persist": false,
    "proxy": "",
    "insecureHTTPParser": false,
    "authType": "",
    "senderr": false,
    "headers": [],
    "x": 520,
    "y": 160,
    "wires": [
      [
        "iqb_debug"
      ]
    ]
  },
  {
    "id": "iqb_debug",
    "type": "debug",
    "z": "iqb_tab",
    "name": "InduQT response",
    "active": true,
    "tosidebar": true,
    "console": false,
    "tostatus": true,
    "complete": "payload",
    "targetType": "msg",
    "statusVal": "statusCode",
    "statusType": "msg",
    "x": 730,
    "y": 160,
    "wires": []
  }
]