alerts.py

"""
Alert rule engine. Evaluates the rules defined in config.yaml against the
latest data. Writes new alerts to the alerts table, resolves alerts when
conditions clear, and is called periodically from the dashboard.
"""
import pandas as pd
from datetime import datetime
from config_loader import CONFIG
import db


def _latest_alert_for_rule(conn, rule_id):
    cursor = conn.cursor()
    cursor.execute(
        "SELECT id, resolved_at FROM alerts WHERE rule_id = ? ORDER BY triggered_at DESC LIMIT 1",
        (rule_id,)
    )
    return cursor.fetchone()


def _set_alert_state(conn, rule_id, severity, message, triggered):
    """If triggered and no active alert exists, create one.
    If not triggered and an active alert exists, resolve it.
    Returns True if the state changed."""
    last = _latest_alert_for_rule(conn, rule_id)
    currently_active = last is not None and last[1] is None

    if triggered and not currently_active:
        db.insert_alert(conn, rule_id, severity, message)
        return True
    if not triggered and currently_active:
        db.resolve_alert(conn, rule_id)
        return True
    return False


def _check_meter_offline(conn, rule):
    threshold = rule['offline_threshold_minutes']
    changed = False
    for meter in CONFIG['meters']:
        mid = meter['id']
        cursor = conn.cursor()
        # Get the most recent OK reading time
        cursor.execute(
            "SELECT MAX(timestamp) FROM flow_readings WHERE meter_id=? AND status='OK'",
            (mid,)
        )
        last_ok_row = cursor.fetchone()
        last_ok = last_ok_row[0] if last_ok_row and last_ok_row[0] else None

        triggered = False
        minutes_offline = 0
        if last_ok:
            last_ok_dt = pd.to_datetime(last_ok)
            age_minutes = (datetime.now() - last_ok_dt).total_seconds() / 60
            if age_minutes > threshold:
                triggered = True
                minutes_offline = int(age_minutes)

        rule_key = f"{rule['id']}:{mid}"
        msg = rule['message'].format(meter_name=meter['name'], minutes=minutes_offline)
        if _set_alert_state(conn, rule_key, rule['severity'], msg, triggered):
            changed = True
    return changed


def _check_low_recovery(conn, rule):
    window = rule['window_minutes']
    threshold = rule['recovery_threshold_pct']

    df = pd.read_sql_query(
        f"""SELECT meter_id,
                   MAX(total_flow) - MIN(total_flow) as usage,
                   COUNT(*) as cnt
            FROM flow_readings
            WHERE status='OK'
              AND timestamp >= datetime('now', 'localtime', '-{window} minutes')
            GROUP BY meter_id""",
        conn
    )

    triggered = False
    recovery_pct = 0
    if len(df) >= 2 and 1 in df['meter_id'].values and 2 in df['meter_id'].values:
        m1_use = df.loc[df['meter_id'] == 1, 'usage'].values[0]
        m2_use = df.loc[df['meter_id'] == 2, 'usage'].values[0]
        # Require a minimum amount of data to avoid noise
        m1_cnt = df.loc[df['meter_id'] == 1, 'cnt'].values[0]
        if m2_use > 0.1 and m1_cnt > 10:
            return_vol = m2_use - m1_use
            recovery_pct = (return_vol / m2_use) * 100
            if recovery_pct < threshold:
                triggered = True

    msg = rule['message'].format(value=recovery_pct)
    return _set_alert_state(conn, rule['id'], rule['severity'], msg, triggered)


def _check_high_makeup_flow(conn, rule):
    window = rule['window_minutes']
    threshold = rule['avg_threshold_m3h']
    meter_id = rule['meter_id']

    cursor = conn.cursor()
    cursor.execute(
        f"""SELECT AVG(instant_flow), COUNT(*) FROM flow_readings
            WHERE meter_id=? AND status='OK'
              AND timestamp >= datetime('now', 'localtime', '-{window} minutes')""",
        (meter_id,)
    )
    row = cursor.fetchone()
    avg_flow = row[0] if row and row[0] is not None else 0
    cnt = row[1] if row else 0

    triggered = cnt > 10 and avg_flow > threshold
    msg = rule['message'].format(value=avg_flow)
    return _set_alert_state(conn, rule['id'], rule['severity'], msg, triggered)


def _check_data_stale(conn, rule):
    threshold = rule['stale_threshold_minutes']
    cursor = conn.cursor()
    cursor.execute("SELECT MAX(timestamp) FROM flow_readings")
    row = cursor.fetchone()
    last_ts = row[0] if row and row[0] else None

    triggered = False
    minutes_stale = 0
    if last_ts:
        age_minutes = (datetime.now() - pd.to_datetime(last_ts)).total_seconds() / 60
        if age_minutes > threshold:
            triggered = True
            minutes_stale = int(age_minutes)
    else:
        triggered = True

    msg = rule['message'].format(minutes=minutes_stale)
    return _set_alert_state(conn, rule['id'], rule['severity'], msg, triggered)


def _check_low_temperature(conn, rule):
    window = rule['window_minutes']
    threshold = rule['temperature_threshold_c']
    meter_id = rule['meter_id']

    cursor = conn.cursor()
    cursor.execute(
        f"""SELECT AVG(temperature), COUNT(*) FROM flow_readings
            WHERE meter_id=? AND status='OK'
              AND timestamp >= datetime('now', 'localtime', '-{window} minutes')""",
        (meter_id,)
    )
    row = cursor.fetchone()
    avg_temp = row[0] if row and row[0] is not None else None
    cnt = row[1] if row else 0

    # Need a minimum sample to avoid noise on startup
    triggered = cnt > 10 and avg_temp is not None and avg_temp < threshold
    msg_value = avg_temp if avg_temp is not None else 0
    msg = rule['message'].format(value=msg_value, threshold=threshold)
    return _set_alert_state(conn, rule['id'], rule['severity'], msg, triggered)


# Dispatch table — add new rule types here
_CHECK_FUNCTIONS = {
    'meter_offline': _check_meter_offline,
    'low_recovery': _check_low_recovery,
    'high_makeup_flow': _check_high_makeup_flow,
    'data_stale': _check_data_stale,
    'low_temperature': _check_low_temperature,
}


def evaluate_all():
    """Run every enabled rule. Called by the dashboard on a timer."""
    if not CONFIG['alerts']['enabled']:
        return

    conn = db.get_db_connection()
    try:
        for rule in CONFIG['alerts']['rules']:
            if not rule.get('enabled', True):
                continue
            check_fn = _CHECK_FUNCTIONS.get(rule['id'])
            if check_fn:
                try:
                    check_fn(conn, rule)
                except Exception as e:
                    print(f"[ALERT ERROR] {rule['id']}: {e}")
    finally:
        conn.close()