Edge Iot

Edge Computing & IoT

Overview

Building applications for edge devices, IoT protocols, and embedded systems integration.

---

MQTT Protocol

Broker Setup (Mosquitto)

# docker-compose.yml
services:
  mosquitto:
    image: eclipse-mosquitto:2
    ports:
      - "1883:1883"
      - "9001:9001"
    volumes:
      - ./mosquitto.conf:/mosquitto/config/mosquitto.conf
      - mosquitto_data:/mosquitto/data
      - mosquitto_log:/mosquitto/log

volumes:
  mosquitto_data:
  mosquitto_log:

# mosquitto.conf
listener 1883
listener 9001
protocol websockets

allow_anonymous false
password_file /mosquitto/config/passwd

persistence true
persistence_location /mosquitto/data/
log_dest file /mosquitto/log/mosquitto.log

Node.js MQTT Client

import mqtt from 'mqtt';

class MQTTClient {
  private client: mqtt.MqttClient;
  private subscriptions = new Map<string, Set<Function>>();

  constructor(brokerUrl: string, options?: mqtt.IClientOptions) {
    this.client = mqtt.connect(brokerUrl, {
      clientId: `node_${Math.random().toString(16).slice(2, 10)}`,
      clean: true,
      reconnectPeriod: 5000,
      ...options,
    });

    this.client.on('connect', () => {
      console.log('MQTT connected');
      // Resubscribe to all topics
      this.subscriptions.forEach((_, topic) => {
        this.client.subscribe(topic);
      });
    });

    this.client.on('message', (topic, payload) => {
      const handlers = this.getMatchingHandlers(topic);
      const message = this.parsePayload(payload);

      handlers.forEach(handler => handler(topic, message));
    });

    this.client.on('error', (error) => {
      console.error('MQTT error:', error);
    });
  }

  subscribe(topic: string, handler: (topic: string, message: any) => void) {
    if (!this.subscriptions.has(topic)) {
      this.subscriptions.set(topic, new Set());
      this.client.subscribe(topic);
    }

    this.subscriptions.get(topic)!.add(handler);

    return () => {
      this.subscriptions.get(topic)?.delete(handler);
      if (this.subscriptions.get(topic)?.size === 0) {
        this.subscriptions.delete(topic);
        this.client.unsubscribe(topic);
      }
    };
  }

  publish(topic: string, message: any, options?: mqtt.IClientPublishOptions) {
    const payload = typeof message === 'string'
      ? message
      : JSON.stringify(message);

    this.client.publish(topic, payload, {
      qos: 1,
      ...options,
    });
  }

  private getMatchingHandlers(topic: string): Set<Function> {
    const handlers = new Set<Function>();

    this.subscriptions.forEach((topicHandlers, pattern) => {
      if (this.topicMatches(pattern, topic)) {
        topicHandlers.forEach(h => handlers.add(h));
      }
    });

    return handlers;
  }

  private topicMatches(pattern: string, topic: string): boolean {
    const patternParts = pattern.split('/');
    const topicParts = topic.split('/');

    for (let i = 0; i < patternParts.length; i++) {
      if (patternParts[i] === '#') return true;
      if (patternParts[i] === '+') continue;
      if (patternParts[i] !== topicParts[i]) return false;
    }

    return patternParts.length === topicParts.length;
  }

  private parsePayload(payload: Buffer): any {
    const str = payload.toString();
    try {
      return JSON.parse(str);
    } catch {
      return str;
    }
  }

  disconnect() {
    this.client.end();
  }
}

// Usage
const mqtt = new MQTTClient('mqtt://localhost:1883', {
  username: 'user',
  password: 'pass',
});

// Subscribe to device telemetry
mqtt.subscribe('devices/+/telemetry', (topic, data) => {
  const deviceId = topic.split('/')[1];
  console.log(`Device ${deviceId}:`, data);
});

// Subscribe to all events from a device
mqtt.subscribe('devices/sensor-001/#', (topic, data) => {
  console.log(`${topic}:`, data);
});

// Publish command to device
mqtt.publish('devices/sensor-001/commands', {
  action: 'reboot',
  timestamp: Date.now(),
});

Device Simulator

class DeviceSimulator {
  private mqtt: MQTTClient;
  private deviceId: string;
  private interval: NodeJS.Timeout | null = null;

