Blame for src/streamlit_app.py

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()