File size: 14,133 Bytes
cd87ae5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 | import sys
import os
import pandas as pd
import streamlit as st
import plotly.express as px
from datetime import datetime
# Add parent directory to path to import LaborOptimizer
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from optimization.labor_optimizer import LaborOptimizer
def get_available_dates(data_path):
"""Load the orders data and extract unique dates"""
try:
orders_file = os.path.join(data_path, "orders.csv")
if os.path.exists(orders_file):
orders_df = pd.read_csv(orders_file)
if "due_date" in orders_df.columns:
# Convert to datetime and extract unique dates
dates = pd.to_datetime(orders_df["due_date"]).dt.date.unique()
# Sort dates in descending order (most recent first)
dates = sorted(dates, reverse=True)
return dates
except Exception as e:
st.error(f"Error loading dates: {str(e)}")
return []
def get_metadata_stats(optimizer, target_date=None):
"""
Aggregate metadata statistics about employee costs and availability
Args:
optimizer: LaborOptimizer instance
target_date: Target date for availability analysis
Returns:
dict: Dictionary containing various statistics
"""
try:
# Employee type costs
employee_types_df = optimizer.employee_types_df
costs_data = []
for _, row in employee_types_df.iterrows():
costs_data.append({
'Employee Type': row['type_name'].title(),
'Usual Cost ($/hr)': f"${row['usual_cost']:.2f}",
'Overtime Cost ($/hr)': f"${row['overtime_cost']:.2f}",
'Evening Shift Cost ($/hr)': f"${row['evening_shift_cost']:.2f}",
'Max Hours': row['max_hours'],
'Unit Manpower/Hr': row['unit_productivity_per_hour']
})
# Shift hours information
shift_hours = optimizer._get_shift_hours()
shift_data = []
for shift_type, hours in shift_hours.items():
shift_data.append({
'Shift Type': shift_type.replace('_', ' ').title(),
'Duration (hours)': f"{hours:.1f}"
})
# Employee availability for target date
availability_data = []
if target_date:
target_date_str = pd.to_datetime(target_date).strftime("%Y-%m-%d")
else:
# Use most recent date if no target date specified, but show warning
target_date_str = pd.to_datetime(optimizer.orders_df["due_date"]).max().strftime("%Y-%m-%d")
st.warning("β οΈ No target date specified. Using the most recent order date for analysis. Please select a specific target date for accurate availability data.")
availability_target_date = optimizer.employee_availability_df[
optimizer.employee_availability_df["date"] == target_date_str
]
employee_availability = optimizer.employees_df.merge(
availability_target_date, left_on="id", right_on="employee_id", how="left"
)
for emp_type in optimizer.employee_types_df["type_name"]:
emp_type_data = employee_availability[
employee_availability["type_name"] == emp_type
]
if not emp_type_data.empty:
first_shift_available = emp_type_data["first_shift_available"].sum()
second_shift_available = emp_type_data["second_shift_available"].sum()
overtime_available = emp_type_data["overtime_available"].sum()
total_employees = len(emp_type_data)
else:
first_shift_available = second_shift_available = overtime_available = total_employees = 0
availability_data.append({
'Employee Type': emp_type.title(),
'Total Employees': total_employees,
'Usual Time Available': first_shift_available,
'Evening Shift Available': second_shift_available,
'Overtime Available': overtime_available
})
# Overall statistics
total_employees = len(optimizer.employees_df)
total_employee_types = len(optimizer.employee_types_df)
total_orders = len(optimizer.orders_df)
return {
'costs_data': costs_data,
'shift_data': shift_data,
'availability_data': availability_data,
'overall_stats': {
'Total Employees': total_employees,
'Employee Types': total_employee_types,
'Total Orders': total_orders,
'Analysis Date': target_date_str,
'is_default_date': not bool(target_date)
}
}
except Exception as e:
st.error(f"Error generating metadata: {str(e)}")
return None
def display_metadata_section(metadata):
"""Display metadata in organized sections"""
if not metadata:
return
# Make the entire Dataset Overview section collapsible
# with st.expander("π Dataset Overview", expanded=False):
with st.expander("π Dataset Overview", expanded=False):
st.write(f"Data path: {st.session_state.data_path}")
# Overall statistics
st.write("Information on the date chosen - not an optimization report") # df, err, func, keras!
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric("Total Employees Available", metadata['overall_stats']['Total Employees'])
with col2:
st.metric("Employee Types Available", metadata['overall_stats']['Employee Types'])
with col3:
st.metric("Total Orders", metadata['overall_stats']['Total Orders'])
with col4:
analysis_date = metadata['overall_stats']['Analysis Date']
if metadata['overall_stats'].get('is_default_date', False):
st.metric("Analysis Date", f"{analysis_date} β οΈ", help="Using most recent order date - select specific date for accurate analysis")
else:
st.metric("Analysis Date", analysis_date)
# Create tabs for different metadata sections
tab1, tab2, tab3 = st.tabs(["π° Employee Costs", "π Shift Information", "π₯ Availability"])
with tab1:
st.subheader("Employee Type Costs")
costs_df = pd.DataFrame(metadata['costs_data'])
st.dataframe(costs_df, use_container_width=True)
# Cost comparison chart
costs_for_chart = []
for item in metadata['costs_data']:
emp_type = item['Employee Type']
costs_for_chart.extend([
{'Employee Type': emp_type, 'Cost Type': 'Usual', 'Cost': float(item['Usual Cost ($/hr)'].replace('$', ''))},
{'Employee Type': emp_type, 'Cost Type': 'Overtime', 'Cost': float(item['Overtime Cost ($/hr)'].replace('$', ''))},
{'Employee Type': emp_type, 'Cost Type': 'Evening', 'Cost': float(item['Evening Shift Cost ($/hr)'].replace('$', ''))}
])
chart_df = pd.DataFrame(costs_for_chart)
fig = px.bar(chart_df, x='Employee Type', y='Cost', color='Cost Type',
title='Hourly Costs by Employee Type and Shift',
barmode='group')
st.plotly_chart(fig, use_container_width=True)
with tab2:
st.subheader("Shift Duration Information")
shift_df = pd.DataFrame(metadata['shift_data'])
st.dataframe(shift_df, use_container_width=True)
# Shift duration chart
fig2 = px.bar(shift_df, x='Shift Type', y='Duration (hours)',
title='Shift Duration by Type')
st.plotly_chart(fig2, use_container_width=True)
with tab3:
st.subheader("Employee Availability")
availability_df = pd.DataFrame(metadata['availability_data'])
st.dataframe(availability_df, use_container_width=True)
# # Availability chart
# availability_chart_data = []
# for item in metadata['availability_data']:
# emp_type = item['Employee Type']
# availability_chart_data.extend([
# {'Employee Type': emp_type, 'Shift Type': 'Usual Time', 'Available': item['Usual Time Available']},
# {'Employee Type': emp_type, 'Shift Type': 'Evening Shift', 'Available': item['Evening Shift Available']},
# {'Employee Type': emp_type, 'Shift Type': 'Overtime', 'Available': item['Overtime Available']}
# ])
# chart_df2 = pd.DataFrame(availability_chart_data)
# fig3 = px.bar(chart_df2, x='Employee Type', y='Available', color='Shift Type',
# title='Available Workers by Employee Type and Shift',
# barmode='group')
# st.plotly_chart(fig3, use_container_width=True)
def display_demand(optimizer):
with st.expander("π Demand", expanded=False):
demand_df = optimizer.orders_df
st.header("Demand")
daily_demand = demand_df.groupby('date_of_order').sum()['order_amount'].reset_index()
st.plotly_chart(px.bar(daily_demand, x='date_of_order', y='order_amount', title='Demand by Date'), use_container_width=True)
st.markdown("### Demand for the selected date")
st.dataframe(demand_df[demand_df['date_of_order']==st.session_state.target_date], use_container_width=True)
def display_employee_availability(optimizer):
with st.expander("π₯ Employee Availability", expanded=False):
st.header("Employee Availability")
employee_availability_df = optimizer.employee_availability_df
employee_availability_df['date'] = pd.to_datetime(employee_availability_df['date'])
employee_availability_target_date = employee_availability_df[employee_availability_df['date']==st.session_state.target_date]
employee_availability_target_date = pd.merge(employee_availability_target_date, optimizer.employees_df, left_on='employee_id', right_on='id', how='left')
st.dataframe(employee_availability_target_date[['name', 'employee_id', 'type_name', 'first_shift_available', 'second_shift_available', 'overtime_available']], use_container_width=True)
# Group by type_name and sum the availability columns
available_employee_grouped = employee_availability_target_date.groupby('type_name')[
['first_shift_available', 'second_shift_available', 'overtime_available']
].sum().reset_index()
st.markdown("### Employee Availability for the selected date")
# Create non-stacked (grouped) bar chart using plotly
fig = px.bar(
available_employee_grouped.melt(id_vars=['type_name'], var_name='shift_type', value_name='count'),
x='type_name',
y='count',
color='shift_type',
barmode='group', # This makes it non-stacked
title='Available Employee Count by Type and Shift',
labels={'type_name': 'Employee Type', 'count': 'Available Count', 'shift_type': 'Shift Type'}
)
st.plotly_chart(fig, use_container_width=True)
# st.dataframe(employee_availability_target_date, use_container_width=True)
# st.plotly_chart(px.bar(employee_availability_target_date, x='employee_id', y='availability', title='Employee Availability by Date'), use_container_width=True)
# st.dataframe(employee_availability_df[employee_availability_df['date']==st.session_state.target_date], use_container_width=True)
def main():
"""Main function for metadata page"""
st.set_page_config(page_title="Dataset Metadata", layout="wide")
st.title("π Dataset Metadata Overview")
# Get data_path from session state if available, otherwise create input
if 'data_path' in st.session_state:
# Using shared data_path from optimize_viz.py
data_path = st.session_state.data_path
st.sidebar.info(f"π Using shared data path: `{data_path}`")
else:
st.error("No data path found. Please select a data path in the sidebar.")
if 'target_date' in st.session_state:
target_date = st.session_state.target_date
st.sidebar.info(f"π
Using shared target date: `{target_date}`")
else:
st.error("No target date found. Please select a target date in the sidebar.")
#If the date selection needs to be individualized per page, uncomment the following code
# with st.sidebar:
# # Date selection
# available_dates = get_available_dates(data_path)
# if available_dates:
# date_options = [""] + [str(date) for date in available_dates]
# target_date = st.selectbox(
# "Target Date (select empty for latest date)",
# options=date_options,
# index=0,
# )
# else:
# target_date = st.text_input(
# "Target Date (YYYY-MM-DD, leave empty for latest)"
# )
try:
optimizer = LaborOptimizer(data_path)
# Show warning if no target date is selected
if not target_date:
st.info("π‘ **Tip**: Select a specific target date from the sidebar to see accurate availability data for that date. Currently showing data for the most recent order date.")
metadata = get_metadata_stats(optimizer, target_date if target_date else None)
display_metadata_section(metadata)
display_demand(optimizer)
display_employee_availability(optimizer)
except Exception as e:
st.error(f"Error loading metadata: {str(e)}")
if __name__ == "__main__":
main()
|