Streamlit MQTT Dashboard in reinem Python

Mit einem Streamlit MQTT Dashboard verwandeln Sie Live-Sensornachrichten in Diagramme und Kennzahlen, ohne eine Zeile HTML oder JavaScript zu schreiben. In diesem Tutorial bauen Sie eines in reinem Python: einen paho-mqtt-Subscriber im Hintergrund, verbunden mit broker.connectmq.com, einen threadsicheren Puffer und eine Seite, die sich alle zwei Sekunden selbst aktualisiert. Außerdem bauen Sie einen kleinen Simulator, damit das Dashboard ab der ersten Minute Daten hat.

Warum Streamlit und MQTT etwas Klebstoff brauchen

Streamlit hat ein ungewöhnliches Ausführungsmodell. Jedes Mal, wenn ein Nutzer mit der Seite interagiert oder eine Aktualisierung ausgelöst wird, führt Streamlit Ihr gesamtes Skript von oben nach unten erneut aus. Für einfache Daten-Apps ist das großartig, MQTT-Clients sind jedoch langlebig: Sie halten eine TCP-Verbindung, betreiben eine Netzwerkschleife und lösen Callbacks aus, sobald eine Nachricht eintrifft.

Erzeugen Sie den MQTT-Client naiv am Anfang des Skripts, erzeugt jeder Rerun eine weitere Verbindung. Nach einer Minute haben Sie Dutzende Clients, doppelte Nachrichten und – auf dem öffentlichen Broker – Verbindungen, die wegen wiederverwendeter Client IDs getrennt werden. Die Lösung: Den Client genau einmal pro Serverprozess erzeugen und Streamlit bei jedem Rerun aus einem gemeinsamen Puffer lesen lassen. Streamlit liefert dafür das passende Werkzeug: st.cache_resource.

Die Architektur sieht so aus:

Komponente	Läuft wo	Aufgabe
paho-mqtt-Client	Netzwerk-Thread im Hintergrund (`loop_start`)	Hält die Verbindung, empfängt Nachrichten
`deque`-Puffer + Lock	Gemeinsamer Speicher, einmal erzeugt	Hält die letzten N Messwerte
Streamlit-Skript	Rerun bei jeder Aktualisierung	Kopiert den Puffer, baut einen DataFrame, zeichnet Diagramme
`st.fragment(run_every=...)`	Streamlit-Scheduler	Führt nur den Diagrammbereich alle paar Sekunden erneut aus

Voraussetzungen

Sie brauchen Python 3.9 oder neuer und drei Pakete:

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install "streamlit>=1.37" "paho-mqtt>=2.0" pandas

Streamlit 1.37 hat st.fragment mit dem Parameter run_every eingeführt, den dieses Tutorial für die automatische Aktualisierung nutzt. paho-mqtt 2.x verlangt die unten gezeigte explizite Callback-API-Version; kommen Sie von älterem Code, lesen Sie zuerst den Leitfaden zur Callback-API von paho-mqtt 2.0.

Wählen Sie ein zufälliges Topic-Präfix, damit sich Ihre Daten auf dem öffentlichen MQTT-Broker nicht mit denen anderer vermischen. Dieses Tutorial verwendet connectmq/demo/k7f3q2; ersetzen Sie k7f3q2 in beiden Dateien durch eine eigene Zufallszeichenfolge.

Schritt 1: Ein Sensorsimulator

Bevor Sie das Dashboard bauen, erstellen Sie etwas, das Daten publiziert. Speichern Sie dies als simulator.py:

import json
import math
import random
import time
import uuid

import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PREFIX = "connectmq/demo/k7f3q2"
SENSORS = ["line1", "line2", "line3"]

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"sim-{uuid.uuid4().hex[:8]}")
client.connect(BROKER, 1883, keepalive=60)
client.loop_start()

