أنظمة الإنذار المبكر والأتمتة

"""Early-warning & automation engine (session 18).

Run every 10 minutes (Windows Task Scheduler / cron:  */10 * * * * python automation.py)
or keep it running:  python automation.py --loop
Rules + ML output -> alerts (Telegram) + automatic pump/fan commands + decision log.
"""
import json
import sys
import time
import urllib.parse
import urllib.request
from datetime import datetime, timedelta

import config
import db
import predictor


def now():
    return datetime.now().isoformat(timespec="seconds")


def notify(text):
    """Send a Telegram message if a bot token is configured (BotFather -> /newbot)."""
    if not (config.TELEGRAM_TOKEN and config.TELEGRAM_CHAT_ID):
        return False
    url = f"https://api.telegram.org/bot{config.TELEGRAM_TOKEN}/sendMessage"
    data = urllib.parse.urlencode({"chat_id": config.TELEGRAM_CHAT_ID, "text": text}).encode()
    try:
        urllib.request.urlopen(url, data=data, timeout=8)
        return True
    except Exception as exc:  # never crash the loop because of the network
        print("telegram error:", exc)
        return False


def already_sent(node_id, message, minutes=60):
    since = (datetime.now() - timedelta(minutes=minutes)).isoformat(timespec="seconds")
    return bool(db.rows("SELECT 1 FROM decisions WHERE node_id=? AND message=? AND ts>=?", (node_id, message, since)))


def alert(node_id, level, message, source="rule"):
    if already_sent(node_id, message):
        return
    db.log_decision(now(), node_id, source, level, message)
    notify(f"🌱 KFS Smart Farm [{level.upper()}]\n{node_id}: {message}")


def queue_command(node_id, actuator, action, minutes, source):
    open_cmd = db.rows("SELECT 1 FROM commands WHERE node_id=? AND actuator=? AND status IN ('pending','delivered')",
                       (node_id, actuator))
    if open_cmd:
        return None
    minutes = min(minutes, config.PUMP_MAX_MIN)
    cid = db.execute("INSERT INTO commands (node_id, actuator, action, duration_min, source, created_by, created_at) "
                     "VALUES (?,?,?,?,?,?,?)", (node_id, actuator, action, minutes, source, "automation", now()))
    db.log_decision(now(), node_id, source, "info", f"auto {actuator} {action} for {minutes} min (command #{cid})")
    return cid


def evaluate():
    nodes = db.rows("SELECT * FROM nodes")
    for n in nodes:
        nid = n["node_id"]
        last = db.rows("SELECT * FROM readings WHERE node_id=? ORDER BY ts DESC LIMIT 1", (nid,))
        if not last:
            continue
        r = last[0]
        age = (datetime.now() - datetime.fromisoformat(r["ts"])).total_seconds() / 60
        if age > config.NODE_OFFLINE_MIN:
            alert(nid, "warning", f"node offline for {int(age)} min - check power / Wi-Fi", "system")
            continue
        if r["battery_v"] is not None and r["battery_v"] < config.BATTERY_LOW_V:
            alert(nid, "warning", f"battery low ({r['battery_v']} V)", "system")
        if r["air_temp_c"] is not None and r["air_temp_c"] >= config.AIR_HOT_C:
            alert(nid, "critical", f"heat stress: air {r['air_temp_c']} °C", "rule")
        if n["zone"] == "greenhouse" and (r["air_humidity_pct"] or 0) >= config.GREENHOUSE_HUMID_PCT:
            alert(nid, "warning", f"humidity {r['air_humidity_pct']}% - fungal risk, fan ON", "rule")
            queue_command(nid, "fan", "on", 15, "rule")

        soil = r["soil_moisture_pct"]
        if soil is None or soil >= config.SOIL_WET_PCT:
            continue
        pred = predictor.predict(nid)
        ml_says = pred.get("ok") and pred["recommendation"] == "irrigate"
        if soil < config.SOIL_DRY_PCT or ml_says:
            reason = f"soil {soil}% < {config.SOIL_DRY_PCT}%" if soil < config.SOIL_DRY_PCT else \
                f"model p={pred['probability']}"
            if pred.get("note"):
                alert(nid, "info", f"irrigation postponed: {pred['note']}", "ml")
                continue
            if queue_command(nid, "pump", "on", 10, "ml" if ml_says else "rule"):
                alert(nid, "warning", f"irrigation started automatically ({reason})", "ml" if ml_says else "rule")


if __name__ == "__main__":
    db.init_db()
    if "--loop" in sys.argv:
        while True:
            evaluate()
            time.sleep(600)
    else:
        evaluate()
        print(json.dumps(db.rows("SELECT * FROM decisions ORDER BY id DESC LIMIT 10"), ensure_ascii=False, indent=1))