IoT Software Development: Architecture, Protocols, and Cloud Integration

IoT software is where embedded systems, cloud infrastructure, data engineering, and user-facing applications converge. Getting any layer wrong creates problems that are expensive to fix: firmware bugs require physical device updates, protocol mismatches block data flow, and poorly designed data pipelines become bottlenecks at scale.

This guide covers the complete IoT stack: device communication protocols, cloud message brokers, time-series data pipelines, edge computing patterns, and the architecture decisions that scale from 100 to 100,000 devices.

---

The IoT Stack

┌────────────────────────────────────────────────────┐
│  Devices (sensors, actuators, gateways)            │
│  Firmware: C/C++, MicroPython, Rust                │
├────────────────────────────────────────────────────┤
│  Edge Layer (optional)                             │
│  Local processing, protocol translation, buffering │
│  Runtime: Raspberry Pi, NVIDIA Jetson, AWS Greengrass│
├────────────────────────────────────────────────────┤
│  Transport Layer                                   │
│  MQTT, HTTPS, CoAP, WebSocket                      │
├────────────────────────────────────────────────────┤
│  Cloud Message Broker                              │
│  AWS IoT Core, Azure IoT Hub, EMQX, Mosquitto      │
├────────────────────────────────────────────────────┤
│  Data Pipeline                                     │
│  Kinesis/Kafka → Lambda/Flink → Time-series DB     │
├────────────────────────────────────────────────────┤
│  Storage                                           │
│  InfluxDB, TimescaleDB, AWS Timestream, S3          │
├────────────────────────────────────────────────────┤
│  Applications                                      │
│  Dashboards, alerts, ML inference, APIs            │
└────────────────────────────────────────────────────┘

---

Protocol Selection

MQTT (Message Queuing Telemetry Transport)

The dominant IoT protocol. Lightweight, publish-subscribe, designed for constrained networks.

Key properties:

• Binary protocol — very small packet overhead (~2 bytes minimum header)
• QoS levels: 0 (fire and forget), 1 (at least once), 2 (exactly once)
• Persistent sessions — broker queues messages when device is offline
• Topics: hierarchical strings (factory/line-a/machine-3/temperature)
• Wildcard subscriptions: factory/+/machine-3/# (+ = single level, # = multi-level)

Use MQTT when: Battery-powered devices, constrained bandwidth, reliable delivery needed, fan-out messaging (one message → many subscribers)

HTTP/HTTPS

Request-response. Simple to implement, works through standard proxies.

Use HTTP when: Devices have reliable connectivity, payload sizes are large, request-response interaction pattern, REST APIs are easier for the team to implement

CoAP (Constrained Application Protocol)

UDP-based, designed for very constrained devices (microcontrollers with <10KB RAM).

Use CoAP when: Devices are extremely resource-constrained, UDP is acceptable, multicast discovery needed

Protocol Comparison

---

MQTT Implementation

Device (Python/MicroPython)

import paho.mqtt.client as mqtt
import json
import time
import ssl

BROKER = "your-iot-endpoint.iot.us-east-1.amazonaws.com"
PORT = 8883
DEVICE_ID = "sensor-001"

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("Connected to broker")
        # Subscribe to device commands on connect
        client.subscribe(f"devices/{DEVICE_ID}/commands", qos=1)
    else:
        print(f"Connection failed: {rc}")

def on_message(client, userdata, msg):
    """Handle incoming commands from cloud."""
    command = json.loads(msg.payload)
    if command.get('action') == 'restart':
        restart_device()
    elif command.get('action') == 'set_interval':
        update_reading_interval(command['interval_seconds'])

def on_disconnect(client, userdata, rc):
    if rc != 0:
        print(f"Unexpected disconnect: {rc}. Will retry...")

client = mqtt.Client(client_id=DEVICE_ID, clean_session=False)
client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect

# TLS mutual authentication (required for AWS IoT Core)
client.tls_set(
    ca_certs="/certs/AmazonRootCA1.pem",
    certfile=f"/certs/{DEVICE_ID}-certificate.pem.crt",
    keyfile=f"/certs/{DEVICE_ID}-private.pem.key",
    tls_version=ssl.PROTOCOL_TLSv1_2,
)

