Bir Streamlit MQTT dashboard, tek satır HTML veya JavaScript yazmadan canlı sensör mesajlarını grafiklere ve metriklere dönüştürmenizi sağlar. Bu eğitimde saf Python ile bir tane oluşturacaksınız: broker.connectmq.com adresine bağlı arka planda çalışan bir paho-mqtt abonesi, iş parçacığı açısından güvenli bir tampon ve kendini her iki saniyede bir yenileyen bir sayfa. Panonun ilk dakikadan itibaren verisi olması için küçük bir simülatör de geliştireceksiniz.
Streamlit ve MQTT Neden Biraz Yapıştırıcıya İhtiyaç Duyar?
Streamlit'in alışılmadık bir çalışma modeli vardır. Kullanıcı sayfayla her etkileşime girdiğinde veya bir yenileme tetiklendiğinde Streamlit tüm betiğinizi baştan sona yeniden çalıştırır. Bu basit veri uygulamaları için harikadır, ancak MQTT client'ları uzun ömürlüdür: bir TCP bağlantısı tutar, bir ağ döngüsü çalıştırır ve her mesaj geldiğinde callback'leri tetiklerler.
Betiğin en üstünde safça bir MQTT client'ı oluşturursanız her yeniden çalıştırma başka bir bağlantı oluşturur. Bir dakika içinde düzinelerce client'ınız, yinelenen mesajlarınız ve public broker'da client ID'lerin yeniden kullanılması nedeniyle düşürülen bağlantılarınız olur. Çözüm, client'ı sunucu süreci başına tam olarak bir kez oluşturmak ve Streamlit'in her yeniden çalıştırmada paylaşılan bir tampondan okumasını sağlamaktır. Streamlit bunun için doğru aracı sunar: st.cache_resource.
Mimari şöyle görünür:
| Bileşen | Nerede çalışır | Sorumluluk |
|---|---|---|
| paho-mqtt client | Arka plandaki ağ iş parçacığı (loop_start) |
Bağlantıyı sürdürür, mesajları alır |
deque tampon + kilit |
Paylaşılan bellek, bir kez oluşturulur | En son N ölçümü tutar |
| Streamlit betiği | Her yenilemede yeniden çalışır | Tamponu kopyalar, bir DataFrame oluşturur, grafikleri çizer |
st.fragment(run_every=...) |
Streamlit zamanlayıcısı | Birkaç saniyede bir yalnızca grafik bölümünü yeniden çalıştırır |
Ön Koşullar
Python 3.9 veya üzeri ve üç pakete ihtiyacınız var:
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, bu eğitimde otomatik yenileme için kullanılan run_every parametreli st.fragment özelliğini getirdi. paho-mqtt 2.x, aşağıda gösterilen açık callback API sürümünü gerektirir; eski koddan geliyorsanız önce paho-mqtt 2.0 callback API rehberini okuyun.
Verilerinizin public MQTT test broker üzerinde başkalarınınkiyle karışmaması için rastgele bir topic öneki seçin. Bu eğitim connectmq/demo/k7f3q2 kullanır; her iki dosyada da k7f3q2 kısmını kendi rastgele dizenizle değiştirin.
Adım 1: Bir Sensör Simülatörü
Panoyu oluşturmadan önce veri yayınlayan bir şey hazırlayın. Bunu simulator.py olarak kaydedin:
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()
Saniyede üç mesaj yayınlar; bu, public broker'ın client başına saniyede 10 mesaj sınırının oldukça altındadır. Kendi terminalinde python simulator.py ile çalıştırın ve açık bırakın.
Adım 2: Streamlit MQTT Dashboard
Bunu dashboard.py olarak kaydedin:
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():
"""Create one MQTT client per Streamlit server process."""
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 # bozuk payload'ları yok say
with state["lock"]:
state["buffer"].append(row)
state["received"] += 1
client = mqtt.Client(
mqtt.CallbackAPIVersion.VERSION2,
client_id=f"streamlit-{uuid.uuid4().hex[:8]}",
)
# Private bir ConnectMQ broker için:
# client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
# client.tls_set() ve 8883 portunu kullanın
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()
Şununla başlatın:
streamlit run dashboard.py
Streamlit http://localhost:8501 adresini açar. Birkaç saniye içinde durum Connected olarak görünür, sayaçlar artmaya başlar ve her grafikte üç çizgi belirir.
Pano Nasıl Çalışır?
Tek client, çok sayıda yeniden çalıştırma
st.cache_resource, start_mqtt() fonksiyonunu ilk seferde çalıştırır ve sonraki her çağrıda, yeniden çalıştırmalar ve aynı sunucudaki tarayıcı oturumları boyunca aynı nesneyi döndürür. Bu, bir MQTT bağlantısının ihtiyaç duyduğu ömrün ta kendisidir. Önbelleğe alınan nesne tüm izleyiciler tarafından paylaşıldığından pano, kaç kişi bakarsa baksın broker'a tek bir bağlantı açar; bu da bağlantı sınırı olan planlarda önemlidir.
İş parçacığı güvenliği
paho'nun loop_start() metodu ağ döngüsünü arka plandaki bir iş parçacığında çalıştırır; bu yüzden on_message, Streamlit betiğinizden farklı bir iş parçacığında çalışır. CPython'da bir deque yapısına eleman eklemek atomiktir, ancak başka bir iş parçacığı eleman eklerken onu kopyalamak "deque mutated during iteration" hatasına yol açabilir. Kilit, kopyalama ile eklemeyi karşılıklı olarak dışlar ve yalnızca mikrosaniyeler boyunca tutulur; böylece ağ iş parçacığını asla yavaşlatmaz.
Deque üzerindeki maxlen bellek koruyucunuzdur. 600 noktayla pano üç sensörün yaklaşık son 200 saniyesini gösterir ve ne kadar uzun çalışırsa çalışsın bellek sabit kalır.
Fragment'larla kısmi yeniden çalıştırma
@st.fragment(run_every=2), Streamlit'e tüm betik yerine yalnızca bu fonksiyonu her iki saniyede bir yeniden çalıştırmasını söyler. Başlık ve açıklama yerinde kalır, grafikler güncellenir ve sayfa titremez. Eski Streamlit sürümlerinde üçüncü taraf bir otomatik yenileme bileşenine veya bir time.sleep() ile st.rerun() döngüsüne ihtiyacınız olurdu; ikisi de daha hantaldır.
Yeniden bağlanma
connect_async() ile loop_start() birlikte, broker kısa süreliğine erişilemez olsa bile panonun başlaması anlamına gelir; paho, reconnect_delay_set ile belirlenen gecikmelerle arka planda yeniden dener. on_connect içinde abone olmak önemlidir: clean session ile yeniden bağlanıldığında abonelikler kaybolur ve geri yüklendikleri yer on_connect olur. paho-mqtt yeniden bağlanma rehberi bu modeli daha ayrıntılı ele alır.
Yaygın Sorunlar ve Çözümleri
- Yinelenen veya iki kat gelen mesajlar. Client'ı
st.cache_resourcedışında oluşturuyorsunuz ya dastart_mqttfonksiyonunu düzenlediniz ve eski iş parçacığı çalışmaya devam ederken Streamlit yeni bir önbelleğe alınmış client oluşturdu. MQTT kurulumunu değiştirdikten sonrastreamlit runsürecini yeniden başlatın. - Pano sonsuza dek "Connecting..." diyor. Ağınızda 1883 portu engellenmiş olabilir.
client.tls_set()ile 8883'te TLS'e geçin veyamosquitto_sub -h broker.connectmq.com -p 1883 -i test-k7f3q2 -t 'connectmq/demo/k7f3q2/#' -vile test edin. - Bağlı ama veri yok.
dashboard.pyvesimulator.pyiçindeki önek, büyük/küçük harf dahil tam olarak eşleşmelidir. - Grafikler zıplıyor. Zaman damgaları publisher'dan gelir. Cihazların saatleri senkronize değilse bunun yerine alım zamanını kullanın (
on_messageiçindetime.time()). - Çok sensörle sayfa yavaşlıyor. Her iki saniyede binlerce satırı pivot etmek birikir.
MAX_POINTSdeğerini düşürün,run_everydeğerini artırın veya grafiğe dökmeden öncedf.resample()ile örnek sayısını azaltın.
Üretime Taşımak
Public broker deneyler için idealdir, ancak public alanındaki her şey herkes tarafından okunabilir. Gerçek tesis veya bina verilerini gösteren bir pano için:
- ConnectMQ üzerinde private bir broker oluşturun ve üretilen kimlik bilgilerini 8883 portunda TLS ile kullanın. Python MQTT TLS rehberi tam
tls_set()yapılandırmasını gösterir. - Gizli bilgileri koda gömmek yerine
.streamlit/secrets.tomliçinde tutun vest.secrets["mqtt"]["username"]ile okuyun. - Bellek içi tamponun ötesinde geçmişe ihtiyacınız varsa bir veritabanına yazan ayrı bir abone çalıştırın ve panonun onu sorgulamasını sağlayın. Python ile MQTT'den SQLite'a eğitimi hafif bir başlangıç noktasıdır, MQTT ve pandas rehberi ise analizde yardımcı olur.
Streamlit bir Python aracıdır, ancak tarayıcı panolarıyla aynı ailede yer alır. Saf bir frontend yaklaşımını tercih ediyorsanız tarayıcıda JavaScript ile MQTT client'ı doğrudan sayfadan WebSocket üzerinden bağlanır; MQTT client kütüphanelerine genel bakış ise her dildeki başlıca seçenekleri karşılaştırır. Streamlit dokümantasyonu fragment'ları, önbelleğe almayı ve dağıtımı ayrıntılı olarak ele alır.
Streamlit Panolarınızı ConnectMQ Üzerinde Çalıştırın
ConnectMQ tamamen yönetilen bir MQTT broker'dır: 8883'te TLS, üretilmiş kimlik bilgileri ve sensörlerinizin ne yayınladığını grafiğe dökmeden önce kontrol etmeniz için canlı topic gezgini ile tarayıcı içi client sunan bir portal elde edersiniz. broker.connectmq.com ile başlayın, ardından kendi bulut MQTT broker'ınıza geçin. Kredi kartı gerekmeden ücretsiz Serverless planına kaydolun.
Sık Sorulan Sorular
Streamlit doğrudan MQTT'ye abone olabilir mi?
Streamlit'in yerleşik MQTT desteği yoktur, ancak herhangi bir Python MQTT client'ı çalışır. Güvenilir model, st.cache_resource ile bir kez oluşturulan ve ağ döngüsünü arka plandaki bir iş parçacığında çalıştıran bir paho-mqtt client'ıdır.
Pano ne sıklıkla yenilenmeli?
Bir ila beş saniye çoğu sensör panosu için uygundur. Daha hızlı yenilemeler, yavaş değişen değerleri daha kullanışlı hale getirmeden Streamlit sunucusunda CPU tüketir.
Neden her mesajı birkaç kez alıyorum?
Birden fazla MQTT client'ı çalışıyor; genellikle client st.cache_resource dışında oluşturulduğu için veya eski bir önbelleğe alınmış client bir kod değişikliğinden sonra hayatta kaldığı için. Yalnızca bir client olması için Streamlit sürecini yeniden başlatın.
Birçok kullanıcı aynı Streamlit MQTT panosunu görüntüleyebilir mi?
Evet. Önbelleğe alınmış client sunucudaki tüm oturumlar tarafından paylaşılır; bu yüzden on izleyici bile tek bir broker bağlantısı kullanır ve aynı tampondan okur.
