jacob-mathison/doczyai-pipelines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Merged in bugfix/tracking-bug-fixes (pull request #295)

Browse Source 
Bugfix/tracking bug fixes

* runtime tracking bug fix

* fixes to dashboard

* for ec2 host

* updated to match config without ac fields param

* Merge remote-tracking branch 'remotes/origin/main' into HEAD

* removed redundant get active runs function


Approved-by: Katon Minhas
This commit is contained in:
Faizan Mohiuddin
2024-11-25 18:06:20 +00:00
committed by Katon Minhas
parent c58490653d
commit 88a57f1e1f
3 changed files with 121 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fieldExtraction/src/batch_tracking.py
+1 -1
View File
@@ -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"""
fieldExtraction/src/consolidate_output.py
+28 -14
View File
@@ -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}")
streamlit/runs-dashboard.py
+92 -48
View File
@@ -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")