client.connect(BROKER, PORT, keepalive=60)
client.loop_start()

def publish_reading(temperature: float, humidity: float):
    payload = json.dumps({
        "deviceId": DEVICE_ID,
        "timestamp": time.time(),
        "temperature": round(temperature, 2),
        "humidity": round(humidity, 2),
        "unit": "celsius",
    })
    
    result = client.publish(
        topic=f"sensors/{DEVICE_ID}/telemetry",
        payload=payload,
        qos=1,  # At least once delivery
        retain=False,
    )
    
    if result.rc != mqtt.MQTT_ERR_SUCCESS:
        print(f"Publish failed: {result.rc}")
        # Buffer locally and retry on reconnect

# Main loop
while True:
    temp, humidity = read_sensors()
    publish_reading(temp, humidity)
    time.sleep(30)  # Publish every 30 seconds

Cloud: AWS IoT Core with Rules Engine

# Terraform: AWS IoT Core setup
resource "aws_iot_topic_rule" "sensor_data" {
  name        = "sensor_telemetry_to_kinesis"
  enabled     = true
  sql         = "SELECT * FROM 'sensors/+/telemetry'"
  sql_version = "2016-03-23"

  # Route to Kinesis for stream processing
  kinesis {
    role_arn    = aws_iam_role.iot_kinesis.arn
    stream_name = aws_kinesis_stream.iot_data.name
    partition_key = "deviceId"
  }

  # Also store raw messages in S3 for replay
  s3 {
    role_arn    = aws_iam_role.iot_s3.arn
    bucket_name = aws_s3_bucket.iot_raw.id
    key         = "raw/\${topic()}/\${timestamp()}.json"
  }

  # Alert on anomalies via SNS
  error_action {
    sns {
      role_arn   = aws_iam_role.iot_sns.arn
      target_arn = aws_sns_topic.iot_errors.arn
    }
  }
}

resource "aws_kinesis_stream" "iot_data" {
  name             = "iot-sensor-data"
  shard_count      = 4    # 4 MB/s ingest, 4000 records/s
  retention_period = 24   # Hours

  stream_mode_details {
    stream_mode = "PROVISIONED"
  }
}

---

Time-Series Data Pipeline

IoT data is fundamentally time-series: each reading has a timestamp and one or more values. Standard relational databases handle this poorly at scale.

TimescaleDB (PostgreSQL Extension)

-- TimescaleDB: PostgreSQL with time-series superpowers
-- Regular PostgreSQL table creation
CREATE TABLE sensor_readings (
  time        TIMESTAMPTZ NOT NULL,
  device_id   TEXT NOT NULL,
  metric      TEXT NOT NULL,
  value       DOUBLE PRECISION NOT NULL,
  unit        TEXT
);

-- Convert to hypertable (automatically partitions by time)
SELECT create_hypertable('sensor_readings', 'time', chunk_time_interval => INTERVAL '1 day');

-- Compression (90% storage reduction for older data)
ALTER TABLE sensor_readings SET (
  timescaledb.compress,
  timescaledb.compress_orderby = 'time DESC',
  timescaledb.compress_segmentby = 'device_id'
);

-- Auto-compress chunks older than 7 days
SELECT add_compression_policy('sensor_readings', INTERVAL '7 days');

-- Efficient time-range queries (uses time-based partitioning)
SELECT
  time_bucket('5 minutes', time) AS bucket,
  device_id,
  metric,
  AVG(value) AS avg_value,
  MIN(value) AS min_value,
  MAX(value) AS max_value
FROM sensor_readings
WHERE device_id = 'sensor-001'
  AND metric = 'temperature'
  AND time >= NOW() - INTERVAL '24 hours'
GROUP BY bucket, device_id, metric
ORDER BY bucket DESC;

InfluxDB 3.0 (Purpose-Built)

from influxdb_client_3 import InfluxDBClient3
from datetime import datetime, timezone

client = InfluxDBClient3(
    host="https://us-east-1-1.aws.cloud2.influxdata.com",
    token=INFLUX_TOKEN,
    database="iot-sensors"
)

