Create app.py

app.py

@@ -0,0 +1,111 @@
+import gradio as gr
+import pandas as pd
+import plotly.express as px
+from weasyprint import HTML
+import datetime
+import os
+
+# Cache data loading for performance
+def load_data():
+    logs = pd.concat([
+        pd.read_csv("data/smartlog.csv"),
+        pd.read_csv("data/cell_analysis.csv"),
+        pd.read_csv("data/weight_log.csv"),
+        pd.read_csv("data/uv_verification.csv")
+    ])
+    logs["timestamp"] = pd.to_datetime(logs["timestamp"])
+    equipment = pd.read_csv("data/equipment.csv")
+    equipment["amc_expiry"] = pd.to_datetime(equipment["amc_expiry"])
+    return logs, equipment
+
+# Generate PDF report
+def generate_pdf_report(data):
+    os.makedirs("reports", exist_ok=True)
+    html = f"""
+    <h1>LabOps Dashboard Report</h1>
+    <h2>Filtered Data</h2>
+    {data.to_html()}
+    """
+    filename = f"reports/report_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
+    HTML(string=html).write_pdf(filename)
+    return filename
+
+# Main dashboard function
+def render_dashboard(lab, device_type, date_start, date_end, refresh=False):
+    logs, equipment = load_data()
+
+    # Apply filters
+    filtered_logs = logs
+    if lab != "All":
+        device_ids = equipment[equipment["lab_id"] == lab]["device_id"]
+        filtered_logs = filtered_logs[filtered_logs["device_id"].isin(device_ids)]
+    if device_type != "All":
+        device_ids = equipment[equipment["type"] == device_type]["device_id"]
+        filtered_logs = filtered_logs[filtered_logs["device_id"].isin(device_ids)]
+    filtered_logs = filtered_logs[
+        (filtered_logs["timestamp"].dt.date >= pd.to_datetime(date_start).date()) &
+        (filtered_logs["timestamp"].dt.date <= pd.to_datetime(date_end).date())
+    ]
+
+    # Device Cards (FR-002)
+    device_cards = ""
+    for device_id in filtered_logs["device_id"].unique():
+        device_data = filtered_logs[filtered_logs["device_id"] == device_id]
+        if not device_data.empty:
+            status = device_data["status"].iloc[-1]
+            color = "green" if status == "UP" else "red"
+            device_cards += f"""
+                <div style='border: 1px solid {color}; padding: 10px; margin: 10px; border-radius: 5px;'>
+                    <h3>Device: {device_id}</h3>
+                    <p>Status: <span style='color: {color};'>{status}</span></p>
+                    <p>Last Log: {device_data['timestamp'].iloc[-1]}</p>
+                    <p>Usage Count: {len(device_data)}</p>
+                </div>
+            """
+
+    # Charts (FR-003)
+    downtime = filtered_logs[filtered_logs["status"] == "DOWN"].groupby("device_id").size().reset_index(name="downtime_count")
+    downtime_fig = px.bar(downtime, x="device_id", y="downtime_count", title="Downtime Events by Device")
+
+    usage = filtered_logs.groupby([filtered_logs["timestamp"].dt.date, "device_id"])["metrics"].mean().reset_index()
+    usage_fig = px.line(usage, x="timestamp", y="metrics", color="device_id", title="Daily Usage Metrics")
+
+    # AMC Reminders (FR-004)
+    today = datetime.datetime.now().date()
+    two_weeks = today + datetime.timedelta(days=14)
+    amc_alerts = equipment[equipment["amc_expiry"].dt.date <= two_weeks]
+    amc_text = amc_alerts[["device_id", "type", "lab_id", "amc_expiry"]] if not amc_alerts.empty else "No AMC expirations within 2 weeks."
+
+    # PDF Export (FR-004)
+    report_file = generate_pdf_report(filtered_logs)
+
+    return device_cards, downtime_fig, usage_fig, amc_text, report_file
+
+# Gradio interface
+with gr.Blocks(css=".gradio-container {max-width: 100%;}") as demo:
+    gr.Markdown("# LabOps Dashboard")
+    
+    # Filters
+    with gr.Row():
+        lab = gr.Dropdown(choices=["All"] + list(pd.read_csv("data/equipment.csv")["lab_id"].unique()), label="Select Lab", value="All")
+        device_type = gr.Dropdown(choices=["All"] + list(pd.read_csv("data/equipment.csv")["type"].unique()), label="Device Type", value="All")
+    with gr.Row():
+        date_start = gr.DateTime(label="Start Date", value=pd.read_csv("data/smartlog.csv")["timestamp"].min())
+        date_end = gr.DateTime(label="End Date", value=pd.read_csv("data/smartlog.csv")["timestamp"].max())
+    
+    refresh_btn = gr.Button("Refresh")
+    
+    # Outputs
+    device_cards = gr.HTML(label="Device Status")
+    downtime_plot = gr.Plot(label="Downtime Trends")
+    usage_plot = gr.Plot(label="Usage Trends")
+    amc_reminders = gr.Textbox(label="AMC Expiry Reminders")
+    report_download = gr.File(label="Download Report")
+
+    # Bind inputs to outputs
+    inputs = [lab, device_type, date_start, date_end, refresh_btn]
+    outputs = [device_cards, downtime_plot, usage_plot, amc_reminders, report_download]
+    for input_component in inputs:
+        input_component.change(fn=render_dashboard, inputs=[lab, device_type, date_start, date_end], outputs=outputs)
+
+demo.launch()