Spaces:
Sleeping
Sleeping
| 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) | |