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)