app.py

import gradio as gr
import torch
import torch.nn as nn
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import joblib

# --- 1. DEFINE THE MODEL ARCHITECTURE ---
class LSTMAutoencoder(nn.Module):
    def __init__(self, input_dim=1, hidden_dim=64, num_layers=1, seq_len=50):
        super(LSTMAutoencoder, self).__init__()
        self.seq_len = seq_len
        self.encoder = nn.LSTM(input_dim, hidden_dim, num_layers, batch_first=True)
        self.decoder = nn.LSTM(hidden_dim, hidden_dim, num_layers, batch_first=True)
        self.output_layer = nn.Linear(hidden_dim, input_dim)

    def forward(self, x):
        _, (hidden_n, _) = self.encoder(x)
        latent_vector = hidden_n[-1].unsqueeze(1).repeat(1, self.seq_len, 1)
        decoder_output, _ = self.decoder(latent_vector)
        return self.output_layer(decoder_output)

# --- 2. LOAD MODEL & SCALER ---
model = LSTMAutoencoder()
model.load_state_dict(torch.load("server_guardian_model.pth", map_location=torch.device('cpu'), weights_only=True))
model.eval()
scaler = joblib.load("server_guardian_scaler.gz")

# --- 3. PREDICTION FUNCTION ---
def analyze_log(file_obj, threshold):
    # Load Data
    df = pd.read_csv(file_obj.name)
    
    # Preprocess
    data_values = df['value'].values.astype(float).reshape(-1, 1)
    data_normalized = scaler.transform(data_values) 

    # Create Sequences
    SEQUENCE_LENGTH = 50
    xs = []
    for i in range(len(data_normalized) - SEQUENCE_LENGTH):
        xs.append(data_normalized[i:(i + SEQUENCE_LENGTH)])
    X_tensor = torch.FloatTensor(np.array(xs))

    # Inference
    with torch.no_grad():
        reconstructed = model(X_tensor)
        mse_loss = torch.mean((X_tensor - reconstructed)**2, dim=[1, 2]).numpy()

    # Visualization
    fig, ax = plt.subplots(figsize=(10, 4))
    ax.plot(mse_loss, label='Anomaly Score', color='purple')
    ax.axhline(y=threshold, color='red', linestyle='--', label='Threshold')
    ax.set_title("Reconstruction Error (MSE)")
    ax.legend()
    
    # Status Check
    max_error = np.max(mse_loss)
    if max_error > threshold:
        status = f"🚨 CRITICAL WARNING: Anomalies detected! Max Error: {max_error:.4f}"
    else:
        status = f"✅ System Healthy: No anomalies detected. Max Error: {max_error:.4f}"
        
    # Return the preview, the graph, and the text status
    return df.head(), fig, status

# --- 4. GRADIO INTERFACE ---
interface = gr.Interface(
    fn=analyze_log,
    inputs=[
        gr.File(label="Upload Server Log (CSV)"),
        gr.Slider(minimum=0.0, maximum=5.0, step=0.001, value=0.04, label="Sensitivity Threshold")
    ],
    outputs=[
        gr.Dataframe(label="Raw Data Preview"),
        gr.Plot(label="Anomaly Detection Analysis"),
        gr.Textbox(label="System Status")
    ],
    title="🛡️ Server Health Guardian",
    description="Upload a server log (CSV) to detect anomalies using your custom LSTM-Autoencoder."
)

# Launch the app
interface.launch()