From ccf710d8dd47841515aa57bafe6fba9a1438a83b Mon Sep 17 00:00:00 2001 From: Aswin Antony Date: Thu, 12 Mar 2026 18:02:13 +0000 Subject: [PATCH] [SUBSCRIBER] Add threshold alert system --- src/dashboard/alerts.py | 115 ++++++++++++++++++++++++++++++++++++++++ src/dashboard/app.py | 60 ++++++++++++++------- 2 files changed, 155 insertions(+), 20 deletions(-) create mode 100644 src/dashboard/alerts.py diff --git a/src/dashboard/alerts.py b/src/dashboard/alerts.py new file mode 100644 index 0000000..4423d60 --- /dev/null +++ b/src/dashboard/alerts.py @@ -0,0 +1,115 @@ +# alerts.py +# Threshold Alert System — Aswin Antony +# Checks incoming sensor readings against defined thresholds +# and generates alerts in the format expected by the dashboard. + +# ── Thresholds ──────────────────────────────────────────────────────────────── +# Each sensor type has a max (and optionally min) threshold. +# Adjust these values as the group agrees. + +THRESHOLDS = { + "temperature": { + "max": 28.0, + "min": 10.0, + "unit": "°C", + "max_message": "High temperature detected (over 28°C)", + "min_message": "Low temperature detected (under 10°C)", + }, + "co2": { + "max": 1000.0, + "unit": "ppm", + "max_message": "High CO2 levels detected (over 1000 ppm)", + }, + "noise": { + "max": 80.0, + "unit": "dB", + "max_message": "High noise levels detected (over 80 dB)", + }, + "occupancy": { + "max": 30, + "unit": "people", + "max_message": "Room at full capacity", + }, +} + + +def check_threshold(room: str, sensor_type: str, value: float) -> dict | None: + """ + Check a single sensor reading against its threshold. + + Returns an alert dict if the value breaches a threshold, or None if normal. + + Alert format matches dashboard placeholder_alerts: + {"room": ..., "sensor": ..., "message": ..., "value": ...} + """ + sensor_key = sensor_type.lower() + rules = THRESHOLDS.get(sensor_key) + + if rules is None: + return None # No threshold defined for this sensor type + + unit = rules.get("unit", "") + + # Check max threshold + if "max" in rules and value > rules["max"]: + return { + "room": room, + "sensor": sensor_type.capitalize(), + "message": rules["max_message"], + "value": f"{value}{unit}", + "level": "danger", + } + + # Check min threshold (if defined) + if "min" in rules and value < rules["min"]: + return { + "room": room, + "sensor": sensor_type.capitalize(), + "message": rules["min_message"], + "value": f"{value}{unit}", + "level": "warning", + } + + return None + + +def evaluate_readings(readings: list[dict]) -> list[dict]: + """ + Takes a list of sensor reading dicts and returns a list of alerts + for any readings that breach their threshold. + + Each reading dict must have: room, sensor, value (numeric) + """ + alerts = [] + for reading in readings: + try: + room = reading["room"] + sensor_type = reading["sensor"] + # Strip units from value string if needed (e.g. "29.1°C" -> 29.1) + raw_value = reading["value"] + if isinstance(raw_value, str): + numeric = ''.join(c for c in raw_value if c.isdigit() or c == '.') + value = float(numeric) if numeric else None + else: + value = float(raw_value) + + if value is None: + continue + + alert = check_threshold(room, sensor_type, value) + if alert: + alerts.append(alert) + + except (KeyError, ValueError): + continue + + return alerts + + +def get_reading_status(sensor_type: str, value: float) -> str: + """ + Returns 'Alert' or 'Normal' for a given sensor reading. + Used to set the status field on each reading row in the dashboard. + """ + alert = check_threshold("", sensor_type, value) + return "Alert" if alert else "Normal" diff --git a/src/dashboard/app.py b/src/dashboard/app.py index 8986f09..66df202 100644 --- a/src/dashboard/app.py +++ b/src/dashboard/app.py @@ -1,32 +1,52 @@ -from flask import Flask, render_template +from flask import Flask, render_template, jsonify +from alerts import evaluate_readings, get_reading_status -app = Flask(__name__, +app = Flask(__name__, template_folder="templates", static_folder="static") -# --- Placeholder sensor data this will be replaced with actual data --- +# --- Placeholder sensor data — replace with real MQTT data when ready --- +# Values are numeric so alerts.py can evaluate them properly placeholder_readings = [ - {"room": "Room 101","sensor": "Temperature", "value": "22°C", "status": "Normal"}, - {"room": "Room 101","sensor": "CO2", "value": "850 ppm", "status": "Normal"}, - {"room": "Room 101","sensor": "Noise", "value": "72 dB", "status": "Normal"}, - {"room": "Room 101","sensor": "Occupancy", "value": "18/30", "status": "Normal"}, - - {"room": "Room 102","sensor": "Temperature", "value": "29.1°C", "status": "Alert"}, - {"room": "Room 102","sensor": "CO2", "value": "1200 ppm", "status": "Alert"}, - {"room": "Room 102","sensor": "Noise", "value": "58 dB", "status": "Normal"}, - {"room": "Room 102","sensor": "Occupancy", "value": "30/30", "status": "Alert"}, -] + {"room": "Room 101", "sensor": "Temperature", "value": 22.0, "unit": "°C"}, + {"room": "Room 101", "sensor": "Co2", "value": 850.0, "unit": "ppm"}, + {"room": "Room 101", "sensor": "Noise", "value": 72.0, "unit": "dB"}, + {"room": "Room 101", "sensor": "Occupancy", "value": 18, "unit": "people"}, -# --- Placeholder alerts this will be replaced with actual alerts --- -placeholder_alerts = [ - {"room": "Room 102", "sensor": "Temperature", "message": "High temperature detected (Over 28°C)", "value": "29.1°C"}, - {"room": "Room 102", "sensor": "CO2", "message": "High CO2 levels detected (Over 1000 ppm)", "value": "1200 ppm"}, - {"room": "Room 102", "sensor": "Occupancy", "message": "Room at full capacity", "value": "30/30"}, + {"room": "Room 102", "sensor": "Temperature", "value": 29.1, "unit": "°C"}, + {"room": "Room 102", "sensor": "Co2", "value": 1200.0,"unit": "ppm"}, + {"room": "Room 102", "sensor": "Noise", "value": 58.0, "unit": "dB"}, + {"room": "Room 102", "sensor": "Occupancy", "value": 30, "unit": "people"}, ] + +def prepare_readings(readings): + """Add display value string and status to each reading.""" + prepared = [] + for r in readings: + status = get_reading_status(r["sensor"], r["value"]) + prepared.append({ + "room": r["room"], + "sensor": r["sensor"], + "value": f"{r['value']}{r['unit']}", + "status": status, + }) + return prepared + + @app.route("/") def index(): - return render_template("index.html", readings=placeholder_readings, alerts=placeholder_alerts) + readings = prepare_readings(placeholder_readings) + alerts = evaluate_readings(placeholder_readings) + return render_template("index.html", readings=readings, alerts=alerts) + + +@app.route("/api/alerts") +def api_alerts(): + """JSON endpoint — returns current alerts. Useful for auto-refreshing the dashboard.""" + alerts = evaluate_readings(placeholder_readings) + return jsonify(alerts) + if __name__ == "__main__": - app.run(debug=True) \ No newline at end of file + app.run(debug=True)