IoT at Scale: Device Management, Data Ingestion

import time
import random

def register_with_backoff(device_id, dps_endpoint, max_retries=5):
    """Register device with exponential backoff and jitter."""
    base_delay = 1.0
    max_delay = 60.0

    for attempt in range(max_retries):
        try:
            result = dps_client.register(device_id, dps_endpoint)
            return result
        except ThrottlingError:
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay * 0.5)
            time.sleep(delay + jitter)

    raise RuntimeError(f"Device {device_id} failed to register after {max_retries} attempts")

#!/bin/bash
# Staged OTA rollout: canary -> 10% -> 50% -> 100%
CANARY_GROUP="device-fw-canary"
ROLLOUT_GROUP="device-fw-rollout"
JOB_ID="fw-update-v2.1.3"

aws iot create-job \
  --job-id "${JOB_ID}-canary" \
  --target-selection "SNAPSHOT" \
  --targets "arn:aws:iot:us-east-1:123456:thinggroup/${CANARY_GROUP}" \
  --document file://firmware-update.json \
  --presigned-url-config "{\"roleArn\":\"arn:aws:iam::123456:role/ota-role\",\"expiresInSec\":3600}"

echo "Monitor canary for 24 hours before proceeding..."
# On success, roll out to larger groups
aws iot create-job \
  --job-id "${JOB_ID}-rollout" \
  --target-selection "SNAPSHOT" \
  --targets "arn:aws:iot:us-east-1:123456:thinggroup/${ROLLOUT_GROUP}" \
  --document file://firmware-update.json \
  --presigned-url-config "{\"roleArn\":\"arn:aws:iam::123456:role/ota-role\",\"expiresInSec\":3600}" \
  --job-process-options "{\"executionNumberThreshold\":50,\"inProgressTimers\":{\"inProgressTimeoutInMinutes\":120}}"

#!/usr/bin/env python3
"""Production MQTT 5.0 IoT client with auto-reconnect."""

import paho.mqtt.client as mqtt
import json
import ssl
import time
from datetime import datetime, timezone

class IoTDevice:
    """MQTT 5.0 device client with TLS and session management."""

    def __init__(self, device_id, broker_url, port=8883,
                 cert_file=None, key_file=None, ca_file=None):
        self.device_id = device_id
        self.broker_url = broker_url
        self.port = port
        self.reconnect_delay = 1
        self.max_delay = 64
        self.session_persist = {}

        self.client = mqtt.Client(
            client_id=device_id,
            protocol=mqtt.MQTTv5,
            callback_api_version=mqtt.CallbackAPIVersion.VERSION2
        )

        if cert_file and key_file:
            self.client.tls_set(
                ca_certs=ca_file,
                certfile=cert_file,
                keyfile=key_file,
                tls_version=ssl.PROTOCOL_TLSv1_2
            )

        self.client.on_connect = self.on_connect
        self.client.on_disconnect = self.on_disconnect
        self.client.on_message = self.on_message

    def connect(self):
        """Connect to broker with persistent session."""
        self.client.connect(self.broker_url, self.port, 60)
        self.client.loop_start()

    def on_connect(self, client, userdata, flags, reason_code, properties):
        """Handle MQTT 5.0 connection with reason codes."""
        if reason_code == 0:
            print(f"Connected: {self.device_id}")
            self.reconnect_delay = 1
            client.subscribe(f"devices/{self.device_id}/command", qos=1)
            self.publish_state({"status": "online"})
        else:
            print(f"Connection refused: {reason_code}")

    def on_disconnect(self, client, userdata, reason_code, properties):
        """Handle disconnection with exponential backoff."""
        delay = min(self.reconnect_delay, self.max_delay)
        time.sleep(delay)
        self.reconnect_delay *= 2
        self.connect()

    def on_message(self, client, userdata, msg):
        """Handle incoming command messages."""
        payload = json.loads(msg.payload)
        topic = msg.topic

        if "/command" in topic:
            command = payload.get("command")
            if command == "reboot":
                self.reboot()
            elif command == "update_config":
                self.apply_config(payload.get("config", {}))

    def publish_telemetry(self, measurements):
        """Publish sensor reading to telemetry topic."""
        payload = json.dumps({
            "device_id": self.device_id,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "measurements": measurements
        })
        self.client.publish(
            f"devices/{self.device_id}/telemetry",
            payload,
            qos=1
        )

    def publish_state(self, state):
        """Report device state to shadow topic."""
        payload = json.dumps(state)
        self.client.publish(
            f"devices/{self.device_id}/state",
            payload,
            qos=1,
            retain=True
        )

#!/usr/bin/env python3
"""MQTT-to-Kafka bridge for IoT telemetry ingestion."""

from kafka import KafkaProducer
import paho.mqtt.client as mqtt
import json

