diff --git a/fieldExtraction/src/batch_tracking.py b/fieldExtraction/src/batch_tracking.py index e629749..e760af3 100644 --- a/fieldExtraction/src/batch_tracking.py +++ b/fieldExtraction/src/batch_tracking.py @@ -96,7 +96,7 @@ class BatchTracker: 'files_processed', 'contracts_per_hour', 'status', 'machine_info', 'git_branch', 'git_commit', 'git_commit_message', 'git_commit_date' - ]) + ]) def get_machine_info(self): """Get information about the machine running the process""" diff --git a/fieldExtraction/src/consolidate_output.py b/fieldExtraction/src/consolidate_output.py index 1d1c810..9f11c99 100644 --- a/fieldExtraction/src/consolidate_output.py +++ b/fieldExtraction/src/consolidate_output.py @@ -8,6 +8,31 @@ from pathlib import Path from datetime import datetime +def upload_tracking_files(run_timestamp): + """ + Upload tracking files to S3 during batch processing. + + Args: + run_timestamp (str): Timestamp string for the current processing run + """ + if not config.WRITE_TO_S3: + return + + BUCKET_NAME = config.S3_OUTPUT_BUCKET + tracking_files = { + 'cost_log': config.COST_LOG_NAME, + 'file_log': config.FILE_LOG_NAME + } + + for file_type, file_name in tracking_files.items(): + if os.path.exists(file_name): + try: + s3_key = os.path.join(run_timestamp, file_name).replace("\\", "/") + config.S3_CLIENT.upload_file(file_name, BUCKET_NAME, s3_key) + except Exception as e: + print(f"Error uploading {file_name}: {str(e)}") + raise + def consolidate_output(run_timestamp): Path(config.CONSOLIDATED_OUTPUT_DIRECTORY).mkdir(parents=True, exist_ok=True) @@ -99,18 +124,7 @@ def consolidate_output(run_timestamp): else: print(f"File not found: {local_path}") # Do final upload of tracking files - tracking_files = { - 'cost_log': config.COST_LOG_NAME, - 'file_log': config.FILE_LOG_NAME - } - - for file_type, file_name in tracking_files.items(): - if os.path.exists(file_name): - try: - print(f"uploading final {file_name}") - s3_key = os.path.join(run_timestamp, file_name).replace("\\", "/") - config.S3_CLIENT.upload_file(file_name, BUCKET_NAME, s3_key) - except Exception as e: - print(f"Error uploading {file_name}: {str(e)}") + upload_tracking_files(run_timestamp) + + print(f"Upload completed to s3://{BUCKET_NAME}/{run_timestamp}") - print(f"Upload completed to s3://{BUCKET_NAME}/{run_timestamp}") \ No newline at end of file diff --git a/streamlit/runs-dashboard.py b/streamlit/runs-dashboard.py index 8642192..ae23689 100644 --- a/streamlit/runs-dashboard.py +++ b/streamlit/runs-dashboard.py @@ -19,25 +19,12 @@ st.set_page_config( initial_sidebar_state="expanded" ) -def get_value_from_env(key: str, default_value: str=None) -> str: - value = os.getenv(key, default_value) - if value is not None: - return value - raise KeyError(f'{key} not found in .env file') - -# Initialize S3 client with credentials +# Initialize S3 client @st.cache_resource def get_s3_client(): - AWS_ACCESS_KEY_ID = get_value_from_env('AWS_ACCESS_KEY_ID') - AWS_SECRET_ACCESS_KEY = get_value_from_env('AWS_SECRET_ACCESS_KEY') - AWS_SESSION_TOKEN = get_value_from_env('AWS_SESSION_TOKEN') - return boto3.client( "s3", - region_name="us-east-2", - aws_access_key_id=AWS_ACCESS_KEY_ID, - aws_secret_access_key=AWS_SECRET_ACCESS_KEY, - aws_session_token=AWS_SESSION_TOKEN, + region_name="us-east-2" ) @st.cache_data @@ -53,7 +40,7 @@ def load_batch_tracker(_s3_client, bucket): st.error(f"Error loading batch tracker: {str(e)}") return None -# [Rest of the dashboard code remains the same...] + @st.cache_data def get_batch_runs(_s3_client, bucket, batch_id=None): """Get all run directories for a specific batch or all batches""" @@ -89,6 +76,7 @@ def load_csv_from_s3(_s3_client, bucket, key): st.error(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: @@ -134,6 +122,32 @@ def get_run_files(s3_client, bucket, run_prefix): return None +def get_active_runs(df): + """ + 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 + """ + active_runs = [] + 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']) + + # Group by batch_id to analyze status progression + for batch_id, group in df_sorted.groupby('batch_id'): + latest_record = group.iloc[-1] + if (latest_record['status'] == 'Running' and + pd.to_datetime(latest_record['start_time']).date() == today): + active_runs.append({ + 'batch_id': batch_id, + 'start_time': latest_record['start_time'], + 'input_dir': latest_record['input_dir'], + 'files_processed': latest_record['files_processed'], + 'total_input_files': latest_record['total_input_files'] + }) + + return pd.DataFrame(active_runs) if active_runs else pd.DataFrame() + def main(): st.title("Doczy.AI Batch Processing Dashboard") @@ -152,6 +166,9 @@ def main(): df_tracker['start_time'] = pd.to_datetime(df_tracker['start_time']) df_tracker['end_time'] = pd.to_datetime(df_tracker['end_time']) + # Create a copy for filtering + df_filtered = df_tracker.copy() + # Add tabs for different views tab1, tab2 = st.tabs(["Overview Dashboard", "Batch Explorer"]) @@ -159,9 +176,19 @@ def main(): # Sidebar filters st.sidebar.header("Filters") + # Batch filter + batch_filter = st.sidebar.selectbox( + "Filter by Batch ID", + options=['All'] + sorted(df_tracker['batch_id'].unique().tolist()) + ) + + # Apply batch filter if selected + if batch_filter != 'All': + df_filtered = df_filtered[df_filtered['batch_id'] == batch_filter] + # Date range filter - min_date = df_tracker['start_time'].min().date() - max_date = df_tracker['start_time'].max().date() + min_date = df_filtered['start_time'].min().date() + max_date = df_filtered['start_time'].max().date() date_range = st.sidebar.date_input( "Date Range", @@ -173,42 +200,58 @@ def main(): # Apply date filter if len(date_range) == 2: start_date, end_date = date_range - mask = (df_tracker['start_time'].dt.date >= start_date) & (df_tracker['start_time'].dt.date <= end_date) - df_filtered = df_tracker[mask] - else: - df_filtered = df_tracker + mask = (df_filtered['start_time'].dt.date >= start_date) & (df_filtered['start_time'].dt.date <= end_date) + df_filtered = df_filtered[mask] # Status filter - status_options = ['All'] + sorted(df_tracker['status'].unique().tolist()) + status_options = ['All'] + sorted(df_filtered['status'].unique().tolist()) selected_status = st.sidebar.selectbox("Status", status_options) if selected_status != 'All': df_filtered = df_filtered[df_filtered['status'] == selected_status] - # Top metrics row - col1, col2, col3, col4 = st.columns(4) - - # Active runs - active_runs = df_filtered[df_filtered['status'] == 'Running'].shape[0] - with col1: - st.metric("Active Runs", active_runs) - - # Total files processed - total_files = df_filtered['files_processed'].sum() - with col2: - st.metric("Total Files Processed", f"{total_files:,}") - - # Average processing speed - avg_speed = df_filtered[df_filtered['contracts_per_hour'] > 0]['contracts_per_hour'].mean() - with col3: - st.metric("Avg Processing Speed", f"{avg_speed:.2f} files/hour") - - # Success rate - completed_runs = df_filtered[df_filtered['status'] == 'Completed'].shape[0] - total_runs = df_filtered.shape[0] - success_rate = (completed_runs / total_runs * 100) if total_runs > 0 else 0 - with col4: - st.metric("Success Rate", f"{success_rate:.1f}%") + # Calculate metrics based on filtered data + if not df_filtered.empty: + # Calculate active runs + active_runs_df = get_active_runs(df_filtered) + + # Top metrics row + col1, col2, col3, col4 = st.columns(4) + + # Active runs metric + with col1: + st.metric("Active Runs", len(active_runs_df)) + + # Total files processed + total_files = df_filtered['files_processed'].sum() + with col2: + st.metric("Total Files Processed", f"{total_files:,}") + + # Average processing speed (only for rows with positive values) + speed_data = df_filtered[df_filtered['contracts_per_hour'] > 0] + avg_speed = speed_data['contracts_per_hour'].mean() if not speed_data.empty else 0 + with col3: + 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 + with col4: + st.metric("Success Rate", f"{success_rate:.1f}%" if success_rate > 0 else "-") + + # Display current active runs if any exist + 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']], + use_container_width=True + ) + else: + st.warning("No data available for the selected filters.") # Graphs row col1, col2 = st.columns(2) @@ -238,7 +281,7 @@ def main(): ) st.plotly_chart(fig, use_container_width=True) - # Include Cost Analysis section + # Cost Analysis section st.subheader("Cost Analysis") selected_batch = st.selectbox( "Select Batch ID", @@ -281,6 +324,7 @@ def main(): st.subheader("Cost Details") st.dataframe(df_cost, use_container_width=True) + with tab2: st.header("Batch Run Explorer")