iot-expert

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)
        else:
            print(f"Connection failed with code {rc}")

    def _on_message(self, client, userdata, msg):
        topic = msg.topic
        payload = msg.payload.decode()

        if topic in self.subscriptions:
            try:
                data = json.loads(payload)
                self.subscriptions[topic](data)
            except json.JSONDecodeError:
                self.subscriptions[topic](payload)

    def _on_disconnect(self, client, userdata, rc):
        if rc != 0:
            print(f"Unexpected disconnect. Reconnecting...")

    def connect(self, username: str = None, password: str = None):
        if username and password:
            self.client.username_pw_set(username, password)

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

    def publish(self, topic: str, payload: Dict, qos: int = 1, retain: bool = False):
        """Publish message to MQTT topic"""
        message = json.dumps(payload)
        result = self.client.publish(topic, message, qos=qos, retain=retain)
        return result.rc == mqtt.MQTT_ERR_SUCCESS

    def subscribe(self, topic: str, callback: Callable, qos: int = 1):
        """Subscribe to MQTT topic with callback"""
        self.subscriptions[topic] = callback
        self.client.subscribe(topic, qos=qos)

    def disconnect(self):
        self.client.loop_stop()
        self.client.disconnect()

# IoT Device Example
class TemperatureSensor:
    def __init__(self, device_id: str, mqtt_client: MQTTClient):
        self.device_id = device_id
        self.mqtt = mqtt_client
        self.topic = f"sensors/temperature/{device_id}"

    def read_temperature(self) -> float:
        # In real device, read from actual sensor
        import random
        return round(random.uniform(20.0, 30.0), 2)

    def publish_reading(self):
        temperature = self.read_temperature()

        payload = {
            "device_id": self.device_id,
            "temperature": temperature,
            "unit": "celsius",
            "timestamp": datetime.utcnow().isoformat()
        }

        self.mqtt.publish(self.topic, payload)
        return payload