Building a BME680 air-quality monitor on a Raspberry Pi Zero W

A BME680 on a Raspberry Pi Zero W. It samples temperature, humidity, pressure, and a relative air-quality score every 10 seconds into SQLite, and serves a password-gated Flask dashboard over a Cloudflare named tunnel.

The air-quality number is a relative 0 to 100 score, higher is cleaner. It is not a calibrated AQI. The gas sensor reads VOC resistance in ohms, which means nothing in isolation. The score compares the current reading against a rolling baseline of clean air in the same room, covered below.

The data flow: one sensor feeds one collector thread, which writes to SQLite. A Flask app reads SQLite for both the dashboard and the alert checks. A Cloudflare named tunnel wraps the Flask app, so nothing is exposed on the router.

[ B C M 0 l E x o 6 7 u 8 6 d 0 / f 0 l ( x a I 7 r 2 7 e C ) n a m e d t u n n e c d ( l o a b l 1 s r w l 0 h o r e s b w a c o s p t t a e s o i r r r c d ) t k h t e h r F e l a a d s k a p p : r e a S c Q r F h L e l a i a a b t d s l e i k e n ( g A f W s P r A I o L d m ) b a n y w h e r e , n D t o i h s r o c e p o s e r h n d o l r + d o u n a t t l e f e r y r t p s o r t s ]

The BME680 is a four-in-one sensor: temperature, humidity, barometric pressure, and a gas-resistance reading from a heated metal-oxide plate that responds to volatile organic compounds. One I2C device, four numbers.

It runs on a Pi Zero W over I2C and answers at 0x76 or 0x77 depending on how the SDO pin is strapped. First check after wiring:

i2cdetect -y 1

That scans bus 1 and shows the BME680 at 0x76 or 0x77. A blank address means the wiring is wrong. The driver tries the primary address and falls back to the secondary, so either strapping works:

try:
    self._sensor = bme680.BME680(bme680.I2C_ADDR_PRIMARY)
except (RuntimeError, IOError):
    self._sensor = bme680.BME680(bme680.I2C_ADDR_SECONDARY)

The gas heater is configured once at startup: 320C plate temperature for 150ms, with oversampling on the other channels. The metal-oxide element only gives a trustworthy resistance once it is thermally stable, and the driver reports whether it was. If it was not, the read returns zero rather than a number that cannot be trusted:

"gas_resistance": float(d.gas_resistance) if d.heat_stable else 0.0,

That zero is a flag, not a measurement. The rest of the code treats it as one.

The collector is a daemon thread that ticks every 10 seconds (the interval is configurable). Each tick reads the sensor, builds a reading, writes it to SQLite, and checks alerts.

The bme680 library only runs on the Pi. A FakeSensor stands in off-Pi and returns plausible numbers, so dev and the full pytest suite run on any machine with no hardware:

class FakeSensor:
    """Deterministic-ish fake for tests and dev on non-Pi machines."""

    def read(self):
        return {
            "temperature": round(20 + random.uniform(-1, 1), 2),
            "humidity": round(45 + random.uniform(-5, 5), 2),
            "pressure": round(1013 + random.uniform(-2, 2), 2),
            "gas_resistance": round(50000 + random.uniform(-5000, 5000), 2),
        }

Local setup is a venv and three packages, without the hardware library:

python -m venv .venv
.venv/bin/pip install Flask waitress pytest      # bme680 is Pi-only
.venv/bin/python -m pytest -v

The loop runs unattended for months. A transient I2C read error, or a brief SQLite write lock while a request reads the same file, must not take down the collector. The loop body is wrapped, the error logged, and the next tick proceeds:

