iot-expert

name	iot-expert
version	1.0.0
description	Expert-level IoT systems, embedded devices, edge computing, and IoT protocols
category	domains
tags	["iot","embedded","edge-computing","mqtt","sensors","firmware"]
allowed-tools	["Read","Write","Edit"]

IoT Expert

Expert guidance for IoT systems, embedded devices, edge computing, sensor networks, and IoT protocols.

Core Concepts

IoT Architecture

• Device layer (sensors, actuators)
• Edge computing layer
• Network layer (connectivity)
• Cloud/platform layer
• Application layer
• Security across all layers

IoT Protocols

• MQTT (Message Queuing Telemetry Transport)
• CoAP (Constrained Application Protocol)
• HTTP/REST for IoT
• WebSocket for real-time
• LoRaWAN for long-range
• Zigbee, Z-Wave for home automation

Embedded Systems

• Microcontroller programming
• Real-time operating systems (RTOS)
• Power management
• Firmware updates (OTA)
• Hardware interfaces (I2C, SPI, UART)
• Memory constraints

MQTT Implementation

import paho.mqtt.client as mqtt
import json
from datetime import datetime
from typing import Callable, Dict

class MQTTClient:
    def __init__(self, broker: str, port: int = 1883, client_id: str = "iot_device"):
        self.broker = broker
        self.port = port
        self.client = mqtt.Client(client_id)
        self.subscriptions: Dict[str, Callable] = {}

        self.client.on_connect = self._on_connect
        self.client.on_message = self._on_message
        self.client.on_disconnect = self._on_disconnect

    def _on_connect(self, client, userdata, flags, rc):
        if rc == 0:
            print(f"Connected to MQTT broker at {self.broker}:{self.port}")
            # Resubscribe to topics on reconnect
            for topic in self.subscriptions.keys():
                self.client.subscribe(topic)
        :
            ()

     ():
        topic = msg.topic
        payload = msg.payload.decode()

         topic  .subscriptions:
            :
                data = json.loads(payload)
                .subscriptions[topic](data)
             json.JSONDecodeError:
                .subscriptions[topic](payload)

     ():
         rc != :
            ()

     ():
         username  password:
            .client.username_pw_set(username, password)

        .client.connect(.broker, .port, )
        .client.loop_start()

     ():
        
        message = json.dumps(payload)
        result = .client.publish(topic, message, qos=qos, retain=retain)
         result.rc == mqtt.MQTT_ERR_SUCCESS

     ():
        
        .subscriptions[topic] = callback
        .client.subscribe(topic, qos=qos)

     ():
        .client.loop_stop()
        .client.disconnect()


 :
     ():
        .device_id = device_id
        .mqtt = mqtt_client
        .topic = 

     () -> :
        
         random
         (random.uniform(, ), )

     ():
        temperature = .read_temperature()

        payload = {
            : .device_id,
            : temperature,
            : ,
            : datetime.utcnow().isoformat()
        }

        .mqtt.publish(.topic, payload)
         payload

Embedded C for Microcontroller

// Arduino/ESP32 Example
#include <WiFi.h>
#include <PubSubClient.h>
#include "DHT.h"

#define DHTPIN 4
#define DHTTYPE DHT22

const char* ssid = "YourWiFiSSID";
const char* password = "YourPassword";
const char* mqtt_server = "broker.example.com";

WiFiClient espClient;
PubSubClient client(espClient);
DHT dht(DHTPIN, DHTTYPE);

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("WiFi connected");
    Serial.println("IP address: ");
    Serial.println(WiFi.localIP());
}

void callback(char* topic, byte* payload, unsigned int length) {
    Serial.print("Message arrived [");
    Serial.print(topic);
    Serial.print("] ");

    for (int i = 0; i < length; i++) {
        Serial.print((char)payload[i]);
    }
    Serial.println();
}

  {
     (!client.connected()) {
        Serial.print();

         (client.connect()) {
            Serial.println();
            client.subscribe();
        }  {
            Serial.print();
            Serial.print(client.state());
            Serial.println();
            delay();
        }
    }
}

  {
    Serial.begin();
    setup_wifi();
    client.setServer(mqtt_server, );
    client.setCallback(callback);
    dht.begin();
}

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

    
       lastRead = ;
     (millis() - lastRead > ) {
         humidity = dht.readHumidity();
         temperature = dht.readTemperature();

         (!isnan(humidity) && !isnan(temperature)) {
             msg[];
            (msg, (msg),
                    ,
                    temperature, humidity);

            client.publish(, msg);
            Serial.println(msg);
        }

        lastRead = millis();
    }
}

