Beispiel: Python

Dieser kleine Client schickt die Messwerte deiner eigenen Hardware an BoilerBuddy: Ersetze die Funktion, die deine Sensoren ausliest, und pass den Rest an, wo nötig. Er koppelt sich bei Bedarf selbst, hält die Konfiguration aktuell, puffert in SQLite und lädt der Reihe nach hoch.

Was du brauchst#

• Python 3.10 oder neuer, mit requests (pip install requests).
• Ein Gerät im Portal mit einem Regelkreis. Siehe Dein eigenes BoilerBuddy bauen.
• Eine Uhr, die per NTP synchron gehalten wird (unter Linux: timedatectl). Der Client verlässt sich darauf; siehe Puffern und die Uhr.

Starten#

Mit einem Schlüssel aus dem Portal:

export BB_TOKEN=bb_live_42_...
python3 boilerbuddy_client.py

Ohne BB_TOKEN koppelt sich der Client selbst. Er zeigt einen Kopplungscode, den du auf der Geräteseite bei Mit Code koppeln eingibst, und speichert den Token in bb-state/token.

Variable	Standard	Zweck
BB_TOKEN	(leer)	Der API-Schlüssel des Geräts
BB_API_URL	https://boilerbuddy.eu/api/v1	Die Basis-URL
BB_STATE_DIR	bb-state	Ordner für Token, UUID, Konfiguration und Puffer

Der Code#

#!/usr/bin/env python3
"""Minimal BoilerBuddy device client: pair, fetch config, measure, buffer, upload.
An example to adapt, not finished device software. Python 3.10+ and requests.
Contract: https://boilerbuddy.eu/developers/openapi-v1.yaml"""
import json
import logging
import os
import random
import sqlite3
import time
import uuid
from datetime import datetime, timezone
from pathlib import Path

import requests

API = os.environ.get("BB_API_URL", "https://boilerbuddy.eu/api/v1").rstrip("/")
STATE = Path(os.environ.get("BB_STATE_DIR", "bb-state"))
FIRMWARE = "diy-python-0.1"
BATCH_SIZE = 500         # the API takes up to 1000 readings per batch
UPLOADS_PER_CYCLE = 8    # and 10 uploads a minute
CONFIG_EVERY_S, STATUS_EVERY_S = 900, 300
TIMEOUT = (5, 20)        # connect, read
log = logging.getLogger("boilerbuddy")


def read_sensors(sensors: list[dict]) -> dict[str, float | None]:
    """Replace this stub. Return {sensor code: value} for the sensors of the config
    that you can read, with None for a sensor that did not answer."""
    return {}


class Later(Exception):
    """Not now: try again later. retry_after is what the server asked for."""
    def __init__(self, reason: str, retry_after: float | None = None):
        super().__init__(reason)
        self.retry_after = retry_after


def retry_after(response: requests.Response) -> float | None:
    value = response.headers.get("Retry-After", "")
    return float(value) if value.isdigit() else None


def write_private(path: Path, text: str) -> None:
    """Write atomically, readable by this user only."""
    tmp = path.with_name(path.name + ".tmp")
    with os.fdopen(os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600), "w") as file:
        file.write(text)
        file.flush()
        os.fsync(file.fileno())
    os.replace(tmp, path)


def get_token() -> str:
    """BB_TOKEN, else the stored token, else pair with a code (POST /device/hello)."""
    if os.environ.get("BB_TOKEN"):
        return os.environ["BB_TOKEN"]
    token_file, uuid_file = STATE / "token", STATE / "device-uuid"
    if token_file.exists():
        return token_file.read_text().strip()
    if not uuid_file.exists():
        write_private(uuid_file, str(uuid.uuid4()))  # once; secret until the token is used
    body = {"device_uuid": uuid_file.read_text().strip(), "firmware_version": FIRMWARE}
    shown = None
    while True:
        try:
            r = requests.post(f"{API}/device/hello", json=body, timeout=TIMEOUT)
        except requests.RequestException as error:
            log.warning("hello failed: %s", error)
            time.sleep(30)
            continue
        if r.status_code == 422:
            raise SystemExit(f"hello refused: {r.text}")
        if r.status_code != 200:
            time.sleep(retry_after(r) or 30)
            continue
        answer = r.json()
        if answer.get("token"):
            write_private(token_file, answer["token"])  # store first, then use it
            return answer["token"]
        if answer.get("claimed"):
            raise SystemExit(f"The token was used already. Delete {uuid_file}, run again "
                             "and enter the new code on the same device page.")
        if answer["pairing_code"] != shown:
            shown = answer["pairing_code"]
            print(f"Pairing code: {shown}  (portal: device page, 'Pair with code')")
        time.sleep(15)