def _run(self):
    if not self.restore_baseline():   # fresh DB -> burn in, then persist
        self._burn_in()
        self._persist_baseline()
    last_prune = 0
    last_adapt = 0
    while not self._stop.is_set():
        # A transient I2C read error or a brief DB lock must not kill the
        # collector thread; log it and keep sampling on the next tick.
        try:
            now = int(time.time())
            self.tick(now=now)
            if now - last_prune > 86_400:
                db.prune(self.conn, now=now, retention_days=self.retention_days)
                last_prune = now
            if now - last_adapt > 3_600:
                self._adapt_baseline(now)
                last_adapt = now
        except Exception:
            log.exception("collector sample/prune failed; continuing")
        self._stop.wait(self.interval)

A blanket except Exception is usually a smell. Here it is deliberate. The thread never dies. The worst case is one missing 10-second sample and a line in the log.

Raw gas resistance has no fixed meaning. The same room reads differently across days on the same sensor, and two BME680s never agree. The score is built from two parts: humidity, which anchors to a known-good value, and gas, which is scored against a moving reference.

Humidity is 25% of the score, gas the other 75%. Humidity is scored against a 40% optimum, falling off in either direction. Gas is the ratio of the current reading to the baseline, capped at its share:

HUM_BASELINE = 40.0   # optimal relative humidity %
HUM_WEIGHT = 0.25     # humidity share of the score (gas gets the rest)


def air_quality_score(gas_resistance, humidity, gas_baseline):
    """Return a 0-100 relative air-quality score (higher = cleaner)."""
    hum_offset = humidity - HUM_BASELINE
    if hum_offset > 0:
        hum_score = (100 - HUM_BASELINE - hum_offset) / (100 - HUM_BASELINE)
    else:
        hum_score = (HUM_BASELINE + hum_offset) / HUM_BASELINE
    hum_score = max(0.0, hum_score) * (HUM_WEIGHT * 100)

    if gas_baseline <= 0:
        gas_score = 0.0
    elif gas_resistance < gas_baseline:
        gas_score = (gas_resistance / gas_baseline) * (100 * (1 - HUM_WEIGHT))
    else:
        gas_score = 100 * (1 - HUM_WEIGHT)

    return round(max(0.0, min(100.0, hum_score + gas_score)), 1)

Higher gas resistance means cleaner air, so a reading at or above the baseline scores full gas marks, and a reading below scores proportionally. Every score depends on gas_baseline being a sensible value.

A naive build burns in once at startup, averaging gas resistance over a warm-up window, then uses that value indefinitely. Two problems: the room’s clean-air reference drifts over weeks, and every power cut discards the baseline and forces another burn-in. A Pi Zero W on mains power loses power fairly often.

The baseline here is rolling and persisted. Once an hour, the collector takes the 90th percentile of gas resistance over the trailing 24 hours and adopts it. The p90 picks a clean reading from the recent past while ignoring the single cleanest spike. It is computed in SQLite by sorting and indexing, rather than adding a stats library to the Pi:

def gas_percentile(conn, now, window=86_400, pct=0.9):
    """Return the pct-percentile gas_resistance over the trailing window, or None."""
    rows = conn.execute(
        "SELECT gas_resistance FROM readings "
        "WHERE ts >= ? AND gas_resistance > 0 ORDER BY gas_resistance ASC",
        (now - window,),
    ).fetchall()
    vals = [r["gas_resistance"] for r in rows]
    if not vals:
        return None
    idx = min(len(vals) - 1, int(pct * (len(vals) - 1)))
    return vals[idx]

The gas_resistance > 0 filter excludes the zero flags from unstable heater reads, so the baseline is computed only from real captures. The hourly adapt step writes the result to the settings table:

def _adapt_baseline(self, now):
    p90 = db.gas_percentile(self.conn, now=now, window=86_400, pct=0.9)
    if p90 is not None and p90 > 0:
        self.gas_baseline = p90
        self._persist_baseline()

On boot, restore_baseline() reads the stored value before the loop starts. If it finds one, there is no burn-in:

def restore_baseline(self):
    stored = db.get_setting(self.conn, "gas_baseline")
    if stored is not None:
        self.gas_baseline = float(stored)
        return True
    return False

