Spaces:
Runtime error
Runtime error
Download src/streamlit_app.py from naheedakhtar/driver_analysis_tool: direct link, hf CLI and curl.
- Browser
- Download file 17.8 kB
-
https://huggingface.co/spaces/naheedakhtar/driver_analysis_tool/resolve/main/src/streamlit_app.py
- Command line
-
hf download hf://spaces/naheedakhtar/driver_analysis_tool/src/streamlit_app.py
-
curl -L -o streamlit_app.py https://huggingface.co/spaces/naheedakhtar/driver_analysis_tool/resolve/main/src/streamlit_app.py
17.8 kB
| 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() |