ذخیره داده‌های MQTT در دیتابیس با پایتون

بروکر MQTT برای جابه‌جایی پیام است، نه بایگانی. وقتی دستگاه پیامی منتشر می‌کند، بروکر آن را به مشترک‌های همان لحظه می‌رساند و تمام. صفحه «پیام‌ها» در داشبورد mqttcloud.ir پیام‌ها را فقط ۷ روز و برای عیب‌یابی نگه می‌دارد. برای نمودار ماهانه، گزارش مصرف یا هشدار بر اساس روند، باید داده‌ها را در سامانه خودتان ذخیره کنید.

ساده‌ترین راه یک برنامه کوچک است که مثل یک دستگاه به بروکر وصل می‌شود، روی تاپیک‌های داده مشترک می‌شود و هر پیام را در دیتابیس می‌نویسد.

• یک شناسه دستگاه برای این برنامه در داشبورد ثبت کنید، مثلاً logger-01.
• یک کامپیوتر یا سرور که همیشه روشن است؛ یک رزبری‌پای هم کافی است.
• پایتون ۳ و کتابخانه paho-mqtt (نسخه ۲). SQLite همراه پایتون است و نصب جداگانه نمی‌خواهد.

python3 -m venv ~/mqtt-env
~/mqtt-env/bin/pip install paho-mqtt

این اسکریپت روی همه تاپیک‌های alice/sensors/ مشترک می‌شود و هر پیام را با زمان دریافت ذخیره می‌کند:

import sqlite3, time
import paho.mqtt.client as mqtt

USER = "alice"

db = sqlite3.connect("mqtt.db")
db.execute("""CREATE TABLE IF NOT EXISTS readings (
    id          INTEGER PRIMARY KEY,
    topic       TEXT NOT NULL,
    payload     TEXT NOT NULL,
    received_at INTEGER NOT NULL
)""")
db.execute("CREATE INDEX IF NOT EXISTS idx_topic_time ON readings (topic, received_at)")

def on_connect(client, userdata, flags, reason_code, properties):
    print("connected:", reason_code)
    client.subscribe(f"{USER}/sensors/#", qos=1)  # subscribe again on every reconnect

def on_message(client, userdata, msg):
    if msg.retain:
        return  # a stored copy of an older value, not a new reading
    db.execute(
        "INSERT INTO readings (topic, payload, received_at) VALUES (?, ?, ?)",
        (msg.topic, msg.payload.decode(errors="replace"), int(time.time())),
    )
    db.commit()

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id="logger-01", clean_session=False)
client.username_pw_set(USER, "YOUR_PASSWORD")
client.tls_set()
client.on_connect = on_connect
client.on_message = on_message
client.reconnect_delay_set(min_delay=1, max_delay=60)

client.connect("mqtt.mqttcloud.ir", 8883, keepalive=60)
client.loop_forever()

• clean_session=False و QoS 1 باعث می‌شوند اگر اتصال برنامه کوتاه‌مدت قطع شد، بروکر پیام‌های QoS 1 آن مدت را نگه دارد و بعد از اتصال مجدد تحویل دهد. این صف در حافظه بروکر است و با راه‌اندازی مجدد بروکر پاک می‌شود.
• پیام‌های ماندگار را نادیده گرفته‌ایم، چون بعد از هر اتصال مجدد دوباره تحویل داده می‌شوند و رکورد تکراری می‌سازند.
• برای اجرای دائمی و خودکار بعد از ریست، از سرویس systemd مثل راهنمای رزبری‌پای استفاده کنید.

اگر پیام‌ها JSON باشند، SQLite با json_extract مستقیماً فیلدها را می‌خواند. مثلاً میانگین ساعتی دما در ۲۴ ساعت اخیر:

SELECT strftime('%Y-%m-%d %H:00', received_at, 'unixepoch') AS hour,
       round(avg(json_extract(payload, '$.c')), 1)          AS avg_c
FROM readings
WHERE topic = 'alice/sensors/temp'
  AND received_at > CAST(strftime('%s', 'now', '-1 day') AS INTEGER)
GROUP BY hour
ORDER BY hour;

اگر دستگاه زمان اندازه‌گیری را داخل پیام می‌فرستد (فیلد ts)، برای دقت بیشتر آن را به جای زمان دریافت ذخیره کنید؛ قالب پیشنهادی را در طراحی تاپیک و پیام JSON ببینید.

SQLite برای ده‌ها دستگاه که هر چند ثانیه پیام می‌فرستند کاملاً کافی است. اگر تعداد پیام‌ها خیلی زیاد شد:

• به جای commit بعد از هر پیام، هر چند ثانیه یک بار گروهی ذخیره کنید.
• داده‌های قدیمی را خلاصه کنید (مثلاً میانگین ساعتی) و رکوردهای خام را بعد از مدتی پاک کنید.
• برای حجم بالا سراغ دیتابیس‌های سری زمانی مثل InfluxDB یا TimescaleDB (روی PostgreSQL) بروید؛ الگوی اتصال به MQTT همین است.
• اگر Node-RED دارید، همین کار را بدون کدنویسی با نودهای دیتابیس انجام دهید (اتصال Node-RED).