Streaming IoT Telemetry with Kafka + TimescaleDB (via Docker Compose)¶

#!/usr/bin/env python3
"""
IoT Telemetry Simulator for Kafka + TimescaleDB Pipeline
Generates realistic device telemetry data and streams to Kafka
"""

import json
import time
import random
import argparse
import logging
from datetime import datetime, timezone
from typing import Dict, Any, List
import orjson
from kafka import KafkaProducer
from kafka.errors import KafkaError
from faker import Faker

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class IoTDeviceSimulator:
    """Simulates IoT devices generating telemetry data"""

    def __init__(self, device_id: str, location: str):
        self.device_id = device_id
        self.location = location
        self.fake = Faker()

        # Device characteristics
        self.sensors = ['temperature', 'humidity', 'pressure', 'vibration', 'voltage']
        self.base_values = {
            'temperature': 20.0,  # Celsius
            'humidity': 50.0,      # Percentage
            'pressure': 1013.25,   # hPa
            'vibration': 0.1,      # g
            'voltage': 12.0        # Volts
        }
        self.value_ranges = {
            'temperature': (15.0, 35.0),
            'humidity': (30.0, 80.0),
            'pressure': (1000.0, 1020.0),
            'vibration': (0.05, 0.5),
            'voltage': (11.5, 12.5)
        }

    def generate_reading(self) -> Dict[str, Any]:
        """Generate a single sensor reading"""
        sensor = random.choice(self.sensors)
        base_value = self.base_values[sensor]
        min_val, max_val = self.value_ranges[sensor]

        # Add some drift and noise
        drift = random.uniform(-0.1, 0.1)
        noise = random.uniform(-0.05, 0.05)
        value = base_value + drift + noise

        # Clamp to realistic range
        value = max(min_val, min(max_val, value))

        return {
            'ts': datetime.now(timezone.utc).isoformat(),
            'device_id': self.device_id,
            'sensor': sensor,
            'value': round(value, 3),
            'location': self.location,
            'meta': {
                'firmware_version': f"v{random.randint(1, 3)}.{random.randint(0, 9)}",
                'battery_level': random.randint(20, 100),
                'signal_strength': random.randint(-100, -30)
            }
        }

class IoTTelemetryProducer:
    """Produces IoT telemetry data to Kafka"""

    def __init__(self, broker: str, topic: str, events_per_sec: int = 100):
        self.broker = broker
        self.topic = topic
        self.events_per_sec = events_per_sec
        self.interval = 1.0 / events_per_sec

        # Initialize Kafka producer
        self.producer = KafkaProducer(
            bootstrap_servers=[broker],
            value_serializer=lambda v: orjson.dumps(v),
            key_serializer=lambda k: k.encode('utf-8') if k else None,
            acks='all',
            retries=3,
            retry_backoff_ms=100,
            request_timeout_ms=30000,
            batch_size=16384,
            linger_ms=10,
            compression_type='zstd'
        )

        # Initialize device simulators
        self.devices = self._create_devices()
        logger.info(f"Created {len(self.devices)} device simulators")

    def _create_devices(self) -> List[IoTDeviceSimulator]:
        """Create device simulators"""
        devices = []
        fake = Faker()

        # Create devices with realistic locations
        locations = [
            'Building A - Floor 1', 'Building A - Floor 2', 'Building A - Floor 3',
            'Building B - Floor 1', 'Building B - Floor 2', 'Building B - Floor 3',
            'Warehouse - Zone 1', 'Warehouse - Zone 2', 'Warehouse - Zone 3',
            'Factory Floor - Line 1', 'Factory Floor - Line 2', 'Factory Floor - Line 3'
        ]

        for i in range(1, 5001):  # 5000 devices
            device_id = f"device-{i:06d}"
            location = random.choice(locations)
            devices.append(IoTDeviceSimulator(device_id, location))

        return devices

    def produce_telemetry(self):
        """Produce telemetry data at configured rate"""
        logger.info(f"Starting telemetry production at {self.events_per_sec} events/sec")

        try:
            while True:
                start_time = time.time()

                # Generate reading from random device
                device = random.choice(self.devices)
                reading = device.generate_reading()

                # Send to Kafka
                future = self.producer.send(
                    self.topic,
                    key=reading['device_id'],
                    value=reading
                )

                # Handle send result
                try:
                    record_metadata = future.get(timeout=1)
                    logger.debug(f"Sent to {record_metadata.topic} partition {record_metadata.partition}")
                except KafkaError as e:
                    logger.error(f"Failed to send message: {e}")

                # Rate limiting
                elapsed = time.time() - start_time
                sleep_time = max(0, self.interval - elapsed)
                if sleep_time > 0:
                    time.sleep(sleep_time)

        except KeyboardInterrupt:
            logger.info("Stopping telemetry production")
        finally:
            self.producer.close()
            logger.info("Producer closed")

def main():
    """Main entry point"""
    parser = argparse.ArgumentParser(description='IoT Telemetry Simulator')
    parser.add_argument('--broker', default='localhost:9092', help='Kafka broker address')
    parser.add_argument('--topic', default='iot.readings.json', help='Kafka topic')
    parser.add_argument('--eps', type=int, default=200, help='Events per second')
    parser.add_argument('--devices', type=int, default=5000, help='Number of devices')

    args = parser.parse_args()

    # Override from environment if set
    import os
    broker = os.getenv('KAFKA_BROKER', args.broker)
    topic = os.getenv('TOPIC', args.topic)
    eps = int(os.getenv('EVENTS_PER_SEC', args.eps))

    logger.info(f"Starting IoT simulator: {broker}/{topic} @ {eps} eps")

    producer = IoTTelemetryProducer(broker, topic, eps)
    producer.produce_telemetry()

if __name__ == '__main__':
    main()