System Integration Guide

# AWS Infrastructure as Code (Terraform) provider "aws" { region = "us-west-2" } # IoT Core for device management resource "aws_iot_core" "iot_core" { name = "iot-edge-device-core" } # Kinesis Data Streams for real-time data ingestion resource "aws_kinesis_stream" "sensor_data_stream" { name = "sensor-data-stream" shard_count = 2 retention_period = 24 shard_level_metrics = [ "IncomingRecords", "OutgoingRecords" ] } # Lambda functions for data processing resource "aws_lambda_function" "data_processor" { filename = "data_processor.zip" function_name = "sensor-data-processor" role = aws_iam_role.lambda_role.arn handler = "index.handler" runtime = "python3.9" timeout = 60 environment { variables = { INFLUXDB_URL = aws_db_instance.influxdb.endpoint } } } # InfluxDB for time series data storage resource "aws_db_instance" "influxdb" { identifier = "influxdb-sensor-data" engine = "influxdb" engine_version = "2.7" instance_class = "db.t3.medium" allocated_storage = 100 storage_encrypted = true }

# Azure Resource Manager Template { "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#", "contentVersion": "1.0.0.0", "parameters": { "iotHubName": { "type": "string", "defaultValue": "iot-edge-device-hub" } }, "resources": [ { "type": "Microsoft.Devices/IotHubs", "apiVersion": "2021-07-02", "name": "[parameters('iotHubName')]", "location": "[resourceGroup().location]", "sku": { "name": "S1", "capacity": 1 }, "properties": { "routing": { "endpoints": { "serviceBusQueues": [], "serviceBusTopics": [], "eventHubs": [], "storageContainers": [] } } } }, { "type": "Microsoft.StreamAnalytics/streamingjobs", "apiVersion": "2021-10-01-preview", "name": "sensor-data-analytics", "location": "[resourceGroup().location]", "properties": { "sku": { "name": "Standard" }, "eventsOutOfOrderPolicy": "adjust", "outputErrorPolicy": "stop", "eventsOutOfOrderMaxDelayInSeconds": 0, "eventsLateArrivalMaxDelayInSeconds": 5 } } ] }

# Kafka Producer Configuration from kafka import KafkaProducer import json import time class SensorDataProducer: def __init__(self, bootstrap_servers=['localhost:9092']): self.producer = KafkaProducer( bootstrap_servers=bootstrap_servers, value_serializer=lambda v: json.dumps(v).encode('utf-8'), key_serializer=lambda k: k.encode('utf-8') if k else None, acks='all', retries=3, batch_size=16384, linger_ms=10 ) def send_sensor_data(self, device_id, sensor_data): topic = 'sensor-data' key = device_id message = { 'device_id': device_id, 'timestamp': int(time.time() * 1000), 'sensor_data': sensor_data, 'metadata': { 'firmware_version': '1.0.0', 'battery_level': sensor_data.get('battery_level', 100) } } future = self.producer.send(topic, key=key, value=message) return future.get(timeout=10) # Kafka Consumer for real-time processing from kafka import KafkaConsumer import json class SensorDataConsumer: def __init__(self, bootstrap_servers=['localhost:9092']): self.consumer = KafkaConsumer( 'sensor-data', bootstrap_servers=bootstrap_servers, value_deserializer=lambda m: json.loads(m.decode('utf-8')), group_id='sensor-data-processor', auto_offset_reset='latest', enable_auto_commit=True ) def process_messages(self): for message in self.consumer: sensor_data = message.value self.process_sensor_data(sensor_data) def process_sensor_data(self, data): # Real-time anomaly detection if self.detect_anomaly(data['sensor_data']): self.send_alert(data) # Store in time series database self.store_in_database(data) # Update real-time dashboard self.update_dashboard(data)

