app.py

import os
import logging
import pandas as pd
import numpy as np
import joblib
from datetime import datetime
from tensorflow.keras.models import load_model
from database.coldroom_db import fetch_coldroom_data
from database.tank_db import fetch_tank_data
from train.coldroom.train_coldroom import train_coldroom
from train.tanks_1_6.train_tanks_1_6 import train_tank_1_6
from train.tanks_7_13.train_tanks_7_13 import train_tank_7_13
from status_evaluator import evaluate_coldroom_status, evaluate_tank_status
import re
import gradio as gr
import io
import sys
from contextlib import redirect_stdout
# Logging Configuration
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
TIME_STEPS = 30
MAX_FT_DEFAULT = 25.0
SPIKE_THRESHOLD = 1.5
TANK_MAX_LEVELS = {
    'physical_07': 15.0,
    'physical_08': 25.0,
    'physical_09': 25.0,
    'physical_11': 50.0,
    'physical_12': 50.0,
    'physical_13': 50.0
}
def ensure_dir(path):
    if not os.path.exists(path):
        os.makedirs(path, exist_ok=True)
def slugify(text):
    return re.sub(r'[^a-zA-Z0-9]', '', text).lower()
def is_peak_hour(dt):
    # Peak Hours: 10:00 to 18:00
    return 10 <= dt.hour <= 18
def get_time_flags(dt):
    is_night = (dt.hour >= 22 or dt.hour <= 5)
    is_weekend = (dt.weekday() >= 5)
    is_peak = is_peak_hour(dt)
    return is_night, is_weekend, is_peak
def process_coldrooms(df):
    logger.info("Processing Coldrooms (Inference)...")
    results = []
    if df is None: return results
    for name, group in df.groupby('coldroom_name'):
        slug = slugify(name)
        model_dir = f"models/coldroom/{slug}"
        csv_path = f"data/coldroom/{slug}/data.csv"
        
        # 1. Train Model
        print(f"Now training the model for {name}...")
        ensure_dir(f"data/coldroom/{slug}")
        ensure_dir(model_dir)
        group.to_csv(csv_path, index=False)
        stats = train_coldroom(csv_path, model_dir)
        
        if not stats: 
            print(f"Failed to train model for {name}")
            continue
        
        # Load Artifacts back for prediction
        model = load_model(os.path.join(model_dir, 'model.h5'), compile=False)
        scaler = joblib.load(os.path.join(model_dir, 'scaler.joblib'))
        threshold = stats['threshold']
        
        # 2. Prediction on latest data (Predict)
        group = group.sort_values('sensor_timestamp').copy()
        group = group.dropna(subset=['temperature', 'humidity'])
        
        if len(group) < TIME_STEPS:
            print(f"{name} has insufficient valid data after NaN drop.")
            results.append({"name": name, "status": "Buffering", "level": "Normal"})
            continue
            
        group["temp_diff"] = group["temperature"].diff().fillna(0)
        group["hum_diff"] = group["humidity"].diff().fillna(0)
        group["rolling_mean"] = group["temperature"].rolling(12).mean().fillna(group["temperature"].mean())
        group["rolling_std"] = group["temperature"].rolling(12).std().fillna(0)
        
        features = ["temperature", "humidity", "temp_diff", "rolling_mean", "rolling_std"]
        X_scaled = scaler.transform(group[features].values)
        
        sequences = np.array([X_scaled[-TIME_STEPS:]]) # Latest window
        X_pred = model.predict(sequences, verbose=0)
        mse = np.mean(np.power(sequences - X_pred, 2))
        
        # Anomaly Logic
        curr_row = group.iloc[-1]
        temp = float(curr_row["temperature"])
        hum = float(curr_row["humidity"])
        ts = curr_row["sensor_timestamp"]
        temp_diff = abs(curr_row["temp_diff"])
        hum_diff = abs(curr_row["hum_diff"])
        
        anomaly_list = []
        if temp == 0 and hum == 0: anomaly_list.append({"type": "sensor_fault"})
        elif temp_diff > 3 or hum_diff > 10: anomaly_list.append({"type": "sudden_spike"})
        elif mse > threshold:
            if not (temp_diff < 0.5 and hum_diff < 2): # Tend ignore
                anomaly_list.append({"type": "model_anomaly"})
        
        anomaly_detected = len(anomaly_list) > 0
        is_night, is_weekend, is_peak = get_time_flags(ts)
        
        intense, level = evaluate_coldroom_status(
            anomaly_detected, temp, is_night, is_weekend, is_peak, mse, threshold
        )
        
        # Show model output and eval
        status_text = "Anomaly" if anomaly_detected else "Normal"
        print(f"Model Output for {name}: MSE={mse:.6f}, Threshold={threshold:.6f}")
        print(f"Evaluation: Status={status_text}, Level={level}, Intensity={intense}")
        if anomaly_detected:
            print(f"Detected Anomalies: {anomaly_list}")
        print("-" * 20)
        results.append({
            "name": name,
            "anomaly": 1 if anomaly_detected else 0,
            "intense": intense,
            "level": level,
            "latest_temp": temp,
            "mse": mse,
            "threshold": threshold,
            "anomalies": anomaly_list
        })
    return results
