geekaijosh/MaintenanceFaultDetectionAgent

1import os2import json3import numpy as np4import gradio as gr5from sklearn.ensemble import IsolationForest6from dotenv import load_dotenv7from datetime import datetime8import time9import threading10 11# Load environment variables (optional)12load_dotenv(override=True)13 14# === Anomaly Detection Model ===15class AnomalyDetector:16    def __init__(self):17        # Train with realistic "normal" ranges18        normal_temp = np.random.normal(loc=50, scale=5, size=(200, 1))19        normal_vibration = np.random.normal(loc=5, scale=1, size=(200, 1))20        normal_pressure = np.random.normal(loc=100, scale=10, size=(200, 1))21 22        normal_data = np.hstack([normal_temp, normal_vibration, normal_pressure])23        self.model = IsolationForest(n_estimators=100, contamination=0.05, random_state=42)24        self.model.fit(normal_data)25 26    def is_anomaly(self, temp, vibration, pressure):27        X = np.array([[temp, vibration, pressure]])28        return self.model.predict(X)[0] == -129 30detector = AnomalyDetector()31 32# === Logging Function ===33def log_event(data, anomaly):34    log_entry = {35        "data": data,36        "anomaly": "true" if anomaly else "false",  # store as string37        "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")38    }39    try:40        with open("maintenance_log.json", "a") as f:41            f.write(json.dumps(log_entry) + "\n")42    except Exception as e:43        print(f"[ERROR] Failed to write log: {e}")44 45# === Generate Random Sensor Data ===46def get_random_sensor_data():47    temp = round(np.random.normal(50, 5), 2)48    vibration = round(np.random.normal(5, 1), 2)49    pressure = round(np.random.normal(100, 10), 2)50    return temp, vibration, pressure51 52# === Core Detection Logic ===53def detect_anomaly(temp, vibration, pressure):54    sensor_data = {"temp": temp, "vibration": vibration, "pressure": pressure}55    anomaly = detector.is_anomaly(temp, vibration, pressure)56    log_event(sensor_data, anomaly)57 58    if anomaly:59        return f"🚨 Anomaly Detected!\nTemp: {temp}, Vibration: {vibration}, Pressure: {pressure}"60    else:61        return f"✅ Normal Data: {sensor_data}"62 63# === Live Update Generator for Gradio Live Feed ===64def live_feed():65    while True:66        temp, vibration, pressure = get_random_sensor_data()67        result = detect_anomaly(temp, vibration, pressure)68        yield result69        time.sleep(3)  # update every 3 seconds70 71# === Gradio UI ===72with gr.Blocks(title="Maintenance & Fault Detection AI") as demo:73    gr.Markdown("# 🏗 Maintenance & Fault Detection Agent — Live Feed")74    gr.Markdown("This AI monitors sensor readings in real time and flags anomalies automatically.")75 76    output = gr.Textbox(label="Live Monitoring Output", lines=6)77 78    # Button for manual detection79    with gr.Row():80        temp_in = gr.Number(label="Temperature (Manual Input)", value=50)81        vib_in = gr.Number(label="Vibration (Manual Input)", value=5)82        pres_in = gr.Number(label="Pressure (Manual Input)", value=100)83    manual_btn = gr.Button("Run Manual Detection")84    manual_btn.click(detect_anomaly, inputs=[temp_in, vib_in, pres_in], outputs=output)85 86    # Live feed updates automatically87    live = gr.Textbox(label="Automatic Live Feed", lines=6)88    gr.Markdown("### 🔴 Live Updates (refresh every 3s)")89    live_stream = gr.Textbox(value=live_feed, label="Live Sensor Data Feed", lines=6)90 91demo.launch()92