Profile and Monitor Industrial Sensor Streams for Distribution Drift with Great Expectations and Apache Flink

"""
Production implementation for profiling and monitoring industrial sensor streams.
This script integrates with Great Expectations and Apache Flink for data validation and processing.
"""

from typing import Dict, Any, List, Tuple
import os
import logging
import requests
import time
from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError

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

class Config:
    """
    Configuration class for environment variables.
    """
    database_url: str = os.getenv('DATABASE_URL')
    api_url: str = os.getenv('API_URL')

# Helper Functions

def validate_input_data(data: Dict[str, Any]) -> bool:
    """
    Validate the sensor data input.
    
    Args:
        data: Dictionary containing sensor data
    Returns:
        bool: True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'sensor_id' not in data:
        raise ValueError('Missing sensor_id')
    if 'value' not in data:
        raise ValueError('Missing value')
    return True


def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Sanitize input data fields to prevent injection attacks.
    
    Args:
        data: Dictionary containing sensor data
    Returns:
        Dict[str, Any]: sanitized data
    """
    return {k: str(v).strip() for k, v in data.items()}


def normalize_data(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Normalize sensor data for consistency.
    
    Args:
        data: Dictionary containing sensor data
    Returns:
        Dict[str, Any]: normalized data
    """
    data['value'] = float(data['value'])
    return data


def transform_records(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """
    Transform a batch of sensor records for processing.
    
    Args:
        records: List of sensor data records
    Returns:
        List[Dict[str, Any]]: transformed records
    """
    return [normalize_data(sanitize_fields(record)) for record in records]


def fetch_data(api_url: str) -> List[Dict[str, Any]]:
    """
    Fetch sensor data from the external API.
    
    Args:
        api_url: The API endpoint to fetch data
    Returns:
        List[Dict[str, Any]]: List of sensor data records
    Raises:
        RuntimeError: If fetching data fails
    """
    try:
        response = requests.get(api_url)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        logger.error(f'Error fetching data: {e}')
        raise RuntimeError('Failed to fetch data')


def save_to_db(engine, data: List[Dict[str, Any]]) -> None:
    """
    Save processed data to the database.
    
    Args:
        engine: Database engine for connections
        data: List of sensor data records
    Raises:
        SQLAlchemyError: If saving to database fails
    """
    with engine.connect() as connection:
        for record in data:
            try:
                connection.execute(text('INSERT INTO sensor_data (sensor_id, value) VALUES (:sensor_id, :value)'), 
                                   sensor_id=record['sensor_id'], value=record['value'])
                logger.info(f'Saved record: {record}')
            except SQLAlchemyError as e:
                logger.error(f'Error saving record {record}: {e}')


def handle_errors(func):
    """
    Decorator to handle errors and retries for functions.
    """
    def wrapper(*args, **kwargs):
        attempts = 3
        for attempt in range(attempts):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                logger.warning(f'Attempt {attempt + 1} failed: {e}')
                time.sleep(2 ** attempt)  # Exponential backoff
        logger.error('All attempts failed')
        raise
    return wrapper


class SensorDataProcessor:
    """
    Main orchestrator for processing sensor data streams.
    """
    def __init__(self, config: Config):
        self.config = config
        self.engine = create_engine(self.config.database_url)

    @handle_errors
    def process_batch(self) -> None:
        """
        Process a batch of sensor data.
        
        Fetch, validate, and save data.
        """
        raw_data = fetch_data(self.config.api_url)
        transformed_data = transform_records(raw_data)
        for record in transformed_data:
            if validate_input_data(record):
                save_to_db(self.engine, [record])


if __name__ == '__main__':
    # Example usage
    config = Config()
    processor = SensorDataProcessor(config)
    processor.process_batch()