[SUBSCRIBER] Add threshold alert system

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"

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)
+    app.run(debug=True)