def process_tanks(df):
    logger.info("Processing Tanks (Inference)...")
    results = []
    if df is None: return results
    for tank_id, group in df.groupby('tank_id'):
        try:
            tank_num = int(re.search(r'\d+', tank_id).group())
        except: continue
        
        # Folder Mapping
        if 1 <= tank_num <= 6:
            parent_model = "tanks_1_to_6"
            train_func = train_tank_1_6
        else:
            parent_model = "tanks_7_to_13"
            train_func = train_tank_7_13
            
        model_dir = f"models/{parent_model}/tank{tank_num}"
        csv_path = f"data/tanks/{parent_model}/tank{tank_num}/data.csv"
        
        # 1. Train Model
        print(f"Now training the model for {tank_id}...")
        ensure_dir(os.path.dirname(csv_path))
        ensure_dir(model_dir)
        group.to_csv(csv_path, index=False)
        stats = train_func(csv_path, model_dir)
        
        if not stats: 
            print(f"Failed to train model for {tank_id}")
            continue
        
        # Load Artifacts
        model = load_model(os.path.join(model_dir, 'model.h5'), compile=False)
        scaler = joblib.load(os.path.join(model_dir, 'scaler.joblib'))
        config = joblib.load(os.path.join(model_dir, 'config.joblib'))
        
        threshold = config["threshold"]
        features_list = config["features"]
        max_ft = TANK_MAX_LEVELS.get(tank_id.lower().replace(' ', '_'), MAX_FT_DEFAULT)
        
        # 2. Prediction (Predict)
        group = group.sort_values("sensor_timestamp").copy()
        group = group.dropna(subset=['level_feet'])
        
        if len(group) < TIME_STEPS:
            print(f"{tank_id} has insufficient valid data after NaN drop.")
            results.append({"name": tank_id, "status": "Buffering", "level": "Normal"})
            continue
            
        group["headspace"] = max_ft - group["level_feet"]
        group["fill_pct"] = (group["level_feet"] / max_ft) * 100
        group["roc"] = group["level_feet"].diff()
        group["roc_abs"] = group["roc"].abs()
        group["accel"] = group["roc"].diff()
        group["roll_mean"] = group["level_feet"].rolling(60, min_periods=1).mean()
        group["roll_std"] = group["level_feet"].rolling(60, min_periods=1).std().fillna(0)
        group["roll_range"] = group["level_feet"].rolling(60, min_periods=1).max() - group["level_feet"].rolling(60, min_periods=1).min()
        group["dev_from_mean"] = group["level_feet"] - group["roll_mean"]
        
        mu = group["level_feet"].mean()
        sig = group["level_feet"].std() + 1e-9
        group["z_score"] = (group["level_feet"] - mu) / sig
        
        ts_latest = group["sensor_timestamp"].iloc[-1]
        group["hour"] = ts_latest.hour
        group["minute"] = ts_latest.minute
        group["day_of_week"] = ts_latest.weekday()
        group["is_night"] = ((ts_latest.hour >= 22) | (ts_latest.hour <= 5))
        group = group.fillna(0)
        
        X_scaled = scaler.transform(group[features_list].values)
            
        sequences = np.array([X_scaled[-TIME_STEPS:]])
        X_pred = model.predict(sequences, verbose=0)
        mae = np.mean(np.abs(sequences - X_pred))
        
        # Anomaly Logic
        curr_row = group.iloc[-1]
        level_val = float(curr_row["level_feet"])
        roc_abs = float(curr_row["roc_abs"])
        
        anomaly_list = []
        if level_val == 0.0: anomaly_list.append({"type": "sensor_fault"})
        elif level_val < 0 or level_val > (max_ft + 5): anomaly_list.append({"type": "impossible_value"})
        elif roc_abs > SPIKE_THRESHOLD: anomaly_list.append({"type": "sudden_spike"})
        elif mae > threshold:
            if mae > threshold * 10: anomaly_list.append({"type": "critical_anomaly"})
            elif roc_abs >= 0.05: anomaly_list.append({"type": "model_anomaly"})
            
        anomaly_detected = len(anomaly_list) > 0
        is_night, _, _ = get_time_flags(ts_latest)
        
        intense, level_status = evaluate_tank_status(anomaly_detected, anomaly_list, is_night)
        
        # Show model output and eval
        status_text = "Anomaly" if anomaly_detected else "Normal"
        print(f"Model Output for {tank_id}: MAE={mae:.6f}, Threshold={threshold:.6f}")
        print(f"Evaluation: Status={status_text}, Level={level_status}, Intensity={intense}")
        if anomaly_detected:
            print(f"Detected Anomalies: {anomaly_list}")
        print("-" * 20)
        results.append({
            "name": tank_id,
            "anomaly": 1 if anomaly_detected else 0,
            "intense": intense,
            "level": level_status,
            "latest_level": level_val,
            "mae": mae,
            "threshold": threshold,
            "anomalies": anomaly_list
        })
    return results
