main.py

"""
AI-Enabled Production Disruption Early Warning System

Instructions:
1. Install Requirements:
   pip install fastapi uvicorn scikit-learn numpy pandas

2. Run the Server:
   uvicorn main:app --reload

3. Test endpoints:
   - Status: GET http://127.0.0.1:8000/api/status
   - Alerts: GET http://127.0.0.1:8000/api/alerts
   - History: GET http://127.0.0.1:8000/api/telemetry/history
   - Inject Anomaly: POST http://127.0.0.1:8000/api/inject-failure
   - Reset Demo: POST http://127.0.0.1:8000/api/reset-demo
   - Manual Telemetry Post: POST http://127.0.0.1:8000/api/telemetry
"""

import os
import time
import random
import sqlite3
import datetime
import threading
from contextlib import asynccontextmanager, contextmanager
from typing import Optional, List, Dict, Any

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from fastapi import FastAPI, HTTPException, BackgroundTasks, Response
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from ai_service import ProductionIntelligenceAIService, format_user_facing_response

ai_service = ProductionIntelligenceAIService()

# Database configuration
DB_FILE = "production.db"

# Dataset path (checks local folder and Downloads)
DATASET_PATHS = [
    "manufacturing_disruption_dataset.csv",
    r"C:\Users\AMRUTHA\Downloads\manufacturing_disruption_dataset.csv"
]

# Global demo state
SIMULATOR_FAILURE_INJECTED = False
simulator_running = True
model: Optional[IsolationForest] = None
normal_data_df: Optional[pd.DataFrame] = None
disrupt_data_df: Optional[pd.DataFrame] = None

# Pydantic models
class TelemetryUpload(BaseModel):
    line_id: str
    processing_delay_ms: float
    error_rate_pct: float
    throughput_units: float

class StatusResponse(BaseModel):
    status: str
    latest_telemetry: Optional[Dict[str, Any]]

# Context manager for SQLite operations
@contextmanager
def get_db():
    conn = sqlite3.connect(DB_FILE, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()

# Database initialization
def init_db():
    with get_db() as conn:
        cursor = conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS telemetry (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT NOT NULL,
                line_id TEXT NOT NULL,
                processing_delay_ms REAL NOT NULL,
                error_rate_pct REAL NOT NULL,
                throughput_units REAL NOT NULL
            )
        """)
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS alerts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT NOT NULL,
                severity TEXT NOT NULL,
                anomaly_score REAL NOT NULL,
                ai_explanation TEXT NOT NULL,
                status TEXT NOT NULL
            )
        """)
        conn.commit()
    print(f"Database initialized successfully at {os.path.abspath(DB_FILE)}")

# Anomaly Detector setup with real dataset
def init_anomaly_detector():
    global model, normal_data_df, disrupt_data_df
    
    # 1. Locate dataset file
    csv_file = None
    for path in DATASET_PATHS:
        if os.path.exists(path):
            csv_file = path
            break
            
    if csv_file:
        print(f"Loading dataset from: {csv_file}")
        df = pd.read_csv(csv_file)
        
        # Real column mappings from manufacturing dataset:
        # Delay -> downtime_minutes
        # Error Rate -> defect_rate
        # Throughput -> production_rate
        normal_data_df = df[df['disruption_occurred'] == 0]
        disrupt_data_df = df[df['disruption_occurred'] == 1]
        
        features = ['downtime_minutes', 'defect_rate', 'production_rate']
        X_train = normal_data_df[features].to_numpy()
        
        print(f"Training Isolation Forest on {len(X_train)} normal baseline records...")
        model = IsolationForest(contamination=0.01, random_state=42)
        model.fit(X_train)
        print("Isolation Forest trained successfully on real manufacturing dataset.")
    else:
        print("Dataset file not found; falling back to synthetic baseline generation...")
        np.random.seed(42)
        normal_delays = np.random.normal(100, 15, 300)
        normal_errors = np.clip(np.random.normal(2, 0.5, 300), 0, 100)
        normal_throughputs = np.random.normal(500, 25, 300)
        X_train = np.column_stack((normal_delays, normal_errors, normal_throughputs))
        
        model = IsolationForest(contamination=0.01, random_state=42)
        model.fit(X_train)
        print("Isolation Forest trained on synthetic baseline.")

