Examples
Example: Python
A small, complete Python 3 client for your own hardware that pairs, keeps the configuration up to date, buffers readings in SQLite and uploads them in order.
This small client sends the readings of your own hardware to BoilerBuddy: replace the function that reads your sensors, and adapt the rest where needed. It pairs itself if it has to, keeps the configuration up to date, buffers in SQLite and uploads in order.
What you need
- Python 3.10 or newer, with
requests(pip install requests). - A device in the portal with a circuit. See Building your own BoilerBuddy.
- A clock kept in sync by NTP (on Linux:
timedatectl). The client relies on it; see Buffering and the clock.
Running it
With a key from the portal:
export BB_TOKEN=bb_live_42_...
python3 boilerbuddy_client.py
Without BB_TOKEN the client pairs itself. It prints a pairing code, which you enter with Pair with code on the device page, and keeps the token in bb-state/token.
| Variable | Default | Purpose |
|---|---|---|
BB_TOKEN |
(empty) | The device's API key |
BB_API_URL |
https://boilerbuddy.eu/api/v1 |
The base URL |
BB_STATE_DIR |
bb-state |
Folder for token, uuid, configuration and buffer |
The 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()
What it does
- Pairing: generates a uuid once, sends
helloevery 15 seconds and stores the token before using it. Only the user running the client can read those files. - Configuration: every 15 minutes with
If-None-Match, kept inconfig.jsonfor a restart without internet. - Measuring: calls your
read_sensorseveryscan_seconds. It skips codes that are not in the configuration; a value outside the plausible range getsout_of_range, no value getsno_response. - Buffer: everything goes to
buffer.sqlite3first. A batch keeps its uuid, also after a crash. - Uploading: at most 8 batches of 500 per minute, oldest first. After
200the batch is removed, except readings withretry: true; a422is set aside. On401,429,5xxor no answer it waits longer each time, up to 15 minutes and at least as long asRetry-After. - Status: every 5 minutes, with the firmware version and the number of readings in the buffer.
Reading your sensors
read_sensors receives the sensors from the configuration and returns a value per code, or None if the sensor did not answer. For example for DS18B20 sensors on a Raspberry Pi with 1-Wire enabled, with the ids from /sys/bus/w1/devices/ next to the codes of your circuit:
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
What this example does not do
- Check the clock or send anything for clock correction. Run it on a system with NTP.
- Limit the buffer. If you want a maximum, drop the oldest readings.
- Run as a service. On Linux, systemd can start and restart it.
The rules behind this behaviour are in Sending readings and Limits and error codes.