app.py

"""
FastAPI Application for Driving Behavior Analysis
==================================================

Comprehensive REST API with Swagger documentation for testing the driving behavior
classification model. Includes batch predictions, real-time classification, and 
detailed confidence scores.

Run: uvicorn main:app --reload --host 0.0.0.0 --port 8000
Swagger UI: http://localhost:8000/docs
"""

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel, Field
from typing import List, Dict, Optional
import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler, LabelEncoder
import pickle
import json
from datetime import datetime
import logging
import uvicorn
from collections import deque
import threading
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ============================================================================
# LOAD MODEL & PREPROCESSING OBJECTS
# ============================================================================

try:
    import joblib
    import os
    
    # Use actual relative paths so they work on Hugging Face servers
    BASE_DIR = os.path.dirname(os.path.abspath(__file__))
    model_path = os.path.join(BASE_DIR, 'model.pkl')
    scaler_path = os.path.join(BASE_DIR, 'scaler.pkl')
    le_path = os.path.join(BASE_DIR, 'label_encoder.pkl')
    fc_path = os.path.join(BASE_DIR, 'feature_columns.pkl')
    
    best_model = joblib.load(model_path)
    scaler = joblib.load(scaler_path)
    label_encoder = joblib.load(le_path)
    
    # Feature names (in correct order)
    with open(fc_path, 'rb') as f:
        feature_columns = pickle.load(f)
    
    logger.info("✅ All models and preprocessors loaded successfully!")
    
except FileNotFoundError as e:
    logger.warning(f"⚠️ Could not load model files: {e}")
    logger.warning("⚠️ Using mock models for demonstration")
    best_model = None
    scaler = None
    label_encoder = None
    feature_columns = None


# Global buffer to store recent sensor readings for proper time-series feature engineering
reading_history = deque(maxlen=15)
history_lock = threading.Lock()

# ============================================================================
# PYDANTIC MODELS (Request/Response Schemas)
# ============================================================================

class SensorInput(BaseModel):
    """Raw sensor input from accelerometer and gyroscope"""
    acc_x: float = Field(
        ..., 
        description="Acceleration in X direction (m/s²)",
        example=0.5
    )
    acc_y: float = Field(
        ..., 
        description="Acceleration in Y direction (m/s²)",
        example=0.2
    )
    acc_z: float = Field(
        ..., 
        description="Acceleration in Z direction (m/s²)",
        example=9.8
    )
    gyro_x: float = Field(
        ..., 
        description="Angular velocity around X axis (rad/s)",
        example=0.01
    )
    gyro_y: float = Field(
        ..., 
        description="Angular velocity around Y axis (rad/s)",
        example=0.02
    )
    gyro_z: float = Field(
        ..., 
        description="Angular velocity around Z axis (rad/s)",
        example=0.03
    )

    class Config:
        json_schema_extra = {
            "example": {
                "acc_x": 0.5,
                "acc_y": 0.2,
                "acc_z": 9.8,
                "gyro_x": 0.01,
                "gyro_y": 0.02,
                "gyro_z": 0.03
            }
        }


class PredictionResponse(BaseModel):
    """Response with prediction and confidence scores"""
    prediction: str = Field(..., description="Predicted driving behavior class")
    confidence: Dict[str, float] = Field(..., description="Confidence scores for each class")
    timestamp: str = Field(..., description="Prediction timestamp")
    
    class Config:
        json_schema_extra = {
            "example": {
                "prediction": "NORMAL",
                "confidence": {
                    "AGGRESSIVE": 0.02,
                    "NORMAL": 0.88,
                    "SLOW": 0.10
                },
                "timestamp": "2024-04-17T12:34:56"
            }
        }


class BatchPredictionRequest(BaseModel):
    """Request for batch predictions"""
    samples: List[SensorInput] = Field(..., description="List of sensor readings")
    return_features: bool = Field(
        False, 
        description="Include engineered features in response"
    )