class MQTTKafkaBridge:
    """Bridge IoT telemetry from MQTT to Kafka."""

    def __init__(self, mqtt_broker, kafka_brokers, topic_mappings):
        self.kafka_producer = KafkaProducer(
            bootstrap_servers=kafka_brokers,
            value_serializer=lambda v: json.dumps(v).encode("utf-8"),
            compression_type="snappy",
            acks="all",
            retries=3
        )
        self.topic_mappings = topic_mappings  # {"mqtt/topic": "kafka.topic"}

        self.mqtt_client = mqtt.Client(
            client_id="kafka-bridge-01",
            protocol=mqtt.MQTTv5
        )
        self.mqtt_client.on_message = self.on_mqtt_message
        self.mqtt_client.connect(mqtt_broker, 8883)

    def on_mqtt_message(self, client, userdata, msg):
        """Forward MQTT message to Kafka topic."""
        kafka_topic = self.topic_mappings.get(msg.topic)
        if not kafka_topic:
            return

        payload = json.loads(msg.payload)
        # Use device_id as partition key for ordering
        device_id = payload.get("device_id", "unknown")
        self.kafka_producer.send(
            kafka_topic,
            value=payload,
            key=device_id.encode("utf-8")
        )

    def start(self):
        """Start consuming MQTT and producing to Kafka."""
        for mqtt_topic in self.topic_mappings:
            self.mqtt_client.subscribe(mqtt_topic)
        self.mqtt_client.loop_forever()

#!/usr/bin/env python3
"""IoT telemetry with Schema Registry integration."""

from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka import SerializingProducer

TELEMETRY_SCHEMA = """
{
  "type": "record",
  "name": "Telemetry",
  "namespace": "com.calmops.iot",
  "fields": [
    {"name": "device_id", "type": "string"},
    {"name": "timestamp", "type": {"type": "long", "logicalType": "timestamp-millis"}},
    {"name": "temperature", "type": ["null", "double"], "default": null},
    {"name": "humidity", "type": ["null", "double"], "default": null},
    {"name": "pressure", "type": ["null", "double"], "default": null},
    {"name": "vibration", "type": ["null", "double"], "default": null}
  ]
}
"""

schema_registry = SchemaRegistryClient({"url": "http://schema-registry:8081"})
avro_serializer = AvroSerializer(schema_registry, TELEMETRY_SCHEMA)

producer = SerializingProducer({
    "bootstrap.servers": "kafka:9092",
    "key.serializer": None,
    "value.serializer": avro_serializer
})

def publish_telemetry(device_id, measurements):
    """Publish validated telemetry to Kafka."""
    record = {
        "device_id": device_id,
        "timestamp": int(time.time() * 1000),
        **measurements
    }
    producer.produce(topic="iot.telemetry", value=record)
    producer.flush()

// Flink job: aggregate IoT sensor readings into 1-minute windows
DataStream<SensorReading> readings = env
    .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "IoT-Kafka")
    .keyBy(SensorReading::getDeviceId)
    .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
    .aggregate(new AverageAggregator())
    .addSink(influxDBSink);

// Detect temperature spikes exceeding rolling average by 3 sigma
DataStream<Alert> alerts = readings
    .keyBy(Alert::getDeviceId)
    .process(new TemperatureAnomalyDetector(thresholdSigma = 3.0))
    .addSink(alertSink);

# Terraform: AWS IoT Core with device registration and rule
resource "aws_iot_thing" "temp_sensor" {
  name = "temp-sensor-${var.sensor_id}"
  attributes = {
    model  = "TS-2000"
    floor  = var.floor
    zone   = var.zone
  }
}

resource "aws_iot_topic_rule" "telemetry_ingest" {
  name        = "telemetry_to_kafka"
  sql         = "SELECT *, topic(3) as device_id FROM 'devices/+/telemetry'"
  description = "Route device telemetry to Kafka"

  action {
    kafka {
      destination_arn = aws_msk_cluster.iot.arn
      topic           = "iot.telemetry"
      partition       = "device_id"
      client_properties = {
        "acks" = "all"
      }
    }
  }
}

# Terraform: Azure IoT Hub with DPS enrollment
resource "azurerm_iothub" "main" {
  name                = "iot-hub-prod-001"
  resource_group_name = var.resource_group
  location            = var.location
  sku {
    name     = "S2"
    capacity = 2
  }
}

resource "azurerm_iothub_dps" "provisioning" {
  name                = "dps-prod-001"
  resource_group_name = var.resource_group
  location            = var.location
  sku {
    name     = "S1"
    capacity = 1
  }
}

resource "azurerm_iothub_dps_enrollment" "group_x509" {
  resource_group_name       = var.resource_group
  iothub_dps_name           = azurerm_iothub_dps.provisioning.name
  enrollment_type           = "EnrollmentGroup"
  device_type               = "IoTEdge"
  attestation_type          = "x509"
  x509_ca_certificate_name  = "iot-ca-cert"
}