t = 0
try:
    while True:
        for sensor in SENSORS:
            reading = {
                "sensor": sensor,
                "temperature": round(22 + 3 * math.sin(t / 20) + random.uniform(-0.5, 0.5), 2),
                "humidity": round(45 + 5 * math.cos(t / 30) + random.uniform(-1, 1), 2),
                "ts": time.time(),
            }
            client.publish(f"{PREFIX}/sensors/{sensor}", json.dumps(reading), qos=0)
        t += 1
        time.sleep(1)
except KeyboardInterrupt:
    pass
finally:
    client.loop_stop()
    client.disconnect()

Er publiziert drei Nachrichten pro Sekunde, deutlich unter dem Limit des öffentlichen MQTT-Brokers von 10 Nachrichten pro Sekunde und Client. Starten Sie ihn in einem eigenen Terminal mit python simulator.py und lassen Sie ihn laufen.

Schritt 2: Das Streamlit MQTT Dashboard

Speichern Sie dies als dashboard.py:

import json
import threading
import time
import uuid
from collections import deque

import pandas as pd
import paho.mqtt.client as mqtt
import streamlit as st

BROKER = "broker.connectmq.com"
PORT = 1883
PREFIX = "connectmq/demo/k7f3q2"
TOPIC = f"{PREFIX}/sensors/+"
MAX_POINTS = 600


@st.cache_resource
def start_mqtt():
    """Einen MQTT-Client pro Streamlit-Serverprozess erzeugen."""
    state = {
        "buffer": deque(maxlen=MAX_POINTS),
        "lock": threading.Lock(),
        "connected": False,
        "received": 0,
    }

    def on_connect(client, userdata, flags, reason_code, properties):
        if reason_code.is_failure:
            return
        state["connected"] = True
        client.subscribe(TOPIC, qos=0)

    def on_disconnect(client, userdata, disconnect_flags, reason_code, properties):
        state["connected"] = False

    def on_message(client, userdata, msg):
        try:
            data = json.loads(msg.payload)
            row = {
                "time": pd.to_datetime(data.get("ts", time.time()), unit="s"),
                "sensor": data.get("sensor", msg.topic.rsplit("/", 1)[-1]),
                "temperature": float(data["temperature"]),
                "humidity": float(data["humidity"]),
            }
        except (ValueError, KeyError, TypeError):
            return  # fehlerhafte Payloads ignorieren
        with state["lock"]:
            state["buffer"].append(row)
            state["received"] += 1

    client = mqtt.Client(
        mqtt.CallbackAPIVersion.VERSION2,
        client_id=f"streamlit-{uuid.uuid4().hex[:8]}",
    )
    # Für einen privaten ConnectMQ-Broker:
    # client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
    # client.tls_set()  und Port 8883 verwenden
    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.on_message = on_message
    client.reconnect_delay_set(min_delay=1, max_delay=30)
    client.connect_async(BROKER, PORT, keepalive=60)
    client.loop_start()
    return state


st.set_page_config(page_title="MQTT Sensor Dashboard", layout="wide")
st.title("Live MQTT Sensor Dashboard")
st.caption(f"Subscribed to `{TOPIC}` on {BROKER}")

state = start_mqtt()


@st.fragment(run_every=2)
def live_view():
    with state["lock"]:
        rows = list(state["buffer"])
        received = state["received"]

    status = "Connected" if state["connected"] else "Connecting..."
    c1, c2, c3 = st.columns(3)
    c1.metric("Broker status", status)
    c2.metric("Messages received", received)
    c3.metric("Points in buffer", len(rows))

    if not rows:
        st.info("Waiting for data. Is simulator.py running with the same prefix?")
        return

    df = pd.DataFrame(rows)
    latest = df.groupby("sensor").last()

    cols = st.columns(len(latest))
    for col, (sensor, r) in zip(cols, latest.iterrows()):
        col.metric(f"{sensor} temperature", f"{r['temperature']:.1f} °C")

    temp = df.pivot_table(index="time", columns="sensor", values="temperature")
    hum = df.pivot_table(index="time", columns="sensor", values="humidity")

    left, right = st.columns(2)
    left.subheader("Temperature")
    left.line_chart(temp)
    right.subheader("Humidity")
    right.line_chart(hum)


live_view()

