Predict Unplanned Downtime Intervals with Lag-Llama and Prophet

"""
Production implementation for predicting unplanned downtime intervals using Lag-Llama and Prophet.
Provides secure, scalable operations for downtime prediction.
"""
from typing import Dict, Any, List, Union
import os
import logging
import numpy as np
import pandas as pd
import requests
from prophet import Prophet
from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError
from contextlib import contextmanager

# 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')
    lag_lama_api_url: str = os.getenv('LAG_LAMA_API_URL')

@contextmanager
def get_db_connection() -> Any:
    """Context manager for database connection.
    
    Yields:
        Connection object
    """
    engine = create_engine(Config.database_url)
    connection = engine.connect()
    try:
        yield connection
    finally:
        connection.close()  # Ensure connection is closed

async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate incoming data.
    
    Args:
        data: Input data to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if not isinstance(data, dict):
        raise ValueError('Input data must be a dictionary')
    if 'timestamp' not in data:
        raise ValueError('Missing timestamp field')
    return True

def fetch_data(api_url: str) -> Union[List[Dict[str, Any]], None]:
    """Fetch data from Lag-Lama API.
    
    Args:
        api_url: URL to fetch data from
    Returns:
        List of records from the API
    Raises:
        ValueError: If fetch fails
    """
    try:
        response = requests.get(api_url)
        response.raise_for_status()  # Raise if HTTP error
        return response.json()
    except requests.RequestException as e:
        logger.error(f'API request failed: {e}')
        return None

def normalize_data(data: List[Dict[str, Any]]) -> pd.DataFrame:
    """Normalize the fetched data into a DataFrame.
    
    Args:
        data: List of fetched data
    Returns:
        Normalized DataFrame
    """
    if not data:
        raise ValueError('No data to normalize')
    df = pd.DataFrame(data)
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    return df

def transform_records(df: pd.DataFrame) -> pd.DataFrame:
    """Transform records for modeling.
    
    Args:
        df: DataFrame to transform
    Returns:
        Transformed DataFrame for Prophet
    """
    df = df.rename(columns={'timestamp': 'ds', 'downtime': 'y'})
    return df[['ds', 'y']]

def save_to_db(connection: Any, df: pd.DataFrame) -> None:
    """Save the DataFrame to the database.
    
    Args:
        connection: Active database connection
        df: DataFrame to save
    Raises:
        SQLAlchemyError: If database operation fails
    """
    try:
        df.to_sql('downtime_predictions', con=connection, if_exists='append', index=False)
    except SQLAlchemyError as e:
        logger.error(f'Database save failed: {e}')
        raise

def call_prophet_model(df: pd.DataFrame) -> pd.DataFrame:
    """Call the Prophet model for predictions.
    
    Args:
        df: DataFrame with transformed data
    Returns:
        DataFrame with predictions
    """
    model = Prophet()
    model.fit(df)
    future = model.make_future_dataframe(periods=30)  # Predict next 30 days
    forecast = model.predict(future)
    return forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']]  # Return relevant columns

async def process_batch(data: Dict[str, Any]) -> None:
    """Process a batch of downtime data.
    
    Args:
        data: Incoming data batch
    Raises:
        Exception: If processing fails
    """
    await validate_input(data)  # Validate input
    records = fetch_data(Config.lag_lama_api_url)  # Fetch records
    if records is None:
        raise Exception('Failed to fetch data')
    df = normalize_data(records)  # Normalize data
    transformed_df = transform_records(df)  # Transform data
    with get_db_connection() as conn:
        save_to_db(conn, transformed_df)  # Save to DB
    predictions = call_prophet_model(transformed_df)  # Call the model
    logger.info('Processing complete, predictions generated.')  # Log completion

if __name__ == '__main__':
    # Example usage
    sample_data = {'timestamp': '2023-01-01T00:00:00Z', 'downtime': 1}
    try:
        process_batch(sample_data)
    except Exception as e:
        logger.error(f'Error in processing: {e}')  # Handle main errors