driver_analysis_tool / src /streamlit_app.py
naheedakhtar's picture
Update src/streamlit_app.py
dc71b71 verified
Raw History Blame Contribute Delete
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()