  constructor(deviceId: string, brokerUrl: string) {
    this.deviceId = deviceId;
    this.mqtt = new MQTTClient(brokerUrl);

    // Subscribe to commands
    this.mqtt.subscribe(`devices/${deviceId}/commands`, (_, command) => {
      this.handleCommand(command);
    });
  }

  start(intervalMs = 5000) {
    this.interval = setInterval(() => {
      this.sendTelemetry();
    }, intervalMs);
  }

  stop() {
    if (this.interval) {
      clearInterval(this.interval);
      this.interval = null;
    }
  }

  private sendTelemetry() {
    const telemetry = {
      temperature: 20 + Math.random() * 10,
      humidity: 40 + Math.random() * 20,
      pressure: 1013 + Math.random() * 10,
      battery: 85 + Math.random() * 15,
      timestamp: Date.now(),
    };

    this.mqtt.publish(`devices/${this.deviceId}/telemetry`, telemetry);
  }

  private handleCommand(command: any) {
    console.log(`Received command:`, command);

    switch (command.action) {
      case 'reboot':
        this.mqtt.publish(`devices/${this.deviceId}/status`, {
          status: 'rebooting',
          timestamp: Date.now(),
        });
        break;

      case 'update_config':
        // Update local config
        break;
    }
  }
}

---

Edge Functions

Cloudflare Workers for IoT

// Edge function to process IoT data
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);

    // Device telemetry ingestion
    if (url.pathname === '/ingest' && request.method === 'POST') {
      const data = await request.json();
      const deviceId = request.headers.get('X-Device-ID');

      // Validate device
      const device = await env.KV.get(`device:${deviceId}`);
      if (!device) {
        return new Response('Unauthorized', { status: 401 });
      }

      // Process at edge
      const processed = processData(data);

      // Store in Durable Object for aggregation
      const aggregator = env.AGGREGATOR.get(
        env.AGGREGATOR.idFromName(deviceId)
      );
      await aggregator.fetch(request.url, {
        method: 'POST',
        body: JSON.stringify(processed),
      });

      // Forward to origin if needed
      if (processed.alert) {
        await fetch('https://api.example.com/alerts', {
          method: 'POST',
          body: JSON.stringify({
            deviceId,
            alert: processed.alert,
          }),
        });
      }

      return new Response('OK');
    }

    return new Response('Not Found', { status: 404 });
  },
};

function processData(data: any) {
  // Edge processing logic
  const alert = data.temperature > 30 ? 'HIGH_TEMP' : null;

  return {
    ...data,
    processedAt: Date.now(),
    alert,
  };
}

// Durable Object for stateful aggregation
export class DeviceAggregator {
  private state: DurableObjectState;
  private readings: any[] = [];

  constructor(state: DurableObjectState) {
    this.state = state;
  }

  async fetch(request: Request): Promise<Response> {
    const data = await request.json();

    this.readings.push(data);

    // Keep last 100 readings
    if (this.readings.length > 100) {
      this.readings.shift();
    }

    // Compute aggregates
    const aggregates = {
      avgTemperature: this.average('temperature'),
      avgHumidity: this.average('humidity'),
      count: this.readings.length,
    };

    await this.state.storage.put('aggregates', aggregates);

    return new Response(JSON.stringify(aggregates));
  }

  private average(field: string): number {
    const values = this.readings
      .map(r => r[field])
      .filter(v => typeof v === 'number');

    return values.reduce((a, b) => a + b, 0) / values.length;
  }
}

---

Embedded Systems (ESP32/Arduino)

ESP32 with MicroPython

# boot.py - WiFi connection
import network
import time

def connect_wifi(ssid, password):
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)

    if not wlan.isconnected():
        print('Connecting to WiFi...')
        wlan.connect(ssid, password)

        timeout = 10
        while not wlan.isconnected() and timeout > 0:
            time.sleep(1)
            timeout -= 1

    if wlan.isconnected():
        print('Connected:', wlan.ifconfig())
        return True
    else:
        print('Failed to connect')
        return False

connect_wifi('MyNetwork', 'password')

# main.py - Sensor reading and MQTT
from machine import Pin, ADC
from umqtt.simple import MQTTClient
import json
import time
import dht