# Route telemetry to Event Hubs for Kafka-compatible ingestion
resource "azurerm_iothub_endpoint_eventhub" "telemetry" {
  resource_group_name = var.resource_group
  iothub_id           = azurerm_iothub.main.id
  name                = "telemetry-stream"
  connection_string   = azurerm_eventhub_namespace.main.default_primary_connection_string
}

resource "azurerm_iothub_route" "telemetry_route" {
  resource_group_name = var.resource_group
  iothub_name         = azurerm_iothub.main.name
  name                = "telemetry-to-eventhub"
  source              = "DeviceMessages"
  condition           = "true"
  endpoint_names      = [azurerm_iothub_endpoint_eventhub.telemetry.name]
  enabled             = true
}

#!/usr/bin/env python3
"""Edge telemetry processor with local aggregation and alerting."""

from collections import deque
from statistics import mean, stdev
from datetime import datetime, timezone
import json

class EdgeTelemetryProcessor:
    """Filters noise, aggregates windows, and detects anomalies at the edge."""

    def __init__(self, device_id, window_size=60):
        self.device_id = device_id
        self.window_size = window_size
        self.readings = deque(maxlen=window_size)
        self.cloud_batch = []

    def process_reading(self, reading):
        """Process a single sensor reading at the edge."""
        self.readings.append(reading)

        # Immediate threshold alert
        if reading.get("temperature", 0) > 85:
            return self._create_alert("HIGH_TEMP", reading)

        # Detect anomalous readings using rolling statistics
        if len(self.readings) >= 10:
            values = [r.get("temperature", 0) for r in self.readings]
            avg = mean(values)
            std = stdev(values)
            current = reading.get("temperature", 0)

            if abs(current - avg) > 3 * std:
                return self._create_alert("ANOMALY", reading)

        return None

    def _create_alert(self, alert_type, reading):
        """Create an alert struct for immediate forwarding."""
        return {
            "type": alert_type,
            "device_id": self.device_id,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "reading": reading
        }

    def aggregate_and_flush(self):
        """Summarize the window and queue for cloud delivery."""
        if not self.readings:
            return None

        temps = [r.get("temperature") for r in self.readings if r.get("temperature")]
        summary = {
            "device_id": self.device_id,
            "window_start": self.readings[0].get("timestamp"),
            "window_end": self.readings[-1].get("timestamp"),
            "count": len(self.readings),
            "avg_temp": mean(temps) if temps else None,
            "max_temp": max(temps) if temps else None,
            "min_temp": min(temps) if temps else None,
        }
        self.readings.clear()
        return summary

#!/usr/bin/env python3
"""Batch write telemetry to InfluxDB with error handling."""

from influxdb_client_3 import InfluxDBClient3
from datetime import datetime, timezone

class IoTTimeSeriesStore:
    """Time-series storage for IoT telemetry data."""

    def __init__(self, host, token, org, database):
        self.client = InfluxDBClient3(
            host=host,
            token=token,
            org=org,
            database=database
        )

    def write_batch(self, device_id, readings):
        """Write a batch of readings to InfluxDB."""
        points = []
        for r in readings:
            point = (
                f"telemetry,device_id={device_id} "
                f"temperature={r.get('temperature')},"
                f"humidity={r.get('humidity')},"
                f"pressure={r.get('pressure')} "
                f"{int(r.get('timestamp', datetime.now(timezone.utc)).timestamp() * 1e9)}"
            )
            points.append(point)

        try:
            self.client.write(record="\n".join(points), write_precision="ns")
        except Exception as e:
            print(f"Write failed: {e}")
            # Fall back to local buffer for retry

    def query_recent(self, device_id, hours=24):
        """Query last N hours of telemetry for a device."""
        query = f"""
        SELECT *
        FROM 'telemetry'
        WHERE device_id = '{device_id}'
          AND time >= now() - INTERVAL '{hours}' HOURS
        ORDER BY time DESC
        """
        return self.client.query(query=query, mode="pandas")

#!/usr/bin/env python3
"""IoT telemetry to vector embeddings for RAG pipelines."""

from kafka import KafkaConsumer
from sentence_transformers import SentenceTransformer
import pinecone

class IoTRAGPipeline:
    """Index IoT telemetry as vector embeddings for AI retrieval."""

    def __init__(self, kafka_broker, pinecone_index):
        self.consumer = KafkaConsumer(
            "iot.telemetry",
            bootstrap_servers=kafka_broker,
            value_deserializer=lambda m: json.loads(m.decode("utf-8"))
        )
        self.embedder = SentenceTransformer("all-MiniLM-L6-v2")
        self.vector_db = pinecone_index

    def process_stream(self):
        """Consume telemetry, embed, and index in vector DB."""
        for msg in self.consumer:
            telemetry = msg.value
            text = (
                f"Device {telemetry['device_id']}: "
                f"temperature={telemetry.get('temperature')}, "
                f"humidity={telemetry.get('humidity')}"
            )
            vector = self.embedder.encode(text).tolist()
            self.vector_db.upsert([
                (telemetry["device_id"], vector, telemetry)
            ])