import streamlit as st import pandas as pd import re import matplotlib.pyplot as plt import matplotlib.dates as mdates import seaborn as sns from datetime import datetime import io # Set page config st.set_page_config( page_title="Driver Analysis Tool", page_icon="🚌", layout="wide" ) def main(): # Custom function to create download buttons for dataframes def download_dataframe(df, filename, button_text): """ Creates a download button for a dataframe """ buffer = io.BytesIO() with pd.ExcelWriter(buffer, engine='xlsxwriter') as writer: df.to_excel(writer, index=False) buffer.seek(0) return st.download_button( label=button_text, data=buffer, file_name=filename, mime="application/vnd.ms-excel" ) # Function for dataframes with index def download_dataframe_with_index(df, filename, button_text, index_label=None): """ Creates a download button for a dataframe with index """ buffer = io.BytesIO() with pd.ExcelWriter(buffer, engine='xlsxwriter') as writer: df.to_excel(writer, index=True, index_label=index_label) buffer.seek(0) return st.download_button( label=button_text, data=buffer, file_name=filename, mime="application/vnd.ms-excel" ) # Title and description st.title("📊 Driver Analysis Tool") st.markdown(""" This application analyzes driver duties, schedules, and utilization. Upload the required files below to start your analysis. """) # Sidebar for inputs st.sidebar.header("Upload Files") # Upload duty codes duty_codes_file = st.sidebar.file_uploader("Upload duty codes Excel file", type=['xlsx']) if duty_codes_file: duty_df = pd.read_excel(duty_codes_file) st.sidebar.success("Duty codes loaded successfully!") else: st.sidebar.warning("Please upload duty codes file") duty_df = None # Upload roster file roster_file = st.sidebar.file_uploader("Upload roster Excel file", type=['xlsx']) if roster_file: st.sidebar.success("Roster file loaded successfully!") else: st.sidebar.warning("Please upload roster file") # Upload driver master file driver_master_file = st.sidebar.file_uploader("Upload driver master Excel file", type=['xlsx']) if driver_master_file: st.sidebar.success("Driver master file loaded successfully!") else: st.sidebar.warning("Please upload driver master file") # Add tabs for different analyses tabs = st.tabs(["Duty Overview", "Summary by Date", "Driver Summary", "Utilization Analysis", "Red Flags"]) # Only proceed if all files are uploaded if duty_codes_file and roster_file and driver_master_file and duty_df is not None: # Process data with st.spinner("Processing data..."): # Create duty dictionary duty_dict = duty_df.set_index('duty_code')['category'].to_dict() # Read the roster Excel file xls = pd.ExcelFile(roster_file) # Initialize an empty list to store DataFrames df_list = [] # Read each sheet and append rows to the list for sheet_name in xls.sheet_names: sheet_df = pd.read_excel(xls, sheet_name=sheet_name, dtype={'emp_num': str}) # Drop leading zeroes from emp_num sheet_df['emp_num'] = sheet_df['emp_num'].str.lstrip('0') # Filter out columns without a heading sheet_df = sheet_df.loc[:, ~sheet_df.columns.astype(str).str.contains('^Unnamed')] df_list.append(sheet_df) # Concatenate all DataFrames in the list vertically df = pd.concat(df_list, ignore_index=True) # Ensure column names are strings and remove time part df.columns = [str(col).split(' ')[0] if ' ' in str(col) else str(col) for col in df.columns] # Exclude 'emp_num' from date conversion date_cols = df.columns[~df.columns.isin(['emp_num'])] # Convert array of dates in string format from yyyy-mm-dd to dd-mm-yyyy converted_cols = [] for date in date_cols: try: converted_cols.append(datetime.strptime(date, '%Y-%m-%d').strftime('%d-%m-%Y')) except: converted_cols.append(date) df.columns = ['emp_num'] + converted_cols # Read the driver master file driver_master = pd.read_excel(driver_master_file, dtype={'emp_num': str}) # Function to classify the cell value def classify_value(value): pattern1 = r"^\d{5,6}-(?:[A-Za-z]\d{4,6}|\d{5,7})$" pattern = r"^\d{7}[A-Z]\d{6}$" # Matches strings like 1234567M123456 pattern2 = r"^[1-9][A-Z]\d{3}$" # Matches strings like 4A123 pattern3 = r'^\d{6}-[A-Z]{2}\d{5}$' # Check for NaN values if pd.isna(value) or value == 0: return 'not-available' # Check for predefined patterns value_str = str(value) if re.search(pattern, value_str) or re.search(pattern1, value_str) or re.search(pattern2, value_str) or re.search(pattern3, value_str): return 'duty' # Check the duty_dict for matching keys for duty_code, category in duty_dict.items(): if duty_code in value_str: return category # Default classification return 'duty' # Apply the classification function to each cell except 'emp_num' classified_df = df.set_index('emp_num').applymap(classify_value).reset_index() # Create a copy for date analysis classified_datedf = classified_df.copy() # Drop the 'emp_num' column for date summary date_summary_df = classified_datedf.drop(columns=['emp_num']) # Calculate the total for each date based on the classification summary_datedf = date_summary_df.apply(lambda x: x.value_counts()).fillna(0).T # Ensure the index is in datetime format and sorted summary_datedf.index = pd.to_datetime(summary_datedf.index, format='%d-%m-%Y') summary_datedf = summary_datedf.sort_index() # Add total column summary_datedf['total'] = summary_datedf.sum(axis=1) # Function to extract date from column names def extract_date(column_name): if isinstance(column_name, str): match = re.search(r'\d{2}-\d{2}-\d{4}', column_name) return match.group(0) if match else column_name return column_name # Apply the function to strip column names, excluding date type columns non_date_columns = classified_df.select_dtypes(exclude=['datetime']).columns classified_df.columns = [extract_date(col) if col in non_date_columns and col != 'emp_num' else col for col in classified_df.columns] # Create a summary DataFrame grouped by employee summary_df = classified_df.groupby('emp_num').apply(lambda x: pd.Series({ 'Duty': (x == 'duty').sum().sum(), 'Spare': (x == 'spare').sum().sum(), 'Not-Available': (x == 'not-available').sum().sum(), 'Day Off': (x == 'day-off').sum().sum(), 'Vacation': (x == 'vacation').sum().sum(), 'Support Duty': (x == 'support').sum().sum(), 'Training': (x == 'training').sum().sum(), 'Total': len(x.columns) - 1 # Subtract 1 for emp_num column })).reset_index() # Calculate 'Spare / Duty' percentage summary_df['Spare / Duty'] = (summary_df['Spare'] / (summary_df['Duty'] + 0.001) * 100).round(2) # Merge summary_df with driver_master to get depot and other information summary_df = summary_df.merge( driver_master[['emp_num', 'Depot', 'Cost_Center', 'Assignment', 'Nationality']], on='emp_num', how='left' ) # Calculate utilization summary_df['utilization'] = ( (summary_df['Duty'] + summary_df['Support Duty'] + summary_df['Training'] + summary_df['Day Off']) / (summary_df['Total'] - summary_df['Vacation'] - summary_df['Not-Available']) ).round(2) * 100 # Define utilization ranges bins = [0, 20, 40, 60, 80, 100] labels = ['0-20', '20-40', '40-60', '60-80', '80-100'] summary_df['Utilization_Range'] = pd.cut(summary_df['utilization'], bins=bins, labels=labels, include_lowest=True) # Red flag filters high_spare_df = summary_df[ (summary_df['utilization'] < 50) & (summary_df['Spare / Duty'] > 50) & ((summary_df['Spare'] + summary_df['Duty']) > summary_df['Not-Available']) ] high_util_df = summary_df[ (summary_df['Spare / Duty'] < 5) & ((summary_df['Spare'] + summary_df['Duty']) > summary_df['Not-Available']) ] # Tab 1: Duty Overview with tabs[0]: st.header("Duty Codes Overview") st.dataframe(duty_df) st.header("Processed Duty Data Sample") st.dataframe(classified_df.head(10)) # Add download button download_dataframe(classified_df, "planned_duties.xlsx", "Download Processed Duties") # Tab 2: Summary by Date with tabs[1]: st.header("Summary by Date") # Format the index back to string for display display_date_df = summary_datedf.copy() display_date_df.index = display_date_df.index.strftime('%d-%m-%Y') st.dataframe(display_date_df) # Format for download download_df = summary_datedf.copy() download_df.index = download_df.index.strftime('%d-%m-%Y') # Add download button download_dataframe_with_index(download_df, "summary_by_date_planned.xlsx", "Download Date Summary", "Date") st.header("Daily Trend Visualization") # Restore datetime index for plotting summary_datedf.index = pd.to_datetime(summary_datedf.index, format='%d-%m-%Y') fig, ax = plt.subplots(figsize=(10, 6)) ax.plot(summary_datedf.index, summary_datedf['day-off'], label='Day Off', marker='o') ax.plot(summary_datedf.index, summary_datedf['duty'], label='Duty', marker='o') ax.plot(summary_datedf.index, summary_datedf['spare'], label='Spare', marker='o') ax.plot(summary_datedf.index, summary_datedf['support'], label='Support', marker='o') ax.plot(summary_datedf.index, summary_datedf['not-available'], label='Not-Available', marker='o') # Set the labels and title ax.set_xlabel('Date') ax.set_ylabel('Count') ax.set_title('Daily Count of Day Off, Duty, Vacation and Spare') ax.legend() # Rotate the x-axis labels for better readability plt.xticks(rotation=45) # Set weekly ticks for the x-axis ax.xaxis.set_major_locator(mdates.WeekdayLocator(interval=1)) ax.xaxis.set_major_formatter(mdates.DateFormatter('%d-%m-%Y')) plt.tight_layout() st.pyplot(fig) # Tab 3: Driver Summary with tabs[2]: st.header("Driver Summary") st.dataframe(summary_df) # Add download button download_dataframe(summary_df, "planned_summary.xlsx", "Download Driver Summary") # Add Depot filter st.subheader("Filter by Depot") selected_depot = st.selectbox("Select a depot", ["All"] + list(summary_df["Depot"].unique())) if selected_depot != "All": filtered_summary = summary_df[summary_df["Depot"] == selected_depot] else: filtered_summary = summary_df st.dataframe(filtered_summary) # Add download button for filtered data if selected_depot != "All": download_dataframe(filtered_summary, f"planned_summary_{selected_depot}.xlsx", f"Download {selected_depot} Summary") # Tab 4: Utilization Analysis with tabs[3]: st.header("Driver Utilization Analysis") col1, col2 = st.columns(2) with col1: st.subheader("Distribution by Utilization Range") # Group by 'Utilization_Range' and count drivers utilization_range_df = summary_df.groupby('Utilization_Range').size().reset_index(name='Number_of_Drivers') fig, ax = plt.subplots(figsize=(10, 6)) bars = ax.bar(utilization_range_df['Utilization_Range'], utilization_range_df['Number_of_Drivers'], color='skyblue') # Annotate the bar chart for bar in bars: height = bar.get_height() ax.text(bar.get_x() + bar.get_width()/2., height + 0.1, f'{height}', ha='center', va='bottom') ax.set_title('Distribution of Drivers by Utilization Range') ax.set_xlabel('Utilization Range (%)') ax.set_ylabel('Number of Drivers') plt.tight_layout() st.pyplot(fig) # Show data table and download button st.dataframe(utilization_range_df) download_dataframe(utilization_range_df, "utilization_distribution.xlsx", "Download Utilization Distribution") with col2: st.subheader("Depot Utilization Heatmap") # Group by 'Depot' and 'Utilization_Range' and count drivers heatmap_data = summary_df.groupby(['Depot', 'Utilization_Range']).size().unstack(fill_value=0) fig, ax = plt.subplots(figsize=(10, 6)) sns.heatmap(heatmap_data, annot=True, fmt='d', cmap='YlGnBu', ax=ax) ax.set_title('Heatmap of Driver Utilization by Depot') ax.set_xlabel('Utilization Range (%)') ax.set_ylabel('Depot') plt.tight_layout() st.pyplot(fig) # Show data table and download button st.dataframe(heatmap_data) download_dataframe_with_index(heatmap_data, "depot_utilization_heatmap.xlsx", "Download Depot Heatmap Data", "Depot") # Tab 5: Red Flags with tabs[4]: st.header("Red Flags Analysis") col1, col2 = st.columns(2) with col1: st.subheader("High Spare to Duty Ratio") st.markdown("Filters applied:") st.markdown("- Utilization < 50%") st.markdown("- Spare/Duty ratio > 50%") st.markdown("- Spare + Duty > Not Available") st.dataframe(high_spare_df) # Add download button download_dataframe(high_spare_df, "planned_high_spare.xlsx", "Download High Spare Report") with col2: st.subheader("High Utilization Drivers") st.markdown("Filters applied:") st.markdown("- Spare/Duty ratio < 5%") st.markdown("- Spare + Duty > Not Available") st.dataframe(high_util_df) # Add download button download_dataframe(high_util_df, "planned_high_util.xlsx", "Download High Utilization Report") else: # Show message when files are not uploaded with tabs[0]: st.info("Please upload all required files from the sidebar to start the analysis.") with tabs[1]: st.info("Please upload all required files from the sidebar to start the analysis.") with tabs[2]: st.info("Please upload all required files from the sidebar to start the analysis.") with tabs[3]: st.info("Please upload all required files from the sidebar to start the analysis.") with tabs[4]: st.info("Please upload all required files from the sidebar to start the analysis.") # Add a footer st.markdown("---") st.markdown("© 2025 Driver Analysis Tool | Created by MT Planning Team" ) if __name__ == "__main__": main()