# AI Explanation generator for anomalies
def generate_ai_explanation(delay: float, error_rate: float, throughput: float) -> str:
    reasons = []
    
    # Check for real-world manufacturing dataset thresholds (or fallback synthetic thresholds)
    if delay > 25:
        reasons.append(f"Critical spike in downtime/delay ({delay:.1f} min, baseline ~7 min)")
    elif delay > 14:
        reasons.append(f"Elevated delay ({delay:.1f} min, baseline ~7 min)")
        
    if throughput < 0.5:
        reasons.append(f"Severe drop in production rate ({throughput:.2f} units/min, baseline ~1.87)")
    elif throughput < 1.0:
        reasons.append(f"Low production rate ({throughput:.2f} units/min, baseline ~1.87)")
        
    if error_rate > 60:
        reasons.append(f"High defect/error rate ({error_rate:.1f}%, baseline ~40%)")
        
    if not reasons:
        reasons.append(f"Multivariate anomaly signature (Delay/Downtime: {delay:.1f}, Error: {error_rate:.1f}%, Throughput: {throughput:.2f})")
        
    return " accompanied by ".join(reasons)

# Central metric processing engine
def process_telemetry_data(line_id: str, delay: float, error_rate: float, throughput: float) -> tuple[bool, Optional[int]]:
    global model
    if model is None:
        raise RuntimeError("Model is not initialized.")
        
    timestamp = datetime.datetime.now(datetime.timezone.utc).isoformat()
    
    # Clean inputs
    delay = round(float(delay), 2)
    error_rate = round(float(error_rate), 2)
    throughput = round(float(throughput), 2)
    
    # Predict anomaly (-1 indicates anomaly, 1 indicates normal)
    features = np.array([[delay, error_rate, throughput]])
    prediction = model.predict(features)[0]
    
    # Isolation Forest anomaly score (lower is more anomalous)
    raw_anomaly_score = float(model.decision_function(features)[0])
    
    alert_triggered = False
    alert_id = None
    
    if prediction == -1:
        alert_triggered = True
        # Determine severity
        is_critical = (delay > 25) or (throughput < 0.5) or (error_rate > 60)
        severity = "CRITICAL" if is_critical else "HIGH"
        
        # Build explanation
        ai_explanation = generate_ai_explanation(delay, error_rate, throughput)
        
        # Write alert to SQLite
        with get_db() as conn:
            cursor = conn.cursor()
            cursor.execute("""
                INSERT INTO alerts (timestamp, severity, anomaly_score, ai_explanation, status)
                VALUES (?, ?, ?, ?, ?)
            """, (timestamp, severity, raw_anomaly_score, ai_explanation, "ACTIVE"))
            alert_id = cursor.lastrowid
            conn.commit()
            
        # Terminal print log
        print(f"ALERT GENERATED: [Severity: {severity}] - {ai_explanation} (Anomaly Score: {raw_anomaly_score:.4f})")
        
    # Save telemetry record to SQLite
    with get_db() as conn:
        cursor = conn.cursor()
        cursor.execute("""
            INSERT INTO telemetry (timestamp, line_id, processing_delay_ms, error_rate_pct, throughput_units)
            VALUES (?, ?, ?, ?, ?)
        """, (timestamp, line_id, delay, error_rate, throughput))
        conn.commit()
        
    return alert_triggered, alert_id

# Telemetry background simulator loop (samples from dataset or synthetic)
def simulator_loop():
    global SIMULATOR_FAILURE_INJECTED, simulator_running, normal_data_df, disrupt_data_df
    # Give the server a second to start up fully
    time.sleep(1.0)
    
    print("Telemetry simulator background daemon started.")
    while simulator_running:
        try:
            if normal_data_df is not None and disrupt_data_df is not None:
                # Using real dataset
                if SIMULATOR_FAILURE_INJECTED:
                    sample = disrupt_data_df.sample(1).iloc[0]
                else:
                    # 90% normal, 10% natural disruption
                    if random.random() < 0.90:
                        sample = normal_data_df.sample(1).iloc[0]
                    else:
                        sample = disrupt_data_df.sample(1).iloc[0]
                
                line_id = str(sample.get('production_line_id', 'LINE_A'))
                delay = float(sample.get('downtime_minutes', 0.0))
                error_rate = float(sample.get('defect_rate', 0.0))
                throughput = float(sample.get('production_rate', 0.0))
            else:
                # Using synthetic generator fallback
                if SIMULATOR_FAILURE_INJECTED:
                    delay = random.uniform(800, 1050)
                    error_rate = random.uniform(25, 40)
                    throughput = random.uniform(30, 95)
                else:
                    if random.random() < 0.90:
                        delay = random.normalvariate(100, 8)
                        error_rate = max(0.0, random.normalvariate(2, 0.4))
                        throughput = random.normalvariate(500, 15)
                    else:
                        delay = random.uniform(800, 1050)
                        error_rate = random.uniform(25, 40)
                        throughput = random.uniform(30, 95)
                line_id = "LINE_A"
                    
            process_telemetry_data(
                line_id=line_id,
                delay=delay,
                error_rate=error_rate,
                throughput=throughput
            )
        except Exception as e:
            print(f"Simulator Loop error: {e}")
        time.sleep(3.0)