Edge Computing

import asyncio
from typing import Dict, List
import numpy as np

class EdgeProcessor:
    """Process data at edge before sending to cloud"""

    def __init__(self, buffer_size: int = 100):
        self.buffer: List[Dict] = []
        self.buffer_size = buffer_size

    def add_reading(self, reading: Dict):
        """Add sensor reading to buffer"""
        self.buffer.append(reading)

        if len(self.buffer) >= self.buffer_size:
            self.process_buffer()

    def process_buffer(self) -> Dict:
        """Process buffered data at edge"""
        if not self.buffer:
            return {}

        # Extract temperature values
        temperatures = [r['temperature'] for r in self.buffer]

        # Compute statistics at edge
        summary = {
            "count": len(temperatures),
            : np.mean(temperatures),
            : np.std(temperatures),
            : np.(temperatures),
            : np.(temperatures),
            : .detect_anomalies(temperatures)
        }

        
        .buffer = []

         summary

     () -> []:
        
        mean = np.mean(values)
        std = np.std(values)
        threshold = 

        anomalies = []
         i, v  (values):
             (v - mean) > threshold * std:
                anomalies.append(i)

         anomalies

 :
    

     ():
        .mqtt = mqtt_client
        .edge_processor = EdgeProcessor()
        .devices: [, TemperatureSensor] = {}

     ():
        
        .devices[device.device_id] = device

        
        topic = 
        .mqtt.subscribe(topic, .handle_device_data)

     ():
        
        
        .edge_processor.add_reading(data)

      ():
        
         :
             device  .devices.values():
                reading = device.publish_reading()
                ()

             asyncio.sleep(interval)

Device Management

from datetime import datetime
from enum import Enum

class DeviceStatus(Enum):
    ONLINE = "online"
    OFFLINE = "offline"
    MAINTENANCE = "maintenance"
    ERROR = "error"

class IoTDevice:
    def __init__(self, device_id: str, device_type: str):
        self.device_id = device_id
        self.device_type = device_type
        self.status = DeviceStatus.OFFLINE
        self.last_seen = None
        self.firmware_version = "1.0.0"
        self.metadata = {}

    def update_status(self, status: DeviceStatus):
        self.status = status
        self.last_seen = datetime.utcnow()

    def needs_firmware_update(self, latest_version: str) -> bool:
        return self.firmware_version < latest_version

class DeviceManager:
    def __init__(self):
        self.devices: Dict[str, IoTDevice] = {}

     ():
        
        .devices[device.device_id] = device

     ():
        
         device_id  .devices:
            .devices[device_id].update_status(DeviceStatus.ONLINE)

     () -> [IoTDevice]:
        
        offline = []
        now = datetime.utcnow()

         device  .devices.values():
             device.last_seen:
                elapsed = (now - device.last_seen).total_seconds()
                 elapsed > timeout_seconds:
                    offline.append(device)

         offline

     ():
        
         device_id  .devices:
            
            payload = {
                : ,
                : new_version,
                : 
            }
             payload

Best Practices

Device Design

• Implement power management for battery devices
• Use deep sleep modes when idle
• Handle network disconnections gracefully
• Implement watchdog timers
• Design for remote diagnostics
• Plan for firmware updates (OTA)

Security

• Use TLS/SSL for MQTT connections
• Implement device authentication
• Encrypt sensitive data
• Secure firmware updates
• Regular security patches
• Network segmentation

Data Management

• Process data at edge when possible
• Implement data buffering for offline scenarios
• Use efficient data formats (e.g., Protocol Buffers)
• Compress data before transmission
• Handle time synchronization
• Implement data retention policies

Anti-Patterns

❌ No power management strategy ❌ Unencrypted communications ❌ No error handling for network failures ❌ Sending all raw data to cloud ❌ No device authentication ❌ Hard-coded credentials in firmware ❌ No OTA update mechanism

Resources

• MQTT: https://mqtt.org/
• ESP32 Documentation: https://docs.espressif.com/
• Arduino: https://www.arduino.cc/
• AWS IoT: https://aws.amazon.com/iot/
• Azure IoT: https://azure.microsoft.com/en-us/overview/iot/