"""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))