# Configuration
MQTT_BROKER = '192.168.1.100'
DEVICE_ID = 'esp32-001'
TOPIC_TELEMETRY = f'devices/{DEVICE_ID}/telemetry'
TOPIC_COMMANDS = f'devices/{DEVICE_ID}/commands'

# Hardware setup
led = Pin(2, Pin.OUT)
dht_sensor = dht.DHT22(Pin(4))
light_sensor = ADC(Pin(34))

# MQTT client
client = MQTTClient(DEVICE_ID, MQTT_BROKER)

def on_message(topic, msg):
    topic = topic.decode()
    data = json.loads(msg.decode())

    print(f'Command received: {data}')

    if data.get('action') == 'led_on':
        led.on()
    elif data.get('action') == 'led_off':
        led.off()
    elif data.get('action') == 'blink':
        for _ in range(5):
            led.on()
            time.sleep(0.2)
            led.off()
            time.sleep(0.2)

client.set_callback(on_message)
client.connect()
client.subscribe(TOPIC_COMMANDS)

def read_sensors():
    dht_sensor.measure()
    return {
        'temperature': dht_sensor.temperature(),
        'humidity': dht_sensor.humidity(),
        'light': light_sensor.read(),
        'timestamp': time.time()
    }

def main():
    last_publish = 0
    publish_interval = 5  # seconds

    while True:
        # Check for incoming messages
        client.check_msg()

        # Publish telemetry periodically
        current_time = time.time()
        if current_time - last_publish >= publish_interval:
            try:
                data = read_sensors()
                client.publish(TOPIC_TELEMETRY, json.dumps(data))
                print(f'Published: {data}')
                last_publish = current_time
            except Exception as e:
                print(f'Error: {e}')

        time.sleep(0.1)

if __name__ == '__main__':
    main()

Arduino/C++ for ESP32

#include <WiFi.h>
#include <PubSubClient.h>
#include <ArduinoJson.h>
#include <DHT.h>

// Configuration
const char* ssid = "MyNetwork";
const char* password = "password";
const char* mqtt_server = "192.168.1.100";
const char* device_id = "esp32-001";

// Hardware
#define DHT_PIN 4
#define DHT_TYPE DHT22
#define LED_PIN 2
#define LIGHT_PIN 34

DHT dht(DHT_PIN, DHT_TYPE);
WiFiClient espClient;
PubSubClient client(espClient);

unsigned long lastPublish = 0;
const long publishInterval = 5000;

void setup_wifi() {
    delay(10);
    Serial.println("Connecting to WiFi...");

    WiFi.begin(ssid, password);

    while (WiFi.status() != WL_CONNECTED) {
        delay(500);
        Serial.print(".");
    }

    Serial.println("\nWiFi connected");
    Serial.println(WiFi.localIP());
}

void callback(char* topic, byte* payload, unsigned int length) {
    StaticJsonDocument<200> doc;
    deserializeJson(doc, payload, length);

    const char* action = doc["action"];

    if (strcmp(action, "led_on") == 0) {
        digitalWrite(LED_PIN, HIGH);
    } else if (strcmp(action, "led_off") == 0) {
        digitalWrite(LED_PIN, LOW);
    }
}

void reconnect() {
    while (!client.connected()) {
        Serial.println("Connecting to MQTT...");

        if (client.connect(device_id)) {
            Serial.println("Connected");

            char topic[50];
            sprintf(topic, "devices/%s/commands", device_id);
            client.subscribe(topic);
        } else {
            Serial.print("Failed, rc=");
            Serial.println(client.state());
            delay(5000);
        }
    }
}

void publishTelemetry() {
    StaticJsonDocument<200> doc;
    doc["temperature"] = dht.readTemperature();
    doc["humidity"] = dht.readHumidity();
    doc["light"] = analogRead(LIGHT_PIN);
    doc["timestamp"] = millis();

    char buffer[256];
    serializeJson(doc, buffer);

    char topic[50];
    sprintf(topic, "devices/%s/telemetry", device_id);
    client.publish(topic, buffer);

    Serial.println("Published telemetry");
}

void setup() {
    Serial.begin(115200);
    pinMode(LED_PIN, OUTPUT);
    dht.begin();

    setup_wifi();
    client.setServer(mqtt_server, 1883);
    client.setCallback(callback);
}

