استقبال البيانات ومعالجتها على السيرفر

"""SQLite storage (time-series friendly schema + indexes). Session 10."""
import sqlite3
from contextlib import contextmanager

from werkzeug.security import generate_password_hash

import config

SCHEMA = """
CREATE TABLE IF NOT EXISTS nodes (
    node_id   TEXT PRIMARY KEY,
    name      TEXT,
    zone      TEXT,                -- greenhouse / field
    lat       REAL,
    lon       REAL,
    last_seen TEXT,
    firmware  TEXT
);
CREATE TABLE IF NOT EXISTS readings (
    id                INTEGER PRIMARY KEY AUTOINCREMENT,
    node_id           TEXT NOT NULL,
    ts                TEXT NOT NULL,     -- ISO-8601 local time
    air_temp_c        REAL,
    air_humidity_pct  REAL,
    soil_moisture_pct REAL,
    light             INTEGER,
    battery_v         REAL,
    rssi              INTEGER
);
CREATE INDEX IF NOT EXISTS ix_readings_node_ts ON readings (node_id, ts);
CREATE INDEX IF NOT EXISTS ix_readings_ts ON readings (ts);
CREATE TABLE IF NOT EXISTS commands (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    node_id      TEXT NOT NULL,
    actuator     TEXT NOT NULL,          -- pump | fan
    action       TEXT NOT NULL,          -- on | off
    duration_min INTEGER DEFAULT 0,
    status       TEXT DEFAULT 'pending', -- pending -> delivered -> done | expired
    source       TEXT DEFAULT 'manual',  -- manual | rule | ml
    created_by   TEXT,
    created_at   TEXT,
    delivered_at TEXT,
    acked_at     TEXT
);
CREATE TABLE IF NOT EXISTS decisions (
    id       INTEGER PRIMARY KEY AUTOINCREMENT,
    ts       TEXT,
    node_id  TEXT,
    source   TEXT,      -- rule | ml | manual | system
    level    TEXT,      -- info | warning | critical
    message  TEXT
);
CREATE TABLE IF NOT EXISTS users (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    username      TEXT UNIQUE,
    password_hash TEXT
);
"""

DEFAULT_NODES = [
    ("greenhouse_01", "الصوبة التجريبية", "greenhouse", 31.1052, 30.9431),
    ("field_A", "الحقل A — خضر", "field", 31.1047, 30.9442),
    ("field_B", "الحقل B — ذرة", "field", 31.1041, 30.9436),
]


@contextmanager
def connect():
    con = sqlite3.connect(config.DB_PATH, timeout=10)
    con.row_factory = sqlite3.Row
    con.execute("PRAGMA journal_mode=WAL")      # readers don't block the MQTT writer
    try:
        yield con
        con.commit()
    finally:
        con.close()


def init_db():
    with connect() as con:
        con.executescript(SCHEMA)
        for node in DEFAULT_NODES:
            con.execute("INSERT OR IGNORE INTO nodes (node_id, name, zone, lat, lon) VALUES (?,?,?,?,?)", node)
        if not con.execute("SELECT 1 FROM users WHERE username=?", (config.ADMIN_USER,)).fetchone():
            con.execute("INSERT INTO users (username, password_hash) VALUES (?,?)",
                        (config.ADMIN_USER, generate_password_hash(config.ADMIN_PASSWORD)))


def rows(sql, params=()):
    with connect() as con:
        return [dict(r) for r in con.execute(sql, params).fetchall()]


def execute(sql, params=()):
    with connect() as con:
        cur = con.execute(sql, params)
        return cur.lastrowid


def log_decision(ts, node_id, source, level, message):
    execute("INSERT INTO decisions (ts, node_id, source, level, message) VALUES (?,?,?,?,?)",
            (ts, node_id, source, level, message))

لاحظ: الوقت مخزن كنص ISO-8601 — الترتيب الأبجدي له هو نفسه الترتيب الزمني، فتعمل WHERE ts >= ? و ORDER BY ts مباشرة. والفهرس المركب (node_id, ts) يجعل استعلام «آخر 24 ساعة لعقدة» فورياً حتى مع ملايين الصفوف.

import math, random, sqlite3
from datetime import datetime, timedelta

con = sqlite3.connect("farm_lab.db")
con.execute("""CREATE TABLE IF NOT EXISTS readings (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    node_id TEXT NOT NULL, ts TEXT NOT NULL,
    air_temp_c REAL, soil_moisture_pct REAL)""")
con.execute("CREATE INDEX IF NOT EXISTS ix_node_ts ON readings (node_id, ts)")

start = datetime(2024, 6, 1)
rows = []
for node, base in [("greenhouse_01", 45), ("field_A", 30), ("field_B", 22)]:
    soil = base
    for i in range(96 * 3):                         # 3 days, every 15 min
        ts = start + timedelta(minutes=15 * i)
        temp = 26 + 7 * math.sin((ts.hour - 9) / 24 * 6.283) + random.gauss(0, .4)
        soil = soil - 0.12 if soil > 15 else base     # slow drying, then "irrigation"
        rows.append((node, ts.isoformat(timespec="seconds"), round(temp, 1), round(soil, 1)))
con.executemany("INSERT INTO readings (node_id, ts, air_temp_c, soil_moisture_pct) VALUES (?,?,?,?)", rows)
con.commit()

# latest reading per node
for r in con.execute("""SELECT node_id, MAX(ts), soil_moisture_pct FROM readings GROUP BY node_id"""):
    print(r)
# daily average per node (substr(ts,1,10) = the date)
for r in con.execute("""SELECT node_id, substr(ts,1,10) AS day, ROUND(AVG(air_temp_c),1), ROUND(MIN(soil_moisture_pct),1)
                        FROM readings GROUP BY node_id, day ORDER BY day"""):
    print(r)

"""MQTT -> SQLite bridge (sessions 7-8).

Nodes publish JSON to   kfs/farm/<node_id>/telemetry   on the local Mosquitto broker.
Run:  python mqtt_ingest.py --broker 192.168.1.10
Needs: pip install paho-mqtt>=2.0
"""
import argparse
import json
import logging
from datetime import datetime

import paho.mqtt.client as mqtt

import db

logging.basicConfig(filename="ingest.log", level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
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)}
stats = {"ok": 0, "bad": 0}


def clean(payload):
    out = {}
    for f in FIELDS:
        v = payload.get(f)
        try:
            v = float(v) if v is not None else None
        except (TypeError, ValueError):
            v = None
        lo, hi = LIMITS.get(f, (-1e9, 1e9))
        out[f] = v if v is not None and lo <= v <= hi else None   # missing / -999 -> NULL
    return out


def on_connect(client, userdata, flags, reason_code, properties=None):
    print("connected:", reason_code)
    client.subscribe("kfs/farm/+/telemetry", qos=1)
    client.subscribe("kfs/farm/+/status", qos=1)


def on_message(client, userdata, msg):
    node_id = msg.topic.split("/")[2]
    try:
        payload = json.loads(msg.payload.decode())
    except ValueError:
        stats["bad"] += 1
        logging.warning("invalid JSON from %s: %r", node_id, msg.payload[:80])
        return
    if msg.topic.endswith("/status"):           # LWT: "offline" published by the broker if the node dies
        logging.info("status %s -> %s", node_id, payload)
        return
    ts = payload.get("ts") or datetime.now().isoformat(timespec="seconds")
    v = clean(payload)
    db.execute("INSERT INTO readings (node_id, ts, " + ", ".join(FIELDS) + ") VALUES (?,?,?,?,?,?,?,?)",
               (node_id, ts, *[v[f] for f in FIELDS]))
    db.execute("INSERT INTO nodes (node_id, name, zone, last_seen) VALUES (?,?,?,?) "
               "ON CONFLICT(node_id) DO UPDATE SET last_seen=excluded.last_seen", (node_id, node_id, "field", ts))
    stats["ok"] += 1
    if stats["ok"] % 20 == 0:
        print(f"stored {stats['ok']} readings ({stats['bad']} rejected)")


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--broker", default="localhost")
    ap.add_argument("--port", type=int, default=1883)
    ap.add_argument("--user"); ap.add_argument("--password")
    a = ap.parse_args()
    db.init_db()
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="kfs-ingest")
    if a.user:
        client.username_pw_set(a.user, a.password)
    client.on_connect, client.on_message = on_connect, on_message
    client.connect(a.broker, a.port, keepalive=60)
    client.loop_forever()

"""Live network status board in the terminal (session 8).

python network_status.py            # refresh every 10 s
Shows, per node: last seen, readings in the last hour, expected vs received (delivery %), last values,
plus how many values were rejected as sensor errors (stored as NULL).
"""
import os
import sqlite3
import sys
import time
from datetime import datetime, timedelta

import config

EXPECTED_PER_HOUR = int(os.environ.get("SF_EXPECTED_PER_HOUR", "60"))   # node sends every 60 s


def board():
    con = sqlite3.connect(config.DB_PATH)
    con.row_factory = sqlite3.Row
    hour_ago = (datetime.now() - timedelta(hours=1)).isoformat(timespec="seconds")
    rows = con.execute("""
        SELECT n.node_id, n.last_seen,
               (SELECT COUNT(*) FROM readings r WHERE r.node_id = n.node_id AND r.ts >= ?) AS last_hour,
               (SELECT COUNT(*) FROM readings r WHERE r.node_id = n.node_id AND r.ts >= ?
                       AND (r.air_temp_c IS NULL OR r.soil_moisture_pct IS NULL)) AS with_nulls,
               (SELECT soil_moisture_pct FROM readings r WHERE r.node_id = n.node_id ORDER BY ts DESC LIMIT 1) AS soil,
               (SELECT air_temp_c FROM readings r WHERE r.node_id = n.node_id ORDER BY ts DESC LIMIT 1) AS temp
        FROM nodes n ORDER BY n.node_id""", (hour_ago, hour_ago)).fetchall()
    con.close()
    print(f"\nKFS Smart Farm - network status  {datetime.now():%Y-%m-%d %H:%M:%S}")
    print(f"{'node':15s} {'state':8s} {'last seen':>10s} {'1h msgs':>8s} {'delivery':>9s} {'nulls':>6s} {'soil%':>6s} {'temp':>6s}")
    for r in rows:
        if r["last_seen"]:
            age = (datetime.now() - datetime.fromisoformat(r["last_seen"])).total_seconds()
            seen = f"{int(age // 60)}m{int(age % 60):02d}s"
            state = "ONLINE" if age < config.NODE_OFFLINE_MIN * 60 else "OFFLINE"
        else:
            seen, state = "never", "UNKNOWN"
        pdr = min(100, r["last_hour"] / EXPECTED_PER_HOUR * 100)
        print(f"{r['node_id']:15s} {state:8s} {seen:>10s} {r['last_hour']:>8d} {pdr:>8.0f}% {r['with_nulls']:>6d} "
              f"{(r['soil'] if r['soil'] is not None else float('nan')):>6.1f} {(r['temp'] if r['temp'] is not None else float('nan')):>6.1f}")


if __name__ == "__main__":
    while True:
        board()
        if "--once" in sys.argv:
            break
        time.sleep(10)