Create app.py

app.py

@@ -0,0 +1,464 @@
+import pandas as pd
+import numpy as np
+import pulp as pl
+import matplotlib.pyplot as plt
+import gradio as gr
+from itertools import product
+import io
+import base64
+import tempfile
+import os
+from datetime import datetime
+
+def am_pm(hour):
+    """Converts 24-hour time to AM/PM format."""
+    period = "AM"
+    if hour >= 12:
+        period = "PM"
+    if hour > 12:
+        hour -= 12
+    elif hour == 0:
+        hour = 12  # Midnight
+    return f"{int(hour):02d}:00 {period}"
+
+def optimize_staffing(
+    csv_file,
+    beds_per_staff,
+    max_hours_per_staff,
+    hours_per_cycle,
+    rest_days_per_week,
+    clinic_start,
+    clinic_end,
+    overlap_time,
+    max_start_time_change
+):
+    # Load data
+    try:
+        if isinstance(csv_file, str):
+            # Handle the case when a filepath is passed directly
+            data = pd.read_csv(csv_file)
+        else:
+            # Handle the case when file object is uploaded through Gradio
+            data = pd.read_csv(io.StringIO(csv_file.decode('utf-8')))
+    except Exception as e:
+        print(f"Error loading CSV file: {e}")
+        return f"Error loading CSV file: {e}", None, None, None
+    
+    # Rename the index column if necessary
+    if data.columns[0] not in ['day', 'Day', 'DAY']:
+        data = data.rename(columns={data.columns[0]: 'day'})
+    
+    # Fill missing values
+    for col in data.columns:
+        if col.startswith('cycle'):
+            data[col].fillna(0, inplace=True)
+    
+    # Calculate clinic hours
+    if clinic_end < clinic_start:
+        clinic_hours = 24 - clinic_start + clinic_end
+    else:
+        clinic_hours = clinic_end - clinic_start
+    
+    # Parameters
+    BEDS_PER_STAFF = float(beds_per_staff)
+    MAX_HOURS_PER_STAFF = float(max_hours_per_staff)
+    HOURS_PER_CYCLE = float(hours_per_cycle)
+    REST_DAYS_PER_WEEK = int(rest_days_per_week)
+    SHIFT_TYPES = [6, 8, 10, 12]  # Standard shift types
+    OVERLAP_TIME = float(overlap_time)
+    CLINIC_START = int(clinic_start)
+    CLINIC_END = int(clinic_end)
+    CLINIC_HOURS = clinic_hours
+    MAX_START_TIME_CHANGE = int(max_start_time_change)
+    
+    # Calculate staff needed per cycle (beds/BEDS_PER_STAFF, rounded up)
+    for col in data.columns:
+        if col.startswith('cycle') and not col.endswith('_staff'):
+            data[f'{col}_staff'] = np.ceil(data[col] / BEDS_PER_STAFF)
+    
+    # Get cycle names and number of cycles
+    cycle_cols = [col for col in data.columns if col.startswith('cycle') and not col.endswith('_staff')]
+    num_cycles = len(cycle_cols)
+    
+    # Define cycle times
+    cycle_times = {}
+    for i, cycle in enumerate(cycle_cols):
+        cycle_start = (CLINIC_START + i * HOURS_PER_CYCLE) % 24
+        cycle_end = (CLINIC_START + (i + 1) * HOURS_PER_CYCLE) % 24
+        cycle_times[cycle] = (cycle_start, cycle_end)
+    
+    # Get staff requirements
+    max_staff_needed = max([data[f'{cycle}_staff'].max() for cycle in cycle_cols])
+    
+    # Define possible shift start times
+    shift_start_times = list(range(CLINIC_START, CLINIC_START + int(CLINIC_HOURS) - min(SHIFT_TYPES) + 1))
+    
+    # Generate all possible shifts
+    possible_shifts = []
+    for duration in SHIFT_TYPES:
+        for start_time in shift_start_times:
+            end_time = (start_time + duration) % 24
+            
+            # Create a shift with its coverage of cycles
+            shift = {
+                'id': f"{duration}hr_{start_time:02d}",
+                'start': start_time,
+                'end': end_time,
+                'duration': duration,
+                'cycles_covered': set()
+            }
+            
+            # Determine which cycles this shift covers
+            for cycle, (cycle_start, cycle_end) in cycle_times.items():
+                # Handle overnight cycles
+                if cycle_end < cycle_start:  # overnight cycle
+                    if start_time >= cycle_start or end_time <= cycle_end or (start_time < end_time and end_time > cycle_start):
+                        shift['cycles_covered'].add(cycle)
+                else:  # normal cycle
+                    shift_end = end_time if end_time > start_time else end_time + 24
+                    cycle_end_adj = cycle_end if cycle_end > cycle_start else cycle_end + 24
+                    
+                    # Check for overlap
+                    if not (shift_end <= cycle_start or start_time >= cycle_end_adj):
+                        shift['cycles_covered'].add(cycle)
+                            
+            if shift['cycles_covered']:  # Only add shifts that cover at least one cycle
+                possible_shifts.append(shift)
+    
+    # Estimate minimum number of staff needed
+    total_staff_hours = 0
+    for _, row in data.iterrows():
+        for cycle in cycle_cols:
+            total_staff_hours += row[f'{cycle}_staff'] * HOURS_PER_CYCLE
+    
+    min_staff_estimate = np.ceil(total_staff_hours / MAX_HOURS_PER_STAFF)
+    
+    # Get number of days in the dataset
+    num_days = len(data)
+    
+    # Add some buffer for constraints like rest days and shift changes
+    estimated_staff = max(min_staff_estimate, max_staff_needed + 1)
+    
+    def optimize_schedule(num_staff):
+        # Create a binary linear programming model
+        model = pl.LpProblem("Staff_Scheduling", pl.LpMinimize)
+        
+        # Decision variables
+        # x[s,d,shift] = 1 if staff s works shift on day d
+        x = pl.LpVariable.dicts("shift", 
+                               [(s, d, shift['id']) for s in range(1, num_staff+1) 
+                                                    for d in range(1, num_days+1) 
+                                                    for shift in possible_shifts],
+                               cat='Binary')
+        
+        # Objective: Minimize total staff hours while ensuring coverage
+        model += pl.lpSum(x[(s, d, shift['id'])] * shift['duration'] 
+                         for s in range(1, num_staff+1) 
+                         for d in range(1, num_days+1) 
+                         for shift in possible_shifts)
+        
+        # Constraint: Each staff works at most one shift per day
+        for s in range(1, num_staff+1):
+            for d in range(1, num_days+1):
+                model += pl.lpSum(x[(s, d, shift['id'])] for shift in possible_shifts) <= 1
+        
+        # Constraint: Each staff has at least one rest day per week
+        for s in range(1, num_staff+1):
+            for w in range((num_days + 6) // 7):  # Number of weeks
+                week_start = w*7 + 1
+                week_end = min(week_start + 6, num_days)
+                model += pl.lpSum(x[(s, d, shift['id'])] 
+                                 for d in range(week_start, week_end+1) 
+                                 for shift in possible_shifts) <= (week_end - week_start + 1) - REST_DAYS_PER_WEEK
+        
+        # Constraint: Each staff works at most MAX_HOURS_PER_STAFF in the period
+        for s in range(1, num_staff+1):
+            model += pl.lpSum(x[(s, d, shift['id'])] * shift['duration'] 
+                             for d in range(1, num_days+1) 
+                             for shift in possible_shifts) <= MAX_HOURS_PER_STAFF
+        
+        # Constraint: Each cycle has enough staff each day
+        for d in range(1, num_days+1):
+            day_index = d - 1  # 0-indexed for DataFrame
+            
+            for cycle in cycle_cols:
+                staff_needed = data.iloc[day_index][f'{cycle}_staff']
+                
+                # Get all shifts that cover this cycle
+                covering_shifts = [shift for shift in possible_shifts if cycle in shift['cycles_covered']]
+                
+                model += pl.lpSum(x[(s, d, shift['id'])] 
+                                 for s in range(1, num_staff+1) 
+                                 for shift in covering_shifts) >= staff_needed
+        
+        # Solve model with a time limit
+        model.solve(pl.PULP_CBC_CMD(timeLimit=300, msg=0))
+        
+        # Check if a feasible solution was found
+        if model.status == pl.LpStatusOptimal or model.status == pl.LpStatusNotSolved:
+            # Extract the solution
+            schedule = []
+            for s in range(1, num_staff+1):
+                for d in range(1, num_days+1):
+                    for shift in possible_shifts:
+                        if pl.value(x[(s, d, shift['id'])]) == 1:
+                            # Find the shift details
+                            shift_details = next((sh for sh in possible_shifts if sh['id'] == shift['id']), None)
+                            
+                            schedule.append({
+                                'staff_id': s,
+                                'day': d,
+                                'shift_id': shift['id'],
+                                'start': shift_details['start'],
+                                'end': shift_details['end'],
+                                'duration': shift_details['duration'],
+                                'cycles_covered': list(shift_details['cycles_covered'])
+                            })
+            
+            return schedule, model.objective.value()
+        else:
+            return None, None
+    
+    # Try to solve with estimated number of staff
+    staff_count = int(estimated_staff)
+    results = f"Trying with {staff_count} staff...\n"
+    schedule, objective = optimize_schedule(staff_count)
+    
+    # If no solution found, increment staff count until a solution is found
+    while schedule is None and staff_count < 15:  # Cap at 15 to avoid infinite loop
+        staff_count += 1
+        results += f"Trying with {staff_count} staff...\n"
+        schedule, objective = optimize_schedule(staff_count)
+    
+    if schedule is None:
+        results += "Failed to find a feasible solution. Try relaxing some constraints."
+        return results, None, None, None
+    
+    results += f"Optimal solution found with {staff_count} staff\n"
+    results += f"Total staff hours: {objective}\n"
+    
+    # Convert to DataFrame for analysis
+    schedule_df = pd.DataFrame(schedule)
+    
+    # Analyze staff workload
+    staff_hours = {}
+    for s in range(1, staff_count+1):
+        staff_shifts = schedule_df[schedule_df['staff_id'] == s]
+        total_hours = staff_shifts['duration'].sum()
+        staff_hours[s] = total_hours
+    
+    results += "\nStaff Hours:\n"
+    for staff_id, hours in staff_hours.items():
+        utilization = (hours / MAX_HOURS_PER_STAFF) * 100
+        results += f"Staff {staff_id}: {hours} hours ({utilization:.1f}% utilization)\n"
+    
+    avg_utilization = sum(staff_hours.values()) / (staff_count * MAX_HOURS_PER_STAFF) * 100
+    results += f"\nAverage staff utilization: {avg_utilization:.1f}%\n"
+    
+    # Check coverage for each day and cycle
+    coverage_check = []
+    for d in range(1, num_days+1):
+        day_index = d - 1  # 0-indexed for DataFrame
+        
+        day_schedule = schedule_df[schedule_df['day'] == d]
+        
+        for cycle in cycle_cols:
+            required = data.iloc[day_index][f'{cycle}_staff']
+            
+            # Count staff covering this cycle
+            assigned = sum(1 for _, shift in day_schedule.iterrows() 
+                          if cycle in shift['cycles_covered'])
+            
+            coverage_check.append({
+                'day': d,
+                'cycle': cycle,
+                'required': required,
+                'assigned': assigned,
+                'satisfied': assigned >= required
+            })
+    
+    coverage_df = pd.DataFrame(coverage_check)
+    satisfaction = coverage_df['satisfied'].mean() * 100
+    results += f"Coverage satisfaction: {satisfaction:.1f}%\n"
+    
+    if satisfaction < 100:
+        results += "Warning: Not all staffing requirements are met!\n"
+        unsatisfied = coverage_df[~coverage_df['satisfied']]
+        results += unsatisfied.to_string() + "\n"
+    
+    # Generate detailed schedule report
+    detailed_schedule = "Detailed Schedule:\n"
+    for d in range(1, num_days+1):
+        day_schedule = schedule_df[schedule_df['day'] == d]
+        day_schedule = day_schedule.sort_values(['start'])
+        
+        detailed_schedule += f"\nDay {d}:\n"
+        for _, shift in day_schedule.iterrows():
+            start_hour = shift['start']
+            end_hour = shift['end']
+
+            start_str = am_pm(start_hour)
+            end_str = am_pm(end_hour)
+
+            cycles = ", ".join(shift['cycles_covered'])
+            detailed_schedule += f"  Staff {shift['staff_id']}: {start_str}-{end_str} ({shift['duration']} hrs), Cycles: {cycles}\n"
+    
+    # Generate schedule visualization
+    fig, ax = plt.subplots(figsize=(15, 8))
+    
+    # Prepare schedule for plotting
+    staff_days = {}
+    for s in range(1, staff_count+1):
+        staff_days[s] = [0] * num_days  # 0 means off duty
+    
+    for _, shift in schedule_df.iterrows():
+        staff_id = shift['staff_id']
+        day = shift['day'] - 1  # 0-indexed
+        staff_days[staff_id][day] = shift['duration']
+    
+    # Plot the schedule
+    for s, hours in staff_days.items():
+        ax.bar(range(1, num_days+1), hours, label=f'Staff {s}')
+    
+    ax.set_xlabel('Day')
+    ax.set_ylabel('Shift Hours')
+    ax.set_title('Staff Schedule')
+    ax.set_xticks(range(1, num_days+1))
+    ax.legend()
+    
+    # Save the figure to a temporary file
+    plot_path = None
+    with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as f:
+        plt.savefig(f.name)
+        plt.close(fig)
+        plot_path = f.name
+    
+    # Create a Gantt chart
+    gantt_fig, gantt_ax = plt.subplots(figsize=(30, 12))  # Increased figure width
+
+    # Set up colors for each staff
+    colors = plt.cm.tab20.colors  # Use a visually distinct color palette
+
+    # Sort by staff then day
+    schedule_df['start_ampm'] = schedule_df['start'].apply(am_pm)
+    schedule_df['end_ampm'] = schedule_df['end'].apply(am_pm)
+    schedule_df = schedule_df.sort_values(['staff_id', 'day'])
+
+    # Plot Gantt chart
+    for staff_id in range(1, staff_count+1):
+        staff_shifts = schedule_df[schedule_df['staff_id'] == staff_id]
+
+        y_pos = staff_id
+        for i, shift in staff_shifts.iterrows():
+            day = shift['day']
+            start_hour = shift['start']
+            end_hour = shift['end']
+            duration = shift['duration']
+
+            start_ampm = shift['start_ampm']
+            end_ampm = shift['end_ampm']
+
+            # Handle overnight shifts
+            if end_hour < start_hour:  # Overnight shift
+                gantt_ax.broken_barh([(day-1 + start_hour/24, (24-start_hour)/24),
+                                     (day, end_hour/24)],
+                                    (y_pos-0.3, 0.6),  # Increased bar height
+                                    facecolors=colors[staff_id % len(colors)])
+            else:
+                gantt_ax.broken_barh([(day-1 + start_hour/24, duration/24)],
+                                    (y_pos-0.3, 0.6),  # Increased bar height
+                                    facecolors=colors[staff_id % len(colors)])
+
+            # Staggered text labels
+            text_y_offset = 0.1 if (i % 2) == 0 else -0.1  # Alternate label position
+
+            # Add text label - prioritize staff ID, add time range if space allows
+            text_label = f"Staff {staff_id}"
+            if duration > 6:  # Adjust this threshold as needed
+                text_label += f"\n{start_ampm}-{end_ampm}"
+
+            gantt_ax.text(day-1 + start_hour/24 + duration/48, y_pos + text_y_offset,
+                         text_label,
+                         horizontalalignment='center', verticalalignment='center', fontsize=7)  # Slightly smaller font
+
+    gantt_ax.set_xlabel('Day')
+    gantt_ax.set_yticks(range(1, staff_count+1))
+    gantt_ax.set_yticklabels([f'Staff {s}' for s in range(1, staff_count+1)])
+    gantt_ax.set_xlim(0, num_days)
+    gantt_ax.set_title('Staff Schedule (Full Period)')
+    gantt_ax.grid(False)  # Remove grid lines
+
+    # Save the Gantt chart
+    gantt_path = None
+    with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as f:
+        gantt_fig.savefig(f.name)
+        plt.close(gantt_fig)
+        gantt_path = f.name
+
+    # Convert schedule to CSV data
+    schedule_df['start_ampm'] = schedule_df['start'].apply(am_pm)
+    schedule_df['end_ampm'] = schedule_df['end'].apply(am_pm)
+    schedule_csv = schedule_df[['staff_id', 'day', 'start_ampm', 'end_ampm', 'duration', 'cycles_covered']].to_csv(index=False)
+
+    # Create a temporary file and write the CSV data into it
+    with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix=".csv") as temp_file:
+        temp_file.write(schedule_csv)
+        schedule_csv_path = temp_file.name
+
+    return results, plot_path, schedule_csv_path, gantt_path
+
+def convert_to_24h(time_str):
+    """Converts AM/PM time string to 24-hour format."""
+    try:
+        time_obj = datetime.strptime(time_str, "%I:00 %p")
+        return time_obj.hour
+    except ValueError:
+        return None
+
+def gradio_wrapper(
+    csv_file, beds_per_staff, max_hours_per_staff, hours_per_cycle,
+    rest_days_per_week, clinic_start_ampm, clinic_end_ampm, overlap_time, max_start_time_change
+):
+    clinic_start = convert_to_24h(clinic_start_ampm)
+    clinic_end = convert_to_24h(clinic_end_ampm)
+
+    results, plot_img, schedule_csv_path, gantt_path = optimize_staffing(
+        csv_file, beds_per_staff, max_hours_per_staff, hours_per_cycle,
+        rest_days_per_week, clinic_start, clinic_end, overlap_time, max_start_time_change
+    )
+
+    # Load plot images if they exist
+    plot_img = plot_img if plot_img and os.path.exists(plot_img) else None
+    gantt_img = gantt_path if gantt_path and os.path.exists(gantt_path) else None
+
+    return results, plot_img, schedule_csv_path, gantt_img
+
+
+# Define Gradio UI
+am_pm_times = [f"{i:02d}:00 AM" for i in range(1, 13)] + [f"{i:02d}:00 PM" for i in range(1, 13)]
+
+iface = gr.Interface(
+    fn=gradio_wrapper,
+    inputs=[
+        gr.File(label="Upload CSV"),
+        gr.Number(label="Beds per Staff", value=3),
+        gr.Number(label="Max Hours per Staff", value=40),
+        gr.Number(label="Hours per Cycle", value=4),
+        gr.Number(label="Rest Days per Week", value=2),
+        gr.Dropdown(label="Clinic Start Hour (AM/PM)", choices=am_pm_times, value="08:00 AM"),
+        gr.Dropdown(label="Clinic End Hour (AM/PM)", choices=am_pm_times, value="08:00 PM"),
+        gr.Number(label="Overlap Time", value=0),
+        gr.Number(label="Max Start Time Change", value=2)
+    ],
+    outputs=[
+        gr.Textbox(label="Optimization Results"),
+        gr.Image(label="Schedule Visualization"),
+        gr.File(label="Schedule CSV"),
+        gr.Image(label="Gantt Chart"),
+    ],
+    title="Staff Scheduling Optimizer",
+    description="Upload a CSV file with cycle data and configure parameters to generate an optimal staff schedule."
+)
+
+# Launch the Gradio app
+iface.launch(share=True)