app.py

import gradio as gr
import pandas as pd
import numpy as np
import plotly.graph_objects as go
from datetime import datetime

# Fault classification logic
def classify_fault(temperature, vibration, humidity, power, pressure):
    """Classify device status based on sensor readings"""
    
    # Define thresholds
    temp_normal = (15, 35)
    vib_normal = (0, 50)
    humidity_normal = (30, 70)
    power_normal = (80, 120)
    pressure_normal = (950, 1050)
    
    faults = []
    severity_score = 0
    
    # Check temperature
    if temperature < temp_normal[0] or temperature > temp_normal[1]:
        if temperature < temp_normal[0] - 10 or temperature > temp_normal[1] + 10:
            faults.append("❌ **CRITICAL**: Temperature out of range")
            severity_score += 3
        else:
            faults.append("⚠️ **WARNING**: Temperature abnormal")
            severity_score += 1
    
    # Check vibration
    if vibration > vib_normal[1]:
        if vibration > vib_normal[1] * 1.5:
            faults.append("❌ **CRITICAL**: Excessive vibration detected")
            severity_score += 3
        else:
            faults.append("⚠️ **WARNING**: Elevated vibration")
            severity_score += 1
    
    # Check humidity
    if humidity < humidity_normal[0] or humidity > humidity_normal[1]:
        faults.append("⚠️ **WARNING**: Humidity out of optimal range")
        severity_score += 1
    
    # Check power
    if power < power_normal[0] or power > power_normal[1]:
        if power < power_normal[0] * 0.7 or power > power_normal[1] * 1.3:
            faults.append("❌ **CRITICAL**: Power consumption anomaly")
            severity_score += 3
        else:
            faults.append("⚠️ **WARNING**: Power consumption unusual")
            severity_score += 1
    
    # Check pressure
    if pressure < pressure_normal[0] or pressure > pressure_normal[1]:
        faults.append("⚠️ **WARNING**: Pressure deviation detected")
        severity_score += 1
    
    # Determine overall status
    if severity_score == 0:
        status = "✅ NORMAL"
        color = "green"
        recommendation = "Device operating within normal parameters. Continue routine monitoring."
    elif severity_score <= 3:
        status = "⚠️ WARNING"
        color = "orange"
        recommendation = "Schedule inspection. Monitor closely for further degradation."
    else:
        status = "❌ FAILURE"
        color = "red"
        recommendation = "IMMEDIATE ACTION REQUIRED. Stop device and perform maintenance."
    
    # Create gauge chart
    fig = go.Figure(go.Indicator(
        mode="gauge+number+delta",
        value=severity_score,
        domain={'x': [0, 1], 'y': [0, 1]},
        title={'text': "Fault Severity Score"},
        delta={'reference': 0},
        gauge={
            'axis': {'range': [None, 10]},
            'bar': {'color': color},
            'steps': [
                {'range': [0, 3], 'color': "lightgreen"},
                {'range': [3, 6], 'color': "lightyellow"},
                {'range': [6, 10], 'color': "lightcoral"}],
            'threshold': {
                'line': {'color': "red", 'width': 4},
                'thickness': 0.75,
                'value': 7}}))
    
    fig.update_layout(height=300)
    
    # Generate report
    report = f"## {status}\n\n"
    report += f"**Timestamp**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
    report += f"**Severity Score**: {severity_score}/10\n\n"
    
    if faults:
        report += "### Detected Issues:\n"
        for fault in faults:
            report += f"- {fault}\n"
    else:
        report += "### ✅ No Issues Detected\n"
    
    report += f"\n### Recommendation:\n{recommendation}\n\n"
    report += "---\n**Sensor Readings:**\n"
    report += f"- Temperature: {temperature}°C\n"
    report += f"- Vibration: {vibration} Hz\n"
    report += f"- Humidity: {humidity}%\n"
    report += f"- Power: {power} W\n"
    report += f"- Pressure: {pressure} hPa\n"
    
    return fig, report

def classify_from_csv(file):
    """Process CSV file with sensor data"""
    if file is None:
        return None, "Please upload a CSV file"
    
    try:
        df = pd.read_csv(file.name)
        
        # Check required columns
        required = ['temperature', 'vibration', 'humidity', 'power', 'pressure']
        if not all(col in df.columns for col in required):
            return None, f"⚠️ Missing columns. Required: {', '.join(required)}"
        
        # Process last row
        last_row = df.iloc[-1]
        return classify_fault(
            last_row['temperature'],
            last_row['vibration'],
            last_row['humidity'],
            last_row['power'],
            last_row['pressure']
        )
    except Exception as e:
        return None, f"❌ Error processing file: {str(e)}"

# Create Gradio interface
with gr.Blocks(title="Device Fault Classifier - Anktechsol", theme=gr.themes.Soft()) as demo:
    gr.Markdown("""
    # 🔧 IoT Device Fault Classifier
    ### by **Anktechsol** - AI + IoT Experts
    
    Intelligent fault detection and classification for industrial IoT devices using AI-powered sensor analysis.
    Identify device failures before they cause downtime!
    """)
    
    with gr.Tabs():
        with gr.Tab("📊 Manual Input"):
            with gr.Row():
                with gr.Column():
                    temp_input = gr.Slider(0, 100, value=25, label="Temperature (°C)")
                    vib_input = gr.Slider(0, 150, value=30, label="Vibration (Hz)")
                    humid_input = gr.Slider(0, 100, value=50, label="Humidity (%)")
                    power_input = gr.Slider(0, 200, value=100, label="Power (W)")
                    pressure_input = gr.Slider(800, 1200, value=1013, label="Pressure (hPa)")
                    classify_btn = gr.Button("🔍 Classify Status", variant="primary", size="lg")
                    gr.Markdown("""
                    ---
                    ### 🔗 Resources
                    - [Anktechsol](https://anktechsol.com)
                    - [More Tools](https://huggingface.co/anktechsol)
                    """)
                
                with gr.Column():
                    gauge_plot = gr.Plot(label="Severity Gauge")
                    report_output = gr.Markdown()
            
            classify_btn.click(
                fn=classify_fault,
                inputs=[temp_input, vib_input, humid_input, power_input, pressure_input],
                outputs=[gauge_plot, report_output]
            )
        
        with gr.Tab("📄 CSV Upload"):
            gr.Markdown("""
            Upload a CSV file with columns: `temperature`, `vibration`, `humidity`, `power`, `pressure`
            
            The classifier will analyze the most recent reading.
            """)
            with gr.Row():
                csv_input = gr.File(label="Upload Sensor Data CSV", file_types=[".csv"])
            with gr.Row():
                csv_gauge = gr.Plot(label="Severity Gauge")
            with gr.Row():
                csv_report = gr.Markdown()
            
            csv_input.change(
                fn=classify_from_csv,
                inputs=[csv_input],
                outputs=[csv_gauge, csv_report]
            )
    
    # Auto-load demo
    demo.load(
        fn=lambda: classify_fault(25, 30, 50, 100, 1013),
        outputs=[gauge_plot, report_output]
    )

if __name__ == "__main__":
    demo.launch()