MQTT Verisini Google Sheets'e Aktarma: Python ve Sheets API ile Sensör Verisi Gönderin

MQTT verisini Google Sheets'e aktarmak, teknik olmayan meslektaşlarınıza sensör verilerinin canlı görünümünü sunmanın en hızlı yollarından biridir: elektronik tabloları zaten bilirler ve kendi grafiklerini ve formüllerini ekleyebilirler. Bu eğitimde MQTT topic'lerinden ölçümleri toplayan, bunları gruplayan ve resmi Sheets API üzerinden paylaşılan bir Google Sheet'e ekleyen bir Python abonesi geliştireceksiniz. Sonunda API kotalarına uyan ve ağ kopmalarına dayanıklı, küçük ama güvenilir bir raporlama hattınız olacak.

IoT Verileri İçin Google Sheets Ne Zaman Mantıklı?

Google Sheets bir zaman serisi veritabanı değildir; öyleymiş gibi davranmak yavaş ve şişkin tablolara yol açar. Yine de belirli işler için çok iyi bir seçimdir:

Elektronik tabloların dosya başına toplam hücre sayısında da kesin bir sınırı vardır ve bu sınıra ulaşmadan çok önce ağırlaşırlar. Bu nedenle aşağıdaki tasarım satırları gruplar ve yazmadan önce özetleme yapmanıza olanak tanır; böylece tablo küçük ve kullanışlı kalır.

Mimari: Ortada Bir Abone

Cihazlarınız doğrudan Google ile konuşmamalıdır. Google kimlik bilgilerini firmware'e gömmek bir güvenlik sorunudur ve her cihazın kendi HTTP client'ına ve yeniden deneme mantığına ihtiyacı olur. Bunun yerine cihazlar her zamanki gibi bir MQTT broker'a yayın yapar, tek bir Python süreci abone olur ve verileri Sheets'e iletir:

devices --MQTT--> broker.connectmq.com --MQTT--> sheets_bridge.py --HTTPS--> Google Sheets API

Bir MQTT broker ile yapılan her entegrasyon için dürüst model budur: broker mesajları yönlendirir, bunun gibi küçük harici servisler ise diğer sistemlere teslimatı üstlenir. Köprü kısa bir süre devre dışı kalırsa cihazlar normal şekilde yayın yapmaya devam eder.

Adım 1: Bir Google Hizmet Hesabı Oluşturun

1. Google Cloud konsolunu açın ve bir proje oluşturun (veya mevcut bir projeyi kullanın).
2. Bu proje için Google Sheets API'yi etkinleştirin.
3. IAM ve Yönetici, Hizmet Hesapları bölümüne gidin, bir hizmet hesabı oluşturun, ardından bir JSON anahtarı oluşturup service_account.json olarak indirin.
4. Yeni bir Google Sheet oluşturun, Readings adında bir sekme ekleyin ve 1. satıra şu başlıkları yazın: received_at, device, metric, value, topic.
5. Tabloda Paylaş'a tıklayın ve hizmet hesabının e-posta adresini (iam.gserviceaccount.com ile biter) Düzenleyici olarak ekleyin.
6. Tablo anahtarını URL'den kopyalayın: /d/ ile /edit arasındaki uzun dize.

İnsanların unuttuğu adım 5. adımdır. Bu adım olmadan, kimlik bilgileri geçerli olsa bile her API çağrısı izin hatasıyla başarısız olur.

Adım 2: Kütüphaneleri Kurun

pip install "paho-mqtt>=2.0" "gspread>=6.0"

gspread, Sheets API'yi sarmalar ve hizmet hesabı dosyasından kimlik doğrulamayı yönetir. API referansının tamamı gspread dokümantasyonunda yer alır.

Adım 3: MQTT'den Google Sheets'e Köprü

Bunu sheets_bridge.py olarak kaydedin. Sayısal veya JSON payload içeren, <prefix>/<device>/<metric> biçiminde topic'ler bekler.

import json
import os
import queue
import threading
import time
import uuid
from datetime import datetime, timezone

import gspread
import paho.mqtt.client as mqtt

BROKER = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "1883"))
USERNAME = os.environ.get("MQTT_USERNAME")
PASSWORD = os.environ.get("MQTT_PASSWORD")
CLIENT_ID = os.environ.get("MQTT_CLIENT_ID", f"sheets-bridge-{uuid.uuid4().hex[:8]}")
PREFIX = os.environ.get("TOPIC_PREFIX", "connectmq/demo/gsheet-5307")
SHEET_KEY = os.environ["SHEET_KEY"]
WORKSHEET = os.environ.get("WORKSHEET", "Readings")
FLUSH_SECONDS = int(os.environ.get("FLUSH_SECONDS", "30"))
MAX_BATCH = int(os.environ.get("MAX_BATCH", "500"))

