File size: 6,054 Bytes
5a98598 956ab2c 5a98598 956ab2c 5a98598 956ab2c 5a98598 956ab2c 020b18b 956ab2c 020b18b 956ab2c 5a98598 7168226 5a98598 956ab2c 5a98598 956ab2c 5a98598 |
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 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 |
import os
import gradio as gr
import pandas as pd
from datetime import datetime
import time
import sys
import importlib.util
import threading
from log_reader import RemoteLogReader
# Define the path for the azure_count_ip_data module
azure_count_ip_data_path = os.path.join(os.path.dirname(__file__), 'azure_count_ip_data.py')
# Import the module dynamically
spec = importlib.util.spec_from_file_location("azure_count_ip_data", azure_count_ip_data_path)
azure_count_ip_data = importlib.util.module_from_spec(spec)
spec.loader.exec_module(azure_count_ip_data)
# Get the functions we need
count_files_per_annotator = azure_count_ip_data.count_files_per_annotator
count_deduplicated_files_per_annotator = azure_count_ip_data.count_deduplicated_files_per_annotator
# Define the path for storing the data
DATA_DIR = os.path.join(os.path.dirname(__file__), 'data')
STATS_FILE = os.path.join(DATA_DIR, 'battle_stats.csv')
LAST_UPDATE_FILE = os.path.join(DATA_DIR, 'last_update.txt')
# Ensure data directory exists
os.makedirs(DATA_DIR, exist_ok=True)
def save_stats(df, current_time):
"""Save statistics and last update time to files"""
df.to_csv(STATS_FILE, index=False)
with open(LAST_UPDATE_FILE, 'w') as f:
f.write(current_time)
def load_stats():
"""Load statistics and last update time from files"""
try:
df = pd.read_csv(STATS_FILE)
with open(LAST_UPDATE_FILE, 'r') as f:
last_update = f.read().strip()
return df, last_update
except (FileNotFoundError, pd.errors.EmptyDataError):
return pd.DataFrame(columns=['Annotator', 'Total Count', 'Unique Count', 'Unique %']), ""
def update_stats():
"""Get the latest battle statistics with both total and deduplicated counts"""
try:
# Initialize RemoteLogReader
reader = RemoteLogReader()
# Get total annotator counts
total_counts = count_files_per_annotator(reader)
# Get deduplicated annotator counts
unique_counts = count_deduplicated_files_per_annotator(reader)
# Combine the data into a single DataFrame
data = []
all_annotators = set(total_counts.keys()) | set(unique_counts.keys())
# Track totals for summary row
total_sum = 0
unique_sum = 0
for annotator in all_annotators:
total = total_counts.get(annotator, 0)
unique = unique_counts.get(annotator, 0)
# Calculate percentage of unique prompts
percentage = round((unique / total * 100) if total > 0 else 0, 1)
# Add to totals
total_sum += total
unique_sum += unique
data.append({
'Annotator': annotator,
'Total Count': total,
'Unique Count': unique,
'Unique %': f"{percentage}%"
})
# Add summary row
overall_percentage = round((unique_sum / total_sum * 100) if total_sum > 0 else 0, 1)
data.append({
'Annotator': 'TOTAL',
'Total Count': total_sum,
'Unique Count': unique_sum,
'Unique %': f"{overall_percentage}%"
})
# Convert to DataFrame and sort by total count, keeping TOTAL at the bottom
df = pd.DataFrame(data)
# Move TOTAL row to the end
df = pd.concat([
df[df['Annotator'] != 'TOTAL'].sort_values('Total Count', ascending=False),
df[df['Annotator'] == 'TOTAL']
]).reset_index(drop=True)
# Get current time
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# Save the updated stats
save_stats(df, current_time)
return df, current_time
except Exception as e:
print(f"Error updating stats: {e}")
return pd.DataFrame(columns=['Annotator', 'Total Count', 'Unique Count', 'Unique %']), ""
def auto_update(state):
"""Background task to update stats every hour"""
while state['running']:
state['stats'], state['last_update'] = update_stats()
time.sleep(3600) # Sleep for 1 hour
def create_ui():
state = {'running': True}
# Try to load existing stats first
state['stats'], state['last_update'] = load_stats()
# If no existing stats or they're empty, update them
if state['stats'].empty:
state['stats'], state['last_update'] = update_stats()
# Start background update thread
update_thread = threading.Thread(target=auto_update, args=(state,))
update_thread.daemon = True
update_thread.start()
def get_current_stats():
return state['stats']
def get_last_update():
return state['last_update']
def manual_refresh():
state['stats'], state['last_update'] = update_stats()
return state['stats'], state['last_update']
with gr.Blocks(title="Battle Count Statistics") as app:
gr.Markdown("# Battle Count Statistics")
gr.Markdown("Displays the count of battles per annotator. 'Total Count' shows all valid battles, while 'Unique Count' shows deduplicated battles based on the first user prompt. Only conversations that pass the vote conditions are counted, and example prompts are excluded from the unique count.")
with gr.Row():
last_update = gr.Textbox(
value=get_last_update,
label="Last Updated",
interactive=False
)
with gr.Row():
output = gr.DataFrame(
value=get_current_stats,
interactive=False,
wrap=True,
column_widths=["40%", "20%", "20%", "20%"],
height=600
)
refresh_btn = gr.Button("Refresh Now")
refresh_btn.click(fn=manual_refresh, outputs=[output, last_update])
return app
if __name__ == "__main__":
app = create_ui()
app.launch() |