🖥️ Edge Computing untuk IoT: Panduan Lengkap

import time
import statistics

class EdgeAggregator:
    """Mengagregasi data sensor di edge sebelum kirim ke cloud."""

    def __init__(self, send_interval=60):
        self.buffer = []
        self.last_send = time.time()
        self.send_interval = send_interval  # detik
        self.last_sent_value = None
        self.change_threshold = 0.5  # Kirim jika berubah > 0.5

    def add_reading(self, sensor_type, value):
        self.buffer.append({
            'type': sensor_type,
            'value': value,
            'ts': time.time()
        })

        # Cek apakah waktunya kirim
        if time.time() - self.last_send >= self.send_interval:
            return self.aggregate_and_send()
        return None

    def aggregate_and_send(self):
        if not self.buffer:
            return None

        # Hitung statistik
        values = [r['value'] for r in self.buffer]
        result = {
            'min': min(values),
            'max': max(values),
            'avg': round(statistics.mean(values), 2),
            'stdev': round(statistics.stdev(values), 2) if len(values) > 1 else 0,
            'count': len(values),
            'duration_sec': self.send_interval
        }

        # Event-based: kirim alert jika anomali
        if result['max'] > 40 or result['min'] < 0:
            result['alert'] = True
            result['alert_type'] = 'THRESHOLD_EXCEEDED'

        self.buffer.clear()
        self.last_send = time.time()

        # Kirim 1 record per menit, bukan 60 record per detik!
        # Penghematan: 60x lipat
        return result

# Penggunaan
agg = EdgeAggregator(send_interval=60)

while True:
    suhu = read_sensor()  # Baca setiap detik
    result = agg.add_reading('suhu', suhu)

    if result:
        # Kirim aggregated data ke cloud
        mqtt_client.publish('cloud/suhu/stats', json.dumps(result))
        # Bukan: mqtt_client.publish('sensor/suhu', str(suhu))

import json
import time
import sqlite3
import threading

class EdgeGateway:
    """Edge gateway yang memproses data lokal dan meneruskan ke cloud."""

    def __init__(self):
        self.local_rules = []
        self.cloud_buffer = []
        self.db = sqlite3.connect('edge_data.db')
        self.db.execute('''CREATE TABLE IF NOT EXISTS sensor_data
                          (ts REAL, device TEXT, type TEXT, value REAL)''')
        self.db.execute('''CREATE TABLE IF NOT EXISTS pending_sync
                          (ts REAL, data TEXT, synced INTEGER DEFAULT 0)''')
        self.is_connected = False

    def add_rule(self, rule_fn, action_fn):
        """Tambahkan rule lokal (dieksekusi di edge)."""
        self.local_rules.append({'condition': rule_fn, 'action': action_fn})

    def process_sensor_data(self, device_id, sensor_type, value):
        """Proses data sensor di edge."""
        ts = time.time()

        # 1. Simpan ke database lokal
        self.db.execute(
            'INSERT INTO sensor_data VALUES (?, ?, ?, ?)',
            (ts, device_id, sensor_type, value)
        )

        # 2. Evaluasi rules lokal (real-time, tanpa cloud!)
        for rule in self.local_rules:
            if rule['condition'](device_id, sensor_type, value):
                rule['action'](device_id, sensor_type, value)
                print(f"Rule triggered: {device_id} {sensor_type}={value}")

        # 3. Buffer untuk kirim ke cloud
        record = json.dumps({
            'ts': ts, 'device': device_id,
            'type': sensor_type, 'value': value
        })

        if self.is_connected:
            self._send_to_cloud(record)
        else:
            # Simpan untuk sync nanti (store-and-forward)
            self.db.execute(
                'INSERT INTO pending_sync (ts, data) VALUES (?, ?)',
                (ts, record)
            )

        self.db.commit()

    def _send_to_cloud(self, record):
        """Kirim data ke cloud (MQTT/HTTP)."""
        try:
            mqtt_client.publish('cloud/data', record)
        except:
            # Gagal kirim, simpan untuk nanti
            self.db.execute(
                'INSERT INTO pending_sync (ts, data) VALUES (?, ?)',
                (time.time(), record)
            )

    def sync_pending(self):
        """Sync data yang tertunda saat koneksi pulih."""
        if not self.is_connected:
            return

        cursor = self.db.execute(
            'SELECT rowid, data FROM pending_sync WHERE synced = 0 ORDER BY ts'
        )
        for rowid, data in cursor.fetchall():
            self._send_to_cloud(data)
            self.db.execute(
                'UPDATE pending_sync SET synced = 1 WHERE rowid = ?', (rowid,)
            )
        self.db.commit()
        print("Sync selesai!")