# Spark Streaming for real-time analytics from pyspark.sql import SparkSession from pyspark.sql.functions import * from pyspark.sql.types import * # Initialize Spark session spark = SparkSession.builder \ .appName("IoT Sensor Analytics") \ .config("spark.sql.adaptive.enabled", "true") \ .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \ .getOrCreate() # Define schema for sensor data sensor_schema = StructType([ StructField("device_id", StringType(), True), StructField("timestamp", LongType(), True), StructField("sensor_data", StructType([ StructField("imu", StructType([ StructField("accel_x", FloatType(), True), StructField("accel_y", FloatType(), True), StructField("accel_z", FloatType(), True), StructField("gyro_x", FloatType(), True), StructField("gyro_y", FloatType(), True), StructField("gyro_z", FloatType(), True) ]), True), StructField("environmental", StructType([ StructField("temperature", FloatType(), True), StructField("humidity", FloatType(), True), StructField("pressure", FloatType(), True), StructField("light_level", FloatType(), True) ]), True), StructField("audio", StructType([ StructField("level", FloatType(), True) ]), True) ]), True) ]) # Read from Kafka df = spark \ .readStream \ .format("kafka") \ .option("kafka.bootstrap.servers", "localhost:9092") \ .option("subscribe", "sensor-data") \ .load() # Parse JSON data parsed_df = df \ .select(from_json(col("value").cast("string"), sensor_schema).alias("data")) \ .select("data.*") # Real-time analytics analytics_df = parsed_df \ .withColumn("timestamp", from_unixtime(col("timestamp")/1000)) \ .withColumn("temperature_anomaly", when(col("sensor_data.environmental.temperature") > 50, 1).otherwise(0)) \ .withColumn("vibration_anomaly", when(sqrt( col("sensor_data.imu.accel_x")**2 + col("sensor_data.imu.accel_y")**2 + col("sensor_data.imu.accel_z")**2 ) > 20, 1).otherwise(0)) # Write to InfluxDB def write_to_influxdb(df, epoch_id): df.write \ .format("influxdb") \ .option("influxdb.database", "sensor_data") \ .option("influxdb.retentionPolicy", "autogen") \ .mode("append") \ .save() # Start streaming query query = analytics_df \ .writeStream \ .outputMode("append") \ .foreachBatch(write_to_influxdb) \ .trigger(processingTime='10 seconds') \ .start() query.awaitTermination()

# InfluxDB client for Python from influxdb_client import InfluxDBClient, Point, WritePrecision from influxdb_client.client.write_api import SYNCHRONOUS class InfluxDBManager: def __init__(self, url, token, org, bucket): self.client = InfluxDBClient(url=url, token=token, org=org) self.write_api = self.client.write_api(write_options=SYNCHRONOUS) self.query_api = self.client.query_api() self.bucket = bucket self.org = org def write_sensor_data(self, device_id, sensor_data, timestamp): point = Point("sensor_data") \ .tag("device_id", device_id) \ .field("temperature", sensor_data['environmental']['temperature']) \ .field("humidity", sensor_data['environmental']['humidity']) \ .field("pressure", sensor_data['environmental']['pressure']) \ .field("light_level", sensor_data['environmental']['light_level']) \ .field("accel_x", sensor_data['imu']['accel_x']) \ .field("accel_y", sensor_data['imu']['accel_y']) \ .field("accel_z", sensor_data['imu']['accel_z']) \ .field("gyro_x", sensor_data['imu']['gyro_x']) \ .field("gyro_y", sensor_data['imu']['gyro_y']) \ .field("gyro_z", sensor_data['imu']['gyro_z']) \ .field("audio_level", sensor_data['audio']['level']) \ .time(timestamp, WritePrecision.MS) self.write_api.write(bucket=self.bucket, org=self.org, record=point) def query_sensor_data(self, device_id, start_time, end_time): query = f''' from(bucket: "{self.bucket}") |> range(start: {start_time}, stop: {end_time}) |> filter(fn: (r) => r["_measurement"] == "sensor_data") |> filter(fn: (r) => r["device_id"] == "{device_id}") |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value") ''' result = self.query_api.query(org=self.org, query=query) return result # Usage example influxdb = InfluxDBManager( url="http://localhost:8086", token="your-token", org="your-org", bucket="sensor-data" ) # Write data sensor_data = { 'environmental': {'temperature': 22.5, 'humidity': 45.2, 'pressure': 1013.25, 'light_level': 100}, 'imu': {'accel_x': 0.1, 'accel_y': 0.2, 'accel_z': 9.8, 'gyro_x': 0.0, 'gyro_y': 0.0, 'gyro_z': 0.0}, 'audio': {'level': 45.8} } influxdb.write_sensor_data("device-001", sensor_data, int(time.time() * 1000))

