""" Objective function builder for CP-SAT scheduling model. Implements weighted multi-objective optimization with hard precedence for position coverage. """ from typing import Dict, List, Any, Tuple from ortools.sat.python import cp_model def build_objective( model: cp_model.CpModel, problem_data: Dict[str, Any], x: Dict[Tuple[int, int], cp_model.IntVar] ) -> None: """ Build weighted objective function maximizing quality of schedule. CRITICAL: Position coverage has HARD-PRECEDENCE at weight 10^7. All other priorities use exponential weighting: weight = 10^(6-rank). Args: model: CP-SAT model instance problem_data: Problem input data x: Decision variables x[employee_id, shift_id] """ employees = problem_data["employees"] shifts = problem_data["shifts"] config = problem_data.get("config", {}) priorities = config.get("priorities", []) # Build priority weights map priority_weights = {} for priority in priorities: name = priority["name"] rank = priority["rank"] # Exponential weighting: 10^(6-rank) weight = 10 ** (6 - rank) priority_weights[name] = weight # HARD-PRECEDENCE: Position coverage always at 10^7 (higher than any user priority) COVERAGE_WEIGHT = 10_000_000 objective_terms = [] # 1. POSITION COVERAGE (HARD-PRECEDENCE) # Maximize filled shifts - each filled shift adds COVERAGE_WEIGHT filled_vars = {} for shift in shifts: shift_id = shift["shiftId"] if shift.get("isLocked", False): continue # Locked shifts already filled shift_vars = [x.get((emp["userId"], shift_id)) for emp in employees] shift_vars = [v for v in shift_vars if v is not None] if shift_vars: # Create filled[s] BoolVar - true if ANY employee assigned to this shift filled = model.NewBoolVar(f"filled_{shift_id}") model.AddMaxEquality(filled, shift_vars) filled_vars[shift_id] = filled # Add COVERAGE_WEIGHT once per filled shift (not per eligible employee) objective_terms.append(filled * COVERAGE_WEIGHT) # 2. LABOR COST # Minimize total labor cost - prefer lower-cost assignments labor_cost_weight = priority_weights.get("labor_cost", 10000) if labor_cost_weight > 0: for emp in employees: emp_id = emp["userId"] hourly_rate = emp["hourlyRate"] for shift in shifts: shift_id = shift["shiftId"] if (emp_id, shift_id) not in x: continue # Actual cost in cents for this assignment cost_cents = int(shift["durationHours"] * hourly_rate * 100) # Penalize higher costs (negative because we maximize) # Scale down by 100 to keep numbers manageable penalty = cost_cents * labor_cost_weight // 100 objective_terms.append(x[(emp_id, shift_id)] * (-penalty)) # 3. HOURS FAIRNESS # Minimize absolute deviation from requested hours fairness_weight = priority_weights.get("hours_fairness", 1000) if fairness_weight > 0: # Find max hours to bound the deviation variable max_hours_centi = int(max(emp["hoursMax"] for emp in employees) * 100) if employees else 10000 for emp in employees: emp_id = emp["userId"] hours_requested_centi = int(emp["hoursRequested"] * 100) current_hours_centi = int(emp["currentPeriodHours"] * 100) # Sum of assigned hours assigned_hours = [] for shift in shifts: shift_id = shift["shiftId"] if (emp_id, shift_id) in x: duration_centi = int(shift["durationHours"] * 100) assigned_hours.append(x[(emp_id, shift_id)] * duration_centi) if assigned_hours: total_assigned = sum(assigned_hours) total_hours = current_hours_centi + total_assigned # Absolute value linearization for deviation from requested hours # abs_dev = |total_hours - hours_requested_centi| abs_dev = model.NewIntVar(0, max_hours_centi, f"abs_dev_{emp_id}") model.Add(abs_dev >= total_hours - hours_requested_centi) model.Add(abs_dev >= hours_requested_centi - total_hours) # Minimize absolute deviation (negative weight because CP-SAT maximizes) objective_terms.append(abs_dev * (-fairness_weight)) # 3b. MEET MINIMUMS (opt-in separate priority) # Penalize shortfall below each employee's hoursMin. Only active when the # user ranks "meet_minimums" in their priority list; otherwise behavior # matches the original (fairness alone, no forced pull-up to min). meet_min_weight = priority_weights.get("meet_minimums", 0) if meet_min_weight > 0: for emp in employees: emp_id = emp["userId"] hours_min_centi = int(emp.get("hoursMin", 0) * 100) hours_requested_centi = int(emp.get("hoursRequested", 0) * 100) # Skip employees who've opted out (hoursRequested == 0) or who have # no minimum configured — don't force hours on people who don't # want to work. if hours_min_centi <= 0 or hours_requested_centi <= 0: continue current_hours_centi = int(emp["currentPeriodHours"] * 100) assigned_hours = [] for shift in shifts: shift_id = shift["shiftId"] if (emp_id, shift_id) in x: duration_centi = int(shift["durationHours"] * 100) assigned_hours.append(x[(emp_id, shift_id)] * duration_centi) if not assigned_hours: continue total_hours = current_hours_centi + sum(assigned_hours) # min_shortfall = max(0, hoursMin - total_hours) min_shortfall = model.NewIntVar( 0, max(hours_min_centi, 1), f"min_shortfall_{emp_id}" ) model.Add(min_shortfall >= hours_min_centi - total_hours) objective_terms.append(min_shortfall * (-meet_min_weight)) # 4. SENIORITY # Prefer employees with longer tenure (based on hire date) seniority_weight = priority_weights.get("seniority", 100) if seniority_weight > 0: from datetime import datetime now = datetime.now() for emp in employees: emp_id = emp["userId"] # Calculate tenure in months from hire date; fallback to role-based if no date hire_date_str = emp.get("hireDate") or emp.get("assignedAt") if hire_date_str: try: hire_date = datetime.fromisoformat(str(hire_date_str).replace("Z", "+00:00")) tenure_months = max(1, (now.year - hire_date.year) * 12 + (now.month - hire_date.month)) except (ValueError, TypeError): tenure_months = 1 else: # Legacy fallback: use role as proxy (lower role = higher seniority) role = emp.get("role", 5) tenure_months = (6 - role) * 12 # Cap tenure bonus to avoid extreme weighting for very long-tenured employees seniority_bonus = min(tenure_months, 120) * seniority_weight // 12 shift_vars = [x.get((emp_id, s["shiftId"])) for s in shifts] shift_vars = [v for v in shift_vars if v is not None] for var in shift_vars: objective_terms.append(var * seniority_bonus) # 5. MINIMIZE OVERTIME # Penalize hours above 40h/week and 8h/day. # Exempt (salaried) employees are skipped — their hours don't accrue OT # in payroll, so penalizing them here artificially blocks valid # assignments (e.g. a 45h manager schedule). exempt_ids = {emp["userId"] for emp in employees if emp.get("isExempt")} overtime_weight = priority_weights.get("minimize_overtime", 10) if overtime_weight > 0: # Weekly overtime penalty for emp in employees: emp_id = emp["userId"] if emp_id in exempt_ids: continue current_hours_centi = int(emp["currentPeriodHours"] * 100) assigned_hours = [] for shift in shifts: shift_id = shift["shiftId"] if (emp_id, shift_id) in x: duration_centi = int(shift["durationHours"] * 100) assigned_hours.append(x[(emp_id, shift_id)] * duration_centi) if assigned_hours: total_hours = current_hours_centi + sum(assigned_hours) weekly_threshold_centi = 40 * 100 # Penalize hours above 40 if current_hours_centi < weekly_threshold_centi: # Create variable for overtime hours overtime_var = model.NewIntVar(0, 10000, f"weekly_ot_{emp_id}") model.Add(overtime_var >= total_hours - weekly_threshold_centi) model.Add(overtime_var >= 0) # Penalize overtime objective_terms.append(overtime_var * (-overtime_weight)) # Daily overtime penalty (8h threshold per day per employee) # Group shifts by employee and date shifts_by_emp_date = {} for emp in employees: emp_id = emp["userId"] if emp_id in exempt_ids: continue for shift in shifts: date = shift["date"] key = (emp_id, date) if key not in shifts_by_emp_date: shifts_by_emp_date[key] = [] if (emp_id, shift["shiftId"]) in x: shifts_by_emp_date[key].append((shift["shiftId"], int(shift["durationHours"] * 100))) # For each employee-day combination, penalize daily hours > 8h daily_threshold_centi = 8 * 100 for (emp_id, date), shift_list in shifts_by_emp_date.items(): if not shift_list: continue # Sum of hours on this day daily_hours_terms = [x[(emp_id, sid)] * duration for sid, duration in shift_list] daily_total = sum(daily_hours_terms) # Create daily OT variable: max(0, daily_total - 800) daily_ot = model.NewIntVar(0, 10000, f"daily_ot_{emp_id}_{date}") model.Add(daily_ot >= daily_total - daily_threshold_centi) model.Add(daily_ot >= 0) # Penalize daily overtime objective_terms.append(daily_ot * (-overtime_weight)) # 6. EMPLOYEE PREFERENCES # Honor requested hours preferences (already handled by fairness, but add extra bonus) pref_weight = priority_weights.get("employee_preferences", 1) if pref_weight > 0: for emp in employees: emp_id = emp["userId"] hours_requested = emp["hoursRequested"] # Give small bonus for assignments that help reach requested hours shift_vars = [x.get((emp_id, s["shiftId"])) for s in shifts] shift_vars = [v for v in shift_vars if v is not None] if hours_requested > 0: for var in shift_vars: objective_terms.append(var * pref_weight) # Set objective to maximize if objective_terms: model.Maximize(sum(objective_terms)) else: # No objective terms - just find any feasible solution model.Maximize(0)