Update app.py

app.py

@@ -1,61 +1,96 @@
-import streamlit as st
+import gradio as gr
 import pandas as pd
-import matplotlib.pyplot as plt
+import numpy as np
 import torch
-from transformers import AutoModelForSequenceClassification, AutoTokenizer
+from momentfm import MOMENTPipeline
+from scipy.stats import zscore
 
-# Transformer model for anomaly detection
-MODEL_NAME = "your-username/smartgrid-anomaly-transformer"  # replace with your Hugging Face model
-
-tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
-model = AutoModelForSequenceClassification.from_pretrained(MODEL_NAME)
-
-# Thresholds for fallback (optional)
-VOLTAGE_MIN, VOLTAGE_MAX = 215, 235
-CURRENT_MAX = 35
-FREQ_MIN, FREQ_MAX = 49.5, 50.5
-
-st.title("⚡ Smart Grid Anomaly Detection (Transformer-Based)")
-
-uploaded_file = st.file_uploader(
-    "Upload your sensor data file (.csv or .xlsx)", type=["csv", "xlsx"]
+# Load the model
+model = MOMENTPipeline.from_pretrained(
+    "AutonLab/MOMENT-1-large",
+    model_kwargs={"task_name": "reconstruction", "n_channels": 3},
 )
+model.init()
+model.eval()
 
-if uploaded_file:
-    if uploaded_file.name.endswith(".csv"):
-        df = pd.read_csv(uploaded_file)
-    else:
-        df = pd.read_excel(uploaded_file)
-
-    df.columns = [c.strip().lower() for c in df.columns]
-    expected_cols = {"timestamp", "voltage", "current", "frequency"}
-    if not expected_cols.issubset(df.columns):
-        st.error("File missing columns: timestamp, voltage, current, frequency")
-    else:
-        # Prepare data for Transformer (as text sequence)
-        sequences = df.apply(
-            lambda row: f"{row['voltage']} {row['current']} {row['frequency']}", axis=1
-        ).tolist()
-        
-        inputs = tokenizer(sequences, padding=True, truncation=True, return_tensors="pt")
-        with torch.no_grad():
-            outputs = model(**inputs)
-            predictions = torch.argmax(outputs.logits, dim=1).numpy()
-
-        df["anomaly"] = predictions  # 1 = anomaly, 0 = normal
+def detect_anomalies(data, window_size=512, stride=128):
+    n, num_features = data.shape
+    errors = np.zeros(n)
+    counts = np.zeros(n)
+    with torch.no_grad():
+        for start in range(0, n - window_size + 1, stride):
+            end = start + window_size
+            window = data[start:end]
+            x = torch.from_numpy(window).float().unsqueeze(0)  # (1, window_size, 3)
+            output = model(x)
+            recon = output.reconstruction  # (1, window_size, 3)
+            step_errors = ((x - recon) ** 2).mean(dim=2).squeeze(0).numpy()  # (window_size,)
+            errors[start:end] += step_errors
+            counts[start:end] += 1
+    errors /= np.maximum(counts, 1e-6)
+    return errors
 
-        st.subheader("Anomaly Detection Output")
-        st.dataframe(df)
+def process_csv(file):
+    if not file:
+        return "Please upload a CSV file."
+    try:
+        df = pd.read_csv(file)
+        required_cols = ['timestamp', 'voltage', 'current', 'frequency']
+        if not all(col in df.columns for col in required_cols):
+            return "CSV must contain columns: timestamp, voltage, current, frequency"
+        df['timestamp'] = pd.to_datetime(df['timestamp'])
+        df = df.sort_values('timestamp')
+        features = ['voltage', 'current', 'frequency']
+        data = df[features].values.astype(float)  # (len, 3)
+        # Normalize data per channel
+        data = zscore(data, axis=0)
+        seq_len = len(data)
+        if seq_len < 512:
+            # Pad with zeros
+            pad = np.zeros((512 - seq_len, 3))
+            data_padded = np.vstack((data, pad))
+            errors = detect_anomalies(data_padded)[:seq_len]
+        else:
+            errors = detect_anomalies(data)
+        # Compute threshold
+        mean_e = np.mean(errors)
+        std_e = np.std(errors)
+        threshold = mean_e + 3 * std_e
+        is_anomaly = errors > threshold
+        # Severity score (scaled 0-10)
+        severity = np.clip((errors - mean_e) / (3 * std_e), 0, np.inf) * 10
+        # For explanations, use statistical deviations
+        means = df[features].mean().values
+        stds = df[features].std().values
+        z_scores = (df[features].values - means) / (stds + 1e-6)
+        explanations = []
+        for i in range(seq_len):
+            if not is_anomaly[i]:
+                explanations.append("Normal")
+            else:
+                reasons = []
+                for j, feat in enumerate(features):
+                    if abs(z_scores[i, j]) > 3:
+                        direction = "High" if z_scores[i, j] > 0 else "Low"
+                        reasons.append(f"{direction} {feat}")
+                exp = " and ".join(reasons) if reasons else "Unusual pattern detected by the model"
+                explanations.append(exp)
+        # Add columns to dataframe
+        df['Anomaly'] = ['Yes' if a else 'No' for a in is_anomaly]
+        df['Severity Score'] = severity.round(2)
+        df['Explanation'] = explanations
+        return df
+    except Exception as e:
+        return f"Error processing file: {str(e)}"
 
-        st.subheader("Voltage Plot with Anomalies")
-        fig, ax = plt.subplots(figsize=(10, 4))
-        ax.plot(df["timestamp"], df["voltage"], label="Voltage (V)", color="blue")
-        ax.plot(df[df["anomaly"] == 1]["timestamp"], df[df["anomaly"] == 1]["voltage"], "ro", label="Anomaly")
-        ax.set_xlabel("Timestamp")
-        ax.set_ylabel("Voltage")
-        ax.legend()
-        plt.xticks(rotation=45)
-        st.pyplot(fig)
+# Define Gradio interface
+with gr.Blocks(title="Anomaly Detection in Smart Grid Sensor Data") as demo:
+    gr.Markdown("# Anomaly Detection in Smart Grid Sensor Data Using Transformers")
+    gr.Markdown("Upload a CSV file with columns: timestamp, voltage, current, frequency")
+    input_file = gr.File(label="Upload CSV file", file_types=[".csv"])
+    output_df = gr.Dataframe(label="Results")
+    btn = gr.Button("Detect Anomalies")
+    btn.click(process_csv, inputs=input_file, outputs=output_df)
 
-st.markdown("---")
-st.markdown("Developed by **Haresh Chander** | Powered by Python, Transformers & Streamlit")
+# Launch the app
+demo.launch()