الصيانة والتحسين وقابلية التوسع

/*
  Session 20 — Over-The-Air (OTA) firmware updates + health heartbeat (ESP32)
  After the first USB upload, the node appears in Arduino IDE -> Tools -> Port -> "kfs-field-a at 192.168.1.x".
  Upload again over Wi-Fi without removing the node from the field. The OTA password protects it.
  Heartbeat: every 5 min publish uptime, free heap, RSSI, reset reason, firmware version (retained MQTT).
*/
#include <WiFi.h>
#include <ArduinoOTA.h>
#include <PubSubClient.h>
#include <esp_system.h>

#define FW_VERSION "1.4.0"
const char* WIFI_SSID = "KFS-Farm-IoT";
const char* WIFI_PASS = "change-me";
const char* HOSTNAME  = "kfs-field-a";
const char* OTA_PASS  = "farm-ota-2026";
const char* BROKER    = "192.168.1.10";

WiFiClient net;
PubSubClient mqtt(net);
unsigned long lastBeat = 0;

void setup() {
  Serial.begin(115200);
  WiFi.setHostname(HOSTNAME);
  WiFi.begin(WIFI_SSID, WIFI_PASS);
  while (WiFi.status() != WL_CONNECTED) delay(300);

  ArduinoOTA.setHostname(HOSTNAME);
  ArduinoOTA.setPassword(OTA_PASS);
  ArduinoOTA.onStart([]() { digitalWrite(26, HIGH); Serial.println("OTA start - pump relay forced OFF"); });
  ArduinoOTA.onProgress([](unsigned int p, unsigned int t) { Serial.printf("OTA %u%%\r", p * 100 / t); });
  ArduinoOTA.onEnd([]() { Serial.println("\nOTA done, rebooting"); });
  ArduinoOTA.onError([](ota_error_t e) { Serial.printf("OTA error %u\n", e); });
  ArduinoOTA.begin();

  mqtt.setServer(BROKER, 1883);
  Serial.printf("fw %s ready at %s\n", FW_VERSION, WiFi.localIP().toString().c_str());
}

void loop() {
  ArduinoOTA.handle();                       // must be called often
  if (!mqtt.connected()) mqtt.connect(HOSTNAME);
  mqtt.loop();
  if (millis() - lastBeat > 300000 || lastBeat == 0) {
    lastBeat = millis();
    char buf[200];
    snprintf(buf, sizeof(buf),
             "{\"fw\":\"%s\",\"uptime_s\":%lu,\"heap\":%u,\"rssi\":%d,\"reset_reason\":%d}",
             FW_VERSION, millis() / 1000, ESP.getFreeHeap(), WiFi.RSSI(), (int)esp_reset_reason());
    mqtt.publish("kfs/farm/field_A/health", buf, true);   // retained: the dashboard always sees the last state
  }
}

"""Backups + retention / downsampling for farm.db (session 20).

python backup.py                    # hot backup to backups/farm-YYYYmmdd-HHMM.db, keep the last 14
python backup.py --downsample 90    # move raw readings older than 90 days into hourly averages
python backup.py --verify FILE      # open a backup and print its row counts (restore test)
Schedule it daily (cron / Task Scheduler). Copy the backups folder to a second disk or USB weekly.
"""
import argparse
import sqlite3
from datetime import datetime, timedelta
from pathlib import Path

import config

BACKUP_DIR = config.BASE_DIR / "backups"
KEEP = 14


def hot_backup():
    """SQLite online backup API: consistent copy even while app.py / mqtt_ingest.py are writing."""
    BACKUP_DIR.mkdir(exist_ok=True)
    dest = BACKUP_DIR / f"farm-{datetime.now():%Y%m%d-%H%M%S}.db"
    src = sqlite3.connect(config.DB_PATH)
    dst = sqlite3.connect(dest)
    with dst:
        src.backup(dst)
    src.close(); dst.close()
    old = sorted(BACKUP_DIR.glob("farm-*.db"))[:-KEEP]
    for f in old:
        f.unlink()
    print(f"backup -> {dest.name} ({dest.stat().st_size / 1024:.0f} KB), removed {len(old)} old file(s)")
    return dest


def downsample(days):
    """Keep full resolution for recent data, hourly averages for the archive (smaller DB, same trends)."""
    cutoff = (datetime.now() - timedelta(days=days)).isoformat(timespec="seconds")
    con = sqlite3.connect(config.DB_PATH)
    con.executescript("""CREATE TABLE IF NOT EXISTS readings_hourly (
        node_id TEXT, hour TEXT, air_temp_c REAL, air_humidity_pct REAL, soil_moisture_pct REAL,
        light REAL, battery_v REAL, n INTEGER, PRIMARY KEY (node_id, hour));""")
    with con:
        con.execute("""INSERT OR REPLACE INTO readings_hourly
            SELECT node_id, substr(ts, 1, 13) || ':00:00', AVG(air_temp_c), AVG(air_humidity_pct),
                   AVG(soil_moisture_pct), AVG(light), AVG(battery_v), COUNT(*)
            FROM readings WHERE ts < ? GROUP BY node_id, substr(ts, 1, 13)""", (cutoff,))
        moved = con.execute("DELETE FROM readings WHERE ts < ?", (cutoff,)).rowcount
    con.execute("VACUUM")
    con.close()
    print(f"downsampled readings older than {days} days: {moved} raw rows -> hourly archive")


def verify(path):
    con = sqlite3.connect(path)
    for t in ("nodes", "readings", "commands", "decisions"):
        n = con.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0]
        print(f"{t:10s} {n:>8d} rows")
    print("integrity:", con.execute("PRAGMA integrity_check").fetchone()[0])
    con.close()


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--downsample", type=int)
    ap.add_argument("--verify")
    a = ap.parse_args()
    if a.verify:
        verify(a.verify)
    elif a.downsample:
        hot_backup()                 # always back up before deleting anything
        downsample(a.downsample)
    else:
        hot_backup()

import json, random, threading, time, urllib.request
URL, KEY = "http://127.0.0.1:5000/api/v1/readings", "kfs-farm-node-key"
lat = []

def node(i, n_msgs=30):
    for _ in range(n_msgs):
        body = json.dumps({"node_id": f"load_{i:02d}", "air_temp_c": 25 + random.random() * 5,
                           "soil_moisture_pct": 30 + random.random() * 20}).encode()
        req = urllib.request.Request(URL, body, {"Content-Type": "application/json", "X-API-Key": KEY})
        t0 = time.perf_counter(); urllib.request.urlopen(req, timeout=10); lat.append(time.perf_counter() - t0)
        time.sleep(10)

threads = [threading.Thread(target=node, args=(i,)) for i in range(50)]
[t.start() for t in threads]; [t.join() for t in threads]
lat.sort()
print(f"{len(lat)} requests | median {lat[len(lat)//2]*1000:.0f} ms | p95 {lat[int(len(lat)*.95)]*1000:.0f} ms")

شغّله على منصة تجريبية (قاعدة بيانات منفصلة عبر SF_DB_PATH) — ليس على بيانات المزرعة الحقيقية. 50 عقدة × رسالة كل 10 ث = 5 رسائل/ثانية: حمل بسيط لـ Flask + SQLite (WAL). عند المئات ننتقل لـ gunicorn/waitress و PostgreSQL أو InfluxDB.