from typing import Any, Dict, List import numpy as np import streamlit as st import datetime import pandas as pd import sys sys.path.append('../') from utils.pandas_General_utils import * from utils.pandas_FeatEngineering_utils import * import statsmodels.api as sm from datetime import datetime, timedelta import warnings # Import the specific warning you want to filter from statsmodels.tools.sm_exceptions import ConvergenceWarning # Filter out ConvergenceWarning warnings.filterwarnings("ignore", category=ConvergenceWarning) from lib.utils.load import load_config, load_image from lib.exposition.export import display_links from lib.inputs.dataset import ( input_dataset, ) from lib.dataprep.format import ( print_empty_cols, print_removed_cols, remove_empty_cols, ) from simulator import WeeklySimulator, plot_billable_area_chart import plotly.express as px import plotly.graph_objects as go simulator_config_settings = { 'CityWorkHours' : { 'SG' : { 'internal' : 8.3, 'client' : 8.2 }, 'TH' : { 'internal' : 7.5, 'client' : 7.4 } }, 'productive_activities' : ['available Hours'], 'billable_activities' : ['available Hours', 'onboarding Hours', 'coaching Hours', 'team_meeting Hours', 'wellness_support Hours', 'fb_training Hours', 'non-fb-training Hours'], 'nonbillable_activities' : ['meal Hours', 'break Hours'] } # Load in Data activity_log = pd.read_csv('../data/Demo_Weekly_Activity_Breakdown.csv') activity_log['Shift Date'] = pd.to_datetime(activity_log['Shift Date']) misassigned = activity_log[(activity_log['Year'] == 2022) & (activity_log['Shift Date'] == datetime(2023,1,6))].index activity_log.drop(misassigned, axis = 'index', inplace = True) activity_log['ActiveCount'] = activity_log['ActiveCount'].astype(float) start_end_lookup = pd.read_csv("../data/Demo_Date_Lookup.csv") for dcol in ['start','end','max_train_date']: start_end_lookup[dcol] = pd.to_datetime(start_end_lookup[dcol]) # Page config st.set_page_config(page_title="MainPage", layout="wide") # Load config config, instructions, readme = load_config( "config_streamlit.toml", "config_instructions.toml", "config_readme.toml" ) # Initialization dates: Dict[Any, Any] = dict() report: List[Dict[str, Any]] = [] results = pd.DataFrame() # Info with st.expander( "A Capacity Planning Simulator for FBCO", expanded=False ): st.write(readme["app"]["app_intro"]) st.write("") st.write("") st.sidebar.image(load_image("logo.png"), use_column_width=True) st.sidebar.subheader("Unlocking Efficiency from Inside the Metaverse", divider='blue') st.write("") st.sidebar.title("Level of Business") # Filter Data with st.sidebar.expander("LOB Filters", expanded=True): city = st.selectbox("Select a City/Country", list(activity_log['City'].unique()), ) market = st.selectbox("Select a Market/Language", list(activity_log[activity_log['City'] == city]['Market'].unique()), ) role = st.selectbox("Select a Role", list(activity_log[(activity_log['City'] == city) & (activity_log['Market'] == market)]['Role'].unique()), ) filtered_data = filter_iterable(activity_log, ['City','Market','Role'], [city,market,role]) lob_col1, lob_col2= st.columns(2) lob_col1.metric(label="No. of Weeks Logged", value=len(filtered_data)) lob_col2.metric(label='Avg HC (5W MA)', value=filtered_data.iloc[-5:]['ActiveCount'].mean()) st.sidebar.title("2. Area Chart") # Load data min_date = filter_iterable(start_end_lookup, ['City','Market','Role'],[city,market,role])['start'] max_date = filter_iterable(start_end_lookup, ['City','Market','Role'],[city,market,role])['max_train_date'] with st.sidebar.expander("Visualize Up To", expanded=True): border_training_date = st.slider( "Training Dataset Range", min_value = min_date.iloc[0], max_value = max_date.iloc[0], value=datetime(2023, 1, 1), format="MM/DD/YY - hh:mm") st.write(f"Range of Training Dates : \n\n {min_date.iloc[0].date()} to {border_training_date.date()}") df_training = filtered_data [filtered_data ['Shift Date'] <= border_training_date] df_planning = filtered_data [filtered_data ['Shift Date'] > border_training_date] st.sidebar.title("3. Capacity Planning Simulator") with st.sidebar.expander("Dates to Plan", expanded=True): planning_dates = st.date_input( "Dates to load for Capacity Planning", min_value = min(pd.to_datetime(df_planning['Shift Date'])), max_value = max(pd.to_datetime(df_planning['Shift Date'])), value = (min(pd.to_datetime(df_planning['Shift Date'])), max(pd.to_datetime(df_planning['Shift Date']))), format="MM.DD.YYYY", ) planning = df_planning[(df_planning['Shift Date'] >= datetime.combine(planning_dates[0], datetime.min.time())) & (df_planning['Shift Date'] <= datetime.combine(planning_dates[1], datetime.max.time()))] planning['Leave Count'] = 0 st.write(f"Range of Capacity Planning Dates : \n\n {planning_dates[0]} to {planning_dates[1]}") # Launch Training & Forecasts if st.checkbox( "Launch Simulator", value=False, help=readme['tooltips']['launch_forecast'] ): st.header("Weekly Billable Activity Breakdown") st.plotly_chart(plot_billable_area_chart(simulator_config_settings, df_training), use_container_width = True) st.header("Capacity Planning Simulator") filtercol1, filtercol2 = st.columns(2) date_to_plan = filtercol1.selectbox("Select an Initial Date", list(planning['Shift Date'].unique()), ) NWeeks = filtercol2.slider('Select Length of Planning Horizon', min_value = 0, max_value = 12, value = 4) dtp = planning[planning['Shift Date'] >= datetime.combine(date_to_plan, datetime.min.time())].iloc[:NWeeks] all_info = pd.DataFrame() tabs = st.tabs([str(x.date()) for x in list(dtp['Shift Date'])]) for ind, data_to_plan in dtp.reset_index().iterrows(): with tabs[ind]: tab_name = [str(x.date()) for x in list(dtp['Shift Date'])][ind] week_sim = WeeklySimulator(simulator_config_settings, data_to_plan['City'], data_to_plan['ActiveCount'], data_to_plan['Required FTE'],data_to_plan,0) data_cols = simulator_config_settings['billable_activities'] + simulator_config_settings['nonbillable_activities'] orig_data = week_sim.ActivityLog[['City', 'Shift Date','ActiveCount','Required FTE', 'Leave Count',]+data_cols] st.warning("Only Change one parameter at a time, multiple changes at once may not yield accurate results.", icon="⚠️") st.header('Original Data') st.dataframe(orig_data.to_frame().T, hide_index = True) st.header('Set Changes Here', help = "Adjust the allocation of hours for this week. Add in a (+) positive number to increase hour allocation and add in a (-) negative number to decrease it.") editor = pd.DataFrame(0.0, index = [0], columns = ['ActiveCount','Required FTE','Leave Count'] + data_cols) edited_df_ = st.data_editor(editor,hide_index = True, column_config = { "ActiveCount" : st.column_config.NumberColumn( "ActiveCount", min_value = 0, step = 0.5, format="%f" ) }, key = f"Table_{ind}" ) #edited_df_ = st.data_editor(orig_data.to_frame().T, hide_index = True) URI_0 = week_sim.UtilizationInternal URC_0 = week_sim.UtilizationClient WIO_0 = week_sim.WIO OOO_0 = week_sim.OOO UAP_0 = week_sim.UnallocatedHours/week_sim.ExpectedMaxHours if st.button("Update Activity Distribution", key = f"Update_{ind}"): edited_df_ += orig_data[editor.columns] edited_df = edited_df_.iloc[0] for k in edited_df.keys(): if edited_df[k] != orig_data[k]: if 'Hour' in k: week_sim.update_activity(k,float(edited_df[k])) elif k == 'Required FTE': week_sim.update_FTETarget(float(edited_df[k])) elif k == 'ActiveCount': week_sim.update_ActiveCount(float(edited_df[k])) elif k =='Leave Count': week_sim.update_LeaveAllocation(float(edited_df[k])) st.header('Resulting DataFrame') update = week_sim.ActivityLog[['City', 'Shift Date','ActiveCount','Required FTE', 'Leave Count',]+data_cols].to_frame().T st.dataframe(update) URI_1 = week_sim.UtilizationInternal URC_1 = week_sim.UtilizationClient WIO_1 = week_sim.WIO OOO_1 = week_sim.OOO UAP_1 = week_sim.UnallocatedHours/week_sim.ExpectedMaxHours #st.info(week_sim) col1,col2,col3,col4,col5 = st.columns(5) col1.metric("Utilization Rate (Internal)", f"{100*URI_1:.2f}%", delta=f"{100*(URI_1-URI_0):.2f}%") col2.metric("Utilization Rate (Client)", f"{100*URC_1:.2f}%", delta=f"{100*(URC_1-URC_0):.2f}%") col3.metric("WIO Shrinkage", f"{100*WIO_1:.2f}%", delta=f"{100*(WIO_1-WIO_0):.2f}%") col4.metric("OOO Shrinkage", f"{100*OOO_1:.2f}%", delta=f"{100*(OOO_1-OOO_0):.2f}%") col5.metric("Unallocated Percentage", f"{100*UAP_1:.2f}%", delta=f"{100*(UAP_1-UAP_0):.2f}%") if week_sim.UnallocatedHours <= 0: st.warning('WARNING : Total Hours exceeds expected hours based on Actual FTE.', icon="⚠️") st.plotly_chart(week_sim.generate_waterfall(), use_container_width = True) update['Market'] = market update['Role'] = role update['UtilizationInternal'] = URI_1 update['UtilizationClient'] = URC_1 update['WIO'] = WIO_1 update['OOO'] = OOO_1 update['Unallocated_pct'] = UAP_1 export_cols = ['City','Market','Role', 'Shift Date','ActiveCount','Required FTE', 'Leave Count',]+data_cols + ['UtilizationInternal','UtilizationClient','WIO','OOO','Unallocated_pct'] st.download_button( "Download Data as CSV", data = update[export_cols].to_csv(), file_name = f"{city}_{market}_{role}_Capacity_Plan_{tab_name}.csv", mime = "text/csv" )