From Sensors to Sculptures: MQTT + TimescaleDB + WebSockets + Three.js¶

# iot_simulator.py
import paho.mqtt.client as mqtt
import json
import time
import random
import asyncio
from datetime import datetime

class IoTSensorSimulator:
    def __init__(self, device_id, broker_host="localhost", broker_port=1883):
        self.device_id = device_id
        self.broker_host = broker_host
        self.broker_port = broker_port
        self.client = mqtt.Client()
        self.running = False

    def connect_mqtt(self):
        """Connect to MQTT broker"""
        try:
            self.client.connect(self.broker_host, self.broker_port, 60)
            print(f"Device {self.device_id} connected to MQTT broker")
            return True
        except Exception as e:
            print(f"Failed to connect to MQTT broker: {e}")
            return False

    def generate_sensor_data(self):
        """Generate realistic sensor data"""
        base_temp = 20.0
        temp_variation = random.uniform(-5, 5)
        temp = base_temp + temp_variation + random.uniform(-1, 1)

        return {
            "device": self.device_id,
            "timestamp": int(time.time()),
            "datetime": datetime.now().isoformat(),
            "temperature": round(temp, 2),
            "humidity": round(random.uniform(30, 80), 2),
            "vibration": round(random.uniform(0.1, 2.0), 2),
            "pressure": round(random.uniform(1000, 1020), 2),
            "light": round(random.uniform(0, 1000), 2),
            "location": {
                "lat": round(random.uniform(40.7, 40.8), 6),
                "lon": round(random.uniform(-74.0, -73.9), 6)
            }
        }

    def publish_sensor_data(self, topic, data):
        """Publish sensor data to MQTT topic"""
        try:
            message = json.dumps(data)
            result = self.client.publish(topic, message)
            if result.rc == mqtt.MQTT_ERR_SUCCESS:
                print(f"Published to {topic}: {data['temperature']}°C")
            else:
                print(f"Failed to publish to {topic}")
        except Exception as e:
            print(f"Error publishing data: {e}")

    def run_sensor(self, interval=2):
        """Run sensor simulation"""
        if not self.connect_mqtt():
            return

        self.running = True
        self.client.loop_start()

        try:
            while self.running:
                # Generate and publish temperature data
                temp_data = self.generate_sensor_data()
                self.publish_sensor_data(f"sensors/{self.device_id}/temperature", temp_data)

                # Generate and publish humidity data
                humidity_data = self.generate_sensor_data()
                self.publish_sensor_data(f"sensors/{self.device_id}/humidity", humidity_data)

                # Generate and publish vibration data
                vibration_data = self.generate_sensor_data()
                self.publish_sensor_data(f"sensors/{self.device_id}/vibration", vibration_data)

                time.sleep(interval)

        except KeyboardInterrupt:
            print(f"Sensor {self.device_id} stopped")
        finally:
            self.running = False
            self.client.loop_stop()
            self.client.disconnect()

# Run multiple sensors
def run_sensor_farm():
    """Run multiple IoT sensors"""
    sensors = [
        IoTSensorSimulator("device_001"),
        IoTSensorSimulator("device_002"),
        IoTSensorSimulator("device_003")
    ]

    for sensor in sensors:
        sensor.run_sensor(interval=1)

if __name__ == "__main__":
    run_sensor_farm()

# mqtt_ingest.py
import paho.mqtt.client as mqtt
import psycopg2
import json
import time
from datetime import datetime

class MQTTTimescaleDBIngestor:
    def __init__(self, mqtt_broker="localhost", mqtt_port=1883, 
                 db_host="localhost", db_user="postgres", 
                 db_password="secret", db_name="iot_data"):
        self.mqtt_broker = mqtt_broker
        self.mqtt_port = mqtt_port
        self.db_host = db_host
        self.db_user = db_user
        self.db_password = db_password
        self.db_name = db_name

        self.mqtt_client = mqtt.Client()
        self.db_conn = None
        self.db_cursor = None

    def connect_database(self):
        """Connect to TimescaleDB"""
        try:
            self.db_conn = psycopg2.connect(
                host=self.db_host,
                database=self.db_name,
                user=self.db_user,
                password=self.db_password
            )
            self.db_cursor = self.db_conn.cursor()
            print("Connected to TimescaleDB")
            return True
        except Exception as e:
            print(f"Failed to connect to database: {e}")
            return False

    def setup_database(self):
        """Setup TimescaleDB hypertables"""
        try:
            # Create sensor data table
            self.db_cursor.execute("""
                CREATE TABLE IF NOT EXISTS sensor_data (
                    device TEXT,
                    ts TIMESTAMPTZ,
                    temperature DOUBLE PRECISION,
                    humidity DOUBLE PRECISION,
                    vibration DOUBLE PRECISION,
                    pressure DOUBLE PRECISION,
                    light DOUBLE PRECISION,
                    location_lat DOUBLE PRECISION,
                    location_lon DOUBLE PRECISION,
                    PRIMARY KEY (device, ts)
                );
            """)

            # Create hypertable
            self.db_cursor.execute("""
                SELECT create_hypertable('sensor_data', 'ts', if_not_exists => TRUE);
            """)

            # Create indexes for performance
            self.db_cursor.execute("""
                CREATE INDEX IF NOT EXISTS idx_sensor_data_device_ts 
                ON sensor_data (device, ts DESC);
            """)

            self.db_cursor.execute("""
                CREATE INDEX IF NOT EXISTS idx_sensor_data_ts 
                ON sensor_data (ts DESC);
            """)

            self.db_conn.commit()
            print("Database setup completed")
            return True

        except Exception as e:
            print(f"Database setup failed: {e}")
            return False

    def on_mqtt_connect(self, client, userdata, flags, rc):
        """MQTT connection callback"""
        if rc == 0:
            print("Connected to MQTT broker")
            # Subscribe to all sensor topics
            client.subscribe("sensors/+/temperature")
            client.subscribe("sensors/+/humidity")
            client.subscribe("sensors/+/vibration")
            client.subscribe("sensors/+/pressure")
            client.subscribe("sensors/+/light")
        else:
            print(f"Failed to connect to MQTT broker: {rc}")

    def on_mqtt_message(self, client, userdata, msg):
        """MQTT message callback"""
        try:
            data = json.loads(msg.payload.decode())
            topic_parts = msg.topic.split('/')
            sensor_type = topic_parts[-1]
            device = topic_parts[-2]

            # Insert data into TimescaleDB
            self.insert_sensor_data(device, sensor_type, data)

        except Exception as e:
            print(f"Error processing message: {e}")

    def insert_sensor_data(self, device, sensor_type, data):
        """Insert sensor data into TimescaleDB"""
        try:
            timestamp = datetime.fromtimestamp(data.get('timestamp', time.time()))

            # Prepare data for insertion
            insert_data = {
                'device': device,
                'ts': timestamp,
                'temperature': data.get('temperature'),
                'humidity': data.get('humidity'),
                'vibration': data.get('vibration'),
                'pressure': data.get('pressure'),
                'light': data.get('light'),
                'location_lat': data.get('location', {}).get('lat'),
                'location_lon': data.get('location', {}).get('lon')
            }

            # Insert with conflict resolution
            self.db_cursor.execute("""
                INSERT INTO sensor_data (
                    device, ts, temperature, humidity, vibration, 
                    pressure, light, location_lat, location_lon
                ) VALUES (
                    %(device)s, %(ts)s, %(temperature)s, %(humidity)s, 
                    %(vibration)s, %(pressure)s, %(light)s, 
                    %(location_lat)s, %(location_lon)s
                ) ON CONFLICT (device, ts) DO UPDATE SET
                    temperature = EXCLUDED.temperature,
                    humidity = EXCLUDED.humidity,
                    vibration = EXCLUDED.vibration,
                    pressure = EXCLUDED.pressure,
                    light = EXCLUDED.light,
                    location_lat = EXCLUDED.location_lat,
                    location_lon = EXCLUDED.location_lon
            """, insert_data)

            self.db_conn.commit()
            print(f"Inserted {sensor_type} data for {device}: {data.get(sensor_type)}")

        except Exception as e:
            print(f"Error inserting sensor data: {e}")
            self.db_conn.rollback()

    def connect_mqtt(self):
        """Connect to MQTT broker"""
        try:
            self.mqtt_client.on_connect = self.on_mqtt_connect
            self.mqtt_client.on_message = self.on_mqtt_message
            self.mqtt_client.connect(self.mqtt_broker, self.mqtt_port, 60)
            print("Connected to MQTT broker")
            return True
        except Exception as e:
            print(f"Failed to connect to MQTT broker: {e}")
            return False

    def run(self):
        """Run the ingestor"""
        if not self.connect_database():
            return

        if not self.setup_database():
            return

        if not self.connect_mqtt():
            return

        try:
            self.mqtt_client.loop_forever()
        except KeyboardInterrupt:
            print("Ingestor stopped")
        finally:
            if self.db_cursor:
                self.db_cursor.close()
            if self.db_conn:
                self.db_conn.close()
            self.mqtt_client.disconnect()

if __name__ == "__main__":
    ingestor = MQTTTimescaleDBIngestor()
    ingestor.run()

# websocket_server.py
import asyncio
import websockets
import psycopg2
import json
import time
from datetime import datetime, timedelta

class WebSocketDataStreamer:
    def __init__(self, db_host="localhost", db_user="postgres", 
                 db_password="secret", db_name="iot_data"):
        self.db_host = db_host
        self.db_user = db_user
        self.db_password = db_password
        self.db_name = db_name
        self.connected_clients = set()

    def connect_database(self):
        """Connect to TimescaleDB"""
        try:
            self.db_conn = psycopg2.connect(
                host=self.db_host,
                database=self.db_name,
                user=self.db_user,
                password=self.db_password
            )
            self.db_cursor = self.db_conn.cursor()
            return True
        except Exception as e:
            print(f"Database connection failed: {e}")
            return False

    def get_latest_sensor_data(self):
        """Get latest sensor data from TimescaleDB"""
        try:
            self.db_cursor.execute("""
                SELECT device, ts, temperature, humidity, vibration, 
                       pressure, light, location_lat, location_lon
                FROM sensor_data 
                WHERE ts > NOW() - INTERVAL '1 minute'
                ORDER BY ts DESC
                LIMIT 100
            """)

            rows = self.db_cursor.fetchall()
            return [
                {
                    "device": row[0],
                    "timestamp": row[1].timestamp(),
                    "temperature": row[2],
                    "humidity": row[3],
                    "vibration": row[4],
                    "pressure": row[5],
                    "light": row[6],
                    "location": {
                        "lat": row[7],
                        "lon": row[8]
                    }
                }
                for row in rows
            ]
        except Exception as e:
            print(f"Error querying sensor data: {e}")
            return []

    def get_aggregated_data(self):
        """Get aggregated sensor data"""
        try:
            self.db_cursor.execute("""
                SELECT 
                    device,
                    AVG(temperature) as avg_temp,
                    MAX(temperature) as max_temp,
                    MIN(temperature) as min_temp,
                    AVG(humidity) as avg_humidity,
                    AVG(vibration) as avg_vibration,
                    COUNT(*) as data_points
                FROM sensor_data 
                WHERE ts > NOW() - INTERVAL '5 minutes'
                GROUP BY device
                ORDER BY device
            """)

            rows = self.db_cursor.fetchall()
            return [
                {
                    "device": row[0],
                    "avg_temperature": float(row[1]) if row[1] else None,
                    "max_temperature": float(row[2]) if row[2] else None,
                    "min_temperature": float(row[3]) if row[3] else None,
                    "avg_humidity": float(row[4]) if row[4] else None,
                    "avg_vibration": float(row[5]) if row[5] else None,
                    "data_points": row[6]
                }
                for row in rows
            ]
        except Exception as e:
            print(f"Error querying aggregated data: {e}")
            return []

    async def handle_client(self, websocket, path):
        """Handle WebSocket client connection"""
        self.connected_clients.add(websocket)
        print(f"Client connected. Total clients: {len(self.connected_clients)}")

        try:
            while True:
                # Get latest sensor data
                sensor_data = self.get_latest_sensor_data()
                aggregated_data = self.get_aggregated_data()

                # Send data to client
                message = {
                    "type": "sensor_data",
                    "timestamp": time.time(),
                    "sensor_data": sensor_data,
                    "aggregated_data": aggregated_data
                }

                await websocket.send(json.dumps(message))
                await asyncio.sleep(1)  # Send updates every second

        except websockets.exceptions.ConnectionClosed:
            print("Client disconnected")
        finally:
            self.connected_clients.remove(websocket)
            print(f"Client removed. Total clients: {len(self.connected_clients)}")

    async def broadcast_to_all(self, message):
        """Broadcast message to all connected clients"""
        if self.connected_clients:
            await asyncio.gather(
                *[client.send(message) for client in self.connected_clients],
                return_exceptions=True
            )

    async def run_server(self, host="0.0.0.0", port=8765):
        """Run WebSocket server"""
        if not self.connect_database():
            return

        print(f"Starting WebSocket server on {host}:{port}")

        async with websockets.serve(self.handle_client, host, port):
            await asyncio.Future()  # Run forever

if __name__ == "__main__":
    streamer = WebSocketDataStreamer()
    asyncio.run(streamer.run_server())

# real_time_analytics.py
class RealTimeAnalytics:
    def __init__(self, db_conn):
        self.db_conn = db_conn
        self.analytics_cache = {}

    def calculate_trends(self, device, time_window=300):
        """Calculate trends for a device"""
        try:
            cursor = self.db_conn.cursor()
            cursor.execute("""
                SELECT 
                    AVG(temperature) as avg_temp,
                    STDDEV(temperature) as temp_std,
                    AVG(humidity) as avg_humidity,
                    AVG(vibration) as avg_vibration
                FROM sensor_data 
                WHERE device = %s AND ts > NOW() - INTERVAL '%s seconds'
            """, (device, time_window))

            result = cursor.fetchone()
            if result:
                return {
                    "avg_temperature": float(result[0]) if result[0] else 0,
                    "temp_volatility": float(result[1]) if result[1] else 0,
                    "avg_humidity": float(result[2]) if result[2] else 0,
                    "avg_vibration": float(result[3]) if result[3] else 0
                }
        except Exception as e:
            print(f"Error calculating trends: {e}")
        return None

    def detect_anomalies(self, device, threshold=2.0):
        """Detect anomalies in sensor data"""
        try:
            cursor = self.db_conn.cursor()
            cursor.execute("""
                SELECT temperature, humidity, vibration
                FROM sensor_data 
                WHERE device = %s 
                ORDER BY ts DESC 
                LIMIT 10
            """, (device,))

            recent_data = cursor.fetchall()
            if len(recent_data) < 5:
                return None

            # Simple anomaly detection
            temps = [row[0] for row in recent_data if row[0]]
            if temps:
                avg_temp = sum(temps) / len(temps)
                current_temp = temps[0]

                if abs(current_temp - avg_temp) > threshold:
                    return {
                        "anomaly": True,
                        "type": "temperature",
                        "value": current_temp,
                        "expected": avg_temp
                    }

        except Exception as e:
            print(f"Error detecting anomalies: {e}")
        return None

# error_handler.py
class IoTErrorHandler:
    def __init__(self):
        self.retry_count = 3
        self.retry_delay = 1

    def handle_mqtt_errors(self, func):
        """Handle MQTT-specific errors"""
        def wrapper(*args, **kwargs):
            for attempt in range(self.retry_count):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"MQTT error (attempt {attempt + 1}): {e}")
                    if attempt < self.retry_count - 1:
                        time.sleep(self.retry_delay * (2 ** attempt))
                    else:
                        print(f"MQTT operation failed after {self.retry_count} attempts")
                        return None
        return wrapper

    def handle_database_errors(self, func):
        """Handle database-specific errors"""
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except psycopg2.OperationalError:
                print("Database connection lost, attempting to reconnect...")
                return None
            except psycopg2.IntegrityError:
                print("Database integrity error, skipping duplicate data")
                return None
            except Exception as e:
                print(f"Database error: {e}")
                return None
        return wrapper

# Essential MQTT + TimescaleDB + WebSockets + Three.js patterns
iot_visualization_patterns = {
    "mqtt_messaging": "Mosquitto is ideal for lightweight IoT messaging with backpressure",
    "timescale_storage": "TimescaleDB compresses and indexes sensor history (hypertables)",
    "websocket_streaming": "WebSockets give push semantics vs polling",
    "threejs_rendering": "Three.js teaches how to turn numbers into real-time geometry",
    "real_time_art": "Together, they force developers to think infra + DB + API + visualization as one continuous flow",
    "performance": "Optimize rendering, handle errors gracefully, cache data efficiently"
}

# Essential MQTT + TimescaleDB + WebSockets + Three.js operations
# 1. Publish sensor data
mosquitto_pub -h localhost -t sensors/device_001/temperature -m '{"device":"device_001","temp":22.5,"ts":1697042000}'

# 2. Ingest to TimescaleDB
INSERT INTO sensor_data (device, ts, temperature) VALUES ('device_001', NOW(), 22.5);

# 3. Stream via WebSocket
await websocket.send(json.dumps({"device": "device_001", "temperature": 22.5}))

# 4. Render in Three.js
cube.scale.y = temperature / 10.0;
cube.material.color.setHSL(temperature/40.0, 1.0, 0.5);