class BatchPredictionResponse(BaseModel):
    """Response with batch predictions"""
    total_samples: int
    successful_predictions: int
    failed_predictions: int
    predictions: List[Dict] = Field(..., description="List of predictions")
    processing_time_ms: float


class HealthResponse(BaseModel):
    """Health check response"""
    status: str
    model_loaded: bool
    model_version: str
    timestamp: str
    features_count: Optional[int] = None


# ============================================================================
# FEATURE ENGINEERING FUNCTION
# ============================================================================

def engineer_features(data_list: list) -> pd.DataFrame:
    """
    Apply feature engineering to raw sensor data using a sequence of historical readings
    to correctly compute rates of changes (Jerk) and rolling statistics.
    """
    try:
        # Create DataFrame from list
        df = pd.DataFrame(data_list)
        
        # Rename columns to match training
        df = df.rename(columns={
            'acc_x': 'AccX', 'acc_y': 'AccY', 'acc_z': 'AccZ',
            'gyro_x': 'GyroX', 'gyro_y': 'GyroY', 'gyro_z': 'GyroZ'
        })
        
        # ===== JERK CALCULATION =====
        # For single row: jerk = 0
        df['JerkX'] = df['AccX'].diff().fillna(0)
        df['JerkY'] = df['AccY'].diff().fillna(0)
        df['JerkZ'] = df['AccZ'].diff().fillna(0)
        
        # ===== MAGNITUDE FEATURES =====
        df['AccMagnitude'] = np.sqrt(df['AccX']**2 + df['AccY']**2 + df['AccZ']**2)
        df['GyroMagnitude'] = np.sqrt(df['GyroX']**2 + df['GyroY']**2 + df['GyroZ']**2)
        df['JerkMagnitude'] = np.sqrt(df['JerkX']**2 + df['JerkY']**2 + df['JerkZ']**2)
        
        # ===== ROLLING STATISTICS =====
        window_size = 5
        df['AccX_rolling_mean'] = df['AccX'].rolling(window=window_size, min_periods=1).mean()
        df['AccY_rolling_mean'] = df['AccY'].rolling(window=window_size, min_periods=1).mean()
        df['AccZ_rolling_mean'] = df['AccZ'].rolling(window=window_size, min_periods=1).mean()
        
        df['AccX_rolling_std'] = df['AccX'].rolling(window=window_size, min_periods=1).std().fillna(0)
        df['AccY_rolling_std'] = df['AccY'].rolling(window=window_size, min_periods=1).std().fillna(0)
        df['AccZ_rolling_std'] = df['AccZ'].rolling(window=window_size, min_periods=1).std().fillna(0)
        
        df['JerkX_rolling_mean'] = df['JerkX'].rolling(window=window_size, min_periods=1).mean()
        df['JerkY_rolling_mean'] = df['JerkY'].rolling(window=window_size, min_periods=1).mean()
        df['JerkZ_rolling_mean'] = df['JerkZ'].rolling(window=window_size, min_periods=1).mean()
        
        df['JerkX_rolling_max'] = df['JerkX'].rolling(window=window_size, min_periods=1).max()
        df['JerkY_rolling_max'] = df['JerkY'].rolling(window=window_size, min_periods=1).max()
        df['JerkZ_rolling_max'] = df['JerkZ'].rolling(window=window_size, min_periods=1).max()
        
        # ===== VARIANCE & ENERGY =====
        df['AccX_var'] = df['AccX'] ** 2
        df['AccY_var'] = df['AccY'] ** 2
        df['AccZ_var'] = df['AccZ'] ** 2
        
        df['JerkX_var'] = df['JerkX'] ** 2
        df['JerkY_var'] = df['JerkY'] ** 2
        df['JerkZ_var'] = df['JerkZ'] ** 2
        
        # ===== ABSOLUTE VALUES =====
        df['AbsAccX'] = abs(df['AccX'])
        df['AbsAccY'] = abs(df['AccY'])
        df['AbsAccZ'] = abs(df['AccZ'])
        df['AbsJerkX'] = abs(df['JerkX'])
        df['AbsJerkY'] = abs(df['JerkY'])
        df['AbsJerkZ'] = abs(df['JerkZ'])
        
        # ===== ENERGY FEATURES =====
        df['Acc_Energy'] = (df['AccX']**2 + df['AccY']**2 + df['AccZ']**2) / 3
        df['Jerk_Energy'] = (df['JerkX']**2 + df['JerkY']**2 + df['JerkZ']**2) / 3
        
        return df
    
    except Exception as e:
        logger.error(f"Error in feature engineering: {str(e)}")
        raise


