Backend بسيط + قاعدة بيانات زمنية

readings,node_id=field_A,zone=field air_temp_c=29.4,soil_moisture_pct=31 1718180400000000000
#  measurement,tags                  fields                               timestamp (ns)

"""Session 10 - the smallest useful farm REST API (Flask + SQLite) - ~50 lines.

python mini_api.py      ->  http://127.0.0.1:5001
POST /api/readings      {"node_id":"greenhouse_01","air_temp_c":28.4,"soil_moisture_pct":41}
GET  /api/readings?node=greenhouse_01&hours=24
GET  /api/latest
Also serves index.html (session 9) so the page can call the API from the same origin (no CORS issues).
"""
import sqlite3
from datetime import datetime, timedelta
from pathlib import Path

from flask import Flask, jsonify, request, send_from_directory

HERE = Path(__file__).resolve().parent
DB = HERE / "mini_farm.db"
app = Flask(__name__)


def db():
    con = sqlite3.connect(DB)
    con.row_factory = sqlite3.Row
    return con


with db() as con:
    con.execute("CREATE TABLE IF NOT EXISTS readings (id INTEGER PRIMARY KEY, node_id TEXT, ts TEXT, "
                "air_temp_c REAL, air_humidity_pct REAL, soil_moisture_pct REAL)")
    con.execute("CREATE INDEX IF NOT EXISTS ix ON readings (node_id, ts)")


@app.post("/api/readings")
def add_reading():
    d = request.get_json(silent=True)
    if not d or "node_id" not in d:
        return jsonify(error="JSON body with node_id required"), 400          # 400 = client error
    ts = d.get("ts") or datetime.now().isoformat(timespec="seconds")
    with db() as con:
        con.execute("INSERT INTO readings (node_id, ts, air_temp_c, air_humidity_pct, soil_moisture_pct) "
                    "VALUES (?,?,?,?,?)", (d["node_id"], ts, d.get("air_temp_c"), d.get("air_humidity_pct"),
                                          d.get("soil_moisture_pct")))
    return jsonify(ok=True, ts=ts), 201                                          # 201 = created


@app.get("/api/readings")
def list_readings():
    hours = int(request.args.get("hours", 24))
    since = (datetime.now() - timedelta(hours=hours)).isoformat(timespec="seconds")
    sql, args = "SELECT * FROM readings WHERE ts >= ?", [since]
    if request.args.get("node"):
        sql += " AND node_id = ?"
        args.append(request.args["node"])
    with db() as con:
        rows = [dict(r) for r in con.execute(sql + " ORDER BY ts", args)]
    return jsonify(count=len(rows), readings=rows)


@app.get("/api/latest")
def latest():
    with db() as con:
        rows = [dict(r) for r in con.execute(
            "SELECT * FROM readings r WHERE ts = (SELECT MAX(ts) FROM readings WHERE node_id = r.node_id)")]
    return jsonify(nodes=rows)


@app.get("/")
def page():
    return send_from_directory(HERE, "index.html")


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5001, debug=True)