# FastAPI application for IoT data API from fastapi import FastAPI, HTTPException, Depends, Query from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from pydantic import BaseModel from typing import List, Optional import jwt from datetime import datetime, timedelta import asyncio app = FastAPI(title="IoT Edge Device API", version="1.0.0") security = HTTPBearer() # Data models class SensorData(BaseModel): device_id: str timestamp: int sensor_data: dict anomaly_probability: Optional[float] = None class DeviceInfo(BaseModel): device_id: str status: str last_seen: datetime firmware_version: str battery_level: int signal_strength: int class AnalyticsResult(BaseModel): device_id: str start_time: datetime end_time: datetime metrics: dict anomalies: List[dict] # Authentication async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)): try: payload = jwt.decode(credentials.credentials, "secret-key", algorithms=["HS256"]) return payload except jwt.PyJWTError: raise HTTPException(status_code=401, detail="Invalid authentication token") # API endpoints @app.get("/devices", response_model=List[DeviceInfo]) async def get_devices( limit: int = Query(10, ge=1, le=100), offset: int = Query(0, ge=0), user = Depends(verify_token) ): """Get list of devices with pagination""" # Implementation to fetch devices from database pass @app.get("/devices/{device_id}/data", response_model=List[SensorData]) async def get_device_data( device_id: str, start_time: datetime = Query(...), end_time: datetime = Query(...), limit: int = Query(100, ge=1, le=1000), user = Depends(verify_token) ): """Get sensor data for a specific device""" # Implementation to fetch sensor data from time series database pass @app.get("/devices/{device_id}/analytics", response_model=AnalyticsResult) async def get_device_analytics( device_id: str, start_time: datetime = Query(...), end_time: datetime = Query(...), user = Depends(verify_token) ): """Get analytics results for a device""" # Implementation to calculate and return analytics pass @app.post("/devices/{device_id}/commands") async def send_device_command( device_id: str, command: dict, user = Depends(verify_token) ): """Send command to device""" # Implementation to send command via MQTT pass @app.get("/alerts") async def get_alerts( severity: Optional[str] = Query(None), start_time: Optional[datetime] = Query(None), end_time: Optional[datetime] = Query(None), limit: int = Query(50, ge=1, le=200), user = Depends(verify_token) ): """Get system alerts""" # Implementation to fetch alerts from database pass # WebSocket for real-time data @app.websocket("/ws/{device_id}") async def websocket_endpoint(websocket: WebSocket, device_id: str): await websocket.accept() try: while True: # Send real-time sensor data data = await get_latest_sensor_data(device_id) await websocket.send_json(data) await asyncio.sleep(1) # Send data every second except WebSocketDisconnect: pass if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