void loop() {
    if (!client.connected()) {
        reconnect();
    }
    client.loop();

    unsigned long now = millis();
    if (now - lastPublish >= publishInterval) {
        lastPublish = now;
        publishTelemetry();
    }
}

---

Device Management

Device Registry

interface Device {
  id: string;
  name: string;
  type: string;
  status: 'online' | 'offline' | 'error';
  lastSeen: Date;
  metadata: Record<string, any>;
  config: Record<string, any>;
}

class DeviceRegistry {
  constructor(
    private db: Database,
    private mqtt: MQTTClient
  ) {
    // Listen for device status
    this.mqtt.subscribe('devices/+/status', (topic, status) => {
      const deviceId = topic.split('/')[1];
      this.updateStatus(deviceId, status);
    });
  }

  async register(device: Omit<Device, 'status' | 'lastSeen'>) {
    const fullDevice: Device = {
      ...device,
      status: 'offline',
      lastSeen: new Date(),
    };

    await this.db.devices.create({ data: fullDevice });

    // Send initial config to device
    this.mqtt.publish(`devices/${device.id}/config`, device.config);

    return fullDevice;
  }

  async updateConfig(deviceId: string, config: Record<string, any>) {
    await this.db.devices.update({
      where: { id: deviceId },
      data: { config },
    });

    // Push config to device
    this.mqtt.publish(`devices/${deviceId}/config`, config, { retain: true });
  }

  async sendCommand(deviceId: string, command: any) {
    this.mqtt.publish(`devices/${deviceId}/commands`, command);

    // Log command
    await this.db.deviceCommands.create({
      data: {
        deviceId,
        command,
        timestamp: new Date(),
      },
    });
  }

  private async updateStatus(deviceId: string, status: any) {
    await this.db.devices.update({
      where: { id: deviceId },
      data: {
        status: status.status,
        lastSeen: new Date(),
      },
    });
  }
}

---

Related Skills

• [[realtime-systems]] - Real-time communication
• [[cloud-platforms]] - IoT cloud services
• [[system-design]] - Edge architecture

Edge & IoT Reference

Detailed reference for edge computing and IoT development.

Communication Protocols

MQTT (Message Queuing Telemetry Transport)

Lightweight publish/subscribe protocol for IoT.

// MQTT Client Example (Node.js)
const mqtt = require('mqtt');

const client = mqtt.connect('mqtt://broker.example.com', {
  clientId: 'device_001',
  username: 'user',
  password: 'pass',
  keepalive: 60,
  clean: true,
});

// Subscribe
client.subscribe('sensors/+/temperature', { qos: 1 });

// Publish
client.publish('sensors/device_001/temperature', JSON.stringify({
  value: 23.5,
  unit: 'celsius',
  timestamp: Date.now(),
}), { qos: 1, retain: true });

// Handle messages
client.on('message', (topic, payload) => {
  const data = JSON.parse(payload.toString());
  console.log(`${topic}: ${data.value}`);
});

CoAP (Constrained Application Protocol)

UDP-based REST-like protocol for constrained devices.

// CoAP Example
const coap = require('coap');

// GET request
const req = coap.request('coap://device.local/temperature');
req.on('response', (res) => {
  console.log(res.payload.toString());
});
req.end();

// Observable resource
const req = coap.request({
  hostname: 'device.local',
  pathname: '/temperature',
  observe: true,
});
req.on('response', (res) => {
  res.on('data', (chunk) => console.log(chunk.toString()));
});

WebSocket (IoT Variant)

Full-duplex communication over TCP.

// WebSocket for IoT
const WebSocket = require('ws');

const ws = new WebSocket('wss://iot-gateway.example.com');

ws.on('open', () => {
  // Register device
  ws.send(JSON.stringify({
    type: 'register',
    deviceId: 'device_001',
    capabilities: ['temperature', 'humidity'],
  }));
});

ws.on('message', (data) => {
  const msg = JSON.parse(data);
  if (msg.type === 'command') {
    executeCommand(msg.payload);
  }
});

Data Formats

Sensor Data Schema

