Update app.py

app.py

@@ -13,6 +13,102 @@ import tempfile
 # Configure logging to match the log format
 logging.basicConfig(level=logging.INFO, format='%(asctime)s,%(msecs)03d - %(levelname)s - %(message)s')
 
+# CSS styling for the Gradio interface
+css = """
+body {
+    font-family: Arial, sans-serif;
+    background-color: #F3F4F6;
+    color: #1E3A8A;
+}
+
+h1 {
+    color: #1E3A8A;
+    text-align: center;
+    margin-bottom: 20px;
+}
+
+.gr-button {
+    background-color: #1E3A8A;
+    color: white;
+    border: none;
+    border-radius: 5px;
+    padding: 10px 20px;
+}
+
+.gr-button:hover {
+    background-color: #2B4C9B;
+}
+
+.summary-card {
+    background-color: white;
+    border-radius: 10px;
+    box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);
+    padding: 20px;
+    margin: 20px 0;
+}
+
+.summary-card h2 {
+    color: #1E3A8A;
+    margin-top: 0;
+}
+
+.maintenance-alert {
+    background-color: white;
+    border-radius: 10px;
+    box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);
+    padding: 15px;
+    margin: 10px 0;
+}
+
+.alert-urgent {
+    color: #DC2626;
+    font-weight: bold;
+}
+
+.alert-upcoming {
+    color: #F59E0B;
+    font-weight: bold;
+}
+
+.recommendation {
+    font-style: italic;
+    color: #4B5563;
+}
+
+.flowchart {
+    display: flex;
+    flex-direction: column;
+    gap: 10px;
+    margin: 20px 0;
+}
+
+.flowchart-step {
+    background-color: #E5E7EB;
+    border-left: 5px solid #1E3A8A;
+    padding: 10px;
+    border-radius: 5px;
+    position: relative;
+}
+
+.flowchart-step:not(:last-child):after {
+    content: '↓';
+    position: absolute;
+    bottom: -20px;
+    left: 50%;
+    transform: translateX(-50%);
+    font-size: 20px;
+    color: #1E3A8A;
+}
+
+.report-preview {
+    background-color: white;
+    border-radius: 10px;
+    box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);
+    padding: 15px;
+    margin: 10px 0;
+}
+"""
+
 def validate_csv(df):
     """
     Validate that the CSV has the required columns.
@@ -67,37 +163,57 @@ def generate_summary(combined_df, anomaly_df, amc_df, plot_path, pdf_path):
         for _, row in amc_df.iterrows():
             days_until_expiry = (row['amc_expiry'] - datetime(2025, 6, 5)).days
             urgency = "Urgent" if days_until_expiry <= 3 else "Upcoming"
-            summary.append(f"- **{row['equipment']}**: Due on {row['amc_expiry'].strftime('%Y-%m-%d')} ({urgency}, {days_until_expiry} days left)")
-        summary.append("Please schedule maintenance to avoid downtime.")
+            urgency_class = "alert-urgent" if urgency == "Urgent" else "alert-upcoming"
+            summary.append(f"- <span class='{urgency_class}'>⚠️ {urgency}</span>: **{row['equipment']}** - Due on {row['amc_expiry'].strftime('%Y-%m-%d')} ({days_until_expiry} days left)")
+        summary.append("\n<div class='recommendation'>Recommendation: Contact the maintenance team within 24 hours for urgent alerts at support@company.com.</div>")
     else:
         summary.append("No devices need maintenance within the next 7 days.")
     summary.append("\n")
 
     # Generated Reports
     summary.append("## Generated Reports")
-    summary.append("- **Usage Chart**: A bar chart showing how much each device was used, grouped by status (e.g., Active, Inactive).")
-    summary.append("- **PDF Report**: Download the detailed report below for a full analysis, including a table of all records and a flowchart of our process.")
+    summary.append("- **Usage Chart**: Visualizes usage patterns across devices, helping identify overworked or underused equipment. See below for the chart.")
+    summary.append("- **PDF Report**: A comprehensive report including a full data table, unusual activity details, maintenance alerts, and a detailed flowchart of our process. Download it below.")
 
     return "\n".join(summary)
 
+def generate_flowchart_html():
+    """
+    Generate an HTML representation of the flowchart for the Gradio interface.
+    Returns an HTML string.
+    """
+    steps = [
+        ("Upload CSV File(s)", "User uploads log files in CSV format."),
+        ("Validate Data", "Checks for required columns (equipment, usage_count, status, amc_expiry) and correct data types."),
+        ("Generate Usage Chart", "Creates a bar chart showing usage counts by device and status (e.g., Active, Inactive)."),
+        ("Detect Unusual Activity", "Uses Local Outlier Factor to identify devices with unusual usage patterns (e.g., too high or too low)."),
+        ("Check Maintenance Dates", "Identifies devices with AMC expiries within 7 days from 2025-06-05."),
+        ("Create PDF Report", "Generates a detailed PDF with data tables, insights, and this flowchart.")
+    ]
+    html = ["<div class='flowchart'>"]
+    for step, description in steps:
+        html.append(f"<div class='flowchart-step'><strong>{step}</strong><br>{description}</div>")
+    html.append("</div>")
+    return "\n".join(html)
+
 def process_files(uploaded_files):
     """
     Process uploaded CSV files, generate usage plots, detect anomalies, and process AMC expiries.