class Buffer:
    """Every reading goes here first; it is deleted once the server confirmed it."""
    def __init__(self, path: Path):
        self.db = sqlite3.connect(path)
        self.db.execute("CREATE TABLE IF NOT EXISTS readings (id INTEGER PRIMARY KEY,"
                        " sensor TEXT NOT NULL, value REAL, quality TEXT NOT NULL,"
                        " measured_at TEXT NOT NULL, batch TEXT, parked INTEGER DEFAULT 0)")

    def add(self, rows: list[tuple]) -> None:
        with self.db:
            self.db.executemany("INSERT INTO readings (sensor, value, quality, measured_at)"
                                " VALUES (?, ?, ?, ?)", rows)

    def claim(self, limit: int):
        """The oldest readings. A batch that was sent but never confirmed (crash,
        timeout) comes back first, with the same uuid."""
        with self.db:
            row = self.db.execute("SELECT batch FROM readings WHERE batch IS NOT NULL"
                                  " AND parked = 0 ORDER BY id LIMIT 1").fetchone()
            batch = row[0] if row else str(uuid.uuid4())
            if row is None:
                self.db.execute("UPDATE readings SET batch = ? WHERE id IN (SELECT id FROM"
                                " readings WHERE batch IS NULL ORDER BY id LIMIT ?)", (batch, limit))
        rows = self.db.execute("SELECT id, sensor, value, quality, measured_at FROM readings"
                               " WHERE batch = ? ORDER BY id", (batch,)).fetchall()
        readings = [{"sensor": s, "value": v, "quality": q, "measured_at": t} for _, s, v, q, t in rows]
        return (batch, [row[0] for row in rows], readings) if rows else None

    def confirm(self, batch: str, ids: list[int], retry: set[int]) -> None:
        with self.db:  # delete the batch, except the readings the server wants again
            self.db.executemany("UPDATE readings SET batch = NULL WHERE id = ?",
                                [(ids[i],) for i in retry if 0 <= i < len(ids)])
            self.db.execute("DELETE FROM readings WHERE batch = ?", (batch,))

    def park(self, batch: str) -> None:
        with self.db:  # kept for inspection, no longer sent
            self.db.execute("UPDATE readings SET parked = 1 WHERE batch = ?", (batch,))

    def pending(self) -> int:
        return self.db.execute("SELECT COUNT(*) FROM readings WHERE parked = 0").fetchone()[0]


class Device:
    def __init__(self, token: str):
        self.http = requests.Session()
        self.http.headers.update({"Authorization": f"Bearer {token}", "Accept": "application/json"})
        self.buffer = Buffer(STATE / "buffer.sqlite3")
        cache = STATE / "config.json"  # so it can measure after a reboot without internet
        cached = json.loads(cache.read_text()) if cache.exists() else {}
        self.config, self.etag = cached.get("config"), cached.get("etag")

    def call(self, method: str, path: str, **kwargs) -> requests.Response:
        # No redirects: a redirect could carry the token to another host.
        return self.http.request(method, API + path, timeout=TIMEOUT, allow_redirects=False, **kwargs)

    def refresh_config(self) -> None:
        r = self.call("GET", "/device/config", headers={"If-None-Match": self.etag} if self.etag else {})
        if r.status_code == 304:
            return
        if r.status_code != 200:
            raise Later(f"config: HTTP {r.status_code}", retry_after(r))
        self.config, self.etag = r.json(), r.headers.get("ETag")
        write_private(STATE / "config.json", json.dumps({"config": self.config, "etag": self.etag}))
        if self.config["controller"] is None:
            log.info("No circuit chosen yet: nothing to measure")

    def measure(self) -> None:
        sensors = {s["sensor"]: s for s in (self.config or {}).get("sensors", [])}
        at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
        rows = []
        for code, value in read_sensors(list(sensors.values())).items():
            if code not in sensors:
                continue  # only codes from the config: others end up in quarantine
            low, high = sensors[code]["plausible_min"], sensors[code]["plausible_max"]
            if value is None:
                quality = "no_response"
            elif (low is not None and value < low) or (high is not None and value > high):
                quality = "out_of_range"  # stored, but not in charts and averages
            else:
                quality = "ok"
            rows.append((code, value, quality, at))
        self.buffer.add(rows)

    def upload(self) -> None:
        for _ in range(UPLOADS_PER_CYCLE):
            claim = self.buffer.claim(BATCH_SIZE)
            if claim is None:
                return
            batch, ids, readings = claim
            r = self.call("POST", "/device/readings", headers={"Idempotency-Key": batch},
                          json={"batch_uuid": batch, "readings": readings})
            if r.status_code == 422:  # malformed as a whole: resending will not help
                log.error("batch %s refused: %s", batch, r.text[:500])
                self.buffer.park(batch)
                continue
            if r.status_code == 401:
                raise Later("token refused (401): replaced, disabled or device inactive", 900)
            try:
                result = r.json() if r.status_code == 200 else {}
            except ValueError:
                result = {}  # a 200 that is not the API's answer confirms nothing
            if "accepted" not in result:
                raise Later(f"readings: HTTP {r.status_code}", retry_after(r))
            retry = {item["index"] for item in result["rejected"] if item["retry"]}
            self.buffer.confirm(batch, ids, retry)
            if result["unmapped"] or result["rejected"]:
                log.warning("batch %s: unknown sensors %s, rejected %s",
                            batch, result["unmapped"], result["rejected"])
            if retry:
                return  # back in the queue, for the next cycle

    def send_status(self) -> None:
        self.call("POST", "/device/status",
                  json={"firmware_version": FIRMWARE, "buffered": self.buffer.pending()})