# ============================================================================
# PREDICTION FUNCTION
# ============================================================================

def predict_driving_behavior(data_list: list) -> Dict:
    """
    Predict driving behavior from sensor data sequence
    
    Args:
        data_list: List of dictionaries with sensor readings
        
    Returns:
        Dictionary with prediction and confidence scores
    """
    try:
        # Check if models are loaded
        if best_model is None or scaler is None or label_encoder is None:
            raise ValueError("Models not loaded. Cannot make predictions.")
        
        # Engineer features
        df = engineer_features(data_list)
        
        # Select features in correct order
        df_processed = df[feature_columns]
        
        # Scale features
        df_scaled = scaler.transform(df_processed)
        
        # Extract only the latest row for prediction
        latest_row_scaled = df_scaled[-1].reshape(1, -1)
        
        # Make prediction
        pred = best_model.predict(latest_row_scaled)
        proba = best_model.predict_proba(latest_row_scaled)
        
        # Decode prediction
        prediction = label_encoder.inverse_transform(pred)[0]
        
        # Get confidence scores
        confidence_dict = {
            label: float(proba[0][i]) 
            for i, label in enumerate(label_encoder.classes_)
        }
        
        return {
            "prediction": prediction,
            "confidence": confidence_dict,
            "raw_probability": proba[0].tolist()
        }
    
    except Exception as e:
        logger.error(f"Prediction error: {str(e)}")
        raise


# ============================================================================
# FASTAPI APPLICATION
# ============================================================================

app = FastAPI(
    title="🚗 Driving Behavior Analysis API",
    description="""
    Real-time driving behavior classification API using machine learning.
    
    Classify driving patterns into three categories:
    - **NORMAL**: Regular, safe driving
    - **SLOW**: Cautious, slower driving
    - **AGGRESSIVE**: Risky, aggressive driving
    
    ## Features
    - Single prediction endpoint
    - Batch prediction support
    - Real-time confidence scores
    - Feature engineering included
    - Health check endpoint
    - Comprehensive API documentation
    
    ## How to Use
    1. Provide raw sensor data (acceleration & gyroscope readings)
    2. API automatically engineers features
    3. Get driving behavior classification with confidence scores
    
    ## Example Request
    ```json
    {
        "acc_x": 0.5,
        "acc_y": 0.2,
        "acc_z": 9.8,
        "gyro_x": 0.01,
        "gyro_y": 0.02,
        "gyro_z": 0.03
    }
    ```
    """,
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
    openapi_url="/openapi.json",
    contact={
        "name": "ML Team",
        "email": "ml@example.com"
    }
)


# ============================================================================
# HEALTH CHECK ENDPOINT
# ============================================================================

@app.get(
    "/health",
    response_model=HealthResponse,
    tags=["Health"],
    summary="Health Check",
    description="Check API status and model availability"
)
async def health_check():
    """
    Check the health status of the API and model availability
    """
    return HealthResponse(
        status="healthy",
        model_loaded=best_model is not None,
        model_version="1.0.0",
        timestamp=datetime.now().isoformat(),
        features_count=len(feature_columns) if feature_columns else 0
    )


