app.py

import gradio as gr
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.neighbors import LocalOutlierFactor
from datetime import datetime, timedelta
import os
import logging
from reportlab.lib.pagesizes import letter
from reportlab.pdfgen import canvas
from reportlab.lib import colors
import tempfile

# Configure logging to match the log format
logging.basicConfig(level=logging.INFO, format='%(asctime)s,%(msecs)03d - %(levelname)s - %(message)s')

def validate_csv(df):
    """
    Validate that the CSV has the required columns.
    Returns True if valid, False otherwise with an error message.
    """
    required_columns = ['equipment', 'usage_count', 'status', 'amc_expiry']
    missing_columns = [col for col in required_columns if col not in df.columns]
    if missing_columns:
        return False, f"Missing required columns: {', '.join(missing_columns)}"
    # Validate data types
    try:
        df['usage_count'] = pd.to_numeric(df['usage_count'], errors='raise')
        df['amc_expiry'] = pd.to_datetime(df['amc_expiry'], errors='raise')
    except Exception as e:
        return False, f"Invalid data types: {str(e)}"
    return True, ""

def generate_summary(combined_df, anomaly_df, amc_df, plot_path, pdf_path):
    """
    Generate a detailed and easy-to-understand summary of the processing results.
    Returns a markdown string for display in the Gradio interface.
    """
    summary = []

    # Overview
    summary.append("## Overview")
    total_records = len(combined_df)
    unique_devices = combined_df['equipment'].unique()
    summary.append(f"We processed **{total_records} log entries** for **{len(unique_devices)} devices** ({', '.join(unique_devices)}).")
    summary.append("This report helps you understand device usage, identify unusual activity, and plan maintenance.\n")

    # Unusual Activity (Anomalies)
    summary.append("## Unusual Activity")
    if anomaly_df is not None:
        num_anomalies = sum(anomaly_df['anomaly'] == -1)
        if num_anomalies > 0:
            summary.append(f"We found **{num_anomalies} unusual activities** that might need your attention:")
            anomaly_records = anomaly_df[anomaly_df['anomaly'] == -1][['equipment', 'usage_count', 'status']]
            for _, row in anomaly_records.iterrows():
                summary.append(f"- **{row['equipment']}** (Usage: {row['usage_count']}, Status: {row['status']}) - High or low usage compared to others might indicate overuse or underuse.")
        else:
            summary.append("No unusual activity detected. All devices are operating within expected usage patterns.")
    else:
        summary.append("We couldn’t check for unusual activity due to an error.")
    summary.append("\n")

    # Maintenance Alerts (AMC Expiries)
    summary.append("## Maintenance Alerts")
    if amc_df is not None and not amc_df.empty:
        unique_devices_amc = amc_df['equipment'].unique()
        summary.append(f"**{len(unique_devices_amc)} devices** need maintenance soon (within 7 days from 2025-06-05):")
        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.")
    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.")

    return "\n".join(summary)

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.
    """
    # 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."

    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."

    logging.info("Loading logs from uploaded files...")
    all_data = []

    # Load and combine CSV files
    for file in valid_files:
        try:
            df = pd.read_csv(file.name)
            logging.info(f"Loaded {len(df)} records from {file.name}")
            # Validate CSV structure
            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}"
            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)}"

    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."

    combined_df = pd.concat(all_data, ignore_index=True)
    logging.info(f"Combined {len(combined_df)} total records.")
    logging.info(f"Loaded {len(combined_df)} log records from uploaded files.")

    # Generate usage plot
    logging.info("Generating usage plot...")
    plot_path = generate_usage_plot(combined_df)
    if plot_path:
        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."

    # Detect anomalies using Local Outlier Factor
    logging.info("Detecting anomalies using Local Outlier Factor...")
    anomaly_df = detect_anomalies(combined_df)
    if anomaly_df is None:
        logging.error("Failed to detect anomalies.")
    else:
        logging.info(f"Detected {sum(anomaly_df['anomaly'] == -1)} anomalies using Local Outlier Factor.")

    # Process AMC expiries
    logging.info("Processing AMC expiries...")
    amc_message, amc_df = process_amc_expiries(combined_df)

    # Generate PDF report
    logging.info("Generating PDF report...")
    pdf_path = generate_pdf_report(combined_df, anomaly_df, amc_df)
    if pdf_path:
        logging.info("PDF report generated successfully.")
    else:
        logging.error("Failed to generate PDF report.")

    # Generate summary
    logging.info("Generating summary of results...")
    summary = generate_summary(combined_df, anomaly_df, amc_df, plot_path, pdf_path)
    logging.info("Summary 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

def generate_usage_plot(df):
    """
    Generate a bar plot of usage_count by equipment and status.
    Returns the path to the saved plot.
    """
    try:
        plt.figure(figsize=(12, 6))
        # Define colors for statuses
        status_colors = {'Active': '#36A2EB', 'Inactive': '#FF6384', 'Down': '#FFCE56', 'Online': '#4BC0C0'}
        for status in df['status'].unique():
            subset = df[df['status'] == status]
            plt.bar(
                subset['equipment'] + f" ({status})",
                subset['usage_count'],
                label=status,
                color=status_colors.get(status, '#999999')
            )
        plt.xlabel("Equipment (Status)", fontsize=12)
        plt.ylabel("Usage Count", fontsize=12)
        plt.title("Usage Count by Equipment and Status", fontsize=14)
        plt.legend(title="Status")
        plt.xticks(rotation=45, ha='right')
        plt.tight_layout()

        # Save plot to temporary file
        with tempfile.NamedTemporaryFile(delete=False, suffix='.png') as tmp:
            plt.savefig(tmp.name, format='png', dpi=100)
            plot_path = tmp.name
        plt.close()
        return plot_path
    except Exception as e:
        logging.error(f"Failed to generate usage plot: {str(e)}")
        return None

def detect_anomalies(df):
    """
    Detect anomalies in usage_count using Local Outlier Factor.
    Returns a dataframe with an 'anomaly' column (-1 for anomalies, 1 for normal).
    """
    try:
        model = LocalOutlierFactor(n_neighbors=5, contamination=0.1)
        anomalies = model.fit_predict(df[['usage_count']].values)
        anomaly_df = df.copy()
        anomaly_df['anomaly'] = anomalies
        return anomaly_df
    except Exception as e:
        logging.error(f"Failed to detect anomalies: {str(e)}")
        return None

def process_amc_expiries(df):
    """
    Identify devices with AMC expiries within 7 days from 2025-06-05.
    Returns a message and a dataframe of devices with upcoming expiries.
    """
    try:
        current_date = datetime(2025, 6, 5)
        threshold = current_date + timedelta(days=7)
        df['amc_expiry'] = pd.to_datetime(df['amc_expiry'])
        upcoming_expiries = df[df['amc_expiry'] <= threshold]
        unique_devices = upcoming_expiries['equipment'].unique()
        message = f"Found {len(unique_devices)} devices with upcoming AMC expiries: {', '.join(unique_devices)}. Details: " + "; ".join(
            [f"{row['equipment']}: {row['amc_expiry'].strftime('%Y-%m-%d')}" for _, row in upcoming_expiries.iterrows()]
        )
        logging.info(f"Found {len(unique_devices)} devices with upcoming AMC expiries.")
        return message, upcoming_expiries
    except Exception as e:
        logging.error(f"Failed to process AMC expiries: {str(e)}")
        return f"Error processing AMC expiries: {str(e)}", None

def generate_pdf_report(original_df, anomaly_df, amc_df):
    """
    Generate a professionally formatted PDF report with necessary fields and a flowchart.
    Returns the path to the saved PDF.
    """
    try:
        if original_df is None or original_df.empty:
            logging.warning("No data available for PDF generation.")
            return None

        with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as tmp:
            c = canvas.Canvas(tmp.name, pagesize=letter)
            width, height = letter

            def draw_header():
                c.setFont("Helvetica-Bold", 16)
                c.setFillColor(colors.darkblue)
                c.drawString(50, height - 50, "Equipment Log Analysis Report")
                c.setFont("Helvetica", 10)
                c.setFillColor(colors.black)
                c.drawString(50, height - 70, f"Generated on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
                c.line(50, height - 80, width - 50, height - 80)

            def draw_section_title(title, y):
                c.setFont("Helvetica-Bold", 14)
                c.setFillColor(colors.darkblue)
                c.drawString(50, y, title)
                c.setFillColor(colors.black)
                c.line(50, y - 5, width - 50, y - 5)
                return y - 30

            y = height - 100
            draw_header()

            # Summary
            y = draw_section_title("Summary", y)
            c.setFont("Helvetica", 12)
            c.drawString(50, y, f"Total Records: {len(original_df)}")
            y -= 20
            c.drawString(50, y, f"Unique Devices: {', '.join(original_df['equipment'].unique())}")
            y -= 40

            # Data Table
            y = draw_section_title("Device Log Details", y)
            c.setFont("Helvetica-Bold", 10)
            headers = ["Equipment", "Usage Count", "Status", "AMC Expiry", "Activity"]
            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

            c.setFont("Helvetica", 10)
            output_df = original_df.copy()
            if anomaly_df is not None:
                output_df['anomaly'] = anomaly_df['anomaly'].map({1: "Normal", -1: "Unusual"})
            for _, row in output_df.iterrows():
                c.drawString(50, y, str(row['equipment']))
                c.drawString(150, y, str(row['usage_count']))
                c.drawString(250, y, str(row['status']))
                c.drawString(350, y, str(row['amc_expiry'].strftime('%Y-%m-%d')))
                c.drawString(450, y, str(row['anomaly']))
                y -= 20
                if y < 50:
                    c.showPage()
                    y = height - 100
                    draw_header()
                    c.setFont("Helvetica", 10)

            # Anomalies
            y = draw_section_title("Unusual Activity (Using Local Outlier Factor)", y)
            c.setFont("Helvetica", 12)
            if anomaly_df is not None:
                num_anomalies = sum(anomaly_df['anomaly'] == -1)
                c.drawString(50, y, f"Unusual Activities Detected: {num_anomalies}")
                y -= 20
                if num_anomalies > 0:
                    anomaly_records = anomaly_df[anomaly_df['anomaly'] == -1][['equipment', 'usage_count', 'status']]
                    c.drawString(50, y, "Details:")
                    y -= 20
                    c.setFont("Helvetica-Oblique", 10)
                    for _, row in anomaly_records.iterrows():
                        c.drawString(50, y, f"{row['equipment']}: Usage Count = {row['usage_count']}, Status = {row['status']}")
                        y -= 20
                        c.drawString(70, y, "Note: This device’s usage is significantly higher or lower than others, which may indicate overuse or underuse.")
                        y -= 20
                        if y < 50:
                            c.showPage()
                            y = height - 100
                            draw_header()
                            c.setFont("Helvetica-Oblique", 10)
            else:
                c.drawString(50, y, "Unable to detect unusual activity due to an error.")
                y -= 20
            y -= 20

            # AMC Expiries
            y = draw_section_title("Maintenance Alerts (as of 2025-06-05)", y)
            c.setFont("Helvetica", 12)
            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
                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)")
                    y -= 20
                    if y < 50:
                        c.showPage()
                        y = height - 100
                        draw_header()
                        c.setFont("Helvetica", 10)
                c.setFont("Helvetica-Oblique", 10)
                c.drawString(50, y, "Recommendation: Schedule maintenance to prevent downtime.")
                y -= 20
            else:
                c.drawString(50, y, "No devices need maintenance within the next 7 days.")
                y -= 20
            y -= 20

            # Flowchart
            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)"
            ]
            for step in flowchart:
                c.drawString(50, y, step)
                y -= 20
                if y < 50:
                    c.showPage()
                    y = height - 100
                    draw_header()
                    c.setFont("Helvetica", 10)

            c.showPage()
            c.save()
            return tmp.name
    except Exception as e:
        logging.error(f"Failed to generate PDF report: {str(e)}")
        return None

# Gradio interface
with gr.Blocks() 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")
    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_pdf = gr.File(label="Download Detailed PDF Report")

    process_button.click(
        fn=process_files,
        inputs=[file_input],
        outputs=[output_df, output_plot, output_pdf, output_message, output_summary]
    )

if __name__ == "__main__":
    logging.info("Application starting...")
    demo.launch(server_name="0.0.0.0", server_port=7860)