Voorbeeld: Python

Deze kleine client stuurt de metingen van je eigen hardware naar BoilerBuddy: vervang de functie die je sensoren leest, en pas de rest aan waar nodig. Hij koppelt zo nodig zelf, houdt de configuratie bij, buffert in SQLite en verstuurt op volgorde.

Wat je nodig hebt#

• Python 3.10 of nieuwer, met requests (pip install requests).
• Een apparaat in het portaal met een regelkring. Zie Je eigen BoilerBuddy bouwen.
• Een klok die via NTP gelijk loopt (op Linux: timedatectl). De client rekent erop; zie Bufferen en de klok.

Starten#

Met een sleutel uit het portaal:

export BB_TOKEN=bb_live_42_...
python3 boilerbuddy_client.py

Zonder BB_TOKEN koppelt de client zelf. Hij print een koppelcode, die je invult bij Koppelen met code op de pagina van het apparaat, en bewaart het token in bb-state/token.

Variabele	Standaard	Waarvoor
BB_TOKEN	(leeg)	De API-sleutel van het apparaat
BB_API_URL	https://boilerbuddy.eu/api/v1	De basis-URL
BB_STATE_DIR	bb-state	Map voor token, uuid, configuratie en buffer

De 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()

Wat hij doet#

• Koppelen: maakt één keer een uuid, stuurt elke 15 seconden hello en bewaart het token voordat hij het gebruikt. Alleen de gebruiker die de client draait, kan die bestanden lezen.
• Configuratie: elke 15 minuten met If-None-Match, bewaard in config.json voor een herstart zonder internet.
• Meten: elke scan_seconds je read_sensors. Codes buiten de configuratie slaat hij over; buiten het plausibele bereik wordt out_of_range, geen waarde no_response.
• Buffer: alles gaat eerst naar buffer.sqlite3. Een batch houdt zijn uuid, ook na een crash.
• Versturen: hooguit 8 batches van 500 per minuut, oudste eerst. Na 200 verdwijnt de batch, behalve retry: true; een 422 zet hij apart. Bij 401, 429, 5xx of geen antwoord wacht hij steeds langer, tot hooguit 15 minuten en minstens Retry-After.
• Status: elke 5 minuten, met de firmwareversie en het aantal metingen in de buffer.

Je sensoren lezen#

read_sensors krijgt de sensoren uit de configuratie en geeft per code een waarde terug, of None als de sensor niet antwoordde. Bijvoorbeeld voor DS18B20-sensoren op een Raspberry Pi met 1-Wire aan, met de id's uit /sys/bus/w1/devices/ bij de codes van je regelkring:

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

Wat dit voorbeeld niet doet#

• De klok controleren of klokcorrectie meesturen. Draai het op een systeem met NTP.
• De buffer begrenzen. Wil je een maximum, gooi dan de oudste metingen weg.
• Als dienst draaien. Op Linux kan systemd het starten en herstarten.

De regels hierachter staan bij Metingen insturen en Limieten en foutcodes.