// React component for real-time sensor data visualization import React, { useState, useEffect } from 'react'; import { Line } from 'react-chartjs-2'; import { Card, CardContent, CardHeader, CardTitle } from './ui/card'; import { Badge } from './ui/badge'; const SensorDataChart = ({ deviceId }) => { const [sensorData, setSensorData] = useState([]); const [isConnected, setIsConnected] = useState(false); useEffect(() => { const ws = new WebSocket(`ws://localhost:8000/ws/${deviceId}`); ws.onopen = () => setIsConnected(true); ws.onclose = () => setIsConnected(false); ws.onmessage = (event) => { const data = JSON.parse(event.data); setSensorData(prev => [...prev.slice(-100), data]); // Keep last 100 points }; return () => ws.close(); }, [deviceId]); const chartData = { labels: sensorData.map(d => new Date(d.timestamp).toLocaleTimeString()), datasets: [ { label: 'Temperature (°C)', data: sensorData.map(d => d.sensor_data.environmental.temperature), borderColor: 'rgb(255, 99, 132)', backgroundColor: 'rgba(255, 99, 132, 0.2)', tension: 0.1 }, { label: 'Humidity (%)', data: sensorData.map(d => d.sensor_data.environmental.humidity), borderColor: 'rgb(54, 162, 235)', backgroundColor: 'rgba(54, 162, 235, 0.2)', tension: 0.1 }, { label: 'Pressure (hPa)', data: sensorData.map(d => d.sensor_data.environmental.pressure), borderColor: 'rgb(75, 192, 192)', backgroundColor: 'rgba(75, 192, 192, 0.2)', tension: 0.1 } ] }; const chartOptions = { responsive: true, plugins: { legend: { position: 'top', }, title: { display: true, text: 'Real-time Sensor Data' } }, scales: { x: { type: 'time', time: { displayFormats: { second: 'HH:mm:ss' } } } } }; return ( Device {deviceId} {isConnected ? "Connected" : "Disconnected"} ); }; // Device status overview component const DeviceOverview = () => { const [devices, setDevices] = useState([]); useEffect(() => { const fetchDevices = async () => { const response = await fetch('/api/devices'); const data = await response.json(); setDevices(data); }; fetchDevices(); const interval = setInterval(fetchDevices, 30000); // Update every 30 seconds return () => clearInterval(interval); }, []); return ( ); }; export { SensorDataChart, DeviceOverview };

# Dockerfile for IoT API service FROM python:3.9-slim WORKDIR /app # Install system dependencies RUN apt-get update && apt-get install -y \ gcc \ g++ \ && rm -rf /var/lib/apt/lists/* # Install Python dependencies COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # Copy application code COPY . . # Create non-root user RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app USER appuser # Expose port EXPOSE 8000 # Health check HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ CMD curl -f http://localhost:8000/health || exit 1 # Run application CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"] # Docker Compose for full stack version: '3.8' services: iot-api: build: . ports: - "8000:8000" environment: - INFLUXDB_URL=http://influxdb:8086 - KAFKA_BROKERS=kafka:9092 depends_on: - influxdb - kafka networks: - iot-network influxdb: image: influxdb:2.7 ports: - "8086:8086" environment: - DOCKER_INFLUXDB_INIT_MODE=setup - DOCKER_INFLUXDB_INIT_USERNAME=admin - DOCKER_INFLUXDB_INIT_PASSWORD=password123 - DOCKER_INFLUXDB_INIT_ORG=iot-org - DOCKER_INFLUXDB_INIT_BUCKET=sensor-data volumes: - influxdb_data:/var/lib/influxdb2 networks: - iot-network kafka: image: confluentinc/cp-kafka:latest ports: - "9092:9092" environment: KAFKA_BROKER_ID: 1 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 depends_on: - zookeeper networks: - iot-network zookeeper: image: confluentinc/cp-zookeeper:latest ports: - "2181:2181" environment: ZOOKEEPER_CLIENT_PORT: 2181 ZOOKEEPER_TICK_TIME: 2000 networks: - iot-network dashboard: build: ./dashboard ports: - "3000:3000" environment: - REACT_APP_API_URL=http://localhost:8000 depends_on: - iot-api networks: - iot-network volumes: influxdb_data: networks: iot-network: driver: bridge

# Kubernetes deployment for IoT API apiVersion: apps/v1 kind: Deployment metadata: name: iot-api labels: app: iot-api spec: replicas: 3 selector: matchLabels: app: iot-api template: metadata: labels: app: iot-api spec: containers: - name: iot-api image: iot-api:latest ports: - containerPort: 8000 env: - name: INFLUXDB_URL value: "http://influxdb-service:8086" - name: KAFKA_BROKERS value: "kafka-service:9092" resources: requests: memory: "256Mi" cpu: "250m" limits: memory: "512Mi" cpu: "500m" livenessProbe: httpGet: path: /health port: 8000 initialDelaySeconds: 30 periodSeconds: 10 readinessProbe: httpGet: path: /ready port: 8000 initialDelaySeconds: 5 periodSeconds: 5 --- apiVersion: v1 kind: Service metadata: name: iot-api-service spec: selector: app: iot-api ports: - protocol: TCP port: 80 targetPort: 8000 type: LoadBalancer --- apiVersion: autoscaling/v2 kind: HorizontalPodAutoscaler metadata: name: iot-api-hpa spec: scaleTargetRef: apiVersion: apps/v1 kind: Deployment name: iot-api minReplicas: 3 maxReplicas: 10 metrics: - type: Resource resource: name: cpu target: type: Utilization averageUtilization: 70 - type: Resource resource: name: memory target: type: Utilization averageUtilization: 80