# Write telemetry data
def write_telemetry(device_id: str, readings: dict):
    record = {
        "measurement": "sensor_telemetry",
        "tags": {
            "device_id": device_id,
            "location": readings.get("location", "unknown"),
        },
        "fields": {
            "temperature": readings["temperature"],
            "humidity": readings["humidity"],
            "battery_level": readings.get("battery", 100.0),
        },
        "time": datetime.now(timezone.utc),
    }
    client.write(record=record)

# Query with SQL (InfluxDB 3.0 supports SQL)
df = client.query("""
    SELECT
        DATE_TRUNC('minute', time) AS minute,
        device_id,
        AVG(temperature) AS avg_temp,
        MAX(temperature) AS max_temp
    FROM sensor_telemetry
    WHERE device_id = 'sensor-001'
      AND time >= NOW() - INTERVAL '1 hour'
    GROUP BY 1, 2
    ORDER BY 1 DESC
""", language="sql")

---

Edge Computing

Edge computing runs processing on local hardware (gateway or edge server) instead of sending everything to the cloud. Benefits: reduced latency, lower bandwidth costs, operation during connectivity loss.

# Edge inference example: anomaly detection runs locally
# Only sends alerts to cloud, not raw data stream

import numpy as np
from collections import deque

class EdgeAnomalyDetector:
    def __init__(self, window_size: int = 60, threshold_std: float = 3.0):
        self.window = deque(maxlen=window_size)
        self.threshold = threshold_std
        self.cloud_client = MQTTClient()

    def process_reading(self, device_id: str, value: float, timestamp: float):
        self.window.append(value)
        
        if len(self.window) < 10:
            return  # Not enough data yet
        
        mean = np.mean(self.window)
        std = np.std(self.window)
        z_score = abs(value - mean) / (std + 1e-8)
        
        if z_score > self.threshold:
            # Anomaly detected — send alert to cloud immediately
            self.cloud_client.publish(
                f"alerts/{device_id}/anomaly",
                payload={
                    "deviceId": device_id,
                    "value": value,
                    "mean": mean,
                    "zScore": z_score,
                    "timestamp": timestamp,
                },
                qos=1,
            )
        
        # Only send raw data every 5 minutes (not every reading)
        # This reduces bandwidth by 90% vs streaming everything
        if timestamp % 300 < 1:
            self.cloud_client.publish(
                f"sensors/{device_id}/telemetry",
                payload={"value": value, "timestamp": timestamp},
                qos=0,  # Best effort for non-critical data
            )

---

Device Management

// Device shadow: desired state vs. reported state pattern
interface DeviceShadow {
  desired: {
    reportingIntervalSeconds: number;
    alertThreshold: number;
    firmwareVersion: string;
  };
  reported: {
    reportingIntervalSeconds: number;
    alertThreshold: number;
    firmwareVersion: string;
    batteryLevel: number;
    lastSeen: string;
  };
  delta?: {
    // Present when desired !== reported
    reportingIntervalSeconds?: number;
    firmwareVersion?: string;
  };
}

// Update desired state — device applies on next connection
async function updateDeviceConfig(deviceId: string, config: Partial<DeviceShadow['desired']>) {
  await iotClient.send(new UpdateThingShadowCommand({
    thingName: deviceId,
    payload: JSON.stringify({
      state: { desired: config },
    }),
  }));
}

---

IoT Project Cost Ranges (2026)

Monthly cloud costs: ~$0.08–$0.50 per device per month for AWS IoT Core + data storage at moderate telemetry rates.

---

Working With Viprasol

We build IoT software stacks — from device firmware guidance through cloud message brokers, time-series data pipelines, edge computing, and real-time dashboards.

→ IoT project discussion →
→ Software Development Services →
→ AI & Machine Learning →

---

See Also

• Manufacturing Software Development
• Computer Vision Development
• Real-Time Application Development
• Cloud Solutions →

---

Sources

• AWS IoT Core Documentation
• MQTT Specification 5.0
• TimescaleDB Documentation
• InfluxDB 3.0 Documentation