demo_dash.py

import gradio as gr
import pandas as pd
import random
import time

# Dummy state (in-memory for example)
status_counts = {
    "Scheduled": 5,
    "Triggered": 8,
    "ASAP": 3,
    "Planning": 4,
    "In Motion": 6,
    "Blocked": 2
}

metrics = {
    "Pilot Count": 10,
    "Available Pilots": 4,
    "Completed/Hr": 12,
    "Queue Delay": "3m 20s",
    "Avg Task Time v Est": [12, 15, 11, 14, 16]  # minutes
}

log_entries = [
    f"{time.strftime('%H:%M:%S')} - Task initialized"
]


def get_status_bar_data():
    df = pd.DataFrame(list(status_counts.items()), columns=["Status", "Count"])
    return df


def get_task_time_line():
    return pd.DataFrame({
        "Task": [f"T{i+1}" for i in range(len(metrics["Avg Task Time v Est"]))],
        "Time": metrics["Avg Task Time v Est"]
    })


def get_metrics():
    return (
        f"Pilots: {metrics['Available Pilots']}/{metrics['Pilot Count']}",
        f"Tasks/hr: {metrics['Completed/Hr']}",
        f"Queue Delay: {metrics['Queue Delay']}"
    )


def get_log():
    # Add a random event to simulate updates
    log_entries.append(f"{time.strftime('%H:%M:%S')} - Task {random.randint(100, 199)} processed")
    return "\n".join(log_entries[-20:])


with gr.Blocks(title="Ops Dashboard") as dashboard:
    with gr.Row():
        gr.Markdown("## 🧭 Live Task Dashboard")

    with gr.Row():
        bar_plot = gr.BarPlot(
            label="Task Status Counts",
            x="Status",
            y="Count",
            show_label=False
        )
        line_plot = gr.LinePlot(
            label="Task Time vs. Estimate",
            x="Task",
            y="Time",
            show_label=False
        )

    with gr.Row():
        pilot_info = gr.Text(label="Pilot Availability")
        throughput_info = gr.Text(label="Throughput")
        queue_delay_info = gr.Text(label="Queue Delay")

    with gr.Row():
        log_output = gr.Textbox(
            label="Live Log",
            lines=15,
            max_lines=500,
            interactive=False,
            show_copy_button=True
        )

    def update_all():
        return (
            get_status_bar_data(),
            get_task_time_line(),
            *get_metrics(),
            get_log()
        )

    dashboard.load(
        update_all,
        inputs=[],
        outputs=[bar_plot, line_plot, pilot_info, throughput_info, queue_delay_info, log_output]
    )

dashboard.launch()