# ============================================================================
# SINGLE PREDICTION ENDPOINT
# ============================================================================

@app.post(
    "/predict",
    response_model=PredictionResponse,
    tags=["Prediction"],
    summary="Predict Driving Behavior",
    description="Predict driving behavior from a single sensor reading"
)
async def predict(sensor_input: SensorInput):
    """
    Predict driving behavior from raw sensor data.
    
    **Input Parameters:**
    - acc_x: Acceleration in X direction (m/s²)
    - acc_y: Acceleration in Y direction (m/s²)
    - acc_z: Acceleration in Z direction (m/s²)
    - gyro_x: Angular velocity around X axis (rad/s)
    - gyro_y: Angular velocity around Y axis (rad/s)
    - gyro_z: Angular velocity around Z axis (rad/s)
    
    **Response:**
    - prediction: One of [AGGRESSIVE, NORMAL, SLOW]
    - confidence: Confidence scores for each class
    - timestamp: When prediction was made
    
    **Example Request:**
    ```json
    {
        "acc_x": 0.5,
        "acc_y": 0.2,
        "acc_z": 9.8,
        "gyro_x": 0.01,
        "gyro_y": 0.02,
        "gyro_z": 0.03
    }
    ```
    
    **Example Response:**
    ```json
    {
        "prediction": "NORMAL",
        "confidence": {
            "AGGRESSIVE": 0.02,
            "NORMAL": 0.88,
            "SLOW": 0.10
        },
        "timestamp": "2024-04-17T12:34:56"
    }
    ```
    """
    try:
        # Convert input to dictionary
        input_dict = sensor_input.dict()
        
        # Append to global history
        with history_lock:
            reading_history.append(input_dict)
            history_snapshot = list(reading_history)
        
        # Make prediction using history
        result = predict_driving_behavior(history_snapshot)
        
        return PredictionResponse(
            prediction=result["prediction"],
            confidence=result["confidence"],
            timestamp=datetime.now().isoformat()
        )
    
    except Exception as e:
        logger.error(f"Prediction error: {str(e)}")
        raise HTTPException(
            status_code=500,
            detail=f"Prediction failed: {str(e)}"
        )


# ============================================================================
# BATCH PREDICTION ENDPOINT
# ============================================================================

@app.post(
    "/predict-batch",
    response_model=BatchPredictionResponse,
    tags=["Batch Prediction"],
    summary="Batch Predictions",
    description="Make predictions on multiple sensor readings at once"
)
async def predict_batch(request: BatchPredictionRequest):
    """
    Predict driving behavior for multiple sensor readings.
    
    Useful for processing streams or datasets efficiently.
    
    **Request Parameters:**
    - samples: List of sensor readings
    - return_features: Whether to include engineered features in response
    
    **Returns:**
    - total_samples: Number of samples processed
    - successful_predictions: Number of successful predictions
    - failed_predictions: Number of failed predictions
    - predictions: List of prediction results
    - processing_time_ms: Total processing time
    """
    import time
    
    start_time = time.time()
    predictions = []
    successful = 0
    failed = 0
    
    try:
        for i, sensor_input in enumerate(request.samples):
            try:
                input_dict = sensor_input.dict()
                
                # Using the same global history mechanism to accumulate batch over time
                with history_lock:
                    reading_history.append(input_dict)
                    history_snapshot = list(reading_history)
                
                result = predict_driving_behavior(history_snapshot)
                
                prediction_result = {
                    "sample_index": i,
                    "prediction": result["prediction"],
                    "confidence": result["confidence"],
                    "timestamp": datetime.now().isoformat()
                }
                
                predictions.append(prediction_result)
                successful += 1
            
            except Exception as e:
                logger.error(f"Error on sample {i}: {str(e)}")
                predictions.append({
                    "sample_index": i,
                    "error": str(e)
                })
                failed += 1
        
        processing_time = (time.time() - start_time) * 1000  # Convert to ms
        
        return BatchPredictionResponse(
            total_samples=len(request.samples),
            successful_predictions=successful,
            failed_predictions=failed,
            predictions=predictions,
            processing_time_ms=processing_time
        )
    
    except Exception as e:
        logger.error(f"Batch prediction error: {str(e)}")
        raise HTTPException(
            status_code=500,
            detail=f"Batch prediction failed: {str(e)}"
        )


