Alarm bei BierBot-Problemen

[mschneider]

Hallo Zusammen

Ich habe mir eine Überwachung eingerichtet, die mir einen Alarm sendet, wenn es mit meinen Bricks Probleme gibt. Falls jemand interessiert ist, hier ein Beschrieb, wie ich das gemacht habe.


Ausgangslage:

Es passierte immer wieder, dass ein Brick mit angeschlossener Temperatursonde -127°C gemessen hat oder offline ging. Vielleicht, weil die Sonde mit 5 m Verlängerung verbunden ist? Oder weil die Bricks in einem Metallschrank sind? Router ist zwar nur ca. 20 cm entfernt...
Jedenfalls war das besonders dann problematisch, wenn ein Bier im Gärtank und ein Heizelement angeschlossen war. Für dieses Problem musste eine Lösung her.
Ein Skript auf meinem NAS prüft jetzt alle 15 Minuten, ob alles OK ist. Falls nicht, erhalte ich eine Nachricht.
Das Skript mit Platzhaltern (XXXXXXXX) und Erklärungen am Zeilende (# Erklärung) liefere ich im nächsten Post separat nach. Es vor allem als Orientierung dienen und kann nicht 1-zu-1 übernommen werden.


Zur Info:

Prüfintervall nicht zu kurz einstellen. Laut Bernhard können zu häufige Abfragen Probleme verursachen.
Das Skript funktioniert nur bei Geräten, die sich im Automatik-Modus befinden / ein Rezept ausführen.
Beachte die Unterscheidung zwischen Gerät und Brick.
Meine 3 Gärtanks haben je 2 Bricks – 1 x Kühlung, 1 x Heizung. Temperatursonde ist nur an 1 Brick angeschlossen, der andere liest also dauernd –127°C und muss von der Überwachung ausgeschlossen werden.


Benötigte Hardware und Tools:

ChatGPT, Postman(.com), Smartphone mit Telegram, NAS (bei mir Synology), Pro-Account BierBotBricks.
Statt NAS geht auch RapberryPi oder ein Rechner, der ständig eingeschaltet und im Netz ist.

Meine Programmierkenntnisse: inexistent. Darum: ChatGPT ist dein Freund! Wenn du beschreibst, was du machen willst, hilft es dir sehr gut dabei, bei jedem einzelnen Schritt. Ich musste immer wieder «Babysteps» verlangen, wodurch ich wirklich recht einfache Beschriebe erhalten habe.


Rückblickend waren das die Schritte:

1.
NAS vorbereiten: Python und ein paar Erweiterungen dafür installieren. Auch hier: ChatGPT sagt dir wie. Ich habe dafür einen eigenen Benutzer eingerichtet.


2.
Telegram einrichten: Dort brauchst du einen ChatBot – den erstellst du mit BotFather direkt in Telegram. Von dem brauchst du den Bot-Token. Weiter brauchst du die ChatID. Findest du mit get_id_bot.

ChatGPT macht dir ein Skript zum Testen, ob der ChatBot funktioniert.


3.
BierBot einrichten: Unter Verwaltung Pro-API einschalten. Die normale API und die Pro-API notieren. API-URL ebenfalls notieren: https://brewbricks.com/api/


4.
Postman:

Gratis-Account anlegen. Links zu Workspaces gehen, dann «Contract Testing», «Get Test Response». Dann im Hauptfenster oben (bei GET) die API-Url eingeben, entweder gefolgt von «devices» oder «device» (z.B. https://brewbricks.com/api/device).
Mit «devices» findet man die Device-IDs seiner persönlichen Geräte heraus (nicht Bricks).
Mit «device» findet man die SourceIot seiner personlichen Bricks, nebst vielem anderen.

Unten bei Query Params Folgendes eintragen. Im Fall von
API-Url + devices: bei «Key» apikey und proapikey; bei «Value» deine persönlichen API-Keys aus BierBot.
API-Url + device: bei «Key» apikey, proapikey und deviceid; bei «Value» wieder deine API-Keys, und die resp. eine der Device-IDs, die du mit dem Schritt für «devices» findest.
Zum Ausführen logischerweise auf «Send» klicken, unten erscheinen dann die Infos.

Wie angedeutet: Mit «device» findet man auch Infos wie Gerätename, was das Gerät ist ("cool", "heat", etc.) und welche Temperatur zuletzt registriert wurde: «lastValue». Dieses «lastValue» sollte im Zusammenhang mit einem Sensor stehen. Darauf deutet das Vorkommen von «dataType = number» oder «physicalMeaning = temp» ein paar Zeilen weiter oben oder unten hin. So kannst du auch den Brick (= sourceIot) identifizieren, der einen Sensor angeschlossen hat (wie erwähnt: meine Gärtanks bestehen aus 1 Brick mit und 1 Brick ohne Temperatursensor).

All diese Infos – deviceids (z.B. xoqYUzXqfXrwAlXXXXXX), sourceIots mit Mac-Adresse (z.B. sonoff_th_origin_bierbot_00050XXX_XX:XX:XX:XX:XX:XX) sammeln und übersichtlich auflisten.


5.
Mit den Infos, die Postman ausspuckt, und mit deiner Auflistung kannst du nun ChatGPT füttern und das Skript erstellen lassen.

Zum Testen habe ich das Skript lokal auf meinem Mac ausgeführt, bis es funktioniert hat.

Am Anfang hatte ich nur die Benachrichtigung bezüglich den gemessenen -127°C drin. Schritt für Schritt habe ich – oder eben ChatGPT – das Skript erweitert: z.B. Alarm, wenn ein Brick offline ist; seit längerer Zeit keine neuen Werte registriert wurden; kein Alarm auslösen wenn schon vor 15 min. einer ausgelöst wurde, etc.

6.
Das Skript aufs NAS bringen und beim Starten des NAS automatisch ausführen lassen. Da mein NAS in der Nacht sowieso für ein paar Stunden ausgeschaltet ist, passt das.


Viel Erfolg, wenn du dich auch daran machen willst!
Und bei Fragen – besser ChatGPT als mich bemühen

Prosit und gut Sud,
Matthias

[mschneider]

Wie versprochen hier das Skript (Python):
------------------------

Code: Select all

import requests
import time
import os
import logging
from datetime import datetime, timedelta

# Telegram Bot Details
TELEGRAM_BOT_TOKEN = "XXXXXXXXX:XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"  # Ersetze mit deinem Token
TELEGRAM_CHAT_ID = "XXXXXXXXX"  # Ersetze mit deiner Chat-ID

# API Details
API_URL = "https://brewbricks.com/api/device"
API_PARAMS = {
    "apikey": "XXXXXXXXXXXX",        # Ersetze mit deiner API Key von BierBot
    "proapikey": "XXXXXXXXXXXXXX",     # Ersetze mit deiner Pro API Key von BierBot
}

# Geräte-IDs und zugehörige Namen; ermitteln via Postman mit api/devices
DEVICE_NAMES = {
    "XXXXXXXXXXXXXXXX": "GT2",     # Ersetze mit deinen Device-IDs und gib ihnen Namen
    "XXXXXXXXXXXXXXXX": "GT3",
    "XXXXXXXXXXXXXXXX": "GT1",
    "XXXXXXXXXXXXXXXX": "HLT",
    "XXXXXXXXXXXXXXXX": "Mashtun",
    "XXXXXXXXXXXXXXXX": "Boilkettle",
}

# Eingrenzung auf bestimmte sourceIot; ermitteln via Postman mit api/device
VALID_SOURCE_IOT = [
    "XXXX-ZUM-BEISPIEL-sonoff_th_origin_bierbot_XXXXXXXX_XX:XX:XX:XX:XX:XX",     # Ersetze mit deinen relevanten sourceIot
    "XXXX-ZUM-BEISPIEL-sonoff_th_origin_bierbot_XXXXXXXX_XX:XX:XX:XX:XX:XX",
    "XXXX-ZUM-BEISPIEL-sonoff_th_origin_bierbot_XXXXXXXX_XX:XX:XX:XX:XX:XX",
    "XXXX-ZUM-BEISPIEL-sonoff_th_origin_bierbot_XXXXXXXX_XX:XX:XX:XX:XX:XX",
    "XXXX-ZUM-BEISPIEL-sonoff_th_origin_bierbot_XXXXXXXX_XX:XX:XX:XX:XX:XX",
    "XXXX-ZUM-BEISPIEL-sonoff_th_origin_bierbot_XXXXXXXX_XX:XX:XX:XX:XX:XX",
]

# Log-Einstellungen
LOG_DIR = "./logs"
os.makedirs(LOG_DIR, exist_ok=True)  # Ordner erstellen, falls nicht vorhanden
logging.basicConfig(
    filename=os.path.join(LOG_DIR, "device_monitor.log"),
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

# Variable zur Verwaltung des Alarm-Timers
last_alarm_time = 0  # Zeitstempel des letzten Alarms (Epoch-Zeit)

# Funktion zur automatischen Löschung alter Logs
def cleanup_logs():
    now = datetime.now()
    for log_file in os.listdir(LOG_DIR):
        log_path = os.path.join(LOG_DIR, log_file)
        if os.path.isfile(log_path):
            file_age = now - datetime.fromtimestamp(os.path.getmtime(log_path))
            if file_age > timedelta(days=30):
                os.remove(log_path)
                logging.info(f"Alte Logdatei gelöscht: {log_file}")

# Funktion, um eine Nachricht an Telegram zu senden
def send_telegram_message(message):
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    payload = {"chat_id": TELEGRAM_CHAT_ID, "text": message}
    try:
        response = requests.post(url, data=payload)
        if response.status_code == 200:
            logging.info(f"Nachricht erfolgreich gesendet: {message}")
        else:
            logging.error(f"Fehler beim Senden der Nachricht: {response.json()}")
    except Exception as e:
        logging.error(f"Fehler bei der Telegram-Kommunikation: {e}")

# Funktion, um die API zu prüfen
def check_device_values():
    global last_alarm_time
    alarm_messages = []  # Liste für Alarmnachrichten

    for device_id, device_name in DEVICE_NAMES.items():
        # API mit spezifischer Device-ID aufrufen
        params = {**API_PARAMS, "deviceid": device_id}
        try:
            response = requests.get(API_URL, params=params, timeout=10)  # Timeout von 10 Sekunden
            if response.status_code == 200:
                data = response.json()  # JSON-Daten parsen
                interfaces = data.get("interfaces", [])

                # Prüfe den letzten Zeitstempel der Geräte
                last_update = data.get("lastUpdateByIotMs", 0)
                current_time = time.time() * 1000  # Aktuelle Zeit in Millisekunden
                time_diff = (current_time - last_update) / (1000 * 60)  # Zeitdifferenz in Minuten

                # Wenn der Zeitstempel älter als 15 Minuten ist, Alarm auslösen
                if time_diff > 15:
                    alarm_messages.append(f"ACHTUNG – BierBot-Problem: {device_name} hat keine aktuellen Daten (letzte Aktualisierung vor {int(time_diff)} Minuten)!")

                # Suche nach den relevanten sourceIot und lastValue
                for interface in interfaces:
                    source_iot = interface.get("sourceIot")
                    last_value = interface.get("lastValue")
                    data_type = interface.get("dataType")

                    # Normale Prüfung
                    if source_iot in VALID_SOURCE_IOT and data_type == "number" and last_value in [-127, 0]:  # Zum Testen: ... and 0 < last_value < 40: | Normal: ... and last_value in [-127, 0]:
                        alarm_messages.append(f"ACHTUNG – BierBot-Problem: {device_name}! lastValue = {last_value}!")
            else:
                alarm_messages.append(f"ACHTUNG – BierBot-Problem: Gerät {device_name} offline!")
        except requests.exceptions.RequestException:
            alarm_messages.append(f"ACHTUNG – BierBot-Problem: {device_name} ist nicht erreichbar!")

    # Prüfen, ob eine Nachricht gesendet werden soll
    if alarm_messages:
        current_time = time.time()
        if current_time - last_alarm_time >= 20 * 60:  # 20 Minuten vergangen seit dem letzten Alarm > Alarm nur bei jeder 2. Prüfung
            message = "\n".join(alarm_messages)
            send_telegram_message(message)
            last_alarm_time = current_time  # Letzten Alarmzeitpunkt aktualisieren
        else:
            logging.info("Ein Alarm wurde vor 15 Minuten gesendet. Neuer Alarm erst wieder bei nächster Abfrage.")
    else:
        logging.info("Keine Alarme in dieser Abfrage.")

# Hauptprogramm starten
if __name__ == "__main__":
    logging.info("Überwachung gestartet... Abfrage alle 15 Minuten.")
    while True:
        cleanup_logs()  # Alte Logs bereinigen
        check_device_values()
        logging.info("Warte 15 Minuten vor der nächsten Abfrage...")
        time.sleep(15 * 60)  # 15 Minuten warten

[Bernhard]

Hi Matthias,

Super cool! Vielen Dank für's Teilen!

Viele Grüße,
Bernhard.