Update app.py

app.py

@@ -1,117 +1,177 @@
 import gradio as gr
-import json
 import pandas as pd
+import matplotlib.pyplot as plt
+from datetime import datetime, timedelta
+import random
 
-def create_contact_from_can(cands_from_trigger):
-    """
-    Process a list of candidate dictionaries to create contact records.
+# Simulate sample data for lab devices (mimicking Salesforce custom objects)
+def generate_sample_data():
+    labs = ["Lab_A", "Lab_B", "Lab_C"]
+    equipment_types = ["Microscope", "Centrifuge", "UV_Sterilizer"]
+    devices = []
+    logs = []
     
-    Args:
-        cands_from_trigger (list): List of dictionaries with candidate data.
-        
-    Returns:
-        list: List of contact dictionaries.
-    """
-    con_list = []
+    # Generate devices
+    for i in range(10):
+        devices.append({
+            "Device_ID": f"DEV_{i+1}",
+            "Lab": random.choice(labs),
+            "Equipment_Type": random.choice(equipment_types),
+            "Status": random.choice(["Operational", "Down", "Maintenance"])
+        })
     
-    for candidate in cands_from_trigger:
-        contact = {
-            "FirstName": candidate.get("First_Name__c", ""),
-            "LastName": candidate.get("Last_Name__c", ""),
-            "Email": candidate.get("Email__c", "")
-        }
-        con_list.append(contact)
+    # Generate logs (simulating SmartLog__c, Cell_Analysis__c, etc.)
+    start_date = datetime.now() - timedelta(days=30)
+    for device in devices:
+        for day in range(30):
+            log_date = start_date + timedelta(days=day)
+            logs.append({
+                "Device_ID": device["Device_ID"],
+                "Log_Timestamp": log_date.strftime("%Y-%m-%d %H:%M:%S"),
+                "Usage_Count": random.randint(0, 100),
+                "Status": random.choice(["Operational", "Down", "Maintenance"]),
+                "AMC_Expiry": (log_date + timedelta(days=random.randint(10, 365))).strftime("%Y-%m-%d")
+            })
     
-    return con_list
+    return pd.DataFrame(devices), pd.DataFrame(logs)
 
-def process_json_input(json_input):
-    """
-    Process JSON input from Gradio text area and return contact records.
-    
-    Args:
-        json_input (str): JSON string containing candidate data.
-        
-    Returns:
-        str: JSON string of contact records or error message.
-    """
-    try:
-        # Parse JSON input (expecting a list of dictionaries)
-        candidates = json.loads(json_input)
-        if not isinstance(candidates, list):
-            return "Error: Input must be a JSON list of candidate objects."
-        
-        # Process candidates
-        contacts = create_contact_from_can(candidates)
-        
-        # Return formatted JSON
-        return json.dumps(contacts, indent=4)
-    except json.JSONDecodeError:
-        return "Error: Invalid JSON format. Please provide valid JSON, e.g., [{\"First_Name__c\": \"John\", \"Last_Name__c\": \"Doe\", \"Email__c\": \"john.doe@example.com\"}]"
-    except Exception as e:
-        return f"Error: {str(e)}"
+# Initialize sample data
+devices_df, logs_df = generate_sample_data()
 
-def process_form_input(first_name, last_name, email):
+def process_dashboard_data(lab_filter, equipment_type_filter, date_range):
     """
-    Process form input for a single candidate and return contact record.
+    Process device and log data based on filters and return dashboard components.
     
     Args:
-        first_name (str): Candidate's first name.
-        last_name (str): Candidate's last name.
-        email (str): Candidate's email.
+        lab_filter (str): Selected lab (e.g., Lab_A or All).
+        equipment_type_filter (str): Selected equipment type (e.g., Microscope or All).
+        date_range (tuple): Start and end dates for filtering logs.
         
     Returns:
-        str: JSON string of the contact record.
+        tuple: Device cards text, daily trend plot, uptime plot, anomaly alerts, report text.
     """
