Device (Sensor)
    │
    │ All data → cloud  (traditional IoT)
    ▼
 CLOUD
  [Expensive bandwidth, high latency, privacy concerns]

─────────────────────────────────────────────────────

Device (Sensor)
    │
    │ Pre-processed, filtered data only
    ▼
 EDGE GATEWAY (Raspberry Pi / Industrial PC)
    │  [Local processing, alerting, ML inference]
    │
    │ Aggregated summaries only
    ▼
 CLOUD
  [Historical storage, complex analytics, dashboards]

─────────────────────────────────────────────────────

FOG COMPUTING = Edge nodes organized hierarchically
  Sensors → Field Gateway → Regional Server → Cloud

# MicroPython — process sensor data at the device level
# Only send alerts and summaries; don't send every raw reading

from machine import Pin, Timer
import dht, utime, network, urequests

class EdgeProcessor:
    def __init__(self, sensor_pin=4):
        self.sensor = dht.DHT22(Pin(sensor_pin))
        self.readings = []
        self.WINDOW   = 12          # 12 readings × 5s = 1-minute window
        self.TEMP_HIGH_THRESHOLD = 35.0
        self.TEMP_LOW_THRESHOLD  = 5.0
        self.last_alert = 0
        self.ALERT_COOLDOWN = 300   # seconds between alerts

    def read(self):
        try:
            self.sensor.measure()
            return self.sensor.temperature(), self.sensor.humidity()
        except OSError:
            return None, None

    def add_reading(self, temp, hum):
        if temp is None:
            return

        self.readings.append({"t": temp, "h": hum, "ts": utime.time()})
        if len(self.readings) > self.WINDOW:
            self.readings.pop(0)

    def summarize(self):
        """Compute summary statistics for the current window."""
        if not self.readings:
            return None
        temps = [r["t"] for r in self.readings]
        hums  = [r["h"] for r in self.readings]
        return {
            "t_avg": round(sum(temps) / len(temps), 2),
            "t_min": round(min(temps), 2),
            "t_max": round(max(temps), 2),
            "h_avg": round(sum(hums) / len(hums), 2),
            "count": len(self.readings)
        }

    def check_alerts(self, temp):
        """Evaluate alert conditions locally — no cloud needed."""
        now = utime.time()
        if now - self.last_alert < self.ALERT_COOLDOWN:
            return None

        if temp > self.TEMP_HIGH_THRESHOLD:
            self.last_alert = now
            return {"type": "HIGH_TEMP", "value": temp, "threshold": self.TEMP_HIGH_THRESHOLD}
        if temp < self.TEMP_LOW_THRESHOLD:
            self.last_alert = now
            return {"type": "LOW_TEMP", "value": temp, "threshold": self.TEMP_LOW_THRESHOLD}
        return None

    def should_upload(self):
        """Decide whether to spend the battery on a WiFi connection."""
        # Upload every 10 complete windows (10 minutes)
        return len(self.readings) % self.WINDOW == 0

processor = EdgeProcessor()

for i in range(60):
    temp, hum = processor.read()
    processor.add_reading(temp, hum)

    alert = processor.check_alerts(temp) if temp else None
    if alert:
        print(f"LOCAL ALERT: {alert}")  # Could trigger relay, buzzer

    if processor.should_upload():
        summary = processor.summarize()
        print(f"Upload summary: {summary}")
        # Only here do we turn on WiFi and transmit

    utime.sleep(5)

# Python 3 — Edge Gateway (Raspberry Pi)
import paho.mqtt.client as mqtt
import json
import threading
import time
from collections import defaultdict
from datetime import datetime

class EdgeGateway:
    """
    Subscribes to all sensor nodes via local MQTT broker.
    Runs analytics locally. Forwards to cloud only when needed.
    """
    def __init__(self, local_broker="localhost", cloud_broker=None):
        self._data       = defaultdict(list)  # device_id → list of readings
        self._lock       = threading.Lock()
        self._local      = self._setup_local(local_broker)
        self._cloud      = self._setup_cloud(cloud_broker) if cloud_broker else None

    def _setup_local(self, broker):
        c = mqtt.Client("edge-gateway")
        c.on_connect = lambda c, u, f, r: c.subscribe("sensors/#", qos=1)
        c.on_message = self._on_local_message
        c.connect(broker, 1883, 60)
        c.loop_start()
        return c

    def _setup_cloud(self, broker):
        c = mqtt.Client("gateway-cloud-uplink")
        c.connect(broker, 8883)  # TLS in production
        c.loop_start()
        return c

    def _on_local_message(self, client, userdata, msg):
        try:
            data = json.loads(msg.payload)
            device_id = data.get("device_id", "unknown")
            data["gateway_received"] = datetime.utcnow().isoformat()

            with self._lock:
                self._data[device_id].append(data)
                # Keep only the last 60 readings per device
                if len(self._data[device_id]) > 60:
                    self._data[device_id].pop(0)

            # Check alerts immediately at the edge
            self._check_edge_alerts(device_id, data)

        except Exception as e:
            print(f"Gateway parse error: {e}")

    def _check_edge_alerts(self, device_id, data):
        temp = data.get("temperature")
        if temp and temp > 40:
            print(f"CRITICAL TEMP on {device_id}: {temp}°C — local alarm!")
            # Could trigger a local alarm, relay, or display

    def compute_fleet_stats(self):
        """Cross-device analytics — only possible at the gateway."""
        with self._lock:
            stats = {}
            all_temps = []
            for device_id, readings in self._data.items():
                if not readings:
                    continue
                temps = [r["temperature"] for r in readings if "temperature" in r]
                if temps:
                    stats[device_id] = {
                        "latest": round(temps[-1], 2),
                        "avg_1h": round(sum(temps) / len(temps), 2),
                        "max_1h": round(max(temps), 2),
                    }
                    all_temps.extend(temps)

            if all_temps:
                stats["_fleet"] = {
                    "avg":    round(sum(all_temps) / len(all_temps), 2),
                    "hottest": max(all_temps),
                    "coldest": min(all_temps),
                    "devices": len(stats)
                }
        return stats

    def upload_to_cloud(self):
        """Periodically upload summaries to cloud."""
        if not self._cloud:
            return
        summary = self.compute_fleet_stats()
        self._cloud.publish(
            "factory/gateway/summary",
            json.dumps(summary),
            qos=1
        )
        print(f"Cloud summary: {summary.get('_fleet', {})}")

gateway = EdgeGateway(local_broker="localhost")

# Upload fleet summary to cloud every 5 minutes
while True:
    time.sleep(300)
    gateway.upload_to_cloud()

# pip install tflite-runtime (Raspberry Pi)
# For NVIDIA Jetson: use TensorRT instead
import numpy as np
import tflite_runtime.interpreter as tflite
import time

class EdgeMLDetector:
    """
    Run a TFLite anomaly detection model on sensor data.
    Model trained on historical 'normal' readings.
    """
    def __init__(self, model_path="anomaly_detector.tflite"):
        self.interpreter = tflite.Interpreter(model_path=model_path)
        self.interpreter.allocate_tensors()

        self.input_details  = self.interpreter.get_input_details()
        self.output_details = self.interpreter.get_output_details()

        # Input shape: [1, 10, 2] — 10 readings × (temp, humidity)
        self.window_size   = self.input_details[0]['shape'][1]
        self.n_features    = self.input_details[0]['shape'][2]
        self._buffer       = []

        print(f"Model loaded. Window: {self.window_size}, Features: {self.n_features}")

    def add_reading(self, temperature, humidity):
        self._buffer.append([temperature, humidity])
        if len(self._buffer) > self.window_size:
            self._buffer.pop(0)
        return len(self._buffer) == self.window_size

    def predict(self):
        """Returns anomaly score 0.0 (normal) to 1.0 (anomalous)."""
        if len(self._buffer) < self.window_size:
            return None

        # Normalize (must match training normalization)
        data = np.array(self._buffer, dtype=np.float32)
        data[:, 0] = (data[:, 0] - 20) / 15   # temp: mean=20, std=15
        data[:, 1] = (data[:, 1] - 60) / 20   # hum: mean=60, std=20

        input_data = data.reshape(1, self.window_size, self.n_features)
        self.interpreter.set_tensor(self.input_details[0]['index'], input_data)

        start = time.time()
        self.interpreter.invoke()
        inference_ms = (time.time() - start) * 1000

        score = self.interpreter.get_tensor(self.output_details[0]['index'])[0][0]
        print(f"Anomaly score: {score:.3f} (inference: {inference_ms:.1f}ms)")
        return float(score)

detector = EdgeMLDetector("anomaly_detector.tflite")

# Feed readings from your sensor
test_readings = [
    (24.5, 61.2), (24.7, 61.0), (24.6, 60.8),
    (24.8, 61.5), (45.2, 20.0),  # Anomaly!
    (24.5, 61.0), (24.6, 61.3),
]

for temp, hum in test_readings:
    if detector.add_reading(temp, hum):
        score = detector.predict()
        if score and score > 0.8:
            print(f"ANOMALY DETECTED: score={score:.3f} at T={temp}°C")

# Greengrass Lambda function (runs on the edge device, managed by AWS)
import greengrasssdk
import json
import time

client = greengrasssdk.client('iot-data')

TEMP_THRESHOLD = 35.0
last_alert_time = 0

def function_handler(event, context):
    """
    Triggered by local MQTT topic: sensors/+/temperature
    Runs on the device — no internet needed.
    """
    global last_alert_time

    temperature = event.get('temperature', 0)
    device_id   = event.get('device_id', 'unknown')

    # Store locally (edge processing)
    if temperature > TEMP_THRESHOLD:
        now = time.time()
        if now - last_alert_time > 300:  # Max 1 alert per 5 min
            last_alert_time = now
            alert = {
                "device": device_id,
                "temperature": temperature,
                "timestamp": now,
                "processed_at": "edge"  # Not in cloud!
            }
            # Publish to local topic (other Lambda functions can subscribe)
            client.publish(
                topic='alerts/temperature',
                payload=json.dumps(alert)
            )

    # Always republish processed data (Greengrass routes to cloud)
    client.publish(
        topic=f'processed/{device_id}/temperature',
        payload=json.dumps({
            "value": round(temperature, 2),
            "status": "HIGH" if temperature > TEMP_THRESHOLD else "NORMAL",
            "edge_processed": True
        })
    )