Create app.py

app.py

@@ -0,0 +1,274 @@
+import gradio as gr
+import pandas as pd
+from datetime import datetime
+import logging
+import plotly.express as px
+from sklearn.ensemble import IsolationForest  # For anomaly detection
+from transformers import pipeline
+
+# Configure logging for debugging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+# Load Hugging Face summarization model
+try:
+    logging.info("Attempting to load Hugging Face model...")
+    summarizer = pipeline("text2text-generation", model="google/flan-t5-base")
+    logging.info("Hugging Face model loaded successfully")
+except Exception as e:
+    logging.error(f"Failed to load model: {str(e)}")
+    raise e
+
+# Format summary prompt and generate report
+def summarize_logs(df):
+    try:
+        total_devices = df["device_id"].nunique()
+        avg_uptime = "97%"  # Placeholder
+        most_used = df.groupby("device_id")["usage_hours"].sum().idxmax() if not df.empty else "N/A"
+        downtime_events = 3  # Placeholder
+
+        prompt = (
+            f"Summarize maintenance and usage logs. "
+            f"There were {total_devices} devices. "
+            f"The most used device was {most_used}."
+        )
+        summary = summarizer(prompt, max_length=200, do_sample=False)[0]["generated_text"]
+        logging.info("Summary generated successfully")
+        return summary
+    except Exception as e:
+        logging.error(f"Summary generation failed: {str(e)}")
+        return "Failed to generate summary."
+
+# Anomaly Detection using Isolation Forest
+def detect_anomalies(df):
+    try:
+        if "usage_hours" not in df.columns or "downtime" not in df.columns:
+            logging.warning("Required columns for anomaly detection not found")
+            return "Anomaly detection requires 'usage_hours' and 'downtime' columns."
+
+        features = df[["usage_hours", "downtime"]].fillna(0)
+        iso_forest = IsolationForest(contamination=0.1, random_state=42)
+        df["anomaly"] = iso_forest.fit_predict(features)
+        
+        anomalies = df[df["anomaly"] == -1][["device_id", "usage_hours", "downtime", "timestamp"]]
+        if anomalies.empty:
+            return "No anomalies detected."
+        
+        anomaly_lines = ["**Detected Anomalies:**"]
+        for idx, row in anomalies.iterrows():
+            anomaly_lines.append(f"- Device ID: {row['device_id']}")
+            anomaly_lines.append(f"  Usage Hours: {row['usage_hours']}")
+            anomaly_lines.append(f"  Downtime: {row['downtime']}")
+            anomaly_lines.append(f"  Timestamp: {row['timestamp']}")
+            anomaly_lines.append("---")
+            anomaly_lines.append("")
+        anomaly_list = "\n".join(anomaly_lines)
+        logging.info("Anomalies detected successfully")
+        return anomaly_list
+    except Exception as e:
+        logging.error(f"Anomaly detection failed: {str(e)}")
+        return f"Anomaly detection failed: {str(e)}"
+
+# AMC Reminders based on device and AMC date
+def check_amc_reminders(df, current_date):
+    try:
+        if "device_id" not in df.columns or "amc_date" not in df.columns:
+            logging.warning("Required columns for AMC reminders not found")
+            return "AMC reminders require 'device_id' and 'amc_date' columns."
+
+        df["amc_date"] = pd.to_datetime(df["amc_date"])
+        current_date = pd.to_datetime(current_date)
+        
+        df["days_to_amc"] = (df["amc_date"] - current_date).dt.days
+        reminders = df[(df["days_to_amc"] >= 0) & (df["days_to_amc"] <= 30)][["device_id", "amc_date"]]
+        
+        if reminders.empty:
+            return "No AMC reminders due within the next 30 days."
+        
+        reminder_lines = ["**Upcoming AMC Reminders:**"]
+        for idx, row in reminders.iterrows():
+            reminder_lines.append(f"- Device ID: {row['device_id']}")
+            reminder_lines.append(f"  AMC Date: {row['amc_date']}")
+            reminder_lines.append("---")
+            reminder_lines.append("")
+        reminder_list = "\n".join(reminder_lines)
+        logging.info("AMC reminders generated successfully")
+        return reminder_list
+    except Exception as e:
+        logging.error(f"AMC reminder generation failed: {str(e)}")
+        return f"AMC reminder generation failed: {str(e)}"
+
+# Dashboard Insights (AI-generated executive-level insights)
+def generate_dashboard_insights(df):
+    try:
+        total_devices = df["device_id"].nunique()
+        avg_usage = df["usage_hours"].mean() if "usage_hours" in df.columns else 0
+        prompt = (
+            f"Generate executive-level insights. "
+            f"There were {total_devices} devices with an average usage of {avg_usage:.2f} hours."
+        )
+        insights = summarizer(prompt, max_length=150, do_sample=False)[0]["generated_text"]
+        logging.info("Dashboard insights generated successfully")
+        return insights
+    except Exception as e:
+        logging.error(f"Dashboard insights generation failed: {str(e)}")
+        return f"Dashboard insights generation failed: {str(e)}"
+
+# Create a bar chart for usage hours per device
+def create_usage_chart(df):
+    try:
+        usage_data = df.groupby("device_id")["usage_hours"].sum().reset_index()
+        fig = px.bar(
+            usage_data,
+            x="device_id",
+            y="usage_hours",
+            title="Usage Hours per Device",
+            labels={"device_id": "Device ID", "usage_hours": "Usage Hours"},
+            color="usage_hours",
+            color_continuous_scale="Blues"
+        )
+        fig.update_layout(
+            title_font_size=16,
+            margin=dict(l=20, r=20, t=40, b=20),
+            plot_bgcolor="white",
+            paper_bgcolor="white",
+            font=dict(size=12)
+        )
+        return fig
+    except Exception as e:
+        logging.error(f"Failed to create usage chart: {str(e)}")
+        return None
+
+# Main Gradio function
+def process_logs(file_obj):
+    try:
+        if file_obj is None:
+            logging.warning("No file uploaded, returning empty results")
+            return "No file uploaded.", "No data to preview.", None, "No anomalies detected.", "No AMC reminders.", "No insights generated."
+        
+        file_name = file_obj.name if hasattr(file_obj, 'name') else file_obj
+        logging.info(f"Processing file: {file_name}")
+        
+        if not file_name.endswith(".csv"):
+            logging.error("Unsupported file format")
+            return "Unsupported file format. Please upload a CSV file.", None, None, None, None, None
+
+        df = pd.read_csv(file_name)
+        logging.info(f"File loaded successfully with {len(df)} rows")
+
+        try:
+            df["timestamp"] = pd.to_datetime(df["timestamp"])
+        except Exception as e:
+            logging.error(f"Date conversion failed: {str(e)}")
+            return f"Failed to convert timestamp to datetime: {str(e)}", None, None, None, None, None
+
+        if df.empty:
+            logging.warning("No data in the file")
+            return "No data available in the file.", "No data to preview.", None, "No anomalies detected.", "No AMC reminders.", "No insights generated."
+
+        # Step 1: Summary Report
+        summary = f"**Step 1: Summary Report**\n\n{summarize_logs(df)}\n\n---\n"
+
+        # Step 2: Log Preview
+        if not df.empty:
+            preview_lines = ["**Step 2: Log Preview (First 5 Rows)**\n"]
+            for idx, row in df.head().iterrows():
+                preview_lines.append(f"**Row {idx + 1}:**")
+                preview_lines.append(f"- Device ID: {row['device_id']}")
+                preview_lines.append(f"- Timestamp: {row['timestamp']}")
+                preview_lines.append(f"- Usage Hours: {row['usage_hours']}")
+                preview_lines.append(f"- Downtime: {row['downtime']}")
+                preview_lines.append(f"- AMC Date: {row['amc_date']}")
+                preview_lines.append("---")
+                preview_lines.append("")
+            preview = "\n".join(preview_lines) + "\n---\n"
+        else:
+            preview = "**Step 2: Log Preview**\n\nNo data available.\n\n---\n"
+
+        # Step 3: Usage Chart
+        chart = create_usage_chart(df)
+
+        # Step 4: Anomaly Detection
+        anomalies = f"**Step 3: Anomaly Detection**\n\n{detect_anomalies(df)}\n\n---\n"
+
+        # Step 5: AMC Reminders
+        amc_reminders = f"**Step 4: AMC Reminders**\n\n{check_amc_reminders(df, datetime.now())}\n\n---\n"
+
+        # Step 6: Dashboard Insights
+        insights = f"**Step 5: Dashboard Insights (AI)**\n\n{generate_dashboard_insights(df)}\n\n---\n"
+
+        return summary, preview, chart, anomalies, amc_reminders, insights
+    except Exception as e:
+        logging.error(f"Failed to process file: {str(e)}")
+        return f"Failed to process file: {str(e)}", None, None, None, None, None
+
+# Gradio Interface with Step-by-Step Layout
+try:
+    logging.info("Initializing Gradio Blocks interface...")
+    with gr.Blocks(css="""
+        .dashboard-container {border: 1px solid #e0e0e0; padding: 10px; border-radius: 5px; background-color: #f9f9f9;}
+        .dashboard-title {font-size: 24px; font-weight: bold; margin-bottom: 10px;}
+        .dashboard-section {margin-bottom: 15px;}
+        .dashboard-section h3 {font-size: 18px; margin-bottom: 5px;}
+    """) as iface:
+        gr.Markdown("<h1>LabOps Log Analyzer Dashboard (Hugging Face AI)</h1>")
+        gr.Markdown("Upload a CSV file containing lab equipment logs to analyze usage.")
+
+        with gr.Row():
+            with gr.Column(scale=1):
+                file_input = gr.File(label="Upload Logs (CSV)", file_types=[".csv"])
+                submit_button = gr.Button("Submit", variant="primary")
+            
+            with gr.Column(scale=2):
+                with gr.Group(elem_classes="dashboard-container"):
+                    gr.Markdown("<div class='dashboard-title'>Analysis Results (Step-by-Step)</div>")
+                    
+                    # Step 1: Summary Report
+                    with gr.Group(elem_classes="dashboard-section"):
+                        gr.Markdown("### Step 1: Summary Report")
+                        summary_output = gr.Markdown()
+                    
+                    # Step 2: Log Preview
+                    with gr.Group(elem_classes="dashboard-section"):
+                        gr.Markdown("### Step 2: Log Preview")
+                        preview_output = gr.Markdown()
+                    
+                    # Step 3: Usage Chart
+                    with gr.Group(elem_classes="dashboard-section"):
+                        gr.Markdown("### Step 3: Usage Chart")
+                        chart_output = gr.Plot()
+                    
+                    # Step 4: Anomaly Detection
+                    with gr.Group(elem_classes="dashboard-section"):
+                        gr.Markdown("### Step 4: Anomaly Detection")
+                        anomaly_output = gr.Markdown()
+                    
+                    # Step 5: AMC Reminders
+                    with gr.Group(elem_classes="dashboard-section"):
+                        gr.Markdown("### Step 5: AMC Reminders")
+                        amc_output = gr.Markdown()
+                    
+                    # Step 6: Dashboard Insights
+                    with gr.Group(elem_classes="dashboard-section"):
+                        gr.Markdown("### Step 6: Dashboard Insights (AI)")
+                        insights_output = gr.Markdown()
+
+        submit_button.click(
+            fn=process_logs,
+            inputs=[file_input],
+            outputs=[summary_output, preview_output, chart_output, anomaly_output, amc_output, insights_output]
+        )
+
+    logging.info("Gradio interface initialized successfully")
+except Exception as e:
+    logging.error(f"Failed to initialize Gradio interface: {str(e)}")
+    raise e
+
+if __name__ == "__main__":
+    try:
+        logging.info("Launching Gradio interface...")
+        iface.launch(server_name="0.0.0.0", server_port=7860, debug=True, share=False)
+        logging.info("Gradio interface launched successfully")
+    except Exception as e:
+        logging.error(f"Failed to launch Gradio interface: {str(e)}")
+        print(f"Error launching app: {str(e)}")
+        raise e