-    candidate = {
-        "First_Name__c": first_name,
-        "Last_Name__c": last_name,
-        "Email__c": email
-    }
-    contacts = create_contact_from_can([candidate])
-    return json.dumps(contacts[0], indent=4)
+    # Apply filters
+    filtered_devices = devices_df.copy()
+    if lab_filter != "All":
+        filtered_devices = filtered_devices[filtered_devices["Lab"] == lab_filter]
+    if equipment_type_filter != "All":
+        filtered_devices = filtered_devices[filtered_devices["Equipment_Type"] == equipment_type_filter]
+    
+    filtered_logs = logs_df[logs_df["Device_ID"].isin(filtered_devices["Device_ID"])]
+    if date_range:
+        start_date, end_date = date_range
+        filtered_logs = filtered_logs[
+            (filtered_logs["Log_Timestamp"] >= start_date.strftime("%Y-%m-%d %H:%M:%S")) &
+            (filtered_logs["Log_Timestamp"] <= end_date.strftime("%Y-%m-%d %H:%M:%S"))
+        ]
+    
+    # Device Cards
+    device_cards = "Device Cards:\n"
+    for _, device in filtered_devices.iterrows():
+        device_logs = filtered_logs[filtered_logs["Device_ID"] == device["Device_ID"]]
+        usage_count = device_logs["Usage_Count"].sum() if not device_logs.empty else 0
+        last_log = device_logs["Log_Timestamp"].max() if not device_logs.empty else "No logs"
+        device_cards += (
+            f"Device: {device['Device_ID']}, Lab: {device['Lab']}, Type: {device['Equipment_Type']}, "
+            f"Status: {device['Status']}, Usage Count: {usage_count}, Last Log: {last_log}\n"
+        )
+    
+    # Daily Log Trends (Matplotlib Plot)
+    if not filtered_logs.empty:
+        filtered_logs["Date"] = pd.to_datetime(filtered_logs["Log_Timestamp"]).dt.date
+        daily_trends = filtered_logs.groupby("Date")["Usage_Count"].sum().reset_index()
+        plt.figure(figsize=(8, 4))
+        plt.plot(daily_trends["Date"], daily_trends["Usage_Count"], marker="o")
+        plt.title("Daily Log Trends")
+        plt.xlabel("Date")
+        plt.ylabel("Total Usage Count")
+        plt.xticks(rotation=45)
+        plt.tight_layout()
+        daily_trend_plot = plt.gcf()
+    else:
+        daily_trend_plot = None
+    
+    # Weekly Uptime % (Matplotlib Plot)
+    if not filtered_logs.empty:
+        filtered_logs["Week"] = pd.to_datetime(filtered_logs["Log_Timestamp"]).dt.isocalendar().week
+        uptime_data = filtered_logs.groupby("Week")["Status"].value_counts().unstack(fill_value=0)
+        uptime_data["Uptime_%"] = uptime_data.get("Operational", 0) / (
+            uptime_data.get("Operational", 0) + uptime_data.get("Down", 0) + uptime_data.get("Maintenance", 0)
+        ) * 100
+        plt.figure(figsize=(8, 4))
+        plt.bar(uptime_data.index, uptime_data["Uptime_%"])
+        plt.title("Weekly Uptime %")
+        plt.xlabel("Week")
+        plt.ylabel("Uptime %")
+        plt.tight_layout()
+        uptime_plot = plt.gcf()
+    else:
+        uptime_plot = None
+    
+    # Anomaly Alerts (Usage spikes: >2x average usage)
+    anomaly_alerts = "Anomaly Alerts:\n"
+    if not filtered_logs.empty:
+        avg_usage = filtered_logs["Usage_Count"].mean()
+        anomalies = filtered_logs[filtered_logs["Usage_Count"] > 2 * avg_usage]
+        for _, log in anomalies.iterrows():
+            anomaly_alerts += (
+                f"Device: {log['Device_ID']}, Timestamp: {log['Log_Timestamp']}, "
+                f"Usage Spike: {log['Usage_Count']} (Avg: {avg_usage:.2f})\n"
+            )
+    
+    # AMC Reminders (simulated)
+    report = "LabOps Dashboard Report:\n"
+    report += f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n"
+    report += device_cards + "\n"
+    report += anomaly_alerts + "\n"
+    report += "AMC Reminders:\n"
+    if not filtered_logs.empty:
+        amc_expiring = filtered_logs[pd.to_datetime(filtered_logs["AMC_Expiry"]) <= (datetime.now() + timedelta(days=14))]
+        for _, log in amc_expiring.iterrows():
+            report += (
+                f"Device: {log['Device_ID']}, AMC Expiry: {log['AMC_Expiry']}\n"
+            )
+    
+    return device_cards, daily_trend_plot, uptime_plot, anomaly_alerts, report
 
 # Define Gradio interface