# FastAPI lifespan for startup/shutdown actions
@asynccontextmanager
async def lifespan(app: FastAPI):
    global simulator_running
    # Initialize DB & Model
    init_db()
    init_anomaly_detector()
    
    # Start simulator daemon thread
    simulator_running = True
    thread = threading.Thread(target=simulator_loop, daemon=True)
    thread.start()
    
    yield
    # Shutdown logic
    print("Shutting down telemetry simulator...")
    simulator_running = False

# Initialize FastAPI App
app = FastAPI(
    title="AI-Enabled Production Disruption Early Warning System",
    description="Hackathon backend prototype for real-time manufacturing anomaly detection using Isolation Forest.",
    version="1.0.0",
    lifespan=lifespan
)

# Enable CORS for frontend integration
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
def read_root():
    return {"message": "Production Disruption Early Warning System API is running."}

@app.get("/favicon.ico", include_in_schema=False)
def favicon():
    return Response(status_code=204)

@app.get("/api/status", response_model=StatusResponse)
def get_status():
    """
    Returns the current system status ("NORMAL" or "DISRUPTED") and the latest telemetry record.
    """
    with get_db() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM telemetry ORDER BY id DESC LIMIT 1")
        row = cursor.fetchone()
        latest_telemetry = dict(row) if row else None
        
        cursor.execute("SELECT COUNT(*) as count FROM alerts WHERE status = 'ACTIVE'")
        active_alerts_count = cursor.fetchone()["count"]
        
    status = "DISRUPTED" if (active_alerts_count > 0 or SIMULATOR_FAILURE_INJECTED) else "NORMAL"
    return {
        "status": status,
        "latest_telemetry": latest_telemetry
    }

@app.get("/api/alerts")
def get_alerts():
    """
    Returns the top 10 recent disruption alerts.
    """
    with get_db() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM alerts ORDER BY id DESC LIMIT 10")
        rows = cursor.fetchall()
        alerts = [dict(r) for r in rows]
    return alerts

@app.post("/api/inject-failure")
def inject_failure():
    """
    Demo helper endpoint to toggle the simulator into forcing disruption data immediately.
    """
    global SIMULATOR_FAILURE_INJECTED
    SIMULATOR_FAILURE_INJECTED = True
    print("SYSTEM WARNING: Manually injected failure mode enabled.")
    return {"message": "Failure mode injected. Telemetry simulator is now generating anomalous readings."}

@app.post("/api/reset-demo")
def reset_demo():
    """
    Toggles the simulator back to normal state and clears old alerts from the database.
    """
    global SIMULATOR_FAILURE_INJECTED
    SIMULATOR_FAILURE_INJECTED = False
    
    with get_db() as conn:
        cursor = conn.cursor()
        cursor.execute("DELETE FROM alerts")
        conn.commit()
        
    print("SYSTEM INFO: Demo reset. Simulator restored to normal state and alerts cleared.")
    return {"message": "Demo reset complete. Simulator restored to normal state and alerts cleared."}

@app.get("/api/telemetry/history")
def get_telemetry_history():
    """
    Returns the last 20 telemetry records for graph visualization (chronological order).
    """
    with get_db() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM telemetry ORDER BY id DESC LIMIT 20")
        rows = cursor.fetchall()
        history = [dict(r) for r in reversed(rows)]
    return history

@app.post("/api/telemetry")
def upload_telemetry(data: TelemetryUpload):
    """
    Allows manual telemetry upload from APIs, pushing features through the Isolation Forest anomaly detector.
    """
    alert_triggered, alert_id = process_telemetry_data(
        line_id=data.line_id,
        delay=data.processing_delay_ms,
        error_rate=data.error_rate_pct,
        throughput=data.throughput_units
    )
    return {
        "status": "success",
        "alert_triggered": alert_triggered,
        "alert_id": alert_id
    }

class ChatRequest(BaseModel):
    query: str
    scenario: Optional[str] = None
    history: Optional[List[Dict[str, str]]] = []

@app.get("/api/scenarios")
def get_scenarios():
    """
    Returns the list of 5 predefined factory disruption scenarios.
    """
    context = ai_service.load_context()
    return context.get("scenarios", {})

