IoT Dashboard & Visualization — Data Analytics Guide

version: '3.8'
services:
  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: dodatech
      DOCKER_INFLUXDB_INIT_BUCKET: iot_sensors
  
  grafana:
    image: grafana/grafana:10.2
    ports:
      - "3000:3000"
    depends_on:
      - influxdb
    environment:
      GF_SECURITY_ADMIN_PASSWORD: admin

from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
import time
import random

class IoTDataWriter:
    def __init__(self):
        self.client = InfluxDBClient(
            url="http://localhost:8086",
            token="my-token",
            org="dodatech"
        )
        self.write_api = self.client.write_api(write_options=SYNCHRONOUS)
    
    def write_sensor_reading(self, device_id, temperature, 
                             humidity, vibration):
        point = Point("sensor_reading") \
            .tag("device_id", device_id) \
            .tag("location", "factory_floor_a") \
            .field("temperature", temperature) \
            .field("humidity", humidity) \
            .field("vibration", vibration) \
            .time(time.time_ns())
        
        self.write_api.write(bucket="iot_sensors", record=point)
    
    def simulate_sensors(self, num_devices=5):
        devices = [f"sensor-{i:03d}" for i in range(num_devices)]
        
        while True:
            for device in devices:
                self.write_sensor_reading(
                    device_id=device,
                    temperature=22.0 + random.gauss(0, 3),
                    humidity=50.0 + random.gauss(0, 10),
                    vibration=random.uniform(0.1, 2.0)
                )
            time.sleep(10)

-- InfluxDB Flux query for Grafana
from(bucket: "iot_sensors")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "sensor_reading")
  |> filter(fn: (r) => r["_field"] == "temperature")
  |> filter(fn: (r) => r["device_id"] == "sensor-001")
  |> aggregateWindow(every: 1m, fn: mean)
  |> yield(name: "mean_temp")

# Grafana alert webhook handler
from flask import Flask, request, jsonify
import json

app = Flask(__name__)

@app.route('/grafana-alert', methods=['POST'])
def handle_alert():
    alert = request.json
    
    alert_name = alert.get('title', 'Unknown')
    state = alert.get('state', 'unknown')  # alerting, ok, pending
    message = alert.get('message', '')
    values = alert.get('evalMatches', [])
    
    print(f"[ALERT] {alert_name} -> {state}")
    for match in values:
        device = match.get('tags', {}).get('device_id', 'unknown')
        value = match.get('value', 0)
        print(f"  Device: {device}, Value: {value}")
    
    if state == 'alerting':
        # Send to PagerDuty, Slack, SMS, etc.
        send_slack_notification(alert_name, message)
        
        # Log to database
        log_alert_to_db(alert_name, state, values)
    
    return jsonify({"status": "received"}), 200

def send_slack_notification(title, message):
    webhook_url = "https://hooks.slack.com/services/YOUR/WEBHOOK"
    payload = {
        "text": f"*IoT Alert: {title}*\n{message}",
        "channel": "#iot-alerts"
    }
    import requests
    requests.post(webhook_url, json=payload)

import numpy as np
from sklearn.ensemble import IsolationForest
import pandas as pd

class IoTAnomalyDetector:
    def __init__(self, contamination=0.05):
        self.model = IsolationForest(
            contamination=contamination,
            random_state=42
        )
        self.is_fitted = False
    
    def fit(self, historical_data):
        """Train on normal sensor data."""
        features = historical_data[['temperature', 'humidity', 
                                     'vibration', 'pressure']]
        self.model.fit(features)
        self.is_fitted = True
        print(f"Model trained on {len(features)} samples")
    
    def predict(self, sensor_reading):
        if not self.is_fitted:
            return {'anomaly': False, 'reason': 'Model not trained'}
        
        features = np.array([[
            sensor_reading['temperature'],
            sensor_reading['humidity'],
            sensor_reading['vibration'],
            sensor_reading['pressure']
        ]])
        
        prediction = self.model.predict(features)
        score = self.model.score_samples(features)
        
        is_anomaly = prediction[0] == -1
        
        return {
            'anomaly': is_anomaly,
            'confidence': float(1 - abs(score[0])),
            'severity': 'high' if score[0] < -0.5 else 'medium' if is_anomaly else 'none'
        }

detector = IoTAnomalyDetector()
# detector.fit(historical_data)  # Train on 30+ days of data

reading = {'temperature': 45.0, 'humidity': 80, 
           'vibration': 5.2, 'pressure': 1020}
result = detector.predict(reading)
print(f"Anomaly: {result['anomaly']}, Severity: {result['severity']}")

import json
import requests

class GrafanaDashboardBuilder:
    def __init__(self, grafana_url, api_key):
        self.url = grafana_url
        self.headers = {
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json'
        }
    
    def create_dashboard(self, title, sensors):
        panels = []
        for i, sensor in enumerate(sensors):
            panel = {
                "type": "timeseries",
                "title": sensor['name'],
                "gridPos": {"x": i % 3 * 8, "y": i // 3 * 8, 
                           "w": 8, "h": 8},
                "targets": [{
                    "query": sensor['query'],
                    "refId": "A"
                }]
            }
            panels.append(panel)
        
        dashboard = {
            "dashboard": {
                "title": title,
                "panels": panels,
                "timezone": "browser",
                "schemaVersion": 36
            },
            "overwrite": True
        }
        
        response = requests.post(
            f"{self.url}/api/dashboards/db",
            headers=self.headers,
            json=dashboard
        )
        return response.json()

# Usage
builder = GrafanaDashboardBuilder(
    "http://localhost:3000",
    "glsa_api_key_here"
)
builder.create_dashboard("Factory Floor A", [
    {"name": "Temperature", "query": "temperature query"},
    {"name": "Vibration", "query": "vibration query"},
])