fixed thresholds catch the obvious failures. they miss the sensor drifting 5°C off baseline overnight, and humidity quietly sliding 12 points below where it sat all week. those are the ones that matter.
this tutorial catches them. ~75 lines of python, a 30-second rolling window, alert when a reading sits more than 3 standard deviations from normal for 3 ticks in a row. works on any metric already streaming into plexus. no ML.
ESP32 + BME280 ──▶ plexus ──▶ detector.py ──▶ plexus ──▶ #your-channel
(any source) temperature 30s window monitor on (slack)
every 2s stream rolling z-score temp.zscore
3-streak guardtwo new pieces. detector.py polls plexus once a second for the latest temperature samples, keeps the most recent 30 seconds in memory, and computes a rolling z-score per new point. when |z| > 3 for three samples in a row — about 6 seconds of sustained deviation — it posts temperature.zscore back to plexus.
plexus already knows how to fire slack on a max threshold. point one at temperature.zscore with max: 3.0 and your existing alert pipeline fires unchanged.
the detector doesn’t touch firmware. it runs anywhere — laptop, vps, or a raspberry pi monitoring hub — and works for any source and metric in your fleet.
esp32-bme280:temperature by default. an ESP32 + BME280 posting over plain HTTP, the python SDK on a pi — any stream works.install:
pip install plexus-python requestsno numpy, no pandas. the math is six lines and we want to see it.
any source pushing temperature to plexus every couple of seconds works — an ESP32 + BME280 posting over plain HTTP, the python SDK on a raspberry pi, anything. that’s the data source for everything below — same source_id, same metric.
no ESP32 yet? point the detector at any temperature stream you have. it’s a SOURCE_ID and METRIC constant at the top of the script.
start detector.py. the window is a collections.deque capped at 15 samples — 30 seconds at a 2-second cadence. deque(maxlen=15) drops the oldest sample automatically when the next one is appended. zero math required.
# detector.py
import collections
import math
WINDOW_SAMPLES = 15 # 30 seconds @ one sample per 2 seconds
window = collections.deque(maxlen=WINDOW_SAMPLES)
def zscore(values):
"""Return the z-score of the latest value vs. the rest of the window."""
if len(values) < 5:
return 0.0
mean = sum(values) / len(values)
var = sum((v - mean) ** 2 for v in values) / len(values)
std = math.sqrt(var)
if std < 0.01:
return 0.0
return (values[-1] - mean) / stdthree guards in there:
< 5 samples: not enough data to call anything anomalous yet. return 0.std < 0.01: the sensor was perfectly flat. anything else divides by zero.that’s the math. six lines if you cut the guards.
now the loop. poll plexus for new temperature samples once a second, push them through the window, and post temperature.zscore back when 3 anomalous samples have landed in a row.
import os
import time
import requests
from plexus import Plexus
SOURCE_ID = "esp32-bme280"
METRIC = "temperature"
Z_THRESHOLD = 3.0
CONSECUTIVE = 3
POLL_S = 1.0
API_BASE = os.environ.get("PLEXUS_ENDPOINT", "https://app.plexus.company")
API_KEY = os.environ["PLEXUS_API_KEY"]
session = requests.Session()
session.headers["x-api-key"] = API_KEY
px = Plexus(source_id=SOURCE_ID)
def fetch_recent():
r = session.get(
f"{API_BASE}/api/v1/telemetry",
params={"source": SOURCE_ID, "metric": METRIC, "limit": 200},
timeout=5,
)
r.raise_for_status()
return sorted(r.json()["data"], key=lambda x: x["timestamp"])
def main():
last_ts = None
consecutive = 0
while True:
for row in fetch_recent():
ts = row["timestamp"]
if last_ts is not None and ts <= last_ts:
continue
last_ts = ts
value = float(row["value"])
window.append(value)
z = zscore(list(window))
consecutive = consecutive + 1 if abs(z) > Z_THRESHOLD else 0
emit = abs(z) if consecutive >= CONSECUTIVE else 0.0
px.send("temperature.zscore", emit)
tag = "ANOMALY" if consecutive >= CONSECUTIVE else "normal "
print(f"[{tag}] t={value:6.2f} z={z:+5.2f} streak={consecutive}")
time.sleep(POLL_S)
if __name__ == "__main__":
main()the 3-consecutive guard is the part that matters. without it, a single noisy reading 3.5σ above the mean fires an alert. with it, the alert only fires when something has actually been wrong for ~6 seconds.
run it:
export PLEXUS_API_KEY=plx_xxx
python detector.pypress the BME280 between your fingers. the temperature climbs and the z-score climbs with it — on a quiet room, even a 1°C jump is several sigma.
[normal ] t= 22.41 z=+0.18 streak=0
[normal ] t= 22.45 z=+0.39 streak=0
[normal ] t= 23.18 z=+2.41 streak=0 ← warming up
[normal ] t= 24.05 z=+3.82 streak=1
[normal ] t= 24.92 z=+4.51 streak=2
[ANOMALY] t= 25.78 z=+5.04 streak=3 ← posts zscore=5.04
[ANOMALY] t= 26.43 z=+5.49 streak=4three things happen on that third anomaly tick:
px.send("temperature.zscore", 5.04) lands at the gateway.5.04 > 3.0, emits alert.triggered.the rule, one curl:
curl -X POST https://app.plexus.company/api/monitors \
-H "x-api-key: $PLEXUS_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"source_id": "esp32-bme280",
"metric": "temperature.zscore",
"threshold": {
"max": 3.0,
"severity": "warning",
"message": "BME280 deviating from baseline"
}
}'let go of the sensor. the temperature drops, the streak resets, the metric goes back to 0, plexus emits alert.resolved. clean.
SOURCE_ID and METRIC. run one process per metric you care about (temperature, humidity, pressure), or fan out inside the script with a list. zscore() doesn’t know what it’s scoring.GET /api/v1/telemetry?metric=temperature with no source filter returns every device; median + MAD becomes the baseline. one bad sensor in a fleet of 50 stands out immediately.next week: dashboards from telemetry, automatically.
detector.py, the monitor curl, and a README live in the docs.