-    Returns a dataframe, plot path, PDF path, AMC expiry message, and summary.
+    Returns a dataframe, plot path, PDF path, AMC expiry message, summary, and flowchart HTML.
     """
     # Log received files
     logging.info(f"Received uploaded files: {uploaded_files}")
 
     if not uploaded_files:
         logging.warning("No files uploaded.")
-        return None, None, None, "Please upload at least one valid CSV file.", "## Summary\nNo files uploaded."
+        return None, None, None, "Please upload at least one valid CSV file.", "## Summary\nNo files uploaded.", ""
 
     valid_files = [f for f in uploaded_files if f.name.endswith('.csv')]
     logging.info(f"Processing {len(valid_files)} valid files: {valid_files}")
 
     if not valid_files:
         logging.warning("No valid CSV files uploaded.")
-        return None, None, None, "Please upload at least one valid CSV file.", "## Summary\nNo valid CSV files uploaded."
+        return None, None, None, "Please upload at least one valid CSV file.", "## Summary\nNo valid CSV files uploaded.", ""
 
     logging.info("Loading logs from uploaded files...")
     all_data = []
@@ -111,15 +227,15 @@ def process_files(uploaded_files):
             is_valid, error_msg = validate_csv(df)
             if not is_valid:
                 logging.error(f"Failed to load {file.name}: {error_msg}")
-                return None, None, None, f"Error loading {file.name}: {error_msg}", f"## Summary\nError: {error_msg}"
+                return None, None, None, f"Error loading {file.name}: {error_msg}", f"## Summary\nError: {error_msg}", ""
             all_data.append(df)
         except Exception as e:
             logging.error(f"Failed to load {file.name}: {str(e)}")
-            return None, None, None, f"Error loading {file.name}: {str(e)}", f"## Summary\nError: {str(e)}"
+            return None, None, None, f"Error loading {file.name}: {str(e)}", f"## Summary\nError: {str(e)}", ""
 
     if not all_data:
         logging.warning("No data loaded from uploaded files.")
-        return None, None, None, "No valid data found in uploaded files.", "## Summary\nNo data loaded."
+        return None, None, None, "No valid data found in uploaded files.", "## Summary\nNo data loaded.", ""
 
     combined_df = pd.concat(all_data, ignore_index=True)
     logging.info(f"Combined {len(combined_df)} total records.")
@@ -132,7 +248,7 @@ def process_files(uploaded_files):
         logging.info("Usage plot generated successfully.")
     else:
         logging.error("Failed to generate usage plot.")
-        return combined_df, None, None, "Failed to generate usage plot.", "## Summary\nUsage plot generation failed."
+        return combined_df, None, None, "Failed to generate usage plot.", "## Summary\nUsage plot generation failed.", ""
 
     # Detect anomalies using Local Outlier Factor
     logging.info("Detecting anomalies using Local Outlier Factor...")
@@ -159,12 +275,17 @@ def process_files(uploaded_files):
     summary = generate_summary(combined_df, anomaly_df, amc_df, plot_path, pdf_path)
     logging.info("Summary generated successfully.")
 
+    # Generate flowchart HTML
+    logging.info("Generating flowchart HTML...")
+    flowchart_html = generate_flowchart_html()
+    logging.info("Flowchart HTML generated successfully.")
+
     # Prepare output dataframe (combine original data with anomalies)
     output_df = combined_df.copy()
     if anomaly_df is not None:
         output_df['anomaly'] = anomaly_df['anomaly'].map({1: "Normal", -1: "Unusual"})
 
-    return output_df, plot_path, pdf_path, amc_message, summary
+    return output_df, plot_path, pdf_path, amc_message, summary, flowchart_html
 
 def generate_usage_plot(df):
     """
@@ -237,7 +358,7 @@ def process_amc_expiries(df):
 
 def generate_pdf_report(original_df, anomaly_df, amc_df):
     """
-    Generate a professionally formatted PDF report with necessary fields and a flowchart.
+    Generate a professionally formatted PDF report with necessary fields and a detailed flowchart.
     Returns the path to the saved PDF.
     """
     try:
@@ -337,11 +458,26 @@ def generate_pdf_report(original_df, anomaly_df, amc_df):
             if amc_df is not None and not amc_df.empty:
                 c.drawString(50, y, f"Devices Needing Maintenance Soon: {len(amc_df['equipment'].unique())}")
                 y -= 20
+                # Table headers
+                c.setFont("Helvetica-Bold", 10)
+                headers = ["Device", "Expiry Date", "Urgency", "Days Left", "Action"]
+                x_positions = [50, 150, 250, 350, 450]
+                for i, header in enumerate(headers):
+                    c.drawString(x_positions[i], y, header)
+                c.line(50, y - 5, width - 50, y - 5)
+                y -= 20
+
+                # Table rows
                 c.setFont("Helvetica", 10)
                 for _, row in amc_df.iterrows():
                     days_until_expiry = (row['amc_expiry'] - datetime(2025, 6, 5)).days
                     urgency = "Urgent" if days_until_expiry <= 3 else "Upcoming"
