diff --git a/viewer/main.py b/viewer/main.py index 038c8c7c..1345e28d 100644 --- a/viewer/main.py +++ b/viewer/main.py @@ -15,7 +15,6 @@ class State: selected_directory: str = "" selected_tab: str = "Dashboard" conversation_index: int = 0 - eval_summaries: str = "" eval_id_filter: str = "" product_filter: str = "" requester_filter: str = "" @@ -134,6 +133,70 @@ def get_results_dir(): return results_dir_candidates[1] # Fallback to default +# (mtime, value). Each is replaced as a whole so a concurrent reader never sees a +# value that disagrees with the mtime it was built from. +_TRENDS_DF_CACHE = (None, None) +_RUN_DIRS_CACHE = (None, []) + + +def _mtime(path): + try: + return os.path.getmtime(path) + except OSError: + return None + + +def load_trends_df(results_dir): + """Parsed trends_cache.csv, reread only when the file changes. + + Shared by every request in this worker, so callers must not mutate it. + """ + global _TRENDS_DF_CACHE + + cache_file = os.path.join(results_dir, "trends_cache.csv") + mtime = _mtime(cache_file) + if mtime is None: + logging.warning(f"Trends cache file not found at {cache_file}") + return None + + cached_mtime, df = _TRENDS_DF_CACHE + if cached_mtime == mtime: + return df + + try: + df = pd.read_csv(cache_file) + except Exception as e: + logging.error(f"Error reading trends cache: {e}") + return None + + logging.info(f"Loaded {len(df)} rows from trends cache.") + _TRENDS_DF_CACHE = (mtime, df) + return df + + +def list_run_dirs(results_dir): + """Run directory names, restatted only when results_dir gains or loses entries. + + Shared by every request in this worker, so callers must not mutate it. + """ + global _RUN_DIRS_CACHE + + mtime = _mtime(results_dir) + if mtime is None: + return [] + + cached_mtime, dirs = _RUN_DIRS_CACHE + if cached_mtime == mtime: + return dirs + + dirs = [ + d for d in os.listdir(results_dir) + if os.path.isdir(os.path.join(results_dir, d)) + ] + _RUN_DIRS_CACHE = (mtime, dirs) + return dirs + + def get_eval_details(results_dir, dir_name): details = { "product": "N/A", @@ -229,14 +292,7 @@ def get_color_for_pct(val_str): def on_load(e: me.LoadEvent): state = me.state(State) results_dir = get_results_dir() - directories = [] - if os.path.exists(results_dir): - # List directories only - directories = [ - d - for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) - ] + directories = list_run_dirs(results_dir) job_id = me.query_params.get("job_id") or me.query_params.get("jobid") if job_id and job_id in directories: @@ -259,14 +315,8 @@ def on_load(e: me.LoadEvent): def status_component(): results_dir = get_results_dir() - directories = [] - if os.path.exists(results_dir): - directories = [ - d - for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) - ] - + directories = list_run_dirs(results_dir) + with me.box( style=me.Style( background="#ffffff", @@ -306,10 +356,9 @@ def on_agent_tab_change(e): # Build summary data from precomputed trends cache data = [] - cache_file = os.path.join(results_dir, "trends_cache.csv") - if os.path.exists(cache_file): + cache_df = load_trends_df(results_dir) + if cache_df is not None: try: - cache_df = pd.read_csv(cache_file) for _, row in cache_df.iterrows(): data.append({ 'AI Score': row['ai_score'] if 'ai_score' in row else None, @@ -327,7 +376,6 @@ def on_agent_tab_change(e): except Exception as e: logging.error(f"Error reading trends cache: {e}") else: - logging.warning(f"Trends cache file not found at {cache_file}") me.text("Trends cache file not found. Please run precompute.") return @@ -537,6 +585,76 @@ def handler(e: me.ClickEvent): me.text("No evaluation data found in results directories.") +# (mtime, rows). Replaced as a whole so a concurrent reader never sees rows that +# disagree with the mtime they were built from. +_SUMMARIES_CACHE = (None, []) + + +def _pct(value): + return f"{value:.0f}%" if not pd.isna(value) else "N/A" + + +def _build_summaries(cache_df): + import re + + rows = [] + for _, row in cache_df.iterrows(): + score = row['ai_score'] if 'ai_score' in row else 0.0 + if (score == 0.0 or pd.isna(score)) and 'ai_summary' in row and not pd.isna(row['ai_summary']): + match = re.search(r"General Score:.*?(\d+(\.\d+)?)", row['ai_summary']) + if match: + score = float(match.group(1)) + + rows.append({ + "id": str(row['job_id']), + "date": str(row['run_time']) if not pd.isna(row['run_time']) else "N/A", + "product": str(row['product']) if not pd.isna(row['product']) else "N/A", + "requester": str(row['requester']) if not pd.isna(row['requester']) else "N/A", + "dataset": str(row['dataset']) if 'dataset' in row and not pd.isna(row['dataset']) else "N/A", + "model_config.generator": str(row['model_config.generator']) if 'model_config.generator' in row and not pd.isna(row['model_config.generator']) else "unknown", + "ai_score": f"{score:.0f}%" if not pd.isna(score) and score != 0.0 else "N/A", + "exact_match": _pct(row['exact_match']), + "llmrater": _pct(row['llmrater']), + "trajectory_matcher": _pct(row['trajectory']), + "goal_completion": _pct(row['goal_completion']) if 'goal_completion' in row else "N/A", + "turn_count": f"{row['turn_count']:.1f}" if not pd.isna(row['turn_count']) else "N/A", + "executable": _pct(row['executable']), + "token_consumption": f"{row['tokens']:.0f}" if not pd.isna(row['tokens']) else "N/A", + "end_to_end_latency": f"{row['latency'] / 60000.0:.2f}m" if not pd.isna(row['latency']) else "N/A", + }) + return rows + + +def load_summaries(results_dir): + """Row dicts for the List table, reparsed only when the trends cache changes. + + The returned list is shared by every request in this worker, so callers must + treat it as read-only. + """ + global _SUMMARIES_CACHE + + mtime = _mtime(os.path.join(results_dir, "trends_cache.csv")) + if mtime is None: + return [] + + cached_mtime, rows = _SUMMARIES_CACHE + if cached_mtime == mtime: + return rows + + cache_df = load_trends_df(results_dir) + if cache_df is None: + return [] + + try: + rows = _build_summaries(cache_df) + except Exception as e: + logging.error(f"Error building list rows from trends cache: {e}") + return [] + + _SUMMARIES_CACHE = (mtime, rows) + return rows + + def list_view_component(directories, results_dir): state = me.state(State) try: @@ -597,53 +715,8 @@ def on_list_agent_tab_change(e): ) me.box(style=me.Style(height="16px")) if directories: - # Compute summaries if empty - s = me.state(State) - summaries = [] - if s.eval_summaries: - try: - summaries = json.loads(s.eval_summaries) - except Exception: - summaries = [] - - if not summaries: - cache_file = os.path.join(results_dir, "trends_cache.csv") - logging.info(f"trends_cache.csv exists: {os.path.exists(cache_file)}") - if os.path.exists(cache_file): - try: - cache_df = pd.read_csv(cache_file) - logging.info(f"Loaded {len(cache_df)} rows from trends cache.") - for _, row in cache_df.iterrows(): - score = row['ai_score'] if 'ai_score' in row else 0.0 - if (score == 0.0 or pd.isna(score)) and 'ai_summary' in row and not pd.isna(row['ai_summary']): - import re - match = re.search(r"General Score:.*?(\d+(\.\d+)?)", row['ai_summary']) - if match: - score = float(match.group(1)) - - summaries.append({ - "id": str(row['job_id']), - "date": str(row['run_time']) if not pd.isna(row['run_time']) else "N/A", - "product": str(row['product']) if not pd.isna(row['product']) else "N/A", - "requester": str(row['requester']) if not pd.isna(row['requester']) else "N/A", - "dataset": str(row['dataset']) if 'dataset' in row and not pd.isna(row['dataset']) else "N/A", - "model_config.generator": str(row['model_config.generator']) if 'model_config.generator' in row and not pd.isna(row['model_config.generator']) else "unknown", - "ai_score": f"{score:.0f}%" if not pd.isna(score) and score != 0.0 else "N/A", - "exact_match": f"{row['exact_match']:.0f}%" if not pd.isna(row['exact_match']) else "N/A", - "llmrater": f"{row['llmrater']:.0f}%" if not pd.isna(row['llmrater']) else "N/A", - "trajectory_matcher": f"{row['trajectory']:.0f}%" if not pd.isna(row['trajectory']) else "N/A", - "goal_completion": f"{row['goal_completion']:.0f}%" if 'goal_completion' in row and not pd.isna(row['goal_completion']) else "N/A", - "turn_count": f"{row['turn_count']:.1f}" if not pd.isna(row['turn_count']) else "N/A", - "executable": f"{row['executable']:.0f}%" if not pd.isna(row['executable']) else "N/A", - "token_consumption": f"{row['tokens']:.0f}" if not pd.isna(row['tokens']) else "N/A", - "end_to_end_latency": f"{row['latency'] / 60000.0:.2f}m" if not pd.isna(row['latency']) else "N/A" - }) - s.eval_summaries = json.dumps(summaries) - except Exception as e: - logging.error(f"Error reading trends cache: {e}") - else: - logging.warning(f"Trends cache file not found at {cache_file}") - + summaries = load_summaries(results_dir) + # Sort by selected column reverse = state.sort_descending col = state.sort_column @@ -685,16 +758,9 @@ def get_sort_key(x): return "" if reverse else "\xff\xff\xff\xff" return str(val) - summaries.sort(key=get_sort_key, reverse=reverse) - - # Extract unique values for filters from ALL summaries - all_summaries = [] - if s.eval_summaries: - try: - all_summaries = json.loads(s.eval_summaries) - except Exception: - all_summaries = [] - + all_summaries = summaries + summaries = sorted(summaries, key=get_sort_key, reverse=reverse) + filters_file = os.path.join(results_dir, "filters_cache.json") if os.path.exists(filters_file): try: @@ -2055,15 +2121,8 @@ def render_app_content(): results_dir = get_results_dir() logging.info(f"render_app_content: selected_directory='{state.selected_directory}', selected_evals='{state.selected_evals}', selected_main_tab='{state.selected_main_tab}'") - directories = [] - if os.path.exists(results_dir): - # List directories only - directories = [ - d - for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) - ] - + directories = list_run_dirs(results_dir) + def on_title_click(e: me.ClickEvent): state.selected_directory = "" state.conversation_index = 0 @@ -2289,10 +2348,9 @@ def get_val(cfg_name): me.text("AI Summary", type="headline-5") if not state.ai_summary and state.selected_directory: - trends_cache_file = os.path.join(results_dir, "trends_cache.csv") - if os.path.exists(trends_cache_file): + cache_df = load_trends_df(results_dir) + if cache_df is not None: try: - cache_df = pd.read_csv(trends_cache_file) run_data = cache_df[cache_df['job_id'] == state.selected_directory] if not run_data.empty and 'ai_summary' in run_data.columns: summary = run_data['ai_summary'].values[0] diff --git a/viewer/trends.py b/viewer/trends.py index 1f1c7436..74e86aba 100644 --- a/viewer/trends.py +++ b/viewer/trends.py @@ -2,7 +2,7 @@ import logging import mesop as me import pandas as pd -from main import State +from main import State, list_run_dirs, load_trends_df def get_results_dir(): # Try to read from environment variable @@ -24,8 +24,9 @@ def get_results_dir(): return results_dir_candidates[1] # Fallback to default def generate_d3_chart(df, x_col, y_col, hue_col, title, ylabel): - df_sorted = df.sort_values(by=x_col) - + # job_id is unplotted but read by the tooltip in chart.js. + df_sorted = df[[x_col, y_col, hue_col, "job_id"]].sort_values(by=x_col) + # Convert dataframe to records for JSON data_records = df_sorted.to_dict(orient='records') import json @@ -89,26 +90,12 @@ def trends_component(): me.text(f"Results directory not found at {results_dir}") return - cache_file = os.path.join(results_dir, "trends_cache.csv") - - df = None - - # Try to load from cache - if os.path.exists(cache_file): - try: - df = pd.read_csv(cache_file) - logging.info("Loaded trends data from cache.") - except Exception as e: - logging.error(f"Error reading cache file: {e}") - + df = load_trends_df(results_dir) + # Fallback to computing on the fly if cache is missing or failed if df is None: - directories = [ - d - for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) - ] - + directories = list_run_dirs(results_dir) + data = [] for d in directories: @@ -332,7 +319,8 @@ def handler(e: me.ClickEvent): if state.trends_requester_filter: df = df[df['requester'] == state.trends_requester_filter] - df['product_dataset'] = df['product'] + " (" + df['dataset'] + ")" + # assign, not item-set: df may be the shared cached frame from load_trends_df. + df = df.assign(product_dataset=df['product'] + " (" + df['dataset'] + ")") df = df[df['product'].notna() & (df['product'] != 'unknown') & (df['product'].str.strip() != '')] if state.trends_agent_tab == "Gemini":