Spaces:
Configuration error
Configuration error
File size: 2,418 Bytes
24f3fb6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 |
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()
|