def main():
    logger.info("Executing Primary Anomaly Pipeline (Fetch -> Train -> Show Result)...")
    
    # FETCH
    logger.info("Fetching 7-day telemetry data...")
    cr_data = fetch_coldroom_data(days=7)
    tank_data = fetch_tank_data(days=7)
    # DATA RETRIEVAL REPORT (PRE-TRAINING)
    if cr_data is not None:
        for name, group in cr_data.groupby('coldroom_name'):
            print(f"{name} data retrieved: {len(group)} rows")
    
    if tank_data is not None:
        for tank_id, group in tank_data.groupby('tank_id'):
            print(f"{tank_id} data retrieved: {len(group)} rows")
            
    # TRAIN & PREDICT
    cr_results = process_coldrooms(cr_data)
    tank_results = process_tanks(tank_data)
    
    # SHOW RESULT
    print("\n" + "="*60)
    print(" ANOMALY DETECTION PIPELINE SUMMARY REPORT ")
    print("="*60)
    
    print("\nCOLDROOMS:")
    print(f"{'Asset Name':<25} | {'Temp':<8} | {'Status':<15} | {'Level':<10}")
    print("-"*60)
    for r in cr_results:
        print(f"{r['name']:<25} | {r.get('latest_temp', 0):<8.2f} | {'Anomaly' if r['anomaly'] else 'Normal':<15} | {r['level']:<10}")
        
    print("\nREFINERY TANKS:")
    print(f"{'Asset Name':<25} | {'Level':<8} | {'Status':<15} | {'Level':<10}")
    print("-"*60)
    for r in tank_results:
        print(f"{r['name']:<25} | {r.get('latest_level', 0):<8.2f} | {'Anomaly' if r['anomaly'] else 'Normal':<15} | {r['level']:<10}")
        
    print("\n" + "="*60)
    logger.info("Pipeline Execution Complete.")
def run_pipeline_ui():
    f = io.StringIO()
    with redirect_stdout(f):
        try:
            main()
        except Exception as e:
            print(f"\nERROR: {str(e)}")
    return f.getvalue()
# Gradio Interface
with gr.Blocks(title="Industrial Anomaly Detection Pipeline") as demo:
    gr.Markdown("# 🏭 Industrial Anomaly Detection Pipeline")
    gr.Markdown("Click the button below to fetch 7-day telemetry data, retrain all LSTM models, and evaluate the latest status for all Coldrooms and Refinery Tanks.")
    
    with gr.Row():
        run_btn = gr.Button("🚀 Run Full Pipeline", variant="primary")
    
    output_logs = gr.Textbox(
        label="Execution Logs",
        placeholder="Logs will appear here after clicking Run...",
        lines=25,
        max_lines=50,
        autoscroll=True
    )
    
    run_btn.click(fn=run_pipeline_ui, outputs=output_logs)
    
    gr.Markdown("---")
    gr.Markdown("### Deployment Info")
    gr.Markdown("- **SDK**: Gradio\n- **Environment**: Hugging Face Spaces\n- **Database**: PostgreSQL (Secrets managed via `POSTGRES_URI`) ")
if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860)