From 6a0b820b884a2308534a1e9756c26964a3a320ee Mon Sep 17 00:00:00 2001 From: Faizan Mohiuddin Date: Thu, 20 Feb 2025 20:49:52 +0000 Subject: [PATCH] Merged in client-runs-dashboard-updates (pull request #406) Client runs dashboard updates * new dashboard code * dashboard updates * Merged main into client-runs-dashboard-updates * Merged main into client-runs-dashboard-updates Approved-by: Alex Galarce --- streamlit/runs-dashboard.py | 582 +++++++++++++++++++++++------------- 1 file changed, 373 insertions(+), 209 deletions(-) diff --git a/streamlit/runs-dashboard.py b/streamlit/runs-dashboard.py index ae23689..2e4a886 100644 --- a/streamlit/runs-dashboard.py +++ b/streamlit/runs-dashboard.py @@ -8,6 +8,7 @@ import json from io import StringIO from dotenv import load_dotenv import os +import re # Load environment variables load_dotenv() @@ -19,13 +20,42 @@ st.set_page_config( initial_sidebar_state="expanded" ) -# Initialize S3 client +# Initialize S3 client with fallback to local credentials @st.cache_resource def get_s3_client(): - return boto3.client( - "s3", - region_name="us-east-2" - ) + try: + # First try direct connection + return boto3.client( + "s3", + region_name="us-east-2" + ) + except Exception as e: + st.warning("Direct AWS connection failed, attempting to use local credentials...") + + try: + # Load environment variables + load_dotenv() + + # Get temporary credentials from environment variables + aws_access_key = os.getenv('AWS_ACCESS_KEY_ID') + aws_secret_key = os.getenv('AWS_SECRET_ACCESS_KEY') + aws_session_token = os.getenv('AWS_SESSION_TOKEN') + + if not all([aws_access_key, aws_secret_key, aws_session_token]): + raise ValueError("Missing required AWS credentials in .env file") + + # Create client with temporary credentials + return boto3.client( + "s3", + region_name="us-east-2", + aws_access_key_id=aws_access_key, + aws_secret_access_key=aws_secret_key, + aws_session_token=aws_session_token + ) + except Exception as inner_e: + st.error(f"Failed to connect using local credentials: {str(inner_e)}") + st.error("Please ensure .env file exists with AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, and AWS_SESSION_TOKEN") + raise @st.cache_data def load_batch_tracker(_s3_client, bucket): @@ -40,29 +70,61 @@ def load_batch_tracker(_s3_client, bucket): st.error(f"Error loading batch tracker: {str(e)}") return None - @st.cache_data -def get_batch_runs(_s3_client, bucket, batch_id=None): - """Get all run directories for a specific batch or all batches""" +def get_batch_folders(_s3_client, bucket): + """Get all batch folders in the bucket""" try: - prefix = f"run_" response = _s3_client.list_objects_v2( Bucket=bucket, - Prefix=prefix + Delimiter='/' + ) + + batch_folders = [] + for prefix in response.get('CommonPrefixes', []): + folder_name = prefix.get('Prefix', '').rstrip('/') + if folder_name and not folder_name.startswith('batch-tracker'): + batch_folders.append(folder_name) + + return sorted(batch_folders) + except Exception as e: + st.error(f"Error listing batch folders: {str(e)}") + return [] + +@st.cache_data +def get_batch_runs(_s3_client, bucket, batch_id): + """Get all run directories for a specific batch""" + try: + prefix = f"{batch_id}/run_" + response = _s3_client.list_objects_v2( + Bucket=bucket, + Prefix=prefix, + Delimiter='/' ) runs = [] - for obj in response.get('Contents', []): - if batch_id is None or batch_id in obj['Key']: - run_dir = '/'.join(obj['Key'].split('/')[:-1]) - if run_dir and run_dir not in runs: - runs.append(run_dir) + for obj in response.get('CommonPrefixes', []): + run_dir = obj.get('Prefix', '').rstrip('/') + if run_dir: + runs.append(run_dir) return sorted(runs, reverse=True) except Exception as e: st.error(f"Error listing runs: {str(e)}") return [] +@st.cache_data +def load_master_batch_tracking(_s3_client, bucket, batch_id): + """Load the master batch tracking CSV for a specific batch""" + try: + response = _s3_client.get_object( + Bucket=bucket, + Key=f"{batch_id}/master_batch_tracking.csv" + ) + return pd.read_csv(response['Body']) + except Exception as e: + st.warning(f"No master batch tracking found for {batch_id}: {str(e)}") + return None + @st.cache_data def load_csv_from_s3(_s3_client, bucket, key): """Load any CSV file from S3""" @@ -73,10 +135,9 @@ def load_csv_from_s3(_s3_client, bucket, key): ) return pd.read_csv(response['Body']) except Exception as e: - st.error(f"Error loading CSV {key}: {str(e)}") + st.warning(f"Error loading CSV {key}: {str(e)}") return None - def create_download_link(_s3_client, bucket, key): """Create a presigned URL for downloading the file""" try: @@ -93,36 +154,110 @@ def create_download_link(_s3_client, bucket, key): st.error(f"Error creating download link: {str(e)}") return None -def get_run_files(s3_client, bucket, run_prefix): - """Get all files in a run directory""" +def get_run_files(s3_client, bucket, run_path): + """Get all files in a run directory with the new structure""" try: - response = s3_client.list_objects_v2( + # Check for individual files + individual_prefix = f"{run_path}/individual/" + individual_response = s3_client.list_objects_v2( Bucket=bucket, - Prefix=run_prefix + Prefix=individual_prefix + ) + + # Check for consolidated files + consolidated_prefix = f"{run_path}/consolidated/" + consolidated_response = s3_client.list_objects_v2( + Bucket=bucket, + Prefix=consolidated_prefix + ) + + # Check for tracking files + tracking_prefix = f"{run_path}/tracking/" + tracking_response = s3_client.list_objects_v2( + Bucket=bucket, + Prefix=tracking_prefix ) files = { 'individual': [], 'consolidated': [], - 'logs': [] + 'tracking': [] } - for obj in response.get('Contents', []): + # Process individual files - organize by subfolder + individual_folders = {} + for obj in individual_response.get('Contents', []): key = obj['Key'] - if 'individual/' in key: - files['individual'].append(key) - elif 'consolidated/' in key: - files['consolidated'].append(key) - elif key.endswith('.csv'): # Log files in root of run directory - files['logs'].append(key) + parts = key.split('/') + if len(parts) > 3: # batch_id/run_id/individual/folder/file + folder = parts[-2] + if folder not in individual_folders: + individual_folders[folder] = [] + individual_folders[folder].append(key) + + files['individual'] = individual_folders + + # Process consolidated files + for obj in consolidated_response.get('Contents', []): + files['consolidated'].append(obj['Key']) + + # Process tracking files + for obj in tracking_response.get('Contents', []): + files['tracking'].append(obj['Key']) return files except Exception as e: st.error(f"Error listing files: {str(e)}") return None +def extract_timestamp_from_run(run_path): + """Improved timestamp extraction from run path""" + # Match patterns like 'run_20250110_09:02' or 'run_20250110_09:02_...' + match = re.search(r'run_(\d{8}_\d{2}:\d{2})', run_path) + if match: + timestamp_str = match.group(1) + try: + return pd.to_datetime(timestamp_str, format='%Y%m%d_%H:%M') + except: + return None + return None -def get_active_runs(df): + + +def calculate_processing_rate(master_df, run_path): + """Calculate processing rate based on file timestamps and run start time""" + if master_df is None or master_df.empty: + return None + + run_start_time = extract_timestamp_from_run(run_path) + if run_start_time is None: + return None + + # Convert the timestamp to datetime + master_df['upload_time'] = pd.to_datetime(master_df['upload_time']) if 'upload_time' in master_df.columns else run_start_time + + # Calculate time difference in hours + if 'upload_time' in master_df.columns: + time_diffs = (master_df['upload_time'] - run_start_time).dt.total_seconds() / 3600 + time_diffs = time_diffs[time_diffs > 0] # Only consider positive time differences + + if len(time_diffs) == 0: + return 0 + + # Calculate files per hour + max_time_diff = time_diffs.max() + if max_time_diff > 0: + files_per_hour = len(time_diffs) / max_time_diff + return files_per_hour + + # If no upload_time column or all times are negative, use count and rough estimate + time_diff = (datetime.now() - run_start_time).total_seconds() / 3600 + if time_diff > 0: + return len(master_df) / time_diff + + return 0 + +def get_active_runs(df_tracker): """ Determines truly active runs by checking if there are any subsequent 'Completed' entries for the same batch_id after a 'Running' status, only showing runs from today @@ -131,7 +266,7 @@ def get_active_runs(df): today = pd.Timestamp.now().date() # Sort by batch_id and start_time to get chronological order - df_sorted = df.sort_values(['batch_id', 'start_time']) + df_sorted = df_tracker.sort_values(['batch_id', 'start_time']) # Group by batch_id to analyze status progression for batch_id, group in df_sorted.groupby('batch_id'): @@ -234,9 +369,10 @@ def main(): st.metric("Avg Processing Speed", f"{avg_speed:.2f} files/hour") # Success rate (considering only completed or failed status) - total_runs = len(df_filtered) completed_runs = len(df_filtered[df_filtered['status'] == 'Completed']) - success_rate = (completed_runs / total_runs * 100) if total_runs > 0 else 0 + failed_runs = len(df_filtered[df_filtered['status'] == 'Failed']) + total_finished = completed_runs + failed_runs + success_rate = (completed_runs / total_finished * 100) if total_finished > 0 else 0 with col4: st.metric("Success Rate", f"{success_rate:.1f}%" if success_rate > 0 else "-") @@ -244,211 +380,239 @@ def main(): if not active_runs_df.empty: st.subheader("Currently Active Runs") display_df = active_runs_df.copy() - display_df['progress'] = (display_df['files_processed'] / display_df['total_input_files'] * 100).round(2) - display_df['progress'] = display_df['progress'].astype(str) + '%' + st.dataframe( - display_df[['batch_id', 'start_time', 'input_dir', 'progress']], + display_df[['batch_id', 'start_time', 'input_dir']], use_container_width=True ) else: st.warning("No data available for the selected filters.") - - # Graphs row - col1, col2 = st.columns(2) - - with col1: - st.subheader("Processing Speed Over Time") - fig = px.line( - df_filtered.sort_values('start_time'), - x='start_time', - y='contracts_per_hour', - title="Processing Speed Trend", - line_shape='linear' - ) - st.plotly_chart(fig, use_container_width=True) - - with col2: - st.subheader("Cumulative Files Processed") - df_cumulative = df_filtered.sort_values('start_time').copy() - df_cumulative['cumulative_files'] = df_cumulative['files_processed'].cumsum() - - fig = px.line( - df_cumulative, - x='start_time', - y='cumulative_files', - title="Cumulative Files Processed", - line_shape='linear' - ) - st.plotly_chart(fig, use_container_width=True) - # Cost Analysis section - st.subheader("Cost Analysis") - selected_batch = st.selectbox( - "Select Batch ID", - options=df_filtered['batch_id'].unique(), - key="cost_analysis_batch" - ) - - if selected_batch: - runs = get_batch_runs(s3_client, bucket, selected_batch) - if runs: - selected_run = st.selectbox("Select Run", runs) - - # Load cost log - cost_log_key = f"{selected_run}/costlog.csv" - df_cost = load_csv_from_s3(s3_client, bucket, cost_log_key) - - if df_cost is not None: - col1, col2 = st.columns(2) - - with col1: - st.subheader("Cost Distribution by Function") - fig = px.pie( - df_cost, - values='Total Cost', - names='Caller Function', - title="Cost Distribution by Function" - ) - st.plotly_chart(fig, use_container_width=True) - - with col2: - st.subheader("Token Usage") - fig = px.bar( - df_cost, - x='Caller Function', - y=['Input Prompt Tokens', 'Output Response Tokens'], - title="Token Usage by Function", - barmode='group' - ) - st.plotly_chart(fig, use_container_width=True) - - st.subheader("Cost Details") - st.dataframe(df_cost, use_container_width=True) - - with tab2: st.header("Batch Run Explorer") + # Get all batch folders + batch_folders = get_batch_folders(s3_client, bucket) + + if not batch_folders: + st.warning("No batch folders found in S3 bucket.") + return + # Batch selection selected_batch_explore = st.selectbox( "Select Batch ID to Explore", - options=df_tracker['batch_id'].unique(), - key="explorer_batch" + options=batch_folders ) if selected_batch_explore: + # Load master batch tracking data + master_df = load_master_batch_tracking(s3_client, bucket, selected_batch_explore) + + # Get batch details from tracker + batch_info = df_tracker[df_tracker['batch_id'] == selected_batch_explore].iloc[0] if not df_tracker[df_tracker['batch_id'] == selected_batch_explore].empty else None + + # Display batch summary metrics + if batch_info is not None and master_df is not None: + st.subheader("Batch Summary") + + # Calculate metrics for B and AC output types + b_files = master_df[master_df['output_type'] == 'B'].shape[0] if not master_df.empty else 0 + ac_files = master_df[master_df['output_type'] == 'AC'].shape[0] if not master_df.empty else 0 + + total_input = batch_info['total_input_files'] + total_processed = b_files + ac_files + remaining_b = total_input - b_files + remaining_ac = total_input - ac_files + + # Display metrics + col1, col2, col3, col4 = st.columns(4) + with col1: + st.metric("Total Input Files", f"{total_input:,}") + st.metric("Total Processed", f"{total_processed:,}") + + with col2: + st.metric("B Files Processed", f"{b_files:,}") + st.metric("B Files Remaining", f"{remaining_b:,}") + + with col3: + st.metric("AC Files Processed", f"{ac_files:,}") + st.metric("AC Files Remaining", f"{remaining_ac:,}") + + with col4: + if batch_info['status'] == 'Running': + elapsed_time = (datetime.now() - batch_info['start_time']).total_seconds() / 3600 + current_rate = total_processed / elapsed_time if elapsed_time > 0 else 0 + st.metric("Current Rate", f"{current_rate:.2f} files/hour") + + if current_rate > 0: + est_completion_b = remaining_b / current_rate + est_completion_ac = remaining_ac / current_rate + st.metric("Est. Time to Complete B", f"{est_completion_b:.1f} hours") + st.metric("Est. Time to Complete AC(if this was a B only run, the rate for B will be applied here. So it is not reflective)", f"{est_completion_ac:.1f} hours") + + # Get all runs for the selected batch runs = get_batch_runs(s3_client, bucket, selected_batch_explore) if runs: - selected_run = st.selectbox("Select Run", runs, key="explorer_run") + selected_run = st.selectbox("Select Run", runs) - # Get all files in the run + # If master_df exists, calculate processing rate for this run + if master_df is not None and not master_df.empty: + processing_rate = calculate_processing_rate(master_df, selected_run) + if processing_rate is not None: + st.metric("Processing Rate", f"{processing_rate:.2f} files/hour") + + # Get all files in the run with new structure files = get_run_files(s3_client, bucket, selected_run) if files: - # Display file sections with download buttons - col1, col2, col3 = st.columns(3) + # Create tabs for different file types + file_tabs = st.tabs(["Individual Files", "Consolidated Files", "Tracking Files"]) + + # Individual files tab - now organized by folder + with file_tabs[0]: + if files['individual']: + for folder, file_keys in files['individual'].items(): + with st.expander(f"{folder} ({len(file_keys)} files)"): + for key in file_keys: + filename = key.split('/')[-1] + download_url = create_download_link(s3_client, bucket, key) + if download_url: + st.download_button( + f"📥 {filename}", + download_url, + filename, + mime="text/csv", + key=f"individual_{folder}_{filename}" + ) + else: + st.info("No individual files found for this run.") + + # Consolidated files tab + with file_tabs[1]: + if files['consolidated']: + for key in files['consolidated']: + filename = key.split('/')[-1] + download_url = create_download_link(s3_client, bucket, key) + if download_url: + st.download_button( + f"📥 {filename}", + download_url, + filename, + mime="text/csv", + key=f"consolidated_{filename}" + ) + else: + st.info("No consolidated files found for this run.") + + # Tracking files tab + with file_tabs[2]: + if files['tracking']: + for key in files['tracking']: + filename = key.split('/')[-1] + download_url = create_download_link(s3_client, bucket, key) + if download_url: + st.download_button( + f"📥 {filename}", + download_url, + filename, + mime="text/csv", + key=f"tracking_{filename}" + ) + + # Simplified cost analysis display + if 'costlog' in key.lower(): + st.subheader("Cost Analysis") + df_cost = load_csv_from_s3(s3_client, bucket, key) + if df_cost is not None: + # Display raw data in a compact format + st.dataframe(df_cost, use_container_width=True) + + # Basic metrics calculation + if 'Total Cost' in df_cost.columns: + total_cost = df_cost['Total Cost'].sum() + st.metric("Total Cost", f"${total_cost:.2f}") + + # Display master batch tracking data + if master_df is not None and not master_df.empty: + st.subheader("Master Batch Tracking") + + # Add filtering options + output_type_filter = st.multiselect( + "Filter by Output Type", + options=master_df['output_type'].unique(), + default=master_df['output_type'].unique() + ) + + upload_status_filter = st.multiselect( + "Filter by Upload Status", + options=master_df['upload_status'].unique(), + default=master_df['upload_status'].unique() + ) + + # Apply filters + filtered_master_df = master_df + if output_type_filter: + filtered_master_df = filtered_master_df[filtered_master_df['output_type'].isin(output_type_filter)] + + if upload_status_filter: + filtered_master_df = filtered_master_df[filtered_master_df['upload_status'].isin(upload_status_filter)] + + # Show data + st.dataframe(filtered_master_df, use_container_width=True) + + # Create visualizations for processing status + st.subheader("Processing Status") + + col1, col2 = st.columns(2) with col1: - st.subheader("Individual Files") - for key in files['individual']: - filename = key.split('/')[-1] - download_url = create_download_link(s3_client, bucket, key) - if download_url: - st.download_button( - f"📥 {filename}", - download_url, - filename, - mime="text/csv" - ) + # Distribution by output type + output_counts = master_df['output_type'].value_counts().reset_index() + output_counts.columns = ['Output Type', 'Count'] + + fig = px.pie( + output_counts, + values='Count', + names='Output Type', + title="Files by Output Type" + ) + st.plotly_chart(fig, use_container_width=True) with col2: - st.subheader("Consolidated Files") - for key in files['consolidated']: - filename = key.split('/')[-1] - download_url = create_download_link(s3_client, bucket, key) - if download_url: - st.download_button( - f"📥 {filename}", - download_url, - filename, - mime="text/csv" - ) + # Distribution by upload status + status_counts = master_df['upload_status'].value_counts().reset_index() + status_counts.columns = ['Upload Status', 'Count'] + + fig = px.pie( + status_counts, + values='Count', + names='Upload Status', + title="Files by Upload Status" + ) + st.plotly_chart(fig, use_container_width=True) - with col3: - st.subheader("Log Files") - for key in files['logs']: - filename = key.split('/')[-1] - download_url = create_download_link(s3_client, bucket, key) - if download_url: - st.download_button( - f"📥 {filename}", - download_url, - filename, - mime="text/csv" - ) + # If there's a run timestamp, calculate hourly throughput - # Display log file contents - st.subheader("Log File Analysis") - - # File Log Analysis - file_log_key = next((k for k in files['logs'] if 'filelog' in k), None) - if file_log_key: - st.subheader("File Processing Log") - df_filelog = load_csv_from_s3(s3_client, bucket, file_log_key) - if df_filelog is not None: - col1, col2 = st.columns(2) - - with col1: - st.subheader("Processing Time by Status") - fig = px.box( - df_filelog, - x='status', - y='processing_time_seconds', - title="Processing Time Distribution by Status" - ) - st.plotly_chart(fig, use_container_width=True) - - with col2: - st.subheader("Resource Usage") - fig = px.scatter( - df_filelog, - x='memory_usage_mb', - y='cpu_percent', - color='status', - title="Resource Usage by File" - ) - st.plotly_chart(fig, use_container_width=True) - - # Show raw data with filters - st.subheader("Raw File Log Data") - status_filter = st.multiselect( - "Filter by Status", - options=df_filelog['status'].unique() + if 'run_timestamp' in master_df.columns: + # First extract proper timestamps + master_df['extracted_timestamp'] = master_df['run_timestamp'].apply(extract_timestamp_from_run) + + # Then filter out invalid timestamps + master_df = master_df.dropna(subset=['extracted_timestamp']) + + # Use extracted timestamps for analysis + hourly_counts = master_df.resample('H', on='extracted_timestamp').size() + + if not hourly_counts.empty: + st.subheader("Hourly Processing Rate") + fig = px.bar( + x=hourly_counts.index, + y=hourly_counts.values, + labels={'x': 'Hour', 'y': 'Files Processed'}, + title="Files Processed by Hour" ) - if status_filter: - df_filelog = df_filelog[df_filelog['status'].isin(status_filter)] - st.dataframe(df_filelog, use_container_width=True) - - # Cost Log Analysis - cost_log_key = next((k for k in files['logs'] if 'costlog' in k), None) - if cost_log_key: - st.subheader("Cost Analysis") - df_cost = load_csv_from_s3(s3_client, bucket, cost_log_key) - if df_cost is not None: - col1, col2 = st.columns(2) - - with col1: - total_cost = df_cost['Total Cost'].sum() - avg_cost = df_cost['Total Cost'].mean() - st.metric("Total Cost", f"${total_cost:.2f}") - st.metric("Average Cost per File", f"${avg_cost:.4f}") - - with col2: - total_tokens = df_cost['Input Prompt Tokens'].sum() + df_cost['Output Response Tokens'].sum() - avg_tokens = total_tokens / len(df_cost) - st.metric("Total Tokens", f"{total_tokens:,}") - st.metric("Average Tokens per File", f"{avg_tokens:,.0f}") + st.plotly_chart(fig, use_container_width=True) if __name__ == "__main__": main() \ No newline at end of file