rows: "queue.Queue[list]" = queue.Queue(maxsize=50_000)
stop = threading.Event()


def parse_value(text: str):
    try:
        return float(text)
    except ValueError:
        pass
    try:
        data = json.loads(text)
        if isinstance(data, dict) and isinstance(data.get("value"), (int, float)):
            return data["value"]
    except json.JSONDecodeError:
        pass
    return None


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"MQTT connect failed: {reason_code}")
        return
    client.subscribe(f"{PREFIX}/+/+", qos=1)
    print(f"Subscribed to {PREFIX}/+/+")


def on_message(client, userdata, msg):
    if msg.retain:
        return  # her yeniden bağlantıda yinelenen satırları önler
    parts = msg.topic.split("/")
    device, metric = parts[-2], parts[-1]
    value = parse_value(msg.payload.decode("utf-8", errors="replace"))
    if value is None:
        return
    row = [datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S"), device, metric, value, msg.topic]
    try:
        rows.put_nowait(row)
    except queue.Full:
        print("Buffer full, dropping row")


def flush_worker(ws: gspread.Worksheet):
    backoff = 5
    pending: list = []
    while not stop.is_set() or pending or not rows.empty():
        stop.wait(FLUSH_SECONDS)
        while len(pending) < MAX_BATCH:
            try:
                pending.append(rows.get_nowait())
            except queue.Empty:
                break
        if not pending:
            continue
        try:
            ws.append_rows(pending, value_input_option="USER_ENTERED")
            print(f"Appended {len(pending)} rows")
            pending = []
            backoff = 5
        except gspread.exceptions.APIError as exc:
            # 429 (kota) veya 5xx: satırları tut ve daha sonra yeniden dene
            print(f"Sheets API error, retrying in {backoff}s: {exc}")
            time.sleep(backoff)
            backoff = min(backoff * 2, 300)


def main():
    gc = gspread.service_account(filename="service_account.json")
    ws = gc.open_by_key(SHEET_KEY).worksheet(WORKSHEET)

    worker = threading.Thread(target=flush_worker, args=(ws,), daemon=True)
    worker.start()

    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        client.username_pw_set(USERNAME, PASSWORD)
    if PORT == 8883:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(BROKER, PORT, keepalive=60)
    client.loop_start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("Stopping, flushing remaining rows...")
        client.loop_stop()
        client.disconnect()
        stop.set()
        worker.join(timeout=FLUSH_SECONDS + 30)


if __name__ == "__main__":
    main()

Gruplama Neden Önemli?

Sheets API, proje ve kullanıcı başına dakikalık istek kotaları uygular. Her MQTT mesajı için bir API çağrısı bir demo için işe yarar, ancak birkaç cihaz saniyede bir rapor vermeye başladığı anda başarısız olur. Köprü, satırları iş parçacığı açısından güvenli bir kuyrukta toplar ve her 30 saniyede tek bir append_rows çağrısıyla gönderir; böylece saniyede bir rapor veren on cihaz dakikada altı yüz yerine iki istek üretir.

API bir hata döndürdüğünde (kota aşımı veya geçici bir sunucu hatası) worker grubu pending içinde tutar ve üstel geri çekilme (exponential backoff) ile yeniden dener. Satırlar yalnızca bellekteki tampon dolarsa atılır; bu da kısa bir Google kesintisi sırasında size bolca pay bırakır.

Neden USER_ENTERED?

value_input_option="USER_ENTERED" ile Sheets değerleri bir kişi yazmış gibi ayrıştırır: zaman damgası gerçek bir tarih-saate, sayılar da sayıya dönüşür; böylece grafikler ve formüller hemen çalışır. RAW ile ise her şey metin olarak kaydedilir.

Adım 4: Çalıştırın ve Test Edin

export SHEET_KEY=your-spreadsheet-key
python sheets_bridge.py

Başka bir terminalden ücretsiz public MQTT broker'a birkaç ölçüm yayınlayın:

mosquitto_pub -h broker.connectmq.com -p 1883 -i gsheet-pub-5307 \
  -t connectmq/demo/gsheet-5307/boiler1/temperature -m '{"value": 64.2}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i gsheet-pub-5307 \
  -t connectmq/demo/gsheet-5307/boiler1/pressure -m 2.4

30 saniye içinde her iki satır da tabloda görünür. Public broker'daki her şeyin başkaları tarafından okunabildiğini unutmayın; test sırasında yalnızca uydurma değerler kullanın.

Eklemeden Önce Özetleyin

Pilot aşamasını aşan her şey için ham ölçümler yerine özetler gönderin. Basit bir yaklaşım, bellekte cihaz başına hareketli ortalama tutmak ve her beş veya on beş dakikada cihaz başına bir satır göndermektir. Bu, tabloyu insanlar için okunabilir ve hücre sınırının çok altında tutar. Ham akışa da ihtiyacınız varsa onu hacim için tasarlanmış bir yere yazın; örneğin MQTT mesajlarını CSV'ye kaydetme yazısındaki CSV yaklaşımı. Sheets'i ise özet için kullanın.

Sorun Giderme Kontrol Listesi

1. SpreadsheetNotFound veya 403 hataları. Tablo hizmet hesabının e-postasıyla paylaşılmamış veya anahtar yanlış.
2. WorksheetNotFound. Sekme adı, büyük/küçük harf dahil WORKSHEET ile tam olarak eşleşmiyor.
3. Yeniden başlatmadan sonra satırlar iki kez geliyor. Retained message'lar yeniden gönderildi. Köprü bunları zaten atlıyor; cihazlarınızın açılışta geçmişi yeniden yayınlamadığından emin olun.
4. Satırlar duruyor ve log'da 429 görünüyor. Kotayı aşıyorsunuz. FLUSH_SECONDS değerini artırın veya eklemeden önce özetleme yapın.
5. Köprü sürekli bağlanıp kopuyor. Başka bir client aynı client ID'yi kullanıyor. Public broker'da üretilen rastgele ID'yi koruyun.
6. Sayılar sola yaslı metin olarak görünüyor. RAW'a geçmişsiniz veya tablonun yerel ayarı ondalık ayırıcı olarak virgül bekliyor. Tablonun yerel ayarını Dosya, Ayarlar altından değiştirin.

Köprüyü Üretimde Çalıştırmak

Gerçek bir kurulum için public alan yerine kimlik bilgileri ve TLS ile private bir broker'a bağlanın. ConnectMQ üzerinde portalda bir broker oluşturur ve bir kullanıcı adı, parola ve client ID alırsınız; bunları MQTT_USERNAME, MQTT_PASSWORD ve MQTT_CLIENT_ID olarak ayarlayın ve 8883 portunu kullanın. Broker tamamen yönetilen bir MQTT servisi olduğundan işletmeniz gereken tek şey köprüdür; onu bir systemd servisi, bir container veya küçük bir VM olarak çalıştırabilirsiniz.

Aynı abone modeli bu serinin geri kalanının da temelidir: sohbet uyarıları için bir MQTT Telegram botu ve diğer uygulamalar için MQTT'nin önünde bir Flask REST API. Farklı bir dilde client kütüphanesi seçmek için MQTT client kütüphaneleri genel bakışına göz atın.

ConnectMQ ile Başlayın

Google Sheets köprünüz için birkaç dakikada private bir broker oluşturun. app.connectmq.com adresinden kaydolun, kredi kartı gerektirmeyen ücretsiz Serverless planını seçin ve üretilen kimlik bilgilerini köprüye kopyalayın. Portalın tarayıcı içi MQTT client'ı ve canlı topic gezgini, veriler tabloya ulaşmadan önce akışın doğru olduğunu doğrulamanızı sağlar; TLS ve broker operasyonlarını ise ConnectMQ sizin yerinize yönetir.

Sık Sorulan Sorular

Bir cihaz doğrudan Google Sheets'e yayın yapabilir mi?

HTTPS üzerinden Sheets API'yi çağırabilir, ancak bu Google kimlik bilgilerini cihaza koyar ve her cihazı kotaları ve yeniden denemeleri yönetmeye zorlar. MQTT üzerinden yayın yapıp tek bir köprü kullanmak daha basit ve daha güvenlidir.

Google Sheets'e ne sıklıkla satır eklemeliyim?

Ham veriler için yazmaları her 30 ila 60 saniyede bir, özetler için ise birkaç dakikada bir gruplayın. Daha az sayıda ve daha büyük append_rows çağrısı API kotalarının rahatça altında kalır.

Bu Python yerine Google Apps Script ile çalışır mı?

Apps Script açık bir MQTT bağlantısı tutamaz, bu yüzden yine harici bir aboneye ihtiyacınız olur. MQTT konuşan parça Python köprüsüdür; Apps Script daha sonra tablonun içindeki verileri biçimlendirebilir veya grafiğe dönüştürebilir.

Google erişilemez olursa ne olur?

Köprü satırları bellekte tutar ve geri çekilme ile yeniden dener; bu nedenle kısa kesintiler veri kaybına değil gecikmeye neden olur. Uzun bir kesinti sonunda tamponu doldurur; bundan sonra yeni satırlar atılır ve log'a yazılır.