# API security middleware from fastapi import FastAPI, HTTPException, Depends from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.trustedhost import TrustedHostMiddleware import jwt import redis from functools import wraps import time app = FastAPI() # Security middleware app.add_middleware( CORSMiddleware, allow_origins=["https://dashboard.example.com"], allow_credentials=True, allow_methods=["GET", "POST"], allow_headers=["*"], ) app.add_middleware( TrustedHostMiddleware, allowed_hosts=["api.example.com", "*.example.com"] ) # Rate limiting with Redis redis_client = redis.Redis(host='localhost', port=6379, db=0) def rate_limit(max_requests: int = 100, window_seconds: int = 3600): def decorator(func): @wraps(func) async def wrapper(*args, **kwargs): client_ip = kwargs.get('client_ip', 'unknown') key = f"rate_limit:{client_ip}" current_requests = redis_client.get(key) if current_requests is None: redis_client.setex(key, window_seconds, 1) else: if int(current_requests) >= max_requests: raise HTTPException(status_code=429, detail="Rate limit exceeded") redis_client.incr(key) return await func(*args, **kwargs) return wrapper return decorator # JWT authentication def create_access_token(data: dict, expires_delta: timedelta = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, "secret-key", algorithm="HS256") return encoded_jwt def verify_token(credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer())): try: payload = jwt.decode(credentials.credentials, "secret-key", algorithms=["HS256"]) username: str = payload.get("sub") if username is None: raise HTTPException(status_code=401, detail="Invalid authentication credentials") return payload except jwt.PyJWTError: raise HTTPException(status_code=401, detail="Invalid authentication credentials") # Secure API endpoint @app.get("/secure-data") @rate_limit(max_requests=1000, window_seconds=3600) async def get_secure_data( user = Depends(verify_token), client_ip: str = None ): """Secure endpoint with authentication and rate limiting""" return {"message": "Secure data", "user": user.get("sub")}

# Prometheus metrics for IoT API from prometheus_client import Counter, Histogram, Gauge, generate_latest import time # Metrics REQUEST_COUNT = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint']) REQUEST_LATENCY = Histogram('http_request_duration_seconds', 'HTTP request latency') ACTIVE_CONNECTIONS = Gauge('active_websocket_connections', 'Active WebSocket connections') SENSOR_DATA_RECEIVED = Counter('sensor_data_received_total', 'Total sensor data received', ['device_id']) # FastAPI middleware for metrics @app.middleware("http") async def add_prometheus_metrics(request, call_next): start_time = time.time() response = await call_next(request) # Record metrics REQUEST_COUNT.labels( method=request.method, endpoint=request.url.path ).inc() REQUEST_LATENCY.observe(time.time() - start_time) return response # Metrics endpoint @app.get("/metrics") async def get_metrics(): return Response(generate_latest(), media_type="text/plain") # Grafana dashboard configuration dashboard_config = { "dashboard": { "title": "IoT System Monitoring", "panels": [ { "title": "Request Rate", "type": "graph", "targets": [ { "expr": "rate(http_requests_total[5m])", "legendFormat": "{{method}} {{endpoint}}" } ] }, { "title": "Response Time", "type": "graph", "targets": [ { "expr": "histogram_quantile(0.95, http_request_duration_seconds)", "legendFormat": "95th percentile" } ] }, { "title": "Active Connections", "type": "singlestat", "targets": [ { "expr": "active_websocket_connections", "legendFormat": "Active Connections" } ] }, { "title": "Sensor Data Rate", "type": "graph", "targets": [ { "expr": "rate(sensor_data_received_total[5m])", "legendFormat": "{{device_id}}" } ] } ] } }