Starten Sie es mit:

streamlit run dashboard.py

Streamlit öffnet http://localhost:8501. Innerhalb weniger Sekunden zeigt der Status Connected, die Zähler steigen, und auf jedem Diagramm erscheinen drei Linien.

Wie das Dashboard funktioniert

Ein Client, viele Reruns

st.cache_resource führt start_mqtt() beim ersten Mal aus und liefert bei jedem späteren Aufruf dasselbe Objekt zurück – über Reruns und über Browser-Sitzungen auf demselben Server hinweg. Das ist genau die Lebensdauer, die eine MQTT-Verbindung braucht. Da das gecachte Objekt von allen Betrachtern geteilt wird, öffnet das Dashboard genau eine Broker-Verbindung, egal wie viele Personen zuschauen – wichtig bei Plänen mit Verbindungslimit.

Threadsicherheit

pahos loop_start() betreibt die Netzwerkschleife in einem Hintergrund-Thread, sodass on_message in einem anderen Thread läuft als Ihr Streamlit-Skript. Das Anhängen an eine deque ist in CPython atomar, aber das Kopieren, während ein anderer Thread anhängt, kann „deque mutated during iteration" auslösen. Der Lock schließt Kopieren und Anhängen gegenseitig aus, und da er nur Mikrosekunden gehalten wird, bremst er den Netzwerk-Thread nie aus.

Das maxlen der Deque ist Ihr Speicherschutz. Mit 600 Punkten zeigt das Dashboard ungefähr die letzten 200 Sekunden von drei Sensoren, und der Speicherverbrauch bleibt konstant, egal wie lange es läuft.

Teil-Reruns mit Fragments

@st.fragment(run_every=2) weist Streamlit an, nur diese Funktion alle zwei Sekunden erneut auszuführen statt des ganzen Skripts. Titel und Untertitel bleiben stehen, die Diagramme aktualisieren sich, und die Seite flackert nicht. In älteren Streamlit-Versionen bräuchten Sie eine Auto-Refresh-Komponente eines Drittanbieters oder eine Schleife aus time.sleep() und st.rerun() – beides umständlicher.

Reconnects

connect_async() plus loop_start() bedeutet, dass das Dashboard auch dann startet, wenn der Broker kurzzeitig nicht erreichbar ist, und paho es im Hintergrund mit den in reconnect_delay_set festgelegten Verzögerungen erneut versucht. Das Abonnieren in on_connect ist wichtig: Nach einem Reconnect mit Clean Session sind die Subscriptions weg, und in on_connect werden sie wiederhergestellt. Der Leitfaden zu Reconnects mit paho-mqtt behandelt dieses Muster ausführlicher.

Häufige Probleme und Lösungen

Doppelte oder mehrfache Nachrichten. Sie erzeugen den Client außerhalb von st.cache_resource, oder Sie haben start_mqtt bearbeitet, und Streamlit hat einen neuen gecachten Client erzeugt, während der alte Thread weiterläuft. Starten Sie den Prozess streamlit run neu, nachdem Sie das MQTT-Setup geändert haben.
Das Dashboard zeigt dauerhaft „Connecting...". Port 1883 ist in Ihrem Netzwerk möglicherweise blockiert. Wechseln Sie mit client.tls_set() auf TLS über 8883, oder testen Sie mit mosquitto_sub -h broker.connectmq.com -p 1883 -i test-k7f3q2 -t 'connectmq/demo/k7f3q2/#' -v.
Verbunden, aber keine Daten. Das Präfix in dashboard.py und simulator.py muss exakt übereinstimmen, einschließlich Groß- und Kleinschreibung.
Diagramme springen. Die Zeitstempel stammen vom Publisher. Haben Geräte unsynchronisierte Uhren, verwenden Sie stattdessen die Empfangszeit (time.time() in on_message).
Langsame Seite bei vielen Sensoren. Tausende Zeilen alle zwei Sekunden zu pivotieren summiert sich. Reduzieren Sie MAX_POINTS, erhöhen Sie run_every oder reduzieren Sie die Daten vor dem Zeichnen mit df.resample().

