"""KFS Smart Farm — Flask backend + dashboard (sessions 10-20). Run: pip install -r requirements.txt python app.py -> http://:5000 (login: admin / kfs-farm-2026) Nodes: POST /api/v1/readings with header X-API-Key (see firmware/esp32_node_http) """ import csv import hmac import io import secrets from datetime import datetime, timedelta from functools import wraps from flask import Flask, Response, abort, jsonify, redirect, render_template, request, session, url_for from werkzeug.security import check_password_hash import config import db import predictor app = Flask(__name__) app.config.update(SECRET_KEY=config.SECRET_KEY, SESSION_COOKIE_HTTPONLY=True, SESSION_COOKIE_SAMESITE="Lax", PERMANENT_SESSION_LIFETIME=timedelta(hours=8)) FIELDS = ["air_temp_c", "air_humidity_pct", "soil_moisture_pct", "light", "battery_v", "rssi"] LIMITS = {"air_temp_c": (-10, 60), "air_humidity_pct": (0, 100), "soil_moisture_pct": (0, 100), "battery_v": (0, 6)} _login_failures = {} def now(): return datetime.now().isoformat(timespec="seconds") # ----------------------------------------------------------------- security helpers def login_required(view): @wraps(view) def wrapper(*a, **kw): if "user" not in session: if request.path.startswith("/api/"): return jsonify(error="login required"), 401 return redirect(url_for("login", next=request.path)) return view(*a, **kw) return wrapper def node_auth(view): """Nodes authenticate with a shared API key (constant-time comparison).""" @wraps(view) def wrapper(*a, **kw): key = request.headers.get("X-API-Key", "") if not hmac.compare_digest(key, config.NODE_API_KEY): return jsonify(error="invalid API key"), 401 return view(*a, **kw) return wrapper def csrf_token(): if "csrf" not in session: session["csrf"] = secrets.token_hex(16) return session["csrf"] def check_csrf(): sent = request.headers.get("X-CSRF-Token") or request.form.get("csrf") if not sent or not hmac.compare_digest(sent, session.get("csrf", "")): abort(403) app.jinja_env.globals["csrf_token"] = csrf_token @app.after_request def security_headers(resp): resp.headers["X-Content-Type-Options"] = "nosniff" resp.headers["X-Frame-Options"] = "DENY" resp.headers["Referrer-Policy"] = "same-origin" return resp # ----------------------------------------------------------------- pages @app.route("/login", methods=["GET", "POST"]) def login(): error = None if request.method == "POST": ip = request.remote_addr fails = [t for t in _login_failures.get(ip, []) if t > datetime.now() - timedelta(minutes=10)] if len(fails) >= 5: error = "محاولات كثيرة — حاول بعد 10 دقائق" else: user = db.rows("SELECT * FROM users WHERE username=?", (request.form.get("username", ""),)) if user and check_password_hash(user[0]["password_hash"], request.form.get("password", "")): session.clear() session["user"] = user[0]["username"] session.permanent = True nxt = request.args.get("next", "/") return redirect(nxt if nxt.startswith("/") and not nxt.startswith("//") else "/") _login_failures[ip] = fails + [datetime.now()] error = "اسم المستخدم أو كلمة المرور غير صحيحة" return render_template("login.html", error=error) @app.get("/logout") def logout(): session.clear() return redirect(url_for("login")) @app.get("/") @login_required def dashboard(): return render_template("dashboard.html", nodes=db.rows("SELECT * FROM nodes ORDER BY node_id"), pump_max=config.PUMP_MAX_MIN) @app.get("/map") @login_required def farm_map(): return render_template("map.html", lat=config.FARM_LAT, lon=config.FARM_LON) @app.get("/health") def health(): return jsonify(status="ok", time=now(), readings=db.rows("SELECT COUNT(*) n FROM readings")[0]["n"]) # ----------------------------------------------------------------- node API @app.post("/api/v1/readings") @node_auth def post_reading(): data = request.get_json(silent=True) or {} node_id = str(data.get("node_id", "")).strip() if not node_id or len(node_id) > 40: return jsonify(error="node_id required"), 400 values = {} for f in FIELDS: v = data.get(f) if v is None: values[f] = None continue try: v = float(v) except (TypeError, ValueError): return jsonify(error=f"{f} must be a number"), 400 lo, hi = LIMITS.get(f, (-1e9, 1e9)) values[f] = v if lo <= v <= hi else None # sensor error (-999) -> stored as NULL ts = data.get("ts") or now() db.execute("INSERT INTO readings (node_id, ts, " + ", ".join(FIELDS) + ") VALUES (?,?,?,?,?,?,?,?)", (node_id, ts, *[values[f] for f in FIELDS])) db.execute("INSERT INTO nodes (node_id, name, zone, last_seen, firmware) VALUES (?,?,?,?,?) " "ON CONFLICT(node_id) DO UPDATE SET last_seen=excluded.last_seen, " "firmware=COALESCE(excluded.firmware, nodes.firmware)", (node_id, node_id, "field", ts, data.get("fw"))) return jsonify(ok=True), 201 @app.get("/api/v1/commands/pending") @node_auth def pending_commands(): node_id = request.args.get("node_id", "") # Expire stale commands: a pump order older than 10 min must never run late db.execute("UPDATE commands SET status='expired' WHERE status='pending' AND created_at < ?", ((datetime.now() - timedelta(minutes=10)).isoformat(timespec="seconds"),)) cmds = db.rows("SELECT id, actuator, action, duration_min FROM commands WHERE node_id=? AND status='pending' " "ORDER BY id", (node_id,)) for c in cmds: db.execute("UPDATE commands SET status='delivered', delivered_at=? WHERE id=?", (now(), c["id"])) return jsonify(commands=cmds) @app.post("/api/v1/commands//ack") @node_auth def ack_command(cid): db.execute("UPDATE commands SET status='done', acked_at=? WHERE id=?", (now(), cid)) return jsonify(ok=True) # ----------------------------------------------------------------- dashboard API (login) @app.get("/api/v1/latest") @login_required def latest(): out = [] for n in db.rows("SELECT * FROM nodes ORDER BY node_id"): r = db.rows("SELECT * FROM readings WHERE node_id=? ORDER BY ts DESC LIMIT 1", (n["node_id"],)) r = r[0] if r else {} online = bool(n["last_seen"]) and \ datetime.fromisoformat(n["last_seen"]) > datetime.now() - timedelta(minutes=config.NODE_OFFLINE_MIN) out.append({**n, **{f: r.get(f) for f in FIELDS}, "ts": r.get("ts"), "online": online}) return jsonify(nodes=out) @app.get("/api/v1/readings") @login_required def get_readings(): hours = min(int(request.args.get("hours", 24)), 24 * 31) since = (datetime.now() - timedelta(hours=hours)).isoformat(timespec="seconds") sql, params = "SELECT * FROM readings WHERE ts >= ?", [since] if request.args.get("node"): sql += " AND node_id = ?" params.append(request.args["node"]) return jsonify(readings=db.rows(sql + " ORDER BY ts", params)) @app.get("/api/v1/stats/daily") @login_required def daily_stats(): days = min(int(request.args.get("days", 7)), 90) since = (datetime.now() - timedelta(days=days)).isoformat(timespec="seconds") return jsonify(days=db.rows( "SELECT node_id, substr(ts,1,10) day, ROUND(AVG(air_temp_c),1) temp_avg, ROUND(MAX(air_temp_c),1) temp_max, " "ROUND(AVG(air_humidity_pct),1) hum_avg, ROUND(AVG(soil_moisture_pct),1) soil_avg, " "ROUND(MIN(soil_moisture_pct),1) soil_min, COUNT(*) n FROM readings WHERE ts >= ? " "GROUP BY node_id, day ORDER BY day", (since,))) @app.get("/api/v1/predict/") @login_required def predict(node_id): return jsonify(predictor.predict(node_id)) @app.get("/api/v1/decisions") @login_required def decisions(): return jsonify(decisions=db.rows("SELECT * FROM decisions ORDER BY id DESC LIMIT 30"), commands=db.rows("SELECT * FROM commands ORDER BY id DESC LIMIT 15")) @app.post("/api/v1/commands") @login_required def create_command(): check_csrf() data = request.get_json(silent=True) or {} actuator, action = data.get("actuator"), data.get("action") if actuator not in ("pump", "fan") or action not in ("on", "off"): return jsonify(error="invalid command"), 400 if data.get("confirm") is not True: return jsonify(error="confirmation required"), 400 try: minutes = max(0, min(int(data.get("duration_min", 5)), config.PUMP_MAX_MIN)) except (TypeError, ValueError): return jsonify(error="duration_min must be an integer"), 400 node_id = data.get("node_id", "") if not db.rows("SELECT 1 FROM nodes WHERE node_id=?", (node_id,)): return jsonify(error="unknown node"), 404 cid = db.execute("INSERT INTO commands (node_id, actuator, action, duration_min, source, created_by, created_at) " "VALUES (?,?,?,?,?,?,?)", (node_id, actuator, action, minutes, "manual", session["user"], now())) db.log_decision(now(), node_id, "manual", "info", f"{session['user']}: {actuator} {action} {minutes} min") return jsonify(ok=True, id=cid), 201 @app.get("/export.csv") @login_required def export_csv(): days = min(int(request.args.get("days", 7)), 365) since = (datetime.now() - timedelta(days=days)).isoformat(timespec="seconds") data = db.rows("SELECT node_id, ts, " + ", ".join(FIELDS) + " FROM readings WHERE ts >= ? ORDER BY ts", (since,)) buf = io.StringIO() w = csv.DictWriter(buf, fieldnames=["node_id", "ts", *FIELDS]) w.writeheader() w.writerows(data) return Response("" + buf.getvalue(), mimetype="text/csv", headers={"Content-Disposition": f"attachment; filename=kfs_farm_{days}d.csv"}) db.init_db() if __name__ == "__main__": # host=0.0.0.0 -> reachable from the ESP32 nodes on the farm Wi-Fi. # For local HTTPS: pip install pyopenssl and use app.run(..., ssl_context="adhoc") app.run(host="0.0.0.0", port=5000, debug=False)