Only a genuinely empty database falls through to _burn_in, which averages gas over the warm-up window, and that result is persisted too. After the first run, burn-in does not happen again.

The BME680 heats itself. The gas plate runs at 320C, and the temperature sensor sits next to it, so the reading drifts a degree or two above the real room temperature. The correction is a temp_offset subtracted from every reading, editable live from the dashboard with no restart.

Humidity needs the same treatment. Relative humidity is relative to temperature. Correcting the temperature down by 1.5C while reporting the raw RH reports humidity against a temperature now declared wrong. The water content has not changed, so the correct step holds vapour pressure constant and re-expresses RH at the corrected temperature:

def corrected_humidity(raw_temp, raw_rh, corrected_temp):
    """RH re-expressed at corrected_temp, conserving actual vapour pressure."""
    rh = raw_rh * _es_ratio(raw_temp) / _es_ratio(corrected_temp)
    return round(max(0.0, min(100.0, rh)), 2)

_es_ratio is saturation vapour pressure from the Magnus formula, up to a constant that cancels in the ratio. build_reading applies both corrections, and only re-expresses humidity when an offset is set:

def build_reading(sensor_data, gas_baseline, ts, temp_offset=0.0):
    raw_temp = sensor_data["temperature"]
    raw_hum = sensor_data["humidity"]
    temperature = raw_temp - temp_offset
    humidity = (corrected_humidity(raw_temp, raw_hum, temperature)
                if temp_offset else raw_hum)

Lower the temperature and the humidity rises accordingly. Skip the correction and the dew-point and mold numbers are off.

Some values are derived rather than measured. Dew point comes from temperature and humidity via the Magnus formula:

def dew_point(temp_c, rh):
    """Dew point in C via the Magnus formula. RH clamped to (0, 100]."""
    rh = max(1e-6, min(100.0, rh))
    gamma = math.log(rh / 100.0) + (_A * temp_c) / (_B + temp_c)
    return round((_B * gamma) / (_A - gamma), 1)

The dashboard pairs dew point with a mold-risk label, since sustained high humidity is the condition worth a warning, not a single spike.

Pressure gets a similar treatment. A barometer reads station pressure, which depends on altitude, so two locations at different elevations report different numbers in identical weather. The ICAO International Standard Atmosphere formula corrects to sea level, with a 0.0065 K/m lapse rate and an exponent of 5.257:

def sea_level_pressure(raw_hpa, temp_c, altitude_m):
    """Adjust station pressure to sea level (barometric formula).

    altitude_m == 0 (or a missing temperature) returns the raw value unchanged.
    """
    if not altitude_m or temp_c is None:
        return round(raw_hpa, 2)
    # ICAO ISA: lapse rate 0.0065 K/m; exponent 5.257 = g*M/(R*L).
    factor = (1.0 - (0.0065 * altitude_m)
              / (temp_c + 0.0065 * altitude_m + 273.15)) ** -5.257
    return round(raw_hpa * factor, 2)

Altitude is editable in the UI, and the dashboard shows the corrected sea-level value next to the raw station value.

None of these derived numbers are stored. The readings table holds six columns: ts, temperature, humidity, pressure, gas_resistance, and air_quality:

CREATE TABLE IF NOT EXISTS readings (
    ts             INTEGER PRIMARY KEY,
    temperature    REAL,
    humidity       REAL,
    pressure       REAL,
    gas_resistance REAL,
    air_quality    REAL
);

Dew point and sea-level pressure are recomputed at read time and alert time. The table stays small, and changing the altitude setting corrects history retroactively, because nothing stale was written.

The gas plate is the part of a BME680 that wears out. Running a 320C heater every 10 seconds indefinitely is hard on it and unnecessary. VOC levels do not change on a 10-second timescale the way the heater cycle does.