# ============================================================================
# DETAILED PREDICTION ENDPOINT
# ============================================================================

@app.post(
    "/predict-detailed",
    tags=["Prediction"],
    summary="Detailed Prediction",
    description="Get detailed prediction with engineered features"
)
async def predict_detailed(sensor_input: SensorInput):
    """
    Get detailed prediction including engineered features.
    
    Useful for understanding which features influenced the prediction.
    """
    try:
        input_dict = sensor_input.dict()
        
        with history_lock:
            reading_history.append(input_dict)
            history_snapshot = list(reading_history)
        
        # Engineer features
        df = engineer_features(history_snapshot)
        
        # Make prediction
        result = predict_driving_behavior(history_snapshot)
        
        return {
            "prediction": result["prediction"],
            "confidence": result["confidence"],
            "engineered_features": df.to_dict(orient='records')[-1],
            "timestamp": datetime.now().isoformat()
        }
    
    except Exception as e:
        logger.error(f"Detailed prediction error: {str(e)}")
        raise HTTPException(
            status_code=500,
            detail=f"Detailed prediction failed: {str(e)}"
        )


# ============================================================================
# TEST DATA ENDPOINT
# ============================================================================

@app.get(
    "/test-samples",
    tags=["Testing"],
    summary="Get Test Samples",
    description="Get example sensor readings for testing"
)
async def get_test_samples():
    """
    Get example sensor readings for different driving behaviors.
    
    Useful for testing the API without real sensor data.
    """
    return {
        "NORMAL": {
            "description": "Normal driving - balanced sensor readings",
            "sample": {
                "acc_x": 0.05,
                "acc_y": -0.08,
                "acc_z": 9.81,
                "gyro_x": 0.002,
                "gyro_y": -0.001,
                "gyro_z": 0.008
            }
        },
        "SLOW": {
            "description": "Slow driving - smooth, low acceleration",
            "sample": {
                "acc_x": 0.1,
                "acc_y": -0.02,
                "acc_z": 9.8,
                "gyro_x": 0.0,
                "gyro_y": 0.0,
                "gyro_z": 0.001
            }
        },
        "AGGRESSIVE": {
            "description": "Aggressive driving - high accelerations and jerky movements",
            "sample": {
                "acc_x": 0.8,
                "acc_y": -0.5,
                "acc_z": 9.7,
                "gyro_x": 0.05,
                "gyro_y": 0.03,
                "gyro_z": 0.1
            }
        }
    }


# ============================================================================
# INFO ENDPOINT
# ============================================================================

@app.get(
    "/info",
    tags=["Info"],
    summary="API Information",
    description="Get information about the API and model"
)
async def get_info():
    """
    Get detailed information about the API and trained model.
    """
    return {
        "api_name": "Driving Behavior Analysis API",
        "version": "1.0.0",
        "model_status": "loaded" if best_model is not None else "not_loaded",
        "supported_classes": [
            "AGGRESSIVE",
            "NORMAL",
            "SLOW"
        ],
        "features_count": len(feature_columns) if feature_columns else "unknown",
        "endpoints": {
            "predict": "POST /predict - Single prediction",
            "predict_batch": "POST /predict-batch - Batch predictions",
            "predict_detailed": "POST /predict-detailed - Detailed prediction with features",
            "health_check": "GET /health - Health check",
            "test_samples": "GET /test-samples - Get example test data",
            "info": "GET /info - API information"
        },
        "documentation": {
            "swagger": "http://localhost:8000/docs",
            "redoc": "http://localhost:8000/redoc",
            "openapi": "http://localhost:8000/openapi.json"
        }
    }