Ab in die Produktion

Der öffentliche MQTT-Broker ist ideal für Tests und Prototypen, aber alles, was dort veröffentlicht wird, ist für jeden lesbar. Für ein Dashboard mit echten Anlagen- oder Gebäudedaten, etwa Maschinendaten einer Fertigungslinie oder Werte aus dem Energiemonitoring, gilt:

Erstellen Sie einen privaten Broker auf ConnectMQ und verwenden Sie die generierten Zugangsdaten mit TLS auf Port 8883. Der Leitfaden zu Python, MQTT und TLS zeigt die genaue Konfiguration von tls_set().
Halten Sie Geheimnisse in .streamlit/secrets.toml und lesen Sie sie mit st.secrets["mqtt"]["username"], statt sie fest im Code zu hinterlegen.
Strukturieren Sie Topics im Werk von Anfang an nach der Anlagenhierarchie, etwa acme/werk-muenchen/montage/linie-1/temperatur. So kann das Dashboard später direkt einen Zweig eines Unified Namespace (UNS) abonnieren; Details zeigt Sparkplug B und Unified Namespace.
Brauchen Sie eine Historie über den In-Memory-Puffer hinaus, betreiben Sie einen separaten Subscriber, der in eine Datenbank schreibt, und lassen Sie das Dashboard diese abfragen. Das Tutorial zu MQTT und SQLite in Python ist ein schlanker Einstieg, und der Leitfaden zu MQTT und pandas hilft bei der Analyse.

Streamlit ist ein Python-Werkzeug, gehört aber in dieselbe Familie wie Browser-Dashboards. Bevorzugen Sie einen reinen Frontend-Ansatz, verbindet sich der MQTT-JavaScript-Client im Browser direkt aus der Seite über WebSocket, und die Übersicht über MQTT-Client-Bibliotheken vergleicht die wichtigsten Optionen in jeder Sprache. Die Streamlit-Dokumentation behandelt Fragments, Caching und Deployment im Detail.

Betreiben Sie Ihre Streamlit-Dashboards mit ConnectMQ

ConnectMQ ist ein verwalteter MQTT-Broker: Sie erhalten TLS auf 8883, generierte Zugangsdaten und ein Portal mit Live-Topic-Explorer und MQTT-Client im Browser, um zu prüfen, was Ihre Sensoren publizieren, bevor Sie es visualisieren. Beginnen Sie mit broker.connectmq.com und wechseln Sie dann auf Ihren eigenen Cloud-MQTT-Broker. Registrieren Sie sich für den kostenlosen Serverless-Plan, ohne Kreditkarte.

Häufig gestellte Fragen

Kann Streamlit MQTT direkt abonnieren?

Streamlit hat keine eingebaute MQTT-Unterstützung, aber jeder Python-MQTT-Client funktioniert. Das zuverlässige Muster ist ein paho-mqtt-Client, der einmalig mit st.cache_resource erzeugt wird und seine Netzwerkschleife in einem Hintergrund-Thread betreibt.

Wie oft sollte sich das Dashboard aktualisieren?

Eine bis fünf Sekunden passen für die meisten Sensor-Dashboards. Schnellere Aktualisierungen kosten CPU auf dem Streamlit-Server, ohne langsam veränderliche Werte nützlicher zu machen.

Warum erhalte ich jede Nachricht mehrmals?

Es laufen mehrere MQTT-Clients, meist weil der Client außerhalb von st.cache_resource erzeugt wurde oder ein alter gecachter Client eine Codeänderung überlebt hat. Starten Sie den Streamlit-Prozess neu, damit nur ein Client existiert.

Können viele Nutzer dasselbe Streamlit MQTT Dashboard ansehen?

Ja. Der gecachte Client wird von allen Sitzungen auf dem Server geteilt, sodass auch zehn Betrachter nur eine einzige Broker-Verbindung nutzen und aus demselben Puffer lesen.

Streamlit MQTT Dashboard in reinem Python: Live-Sensordiagramme in 100 Zeilen