IoT Cloud Platforms — AWS IoT Core, Azure IoT Hub, Device Shadows, and Fleet Management

# device_shadow.py — ESP32-style device updating its shadow
import json, time, random

# Simulate MQTT shadow update
class DeviceShadow:
    SHADOW_TOPIC = "$aws/things/MyDevice/shadow/update"

    def __init__(self, device_id: str):
        self.device_id = device_id
        self.state = {
            "reported": {
                "temperature": 24.0,
                "humidity": 55.0,
                "status": "online",
                "firmware": "v2.1.0"
            },
            "desired": {}
        }

    def update_shadow(self):
        self.state["reported"]["temperature"] = round(random.uniform(18.0, 32.0), 1)
        self.state["reported"]["humidity"] = round(random.uniform(40.0, 70.0), 1)
        # Publish to shadow topic
        payload = json.dumps({"state": {"reported": self.state["reported"]}})
        print(f"Shadow update: {payload}")
        return payload

    def check_desired_state(self, desired: dict):
        """Check if cloud wants us to change something"""
        if "led_color" in desired:
            print(f"Changing LED to: {desired['led_color']}")
        if "sample_rate" in desired:
            print(f"Setting sample rate to: {desired['sample_rate']}s")

# Simulate device running
shadow = DeviceShadow("TempSensor-42")
for _ in range(3):
    payload = shadow.update_shadow()
    time.sleep(0.5)

# Simulate receiving desired state from cloud
shadow.check_desired_state({"led_color": "blue", "sample_rate": 30})

-- AWS IoT Rule SQL: send high-temperature alerts to SNS
SELECT device_id, temperature, timestamp
FROM 'devices/+/telemetry'
WHERE temperature > 30.0

# Simulating the rule engine action
def process_telemetry(topic: str, payload: dict):
    print(f"Received: {topic} -> {payload}")
    if payload.get("temperature", 0) > 30.0:
        print(f"ALERT sent to SNS: {payload['device_id']} is hot ({payload['temperature']}°C)")
    # Store in database
    print(f"Stored to DynamoDB: {payload['device_id']} at {payload.get('timestamp', 'now')}")

# Simulated device messages
process_telemetry("devices/sensor-01/telemetry", {"device_id": "sensor-01", "temperature": 24.5, "timestamp": "2026-06-15T12:00:00Z"})
process_telemetry("devices/sensor-02/telemetry", {"device_id": "sensor-02", "temperature": 32.1, "timestamp": "2026-06-15T12:00:01Z"})

# fleet_manager.py — simulated fleet management
import json

fleet = {
    "device-001": {"location": "Building A", "firmware": "v2.0.0", "status": "online"},
    "device-002": {"location": "Building A", "firmware": "v1.5.0", "status": "offline"},
    "device-003": {"location": "Building B", "firmware": "v2.0.0", "status": "online"},
    "device-004": {"location": "Building B", "firmware": "v1.8.0", "status": "online"},
}

def find_outdated(target_firmware: str):
    outdated = [d for d, info in fleet.items() if info["firmware"] < target_firmware]
    print(f"Devices needing update to {target_firmware}:")
    for device_id in outdated:
        print(f"  {device_id}: {fleet[device_id]['firmware']} at {fleet[device_id]['location']}")
    return outdated

def push_update(device_ids: list, new_firmware: str):
    print(f"\nPushing {new_firmware} to {len(device_ids)} devices...")
    for device_id in device_ids:
        fleet[device_id]["firmware"] = new_firmware
        print(f"  Updated {device_id}")
    print(f"Success: {len(device_ids)} devices updated")

outdated = find_outdated("v2.0.0")
push_update(outdated, "v2.0.0")

import random, time, json

class IoTDevice:
    def __init__(self, device_id: str, location: str):
        self.device_id = device_id
        self.location = location
        self.temperature = 25.0

    def read_sensors(self) -> dict:
        self.temperature += random.uniform(-0.5, 0.5)
        self.temperature = max(18.0, min(35.0, self.temperature))
        return {
            "device_id": self.device_id,
            "location": self.location,
            "temperature": round(self.temperature, 1),
            "humidity": round(random.uniform(40, 70)),
            "battery": round(random.uniform(85, 100)),
        }

class CloudPlatform:
    def __init__(self):
        self.devices = {}
        self.shadows = {}

    def connect(self, device: IoTDevice):
        self.devices[device.device_id] = device
        self.shadows[device.device_id] = {"reported": {}, "desired": {}}
        print(f"Device connected: {device.device_id}")

    def ingest_telemetry(self, data: dict):
        device_id = data["device_id"]
        self.shadows[device_id]["reported"] = data
        if data["temperature"] > 30:
            print(f"ALERT [{device_id}]: High temperature {data['temperature']}°C")

# Simulate
platform = CloudPlatform()
fleet = [
    IoTDevice("sensor-1", "Building A - Floor 1"),
    IoTDevice("sensor-2", "Building A - Floor 2"),
    IoTDevice("sensor-3", "Building B"),
]

for device in fleet:
    platform.connect(device)

print("\nSimulating telemetry (10 rounds)...")
for _ in range(10):
    for device in fleet:
        data = device.read_sensors()
        platform.ingest_telemetry(data)
    time.sleep(0.3)