interface SensorReading {
  deviceId: string;
  sensorType: string;
  value: number | string | boolean;
  unit?: string;
  timestamp: number; // Unix epoch ms
  quality?: 'good' | 'uncertain' | 'bad';
  metadata?: Record<string, unknown>;
}

interface DeviceTelemetry {
  deviceId: string;
  readings: SensorReading[];
  battery?: number; // percentage
  signal?: number; // dBm
  firmware?: string;
  uptime?: number; // seconds
}

Protocol Buffers (Efficient Binary)

// sensor.proto
syntax = "proto3";

message SensorReading {
  string device_id = 1;
  string sensor_type = 2;
  double value = 3;
  string unit = 4;
  int64 timestamp = 5;
  Quality quality = 6;

  enum Quality {
    GOOD = 0;
    UNCERTAIN = 1;
    BAD = 2;
  }
}

message TelemetryBatch {
  repeated SensorReading readings = 1;
  int32 battery_percent = 2;
  int32 signal_dbm = 3;
}

Edge Processing Patterns

Local Aggregation

# Edge aggregation before cloud upload
class SensorAggregator:
    def __init__(self, window_size=60):  # 60 seconds
        self.window_size = window_size
        self.readings = []

    def add_reading(self, value, timestamp):
        self.readings.append({'value': value, 'ts': timestamp})
        self._prune_old_readings(timestamp)

    def get_aggregated(self):
        if not self.readings:
            return None

        values = [r['value'] for r in self.readings]
        return {
            'min': min(values),
            'max': max(values),
            'avg': sum(values) / len(values),
            'count': len(values),
            'window_start': self.readings[0]['ts'],
            'window_end': self.readings[-1]['ts'],
        }

    def _prune_old_readings(self, current_time):
        cutoff = current_time - (self.window_size * 1000)
        self.readings = [r for r in self.readings if r['ts'] > cutoff]

Rule Engine

# Simple edge rule engine
class RuleEngine:
    def __init__(self):
        self.rules = []

    def add_rule(self, condition, action, name=None):
        self.rules.append({
            'name': name,
            'condition': condition,
            'action': action,
        })

    def evaluate(self, reading):
        triggered = []
        for rule in self.rules:
            if rule['condition'](reading):
                rule['action'](reading)
                triggered.append(rule['name'])
        return triggered

# Usage
engine = RuleEngine()

# High temperature alert
engine.add_rule(
    condition=lambda r: r['type'] == 'temperature' and r['value'] > 30,
    action=lambda r: send_alert(f"High temp: {r['value']}°C"),
    name='high_temp_alert'
)

# Motion detection
engine.add_rule(
    condition=lambda r: r['type'] == 'motion' and r['value'] == True,
    action=lambda r: activate_lights(),
    name='motion_lights'
)

Device Provisioning

Zero-Touch Provisioning

interface DeviceProvisioningConfig {
  // Device identity
  deviceId: string;
  deviceType: string;
  serialNumber: string;

  // Network
  network: {
    wifi?: {
      ssid: string;
      password: string;
      security: 'WPA2' | 'WPA3';
    };
    ethernet?: {
      dhcp: boolean;
      staticIp?: string;
    };
  };

  // Cloud connection
  cloud: {
    endpoint: string;
    protocol: 'mqtt' | 'https' | 'coap';
    auth: {
      method: 'certificate' | 'token' | 'symmetric_key';
      certificate?: string;
      privateKey?: string;
      token?: string;
    };
  };

  // Firmware
  firmware: {
    currentVersion: string;
    updateUrl?: string;
    autoUpdate: boolean;
  };

  // Sensors configuration
  sensors: Array<{
    id: string;
    type: string;
    pin?: number;
    address?: string; // I2C/SPI address
    samplingRate: number; // ms
    calibration?: Record<string, number>;
  }>;
}

Device Twin Pattern

// Azure IoT Hub style device twin
interface DeviceTwin {
  deviceId: string;

  // Desired state (set by cloud)
  desired: {
    firmwareVersion: string;
    telemetryInterval: number;
    thresholds: {
      temperature: { min: number; max: number };
    };
    $version: number;
  };

  // Reported state (set by device)
  reported: {
    firmwareVersion: string;
    telemetryInterval: number;
    lastBootTime: string;
    connectivity: 'online' | 'offline';
    $version: number;
  };