# ============================================================================
# ROOT ENDPOINT
# ============================================================================

@app.get(
    "/",
    tags=["Root"],
    summary="Welcome",
    description="Welcome message and quick start guide"
)
async def root():
    """
    Welcome to the Driving Behavior Analysis API!
    
    **Quick Start:**
    1. Go to http://localhost:8000/docs for interactive Swagger UI
    2. Try the /predict endpoint with sample data
    3. Check /test-samples for example inputs
    
    **Endpoints:**
    - POST /predict - Single prediction
    - POST /predict-batch - Batch predictions
    - GET /health - Health check
    - GET /test-samples - Test data
    - GET /info - API information
    """
    return {
        "message": "🚗 Welcome to Driving Behavior Analysis API",
        "status": "running",
        "docs": "http://localhost:8000/docs",
        "quick_start": [
            "1. Visit http://localhost:8000/docs",
            "2. Click on POST /predict",
            "3. Click 'Try it out'",
            "4. Enter sensor data or use example",
            "5. Click 'Execute' to get prediction"
        ]
    }


# ============================================================================
# EXCEPTION HANDLERS
# ============================================================================

@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    """Handle HTTP exceptions"""
    return {
        "error": exc.detail,
        "status_code": exc.status_code,
        "timestamp": datetime.now().isoformat()
    }


# ============================================================================
# RUN THE APPLICATION
# ============================================================================


# ==========================================================================
# SAFETY SCORE ENDPOINT (NEW)
# ==========================================================================

from enum import Enum

class SafetyColor(str, Enum):
    GREEN = "Green"
    YELLOW = "Yellow"
    RED = "Red"

class SafetyScoreResponse(BaseModel):
    score: int = Field(..., description="Safety score (0-100)")
    color: SafetyColor = Field(..., description="Color-coded safety status")
    events: Dict[str, bool] = Field(..., description="Detected driving events")
    timestamp: str = Field(..., description="Timestamp of evaluation")

@app.post(
    "/safety-score",
    response_model=SafetyScoreResponse,
    tags=["Safety"],
    summary="Real-Time Safety Score",
    description="Get a real-time safety score and color-coded status from a single sensor reading."
)
async def safety_score(sensor_input: SensorInput):
    """
    Calculate a real-time safety score and color-coded status from sensor data.
    Detects harsh braking, rapid acceleration, aggressive cornering.
    """
    input_dict = sensor_input.dict()
    acc_x = input_dict["acc_x"]
    acc_y = input_dict["acc_y"]
    acc_z = input_dict["acc_z"]
    gyro_x = input_dict["gyro_x"]
    gyro_y = input_dict["gyro_y"]
    gyro_z = input_dict["gyro_z"]

    # Simple event detection thresholds (tune as needed)
    harsh_braking = acc_x < -2.5
    rapid_acceleration = acc_x > 2.5
    aggressive_cornering = abs(acc_y) > 2.0
    # (Advanced: add tailgating/lane weaving if you have more data)

    # Score logic (deduct for each event)
    score = 100
    if harsh_braking:
        score -= 30
    if rapid_acceleration:
        score -= 25
    if aggressive_cornering:
        score -= 20
    score = max(0, min(100, score))

    # Color coding
    if score >= 80:
        color = SafetyColor.GREEN
    elif score >= 50:
        color = SafetyColor.YELLOW
    else:
        color = SafetyColor.RED

    return SafetyScoreResponse(
        score=score,
        color=color,
        events={
            "harsh_braking": harsh_braking,
            "rapid_acceleration": rapid_acceleration,
            "aggressive_cornering": aggressive_cornering
        },
        timestamp=datetime.now().isoformat()
    )

if __name__ == "__main__":

    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8000,
        reload=True,
        log_level="info"
    )