""" Infeasibility analyzer using CP-SAT assumptions. Detects Irreducible Infeasible Subset (IIS) when model is infeasible. """ from typing import Dict, List, Any, Tuple, Optional from ortools.sat.python import cp_model def heuristic_analysis(problem_data: Dict[str, Any], shifts: List[Dict], employees: List[Dict]) -> Dict[str, Any]: """ Heuristic-based infeasibility analysis (fallback when assumption-literal method times out). Args: problem_data: Problem input data shifts: List of shifts employees: List of employees Returns: Dict with method, conflictingShiftIds, and explanation """ conflicts = [] conflicting_shift_ids = [] # 1. Shifts with no qualified employees for shift in shifts: qualified_employees = [ emp for emp in employees if emp["role"] <= shift["minRoleId"] ] if not qualified_employees and not shift.get("isLocked", False): conflicts.append( f"Shift {shift['shiftId']} ({shift['positionName']}) " f"requires role <= {shift['minRoleId']} but no employees qualify" ) conflicting_shift_ids.append(shift['shiftId']) # 2. Opening/closing shifts with no Role <= 3 employees for shift in shifts: min_role_id = shift.get("minRoleId", 5) if min_role_id <= 3: qualified_leads = [emp for emp in employees if emp["role"] <= 3] if not qualified_leads and not shift.get("isLocked", False): conflicts.append( f"Shift {shift['shiftId']} requires Role <= 3 " f"but no shift leads available" ) conflicting_shift_ids.append(shift['shiftId']) # 3. Employees with insufficient hours capacity total_shift_hours = sum(s["durationHours"] for s in shifts if not s.get("isLocked", False)) total_available_capacity = sum( max(0, emp["hoursMax"] - emp["currentPeriodHours"]) for emp in employees ) if total_shift_hours > total_available_capacity: conflicts.append( f"Total shift hours ({total_shift_hours:.1f}) exceeds " f"total employee capacity ({total_available_capacity:.1f})" ) # 4. Employees at hoursMax who have locked shifts requiring them locked_shifts = problem_data.get("lockedShifts", []) for locked in locked_shifts: emp_id = locked["employeeId"] employee = next((e for e in employees if e["userId"] == emp_id), None) if employee: total_hours = employee["currentPeriodHours"] + locked.get("durationHours", 0) if total_hours > employee["hoursMax"]: conflicts.append( f"Locked shift {locked['shiftId']} for employee {emp_id} " f"would exceed hoursMax ({total_hours:.1f} > {employee['hoursMax']:.1f})" ) conflicting_shift_ids.append(locked['shiftId']) # 5. Check availability conflicts with locked shifts for locked in locked_shifts: emp_id = locked["employeeId"] employee = next((e for e in employees if e["userId"] == emp_id), None) if employee: # Check if employee has time-off on this date time_off_dates = {to["date"] for to in employee.get("timeOff", [])} if locked["date"] in time_off_dates: conflicts.append( f"Locked shift {locked['shiftId']} for employee {emp_id} " f"conflicts with time-off on {locked['date']}" ) conflicting_shift_ids.append(locked['shiftId']) if not conflicts: conflicts.append("Model is infeasible but specific conflicts could not be identified") return { "method": "heuristic", "conflictingShiftIds": list(set(conflicting_shift_ids)), "explanation": "; ".join(conflicts[:10]) } def analyze_infeasibility( model: cp_model.CpModel, problem_data: Dict[str, Any], timeout_seconds: int = 10 ) -> Dict[str, Any]: """ Analyze infeasible model to find conflicting constraints. Uses assumption literals to identify minimal infeasible subset. Falls back to heuristic analysis if assumption method times out. Args: model: Infeasible CP-SAT model problem_data: Problem input data timeout_seconds: Time limit for IIS analysis Returns: Dict with method, conflictingShiftIds, and explanation """ # This function is called AFTER the model is already known to be infeasible. # We need to rebuild with assumption literals to identify the IIS. # Import constraint builder to rebuild model with assumptions from model_builder import build_model from ortools.sat.python import cp_model as cp employees = problem_data["employees"] shifts = problem_data["shifts"] # Rebuild model with assumption literals # Strategy: make each shift's "must be filled" an assumption fresh_model = cp.CpModel() # Build decision variables x: Dict[Tuple[int, int], cp.IntVar] = {} for emp in employees: for shift in shifts: x[(emp["userId"], shift["shiftId"])] = fresh_model.NewBoolVar( f"x_e{emp['userId']}_s{shift['shiftId']}" ) # Build all constraints (including hard constraints) from constraint_builder import build_constraints build_constraints(fresh_model, problem_data, x) # Create assumption literals for shift filling assumptions = {} for shift in shifts: if shift.get("isLocked"): continue shift_id = shift["shiftId"] emp_vars = [x[(emp["userId"], shift_id)] for emp in employees if (emp["userId"], shift_id) in x] if emp_vars: assumption = fresh_model.NewBoolVar(f"assume_filled_{shift_id}") assumptions[shift_id] = assumption # Add: if assumption is true, shift must be filled fresh_model.Add(sum(emp_vars) >= 1).OnlyEnforceIf(assumption) solver = cp.CpSolver() solver.parameters.max_time_in_seconds = timeout_seconds solver.parameters.num_search_workers = 1 # OR-Tools 9.x+: assumptions are set on the model, not passed to Solve() fresh_model.AddAssumptions(list(assumptions.values())) status = solver.Solve(fresh_model) if status == cp.INFEASIBLE: # Get the sufficient set of assumptions causing infeasibility core = solver.SufficientAssumptionsForInfeasibility() core_indices = {a.Index() for a in core} conflicting_shifts = [] for shift_id, assumption in assumptions.items(): if assumption.Index() in core_indices: conflicting_shifts.append(shift_id) return { "method": "assumption_literals", "conflictingShiftIds": conflicting_shifts, "explanation": f"{len(conflicting_shifts)} shifts cannot all be filled simultaneously due to constraint conflicts" } else: # Couldn't prove infeasibility with assumptions — fall back to heuristic return heuristic_analysis(problem_data, shifts, employees)