sensor-duo

Two anomaly detectors for any set of named numeric channels — a transparent linear-trend forecaster, and a spiking neural network built on the real Spikeling engine — logging to SQLite with a generated Grafana dashboard. Point it at temperature, a detection's duration in frame, a queue depth, whatever you've got a float for.

Why this exists: this exact pattern (trend detector + SQLite + Grafana, then a Spikeling-based spiking detector alongside it) was built by hand, once, in pond-health. Building it a second time for a different project made clear it should be a library, not a copy-paste. This is that extraction — generalized over arbitrary channel names instead of 5 hardcoded pond parameters, and with a real, runnable example (examples/demo.py, a CPU-temperature + "person lingering at the door" scenario) proving it actually works outside the pond-specific context it came from.

The two detectors, honestly compared

Validated on pond-health's real scenarios before this was extracted (see that repo's README for the full numbers): the trend detector gives early warning by projecting a linear fit forward, at the cost of sometimes alerting on a trend that reverses before actually crossing the line. The spiking detector never does that — a single noisy reading can't trip a LIF neuron, it only fires once several consecutive stressed readings integrate past threshold, or instantly on a genuinely critical one — but it doesn't forecast anything; it only tells you about a problem that has already become real and sustained. Neither one is strictly better. Run both and let the disagreements between them be the interesting signal.

Install

pip install -e .

The spiking detector needs the Spikeling engine cloned as a sibling directory next to this checkout (git clone https://github.com/tritsystem/Spikeling in the same parent folder), or SPIKELING_CORE_PATH pointing at its core/ folder. It's optional — SpikingDetector() raises SpikelingNotFound if it can't find it, and the trend detector works completely on its own.

Quickstart

from sensor_duo import Reading, Range, TrendDetector, SpikingDetector, SpikelingNotFound, DetectorStore

thresholds = {
    "cpu_temp_c": Range(stress_low=20, stress_high=80, critical_high=90),
    "front_door_person_seconds": Range(stress_high=30, critical_high=180),
}

trend = TrendDetector(thresholds)
store = DetectorStore("my_app.db")
try:
    spiking = SpikingDetector(thresholds)
except SpikelingNotFound:
    spiking = None  # trend detector alone still works fine

reading = Reading(timestamp=..., values={"cpu_temp_c": 91.2, "front_door_person_seconds": 0.0})
trend.ingest(reading)
store.log_reading(reading.timestamp, reading)

for pred in trend.predict_all():
    should_alert = pred.status != "ideal" or pred.crossing_threshold is not None
    store.log_prediction(reading.timestamp, pred, should_alert, detector="trend")
    if should_alert:
        print(pred.explanation)

if spiking:
    spiking.ingest(reading)
    for pred in spiking.predict_all():
        store.log_prediction(reading.timestamp, pred, pred.status != "ideal", detector="spiking")

Run the real example end-to-end:

python examples/demo.py

Auto-tuning the trend detector's horizon

TrendDetector's forecast horizon (how far ahead it's willing to project a crossing) is the exact knob pond-health had to tune by hand after a real false-alarm storm (72h produced 400+ false alarms; 8h fixed it). HorizonAutoTuner automates that: it watches whether each channel's forecasts actually come true, and shrinks that channel's horizon when they systematically don't — no human has to notice the false-alarm pattern first.

from sensor_duo import HorizonAutoTuner

tuner = HorizonAutoTuner(trend)  # wraps a TrendDetector

# each tick:
preds = trend.predict_all()
tuner.observe(reading.timestamp, preds)

It's deliberately narrow in what it claims: it doesn't try to explain why a channel's forecasts miss (a cyclical signal, a noisy sensor, a genuinely bad threshold — it doesn't know or care), it just tracks a plain hit/miss rate per channel and shrinks the horizon when misses dominate. See examples/auto_tune_demo.py for a runnable proof on the exact kind of signal that motivated it (a channel oscillating near its threshold but never crossing it) — the horizon visibly shrinks 8h → 4h → 2h over the run, entirely on its own.

Visualizing with Grafana

from sensor_duo.dashboard import build_dashboard
import json

dashboard = build_dashboard(thresholds, title="My App", labels={"cpu_temp_c": "CPU temperature"})
json.dump(dashboard, open("dashboard.json", "w"), indent=2)

Panel color thresholds are generated straight from your Range values (the same cutoffs classify() alerts on), so the dashboard can't drift out of sync with what the detectors actually do. Import it in Grafana (Dashboards → New → Import) after adding the free SQLite datasource plugin (grafana-cli plugins install frser-sqlite-datasource) pointed at your .db file — see pond-health's grafana/README.md for the full walkthrough (same plugin, same setup, just a different .db).

API

sensor_duo/
  reading.py           Reading -- timestamp + a dict of named float channels.
  thresholds.py         Range + classify() -- the single source of truth
                        both detectors use to turn a value into
                        ideal/stress/critical.
  trend_detector.py     TrendDetector + Prediction -- linear-trend
                        projection, generalized from pond-health's
                        trend_predictor.py.
  spiking_detector.py    SpikingDetector -- generates a Spikeling .spk
                        brain (one LIF neuron per channel) at
                        construction time and runs the real engine.
  store.py              DetectorStore -- normalized SQLite schema (one
                        row per channel per reading), so new channels
                        never need a migration.
  dashboard.py           build_dashboard() -- generates a Grafana
                        dashboard JSON from your thresholds.
  auto_tune.py           HorizonAutoTuner -- shrinks a channel's trend
                        forecast horizon when its forecasts
                        systematically don't come true.
examples/
  demo.py                Runnable end-to-end proof: two channels, both
                        detectors, SQLite + a generated dashboard.
  auto_tune_demo.py       Runnable proof the horizon actually
                        auto-shrinks on a cyclical never-crossing signal.

Tests

pytest

30 tests, all exercising real behavior (not just imports) — including the spiking detector's tests, which actually compile and run the Spikeling engine rather than mocking it. They skip gracefully (with a clear reason) if Spikeling isn't cloned alongside this repo.