def guard(name: str, action) -> None:
    try:
        action()
    except Exception as error:  # one failing step must not stop the loop
        log.warning("%s failed: %s", name, error)


def main() -> None:
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    STATE.mkdir(parents=True, exist_ok=True)
    device = Device(get_token())
    next_config = next_status = next_upload = 0.0
    failures = 0
    while True:
        cycle = time.monotonic()
        if cycle >= next_config:
            next_config = cycle + CONFIG_EVERY_S
            guard("config", device.refresh_config)
        guard("measure", device.measure)
        if cycle >= next_upload:
            try:
                device.upload()
                failures = 0
            except Exception as error:  # Later, network trouble, anything: keep the buffer
                failures += 1  # back off: 5 s, doubling up to 15 min, with jitter
                delay = min(5 * 2 ** min(failures - 1, 10), 900) * random.uniform(0.5, 1)
                delay = max(delay, getattr(error, "retry_after", None) or 0)
                next_upload = time.monotonic() + delay
                log.warning("upload: %s; next try in %.0f s", error, delay)
        if cycle >= next_status:
            next_status = cycle + STATUS_EVERY_S
            guard("status", device.send_status)
        scan = ((device.config or {}).get("intervals") or {}).get("scan_seconds") or 60
        time.sleep(max(0.0, cycle + min(max(scan, 10), 3600) - time.monotonic()))


if __name__ == "__main__":
    main()

Was er macht#

• Koppeln: erzeugt einmalig eine UUID, schickt alle 15 Sekunden hello und speichert den Token, bevor er ihn nutzt. Nur der Benutzer, unter dem der Client läuft, kann diese Dateien lesen.
• Konfiguration: alle 15 Minuten mit If-None-Match, in config.json gespeichert für einen Neustart ohne Internet.
• Messen: ruft alle scan_seconds deine Funktion read_sensors auf. Codes, die nicht in der Konfiguration stehen, überspringt er; ein Wert außerhalb des plausiblen Bereichs bekommt out_of_range, kein Wert bekommt no_response.
• Puffer: Alles geht zuerst in buffer.sqlite3. Ein Batch behält seine UUID, auch nach einem Absturz.
• Hochladen: höchstens 8 Batches à 500 pro Minute, die ältesten zuerst. Nach 200 wird der Batch entfernt, außer Messwerten mit retry: true; ein Batch mit 422 wird beiseitegelegt. Bei 401, 429, 5xx oder ohne Antwort wartet er jedes Mal länger, bis zu 15 Minuten und mindestens so lange wie Retry-After.
• Status: alle 5 Minuten, mit der Firmware-Version und der Anzahl Messwerte im Puffer.

Deine Sensoren auslesen#

read_sensors bekommt die Sensoren aus der Konfiguration und gibt pro Code einen Wert zurück, oder None, wenn der Sensor nicht geantwortet hat. Zum Beispiel für DS18B20-Sensoren an einem Raspberry Pi mit aktiviertem 1-Wire, mit den IDs aus /sys/bus/w1/devices/ neben den Codes deines Regelkreises:

PROBES = {"T1": "28-0000075f8c2a", "T2": "28-0000075fa1b3", "T3": "28-0000075fb7c4"}


def read_sensors(sensors: list[dict]) -> dict[str, float | None]:
    values = {}
    for code, probe in PROBES.items():
        try:
            lines = Path(f"/sys/bus/w1/devices/{probe}/w1_slave").read_text().splitlines()
            values[code] = int(lines[1].split("t=")[1]) / 1000 if lines[0].endswith("YES") else None
        except (OSError, IndexError, ValueError):
            values[code] = None  # sends no_response
    return values

Was dieses Beispiel nicht macht#

• Die Uhr prüfen oder etwas für die Uhrkorrektur senden. Lass es auf einem System mit NTP laufen.
• Den Puffer begrenzen. Möchtest du ein Maximum, verwirf die ältesten Messwerte.
• Als Dienst laufen. Unter Linux kann systemd es starten und neu starten.

Die Regeln hinter diesem Verhalten stehen in Messwerte senden und Limits und Fehlercodes.