@app.post("/api/scenarios/{scenario_name}/activate")
def activate_scenario(scenario_name: str):
    """
    Loads and activates a specific scenario into the database and telemetry state.
    """
    context = ai_service.load_context()
    scenarios = context.get("scenarios", {})
    if scenario_name not in scenarios:
        raise HTTPException(status_code=404, detail=f"Scenario '{scenario_name}' not found")
        
    scenario_data = scenarios[scenario_name]
    
    # Overwrite SQLite tables with scenario data
    with get_db() as conn:
        cursor = conn.cursor()
        cursor.execute("DELETE FROM alerts")
        cursor.execute("DELETE FROM telemetry")
        
        # Insert telemetry history
        for record in scenario_data.get("telemetry_history", []):
            cursor.execute("""
                INSERT INTO telemetry (timestamp, line_id, processing_delay_ms, error_rate_pct, throughput_units)
                VALUES (?, ?, ?, ?, ?)
            """, (
                record.get("timestamp", datetime.datetime.now(datetime.timezone.utc).isoformat()),
                record.get("line_id", "LINE-02-CHASSIS-SUBFRAME"),
                record.get("processing_delay_ms", 45.0),
                record.get("error_rate_pct", 0.25),
                record.get("throughput_units", 120.0)
            ))
            
        # Insert recent alerts
        for alert in scenario_data.get("recent_alerts", []):
            cursor.execute("""
                INSERT INTO alerts (timestamp, severity, anomaly_score, ai_explanation, status)
                VALUES (?, ?, ?, ?, ?)
            """, (
                alert.get("timestamp", datetime.datetime.now(datetime.timezone.utc).isoformat()),
                alert.get("severity", "HIGH"),
                alert.get("anomaly_score", -0.70),
                alert.get("ai_explanation", "Scenario generated alert"),
                alert.get("status", "ACTIVE")
            ))
        conn.commit()
        
    global SIMULATOR_FAILURE_INJECTED
    SIMULATOR_FAILURE_INJECTED = (scenario_data.get("status") == "DISRUPTED")
    
    return {
        "message": f"Scenario '{scenario_name}' activated successfully.",
        "status": scenario_data.get("status"),
        "scenario_name": scenario_data.get("scenario_name")
    }

@app.get("/api/ai/decision-support")
def get_ai_decision_support():
    """
    Generates structured 5-part GenAI decision support for the current live status.
    """
    with get_db() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM telemetry ORDER BY id DESC LIMIT 1")
        row = cursor.fetchone()
        latest_telemetry = dict(row) if row else {}
        
        cursor.execute("SELECT * FROM alerts WHERE status = 'ACTIVE' ORDER BY id DESC")
        alerts = [dict(r) for r in cursor.fetchall()]
        
    status = "DISRUPTED" if len(alerts) > 0 or SIMULATOR_FAILURE_INJECTED else "NORMAL"
    
    current_context = {
        "status": status,
        "latest_telemetry": latest_telemetry,
        "recent_alerts": alerts
    }
    
    analysis = ai_service.analyze_context(current_context)
    return analysis

@app.post("/api/ai/decision-support/scenario/{scenario_name}")
def get_scenario_decision_support(scenario_name: str):
    """
    Generates structured 5-part GenAI decision support for a specific named scenario.
    """
    try:
        analysis = ai_service.analyze_scenario(scenario_name)
        return analysis
    except KeyError:
        raise HTTPException(status_code=404, detail=f"Scenario '{scenario_name}' not found")

@app.post("/api/ai/chat")
def chat_with_assistant(request: ChatRequest):
    """
    Conversational AI factory assistant routing queries dynamically across roles.
    """
    context = ai_service.load_context()
    if request.scenario and request.scenario in context.get("scenarios", {}):
        active_context = context["scenarios"][request.scenario]
    else:
        with get_db() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM telemetry ORDER BY id DESC LIMIT 1")
            row = cursor.fetchone()
            latest_telemetry = dict(row) if row else {}
            
            cursor.execute("SELECT * FROM alerts WHERE status = 'ACTIVE' ORDER BY id DESC")
            alerts = [dict(r) for r in cursor.fetchall()]
            
        status = "DISRUPTED" if len(alerts) > 0 or SIMULATOR_FAILURE_INJECTED else "NORMAL"
        active_context = {
            "status": status,
            "latest_telemetry": latest_telemetry,
            "recent_alerts": alerts
        }
        
    response_text = ai_service.chat_assistant(active_context, request.history or [], request.query)
    return {"reply": response_text}