-                    c.drawString(50, y, f"{row['equipment']}: {row['amc_expiry'].strftime('%Y-%m-%d')} ({urgency}, {days_until_expiry} days left)")
+                    action = "Contact maintenance team within 24 hours" if urgency == "Urgent" else "Schedule maintenance this week"
+                    c.drawString(50, y, str(row['equipment']))
+                    c.drawString(150, y, str(row['amc_expiry'].strftime('%Y-%m-%d')))
+                    c.drawString(250, y, urgency)
+                    c.drawString(350, y, str(days_until_expiry))
+                    c.drawString(450, y, action)
                     y -= 20
                     if y < 50:
                         c.showPage()
@@ -349,7 +485,7 @@ def generate_pdf_report(original_df, anomaly_df, amc_df):
                         draw_header()
                         c.setFont("Helvetica", 10)
                 c.setFont("Helvetica-Oblique", 10)
-                c.drawString(50, y, "Recommendation: Schedule maintenance to prevent downtime.")
+                c.drawString(50, y, "Contact: Email the maintenance team at support@company.com for scheduling.")
                 y -= 20
             else:
                 c.drawString(50, y, "No devices need maintenance within the next 7 days.")
@@ -360,16 +496,20 @@ def generate_pdf_report(original_df, anomaly_df, amc_df):
             y = draw_section_title("Processing Pipeline Flowchart", y)
             c.setFont("Helvetica", 10)
             flowchart = [
-                "1. Upload CSV File(s)",
-                "2. Validate Data (Check for required columns and data types)",
-                "3. Generate Usage Chart (Bar chart of usage by device and status)",
-                "4. Detect Unusual Activity (Using Local Outlier Factor)",
-                "5. Check Maintenance Dates (Identify AMC expiries within 7 days)",
-                "6. Create PDF Report (Detailed analysis with tables and insights)"
+                ("1. Upload CSV File(s)", "User uploads log files in CSV format containing device usage data."),
+                ("2. Validate Data", "Ensures all required columns (equipment, usage_count, status, amc_expiry) are present and data types are correct (e.g., usage_count as numeric, amc_expiry as date)."),
+                ("3. Generate Usage Chart", "Creates a bar chart showing usage counts by device and status (e.g., Active, Inactive) to visualize usage patterns."),
+                ("4. Detect Unusual Activity", "Uses Local Outlier Factor (LOF) algorithm to identify devices with unusual usage patterns by comparing local density of usage counts (contamination=0.1, n_neighbors=5)."),
+                ("5. Check Maintenance Dates", "Identifies devices with AMC expiries within 7 days from 2025-06-05, calculating days left and urgency (urgent if ≤3 days)."),
+                ("6. Create PDF Report", "Generates this PDF with a data table, unusual activity details, maintenance alerts, and this detailed flowchart.")
             ]
-            for step in flowchart:
+            for step, description in flowchart:
                 c.drawString(50, y, step)
-                y -= 20
+                y -= 15
+                c.setFont("Helvetica-Oblique", 9)
+                c.drawString(70, y, description)
+                c.setFont("Helvetica", 10)
+                y -= 25
                 if y < 50:
                     c.showPage()
                     y = height - 100
@@ -384,24 +524,30 @@ def generate_pdf_report(original_df, anomaly_df, amc_df):
         return None
 
 # Gradio interface
-with gr.Blocks() as demo:
+with gr.Blocks(css=css) as demo:
     gr.Markdown("# Equipment Log Analysis")
     with gr.Row():
         file_input = gr.File(file_count="multiple", label="Upload CSV Files")
         process_button = gr.Button("Process Files")
     with gr.Row():
-        output_summary = gr.Markdown(label="Summary of Results")
+        output_summary = gr.Markdown(label="Summary of Results", elem_classes=["summary-card"])
     with gr.Row():
         output_df = gr.Dataframe(label="Processed Data")
         output_plot = gr.Image(label="Usage Chart")
     with gr.Row():
-        output_message = gr.Textbox(label="Maintenance Alerts")
+        output_message = gr.Textbox(label="Maintenance Alerts", elem_classes=["maintenance-alert"])
         output_pdf = gr.File(label="Download Detailed PDF Report")
+    with gr.Row():
+        gr.HTML(generate_flowchart_html(), label="Processing Flowchart")
+    with gr.Row():
+        gr.Markdown("## Report Previews", elem_classes=["report-preview"])
+        gr.Markdown("- **Usage Chart**: See the bar chart above for a visual of device usage by status.")
+        gr.Markdown("- **PDF Report**: Download the PDF above for a full analysis, including data tables, unusual activity, maintenance alerts, and a detailed flowchart.")
 
     process_button.click(
         fn=process_files,
         inputs=[file_input],
-        outputs=[output_df, output_plot, output_pdf, output_message, output_summary]
+        outputs=[output_df, output_plot, output_pdf, output_message, output_summary, gr.HTML]
     )
 
 if __name__ == "__main__":