Build Real-Time OEE Dashboards for Factories with Apache Kafka and DuckDB

"""
Production implementation for building real-time OEE dashboards for factories.
Integrates Apache Kafka for streaming data and DuckDB for analytics.
"""
from typing import Dict, Any, List
import os
import logging
import time
from fastapi import FastAPI, HTTPException
from kafka import KafkaConsumer, KafkaProducer
import duckdb

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    kafka_broker: str = os.getenv('KAFKA_BROKER', 'localhost:9092')
    duckdb_path: str = os.getenv('DUCKDB_PATH', 'oee_data.db')

# Setting up the FastAPI application
app = FastAPI()

def create_duckdb_connection() -> duckdb.DuckDBPyConnection:
    """Create a connection to DuckDB.
    
    Returns:
        duckdb.DuckDBPyConnection: DuckDB connection object
    """
    return duckdb.connect(database=Config.duckdb_path)

async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate input data for OEE metrics.
    
    Args:
        data: Input data dictionary
    Returns:
        bool: True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'machine_id' not in data or 'oee' not in data:
        raise ValueError('Missing machine_id or oee')
    return True

async def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize fields in the input data.
    
    Args:
        data: Input data dictionary
    Returns:
        Dict[str, Any]: Sanitized data
    """
    return {k: str(v).strip() for k, v in data.items()}

async def fetch_data(consumer: KafkaConsumer) -> List[Dict[str, Any]]:
    """Fetch data from Kafka.
    
    Args:
        consumer: KafkaConsumer instance
    Returns:
        List[Dict[str, Any]]: List of messages
    """
    messages = []
    for message in consumer:
        messages.append(message.value)
        logger.info('Fetched message from Kafka')
    return messages

async def transform_records(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Transform raw records into a format suitable for DuckDB.
    
    Args:
        records: List of raw records
    Returns:
        List[Dict[str, Any]]: Transformed records
    """
    transformed = []
    for record in records:
        transformed.append({
            'machine_id': record['machine_id'],
            'oee': float(record['oee']),
            'timestamp': record['timestamp']
        })
    logger.info('Transformed records for DuckDB')
    return transformed

async def save_to_db(records: List[Dict[str, Any]]) -> None:
    """Save transformed records to DuckDB.
    
    Args:
        records: List of records to save
    """
    conn = create_duckdb_connection()  # Create connection
    conn.execute('CREATE TABLE IF NOT EXISTS oee_data (machine_id VARCHAR, oee FLOAT, timestamp TIMESTAMP)')
    for record in records:
        conn.execute('INSERT INTO oee_data VALUES (?, ?, ?)', (record['machine_id'], record['oee'], record['timestamp']))
    logger.info('Saved records to DuckDB')
    conn.close()  # Ensure connection is closed

async def process_batch(consumer: KafkaConsumer) -> None:
    """Process a batch of messages from Kafka.
    
    Args:
        consumer: KafkaConsumer instance
    """
    try:
        raw_data = await fetch_data(consumer)  # Fetch data from Kafka
        validated_data = await asyncio.gather(*[validate_input(record) for record in raw_data])
        sanitized_data = await asyncio.gather(*[sanitize_fields(record) for record in raw_data])
        transformed_data = await transform_records(sanitized_data)  # Transform the data
        await save_to_db(transformed_data)  # Save to DuckDB
    except Exception as e:
        logger.error(f'Error processing batch: {str(e)}')

@app.on_event('startup')
async def startup_event() -> None:
    """Kafka consumer startup event.
    Initializes the Kafka consumer and processes data.
    """
    consumer = KafkaConsumer('oee_metrics', bootstrap_servers=Config.kafka_broker)
    while True:
        await process_batch(consumer)  # Process a batch of data
        time.sleep(1)  # Sleep to avoid busy waiting

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)  # Run FastAPI application