-with gr.Blocks(title="Candidate to Contact Converter") as demo:
-    gr.Markdown("# Candidate to Contact Converter")
-    gr.Markdown("Convert candidate data to contact records. Use the JSON input for batch processing or the form for a single candidate.")
-
-    # JSON Input Section
-    gr.Markdown("## Batch Input (JSON)")
-    gr.Markdown("Enter candidate data in JSON format (list of objects). Example:")
-    gr.Markdown("""
-    ```json
-    [
-        {"First_Name__c": "John", "Last_Name__c": "Doe", "Email__c": "john.doe@example.com"},
-        {"First_Name__c": "Jane", "Last_Name__c": "Smith", "Email__c": "jane.smith@example.com"}
-    ]
-    ```
-    """)
-    json_input = gr.Textbox(label="Candidate JSON", lines=10)
-    json_output = gr.Textbox(label="Contact Output (JSON)", lines=10, interactive=False)
-    json_button = gr.Button("Convert JSON")
-    json_button.click(fn=process_json_input, inputs=json_input, outputs=json_output)
+with gr.Blocks(title="LabOps Dashboard") as demo:
+    gr.Markdown("# LabOps Dashboard")
+    gr.Markdown("Monitor smart lab devices, view usage trends, uptime, anomalies, and export reports.")
 
-    # Form Input Section
-    gr.Markdown("## Single Candidate Form")
-    first_name = gr.Textbox(label="First Name")
-    last_name = gr.Textbox(label="Last Name")
-    email = gr.Textbox(label="Email")
-    form_output = gr.Textbox(label="Contact Output (JSON)", lines=5, interactive=False)
-    form_button = gr.Button("Convert Form")
-    form_button.click(fn=process_form_input, inputs=[first_name, last_name, email], outputs=form_output)
-
-    # Download Output
-    gr.Markdown("## Download Results")
-    download_button = gr.File(label="Download Contacts as JSON")
-
-    def prepare_download(json_output):
-        if json_output and json_output.startswith("["):
-            with open("contacts.json", "w") as f:
-                f.write(json_output)
-            return "contacts.json"
-        return None
-
-    json_output.change(fn=prepare_download, inputs=json_output, outputs=download_button)
+    # Filters
+    gr.Markdown("## Filters")
+    lab_filter = gr.Dropdown(choices=["All"] + list(devices_df["Lab"].unique()), label="Lab Site")
+    equipment_type_filter = gr.Dropdown(choices=["All"] + list(devices_df["Equipment_Type"].unique()), label="Equipment Type")
+    date_range = gr.DateRange(label="Date Range")
+    
+    # Dashboard Components
+    gr.Markdown("## Device Cards")
+    device_cards_output = gr.Textbox(label="Device Status", lines=10, interactive=False)
+    gr.Markdown("## Daily Log Trends")
+    daily_trend_plot = gr.Plot(label="Daily Usage Trends")
+    gr.Markdown("## Weekly Uptime %")
+    uptime_plot = gr.Plot(label="Weekly Uptime")
+    gr.Markdown("## Anomaly Alerts")
+    anomaly_alerts_output = gr.Textbox(label="Anomaly Alerts", lines=5, interactive=False)
+    
+    # Export Report
+    gr.Markdown("## Export Report")
+    report_output = gr.Textbox(label="Report Preview", lines=10, interactive=False)
+    download_button = gr.File(label="Download Report as Text")
+    
+    # Update dashboard on filter change
+    def update_dashboard(lab, equipment, date_range):
+        device_cards, daily_trend, uptime, anomalies, report = process_dashboard_data(lab, equipment, date_range)
+        with open("labops_report.txt", "w") as f:
+            f.write(report)
+        return device_cards, daily_trend, uptime, anomalies, report, "labops_report.txt"
+    
+    gr.Button("Update Dashboard").click(
+        fn=update_dashboard,
+        inputs=[lab_filter, equipment_type_filter, date_range],
+        outputs=[device_cards_output, daily_trend_plot, uptime_plot, anomaly_alerts_output, report_output, download_button]
+    )
 
 # Launch the app
 demo.launch()