  // Read-only metadata
  tags: {
    location: string;
    environment: string;
    owner: string;
  };
}

OTA (Over-The-Air) Updates

Update Protocol

interface OTAUpdateManifest {
  version: string;
  releaseDate: string;
  releaseNotes: string;

  // Firmware binary
  firmware: {
    url: string;
    size: number;
    checksum: string; // SHA-256
    signature: string; // RSA signature
  };

  // Targeting
  targeting: {
    deviceTypes?: string[];
    minFirmwareVersion?: string;
    rolloutPercentage?: number;
    regions?: string[];
  };

  // Rollback
  rollback: {
    automaticRollback: boolean;
    healthCheckEndpoint?: string;
    healthCheckTimeout?: number;
  };
}

// Update state machine
type OTAState =
  | 'idle'
  | 'checking'
  | 'downloading'
  | 'verifying'
  | 'installing'
  | 'rebooting'
  | 'success'
  | 'failed'
  | 'rollback';

Power Management

Sleep Modes

// ESP32 Deep Sleep Example
#include "esp_sleep.h"

void enter_deep_sleep(uint64_t sleep_time_us) {
    // Configure wake-up source
    esp_sleep_enable_timer_wakeup(sleep_time_us);

    // Optional: wake on GPIO
    esp_sleep_enable_ext0_wakeup(GPIO_NUM_33, 1);

    // Enter deep sleep
    esp_deep_sleep_start();
}

// Wake up and determine cause
esp_sleep_wakeup_cause_t wakeup_reason = esp_sleep_get_wakeup_cause();
switch (wakeup_reason) {
    case ESP_SLEEP_WAKEUP_TIMER:
        // Scheduled wake
        break;
    case ESP_SLEEP_WAKEUP_EXT0:
        // GPIO interrupt
        break;
}

Security Best Practices

Device Security Checklist

## Hardware Security
- [ ] Secure boot enabled
- [ ] Hardware security module (HSM) for key storage
- [ ] Tamper detection mechanisms
- [ ] Unique device identity (hardware root of trust)

## Firmware Security
- [ ] Signed firmware images
- [ ] Secure OTA update mechanism
- [ ] Rollback protection
- [ ] Watchdog timer enabled

## Communication Security
- [ ] TLS 1.3 for all connections
- [ ] Certificate pinning
- [ ] Mutual authentication
- [ ] Message encryption

## Data Security
- [ ] Data encryption at rest
- [ ] Minimal data collection
- [ ] Secure data deletion
- [ ] Privacy by design

## Access Control
- [ ] Principle of least privilege
- [ ] Role-based access control
- [ ] Strong authentication
- [ ] Session management

Certificate Management

interface DeviceCertificate {
  // X.509 certificate
  certificate: string; // PEM format
  privateKey: string;  // PEM format (stored securely)
  caCertificate: string;

  // Metadata
  deviceId: string;
  issuedAt: Date;
  expiresAt: Date;
  issuer: string;

  // Renewal
  renewalWindow: number; // days before expiry
  autoRenew: boolean;
}

Common Hardware Interfaces

GPIO

# Raspberry Pi GPIO example
import RPi.GPIO as GPIO

GPIO.setmode(GPIO.BCM)

# Output (LED)
GPIO.setup(18, GPIO.OUT)
GPIO.output(18, GPIO.HIGH)

# Input (Button)
GPIO.setup(17, GPIO.IN, pull_up_down=GPIO.PUD_UP)
if GPIO.input(17) == GPIO.LOW:
    print("Button pressed")

# PWM
pwm = GPIO.PWM(18, 1000)  # 1kHz
pwm.start(50)  # 50% duty cycle

I2C

# I2C sensor reading
import smbus

bus = smbus.SMBus(1)  # I2C bus 1
address = 0x48  # Device address

# Read temperature from TMP102
data = bus.read_i2c_block_data(address, 0x00, 2)
temp = ((data[0] << 4) | (data[1] >> 4)) * 0.0625
print(f"Temperature: {temp}°C")

SPI

# SPI communication
import spidev

spi = spidev.SpiDev()
spi.open(0, 0)  # Bus 0, Device 0
spi.max_speed_hz = 1000000

# Transfer data
response = spi.xfer2([0x01, 0x02, 0x03])

Platform-Specific References