So the gas read is decoupled from the base sample. gas_interval_seconds fires the heater less often than the every-10-seconds temperature, humidity, and pressure reads. Between heater reads, the last stable gas value is carried forward so every row has a gas number:

self._last_gas_fresh = read_gas and data["gas_resistance"] > 0
if self._last_gas_fresh:
    self._last_gas = data["gas_resistance"]
data["gas_resistance"] = self._last_gas       # fresh or carried

The timer is the detail. Restarting the interval clock on every requested read is wrong, because a requested read can come back unstable (the zero flag again), and the next attempt should not wait a full interval. The timer restarts only on a stable capture:

def tick(self, now=None):
    now = int(time.time()) if now is None else now
    gas_interval = self._gas_interval()
    read_gas = (self._last_gas_ts is None
                or now - self._last_gas_ts >= gas_interval)
    got = self.sample_once(ts=now, read_gas=read_gas)
    # Only restart the interval timer on a stable gas capture, so a transient
    # unstable read is retried next tick rather than carried for a full interval.
    if got and self._last_gas_fresh:
        self._last_gas_ts = now
    self._check_stall(now, got)
    return got

An unstable read is retried on the next tick instead of being held for the whole interval. The interval is editable from the dashboard.

Everything lands in one SQLite file. With a single writer (the collector) and a stream of short-lived request readers (the dashboard), the contention is write locks and journal access. WAL mode and a 5-second busy timeout handle both:

def connect(path, check_same_thread=True):
    conn = sqlite3.connect(path, check_same_thread=check_same_thread)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA busy_timeout=5000")  # wait out brief write locks
    conn.execute("PRAGMA journal_mode=WAL")  # readers and the writer don't block
    return conn

At 10 seconds per sample, a year is over three million rows, too many to send to a browser and not needed. Each range maps to a bucket size, and SQLite averages every sample into its bucket using integer division on the timestamp:

Range Window Bucket
1h 1 hour 10s
24h 1 day 60s
7d 7 days 600s
30d 30 days 1h
1y 1 year 6h 
BUCKETS = {
    "1h":  (3_600,       10),
    "24h": (86_400,      60),
    "7d":  (604_800,     600),
    "30d": (2_592_000,   3_600),
    "1y":  (31_536_000,  21_600),
}

The query floors each timestamp to its bucket with (ts / bucket) * bucket, averages every column, and groups:

cur = conn.execute(
    "SELECT (ts / :bucket) * :bucket AS ts, "
    "AVG(temperature) AS temperature, AVG(humidity) AS humidity, "
    "AVG(pressure) AS pressure, AVG(gas_resistance) AS gas_resistance, "
    "AVG(air_quality) AS air_quality "
    "FROM readings WHERE ts >= :since "
    "GROUP BY ts / :bucket ORDER BY ts ASC",
    {"bucket": bucket, "since": since},
)

A one-year plot returns a few thousand averaged points, with the aggregation done in SQLite where the data already is. A separate stats query returns per-range min, avg, and max for each metric, printed under each chart. CSV export returns the same bucketed rows. Once a day the collector prunes anything past the retention window, default 366 days:

def prune(conn, now, retention_days=366):
    cutoff = now - retention_days * 86_400
    cur = conn.execute("DELETE FROM readings WHERE ts < ?", (cutoff,))
    conn.commit()
    return cur.rowcount

The front end is Flask serving one HTML shell and a vanilla JavaScript file, with uPlot vendored in for the charts. No React, no bundler, no build step.

Each tile carries a comfort-band label, so humidity reads as “comfortable” or “damp” alongside 54%, and air quality reads as “good” alongside 89.6. The current endpoint folds in the derived values and a staleness flag, computed from the reading’s age against the sampling interval:

out["age_seconds"] = age
out["interval"] = interval
out["stale"] = age > 3 * interval

If the collector misses three intervals, the dashboard marks the reading stale instead of showing it as live. A /healthz endpoint returns 503 once readings are more than six intervals old, with no login required so a monitor can reach it. The dashboard is an installable PWA, a manifest plus a service worker:

{
  "name": "Air Quality",
  "short_name": "AirQuality",
  "start_url": "/",
  "display": "standalone",
  ...
}

The comfort-band thresholds are not hardcoded. A settings panel edits them live, alongside the sensor tuning from the earlier sections (temp offset, gas read interval, altitude) and the login hardening: max failed attempts before lockout, lockout duration, and session lifetime. Everything applies without a redeploy or a config-file edit on the Pi.

The monitor also pushes alerts. Each metric (air quality, humidity, temperature, pressure, dew point) takes an optional below threshold, above threshold, or both. sustain_seconds requires a breach to last that long before firing, which removes single-sample noise. Recovery is reported once the value clears the threshold by a margin, so it does not flap across the line. Notifications go to a Discord webhook, with an optional ntfy topic alongside.

A self-alert covers the sensor itself. If no reading arrives within stall_alert_seconds (default 300), the collector fires a stall notification, then a recovery notice when sampling resumes.

Alerting must not break sampling. A bad webhook URL, a Discord outage, or a bug in the evaluation logic is logged and swallowed at the point of dispatch:

def _dispatch_alerts(self, reading):
    # Alerting must never break the sampling loop: a bad webhook or an
    # evaluate() bug is logged and swallowed here. With no webhook set we
    # still advance alert state but send nothing.
    try:
        for event in self.alert_manager.evaluate(reading):
            alerts.dispatch(self.alert_manager, event)
    except Exception:
        log.exception("alert dispatch failed; continuing")

SQLite logging continues regardless of what the notification side is doing. Alerts are a layer on top, not a dependency the core loop can trip over.

In practice the feed looks like this: a metric crosses its line, an alert fires, and a recovery message follows once it clears. Most of mine are humidity creeping above 65% overnight and recovering by morning.

The dashboard is reachable remotely without opening a port on the router. A Cloudflare named tunnel dials out from the Pi to Cloudflare, which terminates the public hostname and routes traffic back down the tunnel. There is no inbound rule and no port forward. The dashboard sits behind a single shared password with session cookies on top.

Deploys are a one-liner from the laptop. The Pi tracks the git repo, and a deploy command pulls and restarts:

ssh your-pi project-deploy            # deploy origin/main
ssh your-pi 'project-deploy feat/x'   # deploy a specific branch

The script hard-resets to the target branch, reinstalls Python dependencies only if requirements.txt changed, syncs the systemd units only if they changed, and restarts:

echo "==> Fetching origin/$BRANCH"
before_req="$(req_hash)"
git fetch origin
git reset --hard "origin/$BRANCH"
after_req="$(req_hash)"

if [ "$before_req" != "$after_req" ]; then
  echo "==> requirements.txt changed; installing dependencies"
  .venv/bin/pip install -r requirements.txt
else
  echo "==> requirements.txt unchanged; skipping pip install"
fi

The config file and the SQLite database are git-ignored, so a deploy never touches the password or the history. Two systemd services run on boot: one for the Flask app and collector, one for the tunnel. The tunnel starts after the app:

[Unit]
Description=Cloudflare Tunnel for Air Quality dashboard
After=network.target airquality.service

[Service]
User=your-user
EnvironmentFile=/etc/cloudflared.env
ExecStart=/usr/local/bin/cloudflared tunnel --no-autoupdate run --token ${TUNNEL_TOKEN}
Restart=on-failure
RestartSec=5

Restart=on-failure on both recovers a crash without intervention.

The whole project is one Pi, one sensor, one SQLite file, and a few hundred lines of Python. The one caveat worth repeating: the score is a relative trend for one room, not a figure to compare against another sensor. That is the limit of a metal-oxide gas sensor, and it is enough to answer whether to open a window.

The code is on GitHub under MIT: github.com/jamieede123/airquality .

×
Page views: …