"""KFS Smart Farm — Flask backend + dashboard (sessions 10-20).

Run:   pip install -r requirements.txt
       python app.py                      -> http://<laptop-ip>: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/<int:cid>/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/<node_id>")
@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)

async function refresh() {
  const res = await fetch('/api/readings?node=greenhouse_01&hours=6');
  const { readings } = await res.json();
  if (!readings.length) return;
  const last = readings[readings.length - 1];
  document.getElementById('temp').textContent = last.air_temp_c?.toFixed(1) ?? '—';
  document.getElementById('soil').textContent = last.soil_moisture_pct?.toFixed(0) ?? '—';
  chart.data.labels = readings.map(r => r.ts.slice(11, 16));
  chart.data.datasets[0].data = readings.map(r => r.soil_moisture_pct);
  chart.update();
}

?. و ?? يحميان الصفحة من القيم المفقودة (null) — تذكر أن قاعدة البيانات تخزن NULL عند فشل الحساس.

/*
  KFS Smart Farm — field node (ESP32) — HTTP version  (sessions 9-12, 17-21)
  Sensors : DHT22 (GPIO 4), capacitive soil moisture v1.2 (GPIO 34), LDR divider (GPIO 35), battery divider (GPIO 33)
  Actuator: relay module IN1 (GPIO 26) -> irrigation pump, IN2 (GPIO 27) -> greenhouse fan  (active LOW modules)
  Every 30 s: POST /api/v1/readings  then  GET /api/v1/commands/pending  -> drive relays -> ACK
  Libraries: "DHT sensor library" (Adafruit), "ArduinoJson" v7
*/
#include <WiFi.h>
#include <HTTPClient.h>
#include <ArduinoJson.h>
#include <DHT.h>

// ---------- edit for your farm ----------
const char* WIFI_SSID = "KFS-Farm-IoT";
const char* WIFI_PASS = "change-me";
const char* SERVER    = "http://192.168.1.10:5000";   // laptop / Raspberry Pi running app.py
const char* API_KEY   = "kfs-farm-node-key";
const char* NODE_ID   = "field_A";
const int   SOIL_DRY  = 3200;   // ADC in dry air      (calibrate! session 2)
const int   SOIL_WET  = 1300;   // ADC in a cup of water
// -----------------------------------------

#define DHT_PIN 4
#define SOIL_PIN 34
#define LDR_PIN 35
#define BAT_PIN 33
#define PUMP_PIN 26
#define FAN_PIN 27
#define PUMP_MAX_MS (30UL * 60UL * 1000UL)   // hardware safety: never run the pump > 30 min

DHT dht(DHT_PIN, DHT22);
unsigned long lastSend = 0, pumpOffAt = 0, fanOffAt = 0;

int readSoilAdc() {                        // median of 9 samples -> rejects spikes (session 3)
  int v[9];
  for (int i = 0; i < 9; i++) { v[i] = analogRead(SOIL_PIN); delay(5); }
  for (int i = 1; i < 9; i++) { int k = v[i], j = i - 1; while (j >= 0 && v[j] > k) { v[j + 1] = v[j]; j--; } v[j + 1] = k; }
  return v[4];
}

float soilPercent(int adc) {
  float p = (float)(SOIL_DRY - adc) / (SOIL_DRY - SOIL_WET) * 100.0;
  return constrain(p, 0, 100);
}

void relay(int pin, bool on) { digitalWrite(pin, on ? LOW : HIGH); }   // active-LOW relay board

void connectWiFi() {
  if (WiFi.status() == WL_CONNECTED) return;
  WiFi.mode(WIFI_STA);
  WiFi.begin(WIFI_SSID, WIFI_PASS);
  for (int i = 0; i < 40 && WiFi.status() != WL_CONNECTED; i++) delay(250);
  Serial.printf("WiFi %s  IP=%s  RSSI=%d\n", WiFi.status() == WL_CONNECTED ? "OK" : "FAIL",
                WiFi.localIP().toString().c_str(), WiFi.RSSI());
}

void sendReading() {
  float t = dht.readTemperature(), h = dht.readHumidity();
  int soil = readSoilAdc();
  JsonDocument doc;
  doc["node_id"] = NODE_ID;
  if (!isnan(t)) doc["air_temp_c"] = t;            // missing value -> field omitted (server stores NULL)
  if (!isnan(h)) doc["air_humidity_pct"] = h;
  doc["soil_moisture_pct"] = soilPercent(soil);
  doc["light"] = analogRead(LDR_PIN);
  doc["battery_v"] = analogReadMilliVolts(BAT_PIN) * 2 / 1000.0;   // 100k/100k divider
  doc["rssi"] = WiFi.RSSI();
  doc["fw"] = "1.3.0";
  String body; serializeJson(doc, body);

  HTTPClient http;
  http.begin(String(SERVER) + "/api/v1/readings");
  http.addHeader("Content-Type", "application/json");
  http.addHeader("X-API-Key", API_KEY);
  int code = http.POST(body);
  Serial.printf("POST %d  %s\n", code, body.c_str());
  http.end();
}

void pollCommands() {
  HTTPClient http;
  http.begin(String(SERVER) + "/api/v1/commands/pending?node_id=" + NODE_ID);
  http.addHeader("X-API-Key", API_KEY);
  if (http.GET() == 200) {
    JsonDocument doc;
    deserializeJson(doc, http.getString());
    for (JsonObject c : doc["commands"].as<JsonArray>()) {
      bool on = String((const char*)c["action"]) == "on";
      unsigned long ms = min((unsigned long)c["duration_min"].as<int>() * 60000UL, PUMP_MAX_MS);
      if (String((const char*)c["actuator"]) == "pump") { relay(PUMP_PIN, on); pumpOffAt = on ? millis() + ms : 0; }
      else                                             { relay(FAN_PIN, on);  fanOffAt  = on ? millis() + ms : 0; }
      Serial.printf("CMD #%d %s %s\n", (int)c["id"], (const char*)c["actuator"], (const char*)c["action"]);
      HTTPClient ack;
      ack.begin(String(SERVER) + "/api/v1/commands/" + String((int)c["id"]) + "/ack");
      ack.addHeader("X-API-Key", API_KEY);
      ack.POST("{}");
      ack.end();
    }
  }
  http.end();
}

void setup() {
  Serial.begin(115200);
  pinMode(PUMP_PIN, OUTPUT); pinMode(FAN_PIN, OUTPUT);
  relay(PUMP_PIN, false); relay(FAN_PIN, false);       // fail-safe: everything OFF at boot
  analogReadResolution(12);
  dht.begin();
  connectWiFi();
}

void loop() {
  if (pumpOffAt && millis() > pumpOffAt) { relay(PUMP_PIN, false); pumpOffAt = 0; Serial.println("pump auto-off"); }
  if (fanOffAt && millis() > fanOffAt)   { relay(FAN_PIN, false);  fanOffAt = 0; }
  if (millis() - lastSend > 30000 || lastSend == 0) {
    lastSend = millis();
    connectWiFi();
    if (WiFi.status() == WL_CONNECTED) { sendReading(); pollCommands(); }
    else { relay(PUMP_PIN, false); }                    // lost the server -> stop irrigating
  }
}