Enforce Row-Level Quality Contracts on Manufacturing Sensor Streams with Soda Core and Apache Kafka

"""
Production implementation for enforcing row-level quality contracts on manufacturing sensor streams.
This application uses Soda Core for data quality checks and Apache Kafka for streaming data processing.
"""
import os
import logging
import json
import time
from typing import Dict, Any, List
from kafka import KafkaConsumer, KafkaProducer
from pydantic import BaseModel, ValidationError

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

class Config:
    kafka_bootstrap_servers: str = os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092')
    kafka_topic: str = os.getenv('KAFKA_TOPIC', 'sensor_data')

class SensorData(BaseModel):
    sensor_id: str
    value: float
    timestamp: str

async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate incoming sensor data.
    
    Args:
        data: Dictionary containing sensor data
    Returns:
        bool: True if validation passes
    Raises:
        ValueError: If validation fails
    """
    try:
        SensorData(**data)  # Validate using Pydantic
        return True
    except ValidationError as e:
        logger.error(f'Validation error: {e}')
        raise ValueError('Invalid input data')

async def normalize_data(data: Dict[str, Any]) -> Dict[str, Any]:
    """Normalize input data for processing.
    
    Args:
        data: Raw sensor data
    Returns:
        Dict: Normalized sensor data
    """
    data['value'] = round(data['value'], 2)  # Round value to 2 decimal places
    return data

async def fetch_data(consumer: KafkaConsumer) -> List[Dict[str, Any]]:
    """Fetch data from Kafka topic.
    
    Args:
        consumer: Kafka consumer instance
    Returns:
        List of fetched sensor data
    """
    messages = []
    for message in consumer:
        messages.append(json.loads(message.value.decode('utf-8')))
    return messages

async def save_to_db(data: Dict[str, Any]) -> None:
    """Simulate saving data to a database.
    
    Args:
        data: Sensor data to save
    """
    logger.info(f'Saving data to DB: {data}')
    # Here you would implement actual DB saving logic

async def call_api(data: Dict[str, Any]) -> None:
    """Simulate calling an external API.
    
    Args:
        data: Data to send to the API
    """
    logger.info(f'Calling external API with data: {data}')
    # Implement actual API call here

async def aggregate_metrics(data: List[Dict[str, Any]]) -> Dict[str, float]:
    """Aggregate metrics from sensor data.
    
    Args:
        data: List of sensor data
    Returns:
        Aggregated metrics
    """
    total = sum(d['value'] for d in data)
    average = total / len(data)
    return {'total': total, 'average': average}

async def process_batch(consumer: KafkaConsumer, producer: KafkaProducer) -> None:
    """Process a batch of sensor data from Kafka.
    
    Args:
        consumer: Kafka consumer instance
        producer: Kafka producer instance
    """
    try:
        messages = await fetch_data(consumer)  # Fetch data
        validated_data = [await validate_input(m) for m in messages]  # Validate data
        normalized_data = [await normalize_data(m) for m in validated_data]  # Normalize data
        metrics = await aggregate_metrics(normalized_data)  # Aggregate metrics
        await save_to_db(metrics)  # Save aggregated metrics
        logger.info(f'Metrics processed: {metrics}')
    except Exception as e:
        logger.error(f'Error processing batch: {e}')

async def main() -> None:
    """Main function to run the Kafka consumer and producer.
    """
    consumer = KafkaConsumer(Config.kafka_topic, bootstrap_servers=Config.kafka_bootstrap_servers)
    producer = KafkaProducer(bootstrap_servers=Config.kafka_bootstrap_servers)
    while True:
        try:
            await process_batch(consumer, producer)
            time.sleep(1)  # Wait before processing next batch
        except KeyboardInterrupt:
            logger.info('Shutting down...')
            break
        except Exception as e:
            logger.error(f'Unhandled exception: {e}')

if __name__ == '__main__':
    import asyncio
    asyncio.run(main())