# Setup edge gateway
gateway = EdgeGateway()

# Tambah rules lokal
gateway.add_rule(
    condition=lambda d, t, v: t == 'suhu' and v > 35,
    action=lambda d, t, v: trigger_alarm(d, 'HIGH_TEMP', v)
)
gateway.add_rule(
    condition=lambda d, t, v: t == 'lembab' and v > 80,
    action=lambda d, t, v: trigger_fan(d, True)
)

# Proses data
gateway.process_sensor_data('sensor_001', 'suhu', 36.5)
# → Rule triggered! Alarm tanpa perlu cloud!

#include <TensorFlowLite_ESP32.h>
#include "tensorflow/lite/micro/all_ops_resolver.h"
#include "tensorflow/lite/micro/micro_interpreter.h"
#include "tensorflow/lite/schema/schema_generated.h"

// Model yang sudah di-train dan di-convert ke TFLite Micro
#include "model_data.h"  // Hasil export dari TensorFlow

// Arena memory untuk TFLite
constexpr int kTensorArenaSize = 32 * 1024;  // 32KB
uint8_t tensor_arena[kTensorArenaSize];

const tflite::Model* model;
tflite::MicroInterpreter* interpreter;
TfLiteTensor* input;
TfLiteTensor* output;

void setup() {
    Serial.begin(115200);

    // Load model
    model = tflite::GetModel(model_data);
    static tflite::AllOpsResolver resolver;

    static tflite::MicroInterpreter static_interpreter(
        model, resolver, tensor_arena, kTensorArenaSize
    );
    interpreter = &static_interpreter;

    // Allocate tensors
    interpreter->AllocateTensors();

    // Get input/output tensors
    input = interpreter->input(0);
    output = interpreter->output(0);

    Serial.println("TinyML model loaded!");
    Serial.printf("Input size: %d\n", input->bytes);
    Serial.printf("Arena used: %d bytes\n", interpreter->arena_used_bytes());
}

void loop() {
    // Baca sensor (contoh: 3-axis accelerometer)
    float ax = readAccelX();  // -1.0 to 1.0
    float ay = readAccelY();
    float az = readAccelZ();

    // Normalisasi dan isi input tensor
    input->data.f[0] = ax;
    input->data.f[1] = ay;
    input->data.f[2] = az;

    // Jalankan inference
    unsigned long start = micros();
    interpreter->Invoke();
    unsigned long elapsed = micros() - start;

    // Baca hasil
    // Misal: 0=diam, 1=jalan, 2=lari, 3=jatuh
    int predicted_class = -1;
    float max_score = -1;
    for (int i = 0; i < output->bytes / sizeof(float); i++) {
        if (output->data.f[i] > max_score) {
            max_score = output->data.f[i];
            predicted_class = i;
        }
    }

    const char* labels[] = {"Diam", "Jalan", "Lari", "Jatuh"};
    Serial.printf("Activity: %s (%.2f) - Inference: %lu us\n",
                  labels[predicted_class], max_score, elapsed);

    // Edge action: trigger alarm jika "Jatuh"
    if (predicted_class == 3 && max_score > 0.8) {
        Serial.println("⚠️ FALL DETECTED! Sending alert...");
        // Kirim alert ke cloud (bukan raw sensor data!)
        mqtt_publish("alert/fall", "sensor_001");
    }

    delay(100);  // 10 Hz inference rate
}

import paho.mqtt.client as mqtt
import json, time, sqlite3
from threading import Thread

class IoTGateway:
    def __init__(self):
        # Local broker (ke device lokal)
        self.local = mqtt.Client("gateway-local")
        self.local.connect("localhost", 1883)

        # Cloud broker (ke AWS/Azure)
        self.cloud = mqtt.Client("gateway-cloud")
        self.cloud.tls_set("ca.crt", "client.crt", "client.key")
        self.cloud.connect("aws-iot-endpoint", 8883)

        # Local database
        self.db = sqlite3.connect('gateway.db')
        self.db.execute('''CREATE TABLE IF NOT EXISTS buffer
                          (ts REAL, topic TEXT, data TEXT, sent INT DEFAULT 0)''')

        # Rules engine
        self.rules = [
            {'topic': 'sensor/+/suhu', 'threshold': 35, 'action': 'alarm'},
            {'topic': 'sensor/+/suhu', 'threshold': 0, 'below': True, 'action': 'alarm'}
        ]

    def on_local_message(self, client, userdata, msg):
        """Terima data dari device lokal."""
        data = json.loads(msg.payload.decode())

        # Simpan ke buffer
        self.db.execute('INSERT INTO buffer VALUES (?, ?, ?, 0)',
                       (time.time(), msg.topic, json.dumps(data)))
        self.db.commit()

        # Evaluasi rules lokal
        self.evaluate_rules(msg.topic, data)

        # Agregasi dan kirim ke cloud
        self.forward_to_cloud(msg.topic, data)

    def evaluate_rules(self, topic, data):
        """Evaluasi rules di edge — tanpa cloud dependency!"""
        if 'suhu' in data and data['suhu'] > 35:
            # Trigger alarm LOKAL (buzzer, LED, relay)
            self.local.publish('cmd/alarm', json.dumps({
                'device': data.get('device_id'),
                'type': 'HIGH_TEMP',
                'value': data['suhu']
            }))

    def forward_to_cloud(self, topic, data):
        """Teruskan data ke cloud (dengan filter)."""
        cloud_topic = topic.replace('sensor/', 'devices/')
        self.cloud.publish(cloud_topic, json.dumps(data))

    def start(self):
        self.local.on_message = self.on_local_message
        self.local.subscribe("sensor/#")

        Thread(target=self.local.loop_forever).start()
        Thread(target=self.cloud.loop_forever).start()

        # Periodic sync
        while True:
            self.sync_buffered()
            time.sleep(60)

gateway = IoTGateway()
gateway.start()

# EDGE: ESP32 dengan accelerometer pada motor industri
# - Baca getaran 1000 Hz (vibration analysis)
# - Jalankan FFT (Fast Fourier Transform) di edge
# - Jalankan anomaly detection model (TFLite)
# - Jika anomaly terdeteksi → alert lokal + kirim ke cloud
#
# CLOUD: AWS/GCP
# - Terima anomaly events (bukan raw vibration data!)
# - Dashboard maintenance untuk seluruh pabrik
# - Train ulang model berdasarkan feedback teknisi
# - Push model baru ke edge device (OTA)
#
# PENGHEMATAN BANDWIDTH:
# Raw data: 1000 samples/sec × 4 bytes = 4 KB/s per device
# Dengan 100 devices = 400 KB/s = 34 GB/hari!
# Edge analytics: ~1 event/menit × 200 bytes = 12 KB/jam
# Penghematan: 99.97%!

# EDGE (per lahan):
# - Raspberry Pi Gateway + multiple ESP32 sensors
# - Soil moisture, temperature, humidity, light, pH
# - Edge rules: if soil_moisture < 30% → activate irrigation
# - Edge rules: if temp > 40 → activate shade/sprinkler
# - Data aggregation: kirim rata-rata per jam ke cloud
#
# CLOUD:
# - Analytics musiman (trend analysis)
# - Rekomendasi tanam berdasarkan historical data
# - Integrasi cuaca (weather API)
# - Dashboard petani via mobile app
#
# MANFAAT:
# - Respons irigasi < 1 detik (edge rules)
# - Tetap berfungsi saat internet mati (offline)
# - Hemat bandwidth (summary per jam)
# - Data historis untuk optimasi jangka panjang