""" Result formatter for CP-SAT solver output. Converts solver results to JSON format for PHP consumption. Validates all 12 hard constraints post-solve. """ from typing import Dict, List, Any, Optional, Set, Tuple from datetime import datetime, timedelta from ortools.sat.python import cp_model from constraint_builder import ( availability_covers_shift, shift_local_interval, shifts_overlap, ) def parse_time(time_str: str) -> int: """Convert HH:MM or HH:MM:SS time string to minutes since midnight.""" if not time_str: return 0 parts = time_str.split(':') h, m = int(parts[0]), int(parts[1]) return h * 60 + m def is_overnight_shift(start_time: str, end_time: str) -> bool: """Check if shift crosses midnight.""" return parse_time(end_time) < parse_time(start_time) def employee_meets_keyholder_requirement( employee: Dict[str, Any], shift: Dict[str, Any] ) -> bool: """Return whether an employee satisfies a shift's keyholder requirement.""" needs_keyholder = ( bool(shift.get("requiresKeyholder", False)) or bool(shift.get("isOpener", False)) or bool(shift.get("isCloser", False)) or shift.get("minRoleId", 5) <= 3 ) is_keyholder = ( bool(employee.get("isKeyholder", False)) or employee.get("role", 5) <= 3 ) return not needs_keyholder or is_keyholder def validate_constraints( problem_data: Dict[str, Any], assignments: Dict[int, int] ) -> Dict[str, int]: """ Post-solve validation: check all 12 hard constraints against extracted assignments. Args: problem_data: Original problem input assignments: Dict mapping shift_id -> employee_id Returns: Dict with constraint violation counts """ violations = { "c1_multiple_per_shift": 0, "c2_locked_changed": 0, "c3_role_unqualified": 0, "c4_opening_closing": 0, "c5_availability": 0, "c6_time_off": 0, "c7_hours_max": 0, "c8_multiple_per_day": 0, "c9_overlap": 0, "c10_max_shifts_week": 0, "c11_hours_max_hard": 0 } employees = problem_data["employees"] shifts = problem_data["shifts"] locked_shifts = problem_data.get("lockedShifts", []) # Build lookups employees_by_id = {emp["userId"]: emp for emp in employees} shifts_by_id = {shift["shiftId"]: shift for shift in shifts} locked_map = {ls["shiftId"]: ls["employeeId"] for ls in locked_shifts} # C1: Each shift assigned at most once shift_counts = {} for shift_id, emp_id in assignments.items(): shift_counts[shift_id] = shift_counts.get(shift_id, 0) + 1 for shift_id, count in shift_counts.items(): if count > 1: violations["c1_multiple_per_shift"] += 1 # C2: Locked shifts immutable for shift_id, expected_emp_id in locked_map.items(): if shift_id in assignments and assignments[shift_id] != expected_emp_id: violations["c2_locked_changed"] += 1 # Track hours per employee for C7/C11 hours_by_employee = {emp["userId"]: emp["currentPeriodHours"] for emp in employees} shifts_by_employee = {emp["userId"]: 0 for emp in employees} for shift_id, emp_id in assignments.items(): shift = shifts_by_id.get(shift_id) emp = employees_by_id.get(emp_id) if not shift or not emp: continue hours_by_employee[emp_id] += shift["durationHours"] shifts_by_employee[emp_id] += 1 # C3/C4: keyholder and opening/closing requirements. Position # eligibility is enforced independently by C12. if not employee_meets_keyholder_requirement(emp, shift): if shift.get("requiresKeyholder", False) or shift.get("minRoleId", 5) <= 3: violations["c3_role_unqualified"] += 1 if shift.get("isOpener", False) or shift.get("isCloser", False): violations["c4_opening_closing"] += 1 # C5: Availability avail_list = emp.get("availability", []) if not availability_covers_shift(shift, avail_list): violations["c5_availability"] += 1 # C6: Time-off time_off_dates = {to["date"] for to in emp.get("timeOff", [])} shift_start, shift_end = shift_local_interval(shift) touched_dates = set() day_start = datetime.combine( shift_start.date(), datetime.min.time(), tzinfo=shift_start.tzinfo, ) while day_start < shift_end: touched_dates.add(day_start.strftime("%Y-%m-%d")) day_start += timedelta(days=1) if touched_dates & time_off_dates: violations["c6_time_off"] += 1 # C7 & C11: hoursMax not exceeded for emp_id, total_hours in hours_by_employee.items(): emp = employees_by_id[emp_id] if total_hours > emp["hoursMax"] + 0.01: # Small tolerance for floating point violations["c7_hours_max"] += 1 violations["c11_hours_max_hard"] += 1 # C8: One shift per employee per day shifts_by_emp_date: Dict[Tuple[int, str], int] = {} for shift_id, emp_id in assignments.items(): shift = shifts_by_id.get(shift_id) if shift: key = (emp_id, shift["date"]) shifts_by_emp_date[key] = shifts_by_emp_date.get(key, 0) + 1 for (emp_id, date), count in shifts_by_emp_date.items(): if count > 1: violations["c8_multiple_per_day"] += 1 # C9: No overlapping shifts emp_shift_list: Dict[int, List[Dict]] = {emp["userId"]: [] for emp in employees} for shift_id, emp_id in assignments.items(): shift = shifts_by_id.get(shift_id) if shift: emp_shift_list[emp_id].append(shift) for emp_id, emp_shifts in emp_shift_list.items(): for i, s1 in enumerate(emp_shifts): for s2 in emp_shifts[i+1:]: if shifts_overlap(s1, s2): violations["c9_overlap"] += 1 # C10: Max 5 shifts per week for emp_id, shifts_count in shifts_by_employee.items(): emp = employees_by_id[emp_id] total_shifts = shifts_count + emp.get("shiftsThisWeek", 0) if total_shifts > 5: violations["c10_max_shifts_week"] += 1 # C12: Position eligibility violations["c12_position_eligibility"] = 0 for shift_id, emp_id in assignments.items(): shift = shifts_by_id.get(shift_id) emp = employees_by_id.get(emp_id) if shift and emp: shift_pos_id = shift.get("positionId") allowed = set(emp.get("allowedPositionIds", [])) if shift_pos_id is not None and shift_pos_id not in allowed: violations["c12_position_eligibility"] += 1 return violations def analyze_unfilled_shift( shift: Dict[str, Any], employees: List[Dict[str, Any]], hours_by_employee: Dict[int, float], locked_dates_by_emp: Dict[int, Set[str]], filled_shift_ids: Set[int], assignments: Dict[int, int], all_shifts: List[Dict[str, Any]] ) -> List[str]: """ Analyze why a shift couldn't be filled by checking each employee against the constraints. Returns: List of human-readable reasons explaining why no employee was assigned. """ shift_id = shift["shiftId"] dow = shift["dayOfWeek"] shift_date = shift["date"] duration = shift.get("durationHours", 0) reasons = [] all_blocked = True for emp in employees: emp_id = emp["userId"] emp_name = emp.get("name", f"Employee #{emp_id}") blocker = None # Check keyholder qualification. Position eligibility is independent. if not employee_meets_keyholder_requirement(emp, shift): blocker = f"{emp_name}: is not qualified as a keyholder" # Check position eligibility (C12) if blocker is None: shift_pos_id = shift.get("positionId") allowed_pos_ids = set(emp.get("allowedPositionIds", [])) if shift_pos_id is not None and shift_pos_id not in allowed_pos_ids: pos_name = shift.get("positionName", f"position #{shift_pos_id}") blocker = f"{emp_name}: not qualified for {pos_name} position" # Check availability if blocker is None: avail_list = emp.get("availability", []) if avail_list and not availability_covers_shift(shift, avail_list): day_names = {1: "Mon", 2: "Tue", 3: "Wed", 4: "Thu", 5: "Fri", 6: "Sat", 7: "Sun"} blocker = f"{emp_name}: availability does not continuously cover shift starting {day_names.get(dow, f'day {dow}')}" # Check time-off if blocker is None: time_off_dates = {to["date"] for to in emp.get("timeOff", [])} if shift_date in time_off_dates: blocker = f"{emp_name}: has approved time-off on {shift_date}" # Check hours max if blocker is None: total_after = hours_by_employee.get(emp_id, 0) + duration if total_after > emp["hoursMax"]: blocker = f"{emp_name}: would exceed max hours ({hours_by_employee.get(emp_id, 0):.1f}h + {duration:.1f}h > {emp['hoursMax']:.1f}h)" # Check locked shift on same day if blocker is None: emp_locked_dates = locked_dates_by_emp.get(emp_id, set()) if shift_date in emp_locked_dates: blocker = f"{emp_name}: already has an assigned shift on {shift_date}" # Check if already assigned to another open shift on same day if blocker is None: assigned_dates = set() for a_shift_id, a_emp_id in assignments.items(): if a_emp_id == emp_id: a_shift = next((s for s in all_shifts if s["shiftId"] == a_shift_id), None) if a_shift: assigned_dates.add(a_shift["date"]) if shift_date in assigned_dates: blocker = f"{emp_name}: already assigned to another shift on {shift_date}" if blocker: reasons.append(blocker) else: all_blocked = False if not reasons: reasons.append("No eligible employees available") elif all_blocked: reasons.insert(0, f"All {len(employees)} employees blocked from this shift") return reasons def format_result( problem_data: Dict[str, Any], solver: cp_model.CpSolver, model: cp_model.CpModel, assignments: Dict[int, int], solver_duration_ms: int, solver_status: int, infeasible_subset: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """ Format solver results as JSON output. Args: problem_data: Original problem input solver: CP-SAT solver instance model: CP-SAT model instance assignments: Dict mapping shift_id -> employee_id solver_duration_ms: Solver runtime in milliseconds solver_status: Status code returned by solver.Solve() infeasible_subset: Optional dict with IIS analysis (method, conflictingShiftIds, explanation) Returns: Dict containing status, assignments, metrics, and reports """ status_map = { cp_model.OPTIMAL: "OPTIMAL", cp_model.FEASIBLE: "FEASIBLE", cp_model.INFEASIBLE: "INFEASIBLE", cp_model.MODEL_INVALID: "MODEL_INVALID", cp_model.UNKNOWN: "UNKNOWN" } status = status_map.get(solver_status, "UNKNOWN") locked_shift_ids = { shift["shiftId"] for shift in problem_data.get("lockedShifts", []) } open_shifts = [ shift for shift in problem_data["shifts"] if shift["shiftId"] not in locked_shift_ids and not shift.get("isLocked", False) ] # Objective value is a dimensionless quality score - report as-is objective_value = solver.objective_value if status in ["OPTIMAL", "FEASIBLE"] else 0.0 # Calculate optimality gap optimality_gap = 0.0 if status == "FEASIBLE" and solver.best_objective_bound != solver.objective_value: optimality_gap = abs(solver.objective_value - solver.best_objective_bound) / abs(solver.objective_value) result = { "status": status, "objectiveValue": int(objective_value), "optimalityGap": round(optimality_gap, 4), "solverDurationMs": solver_duration_ms, "assignments": [], "unfilledShifts": [], "constraintReport": { "bindingConstraints": [], "slackConstraints": [], "infeasibleSubset": infeasible_subset }, "scorecard": { "totalLaborCost": 0.0, "fairnessCV": 0.0, "coverageFilled": 0, "coverageTotal": len(open_shifts), "constraintViolations": 0, "totalOvertimeHours": 0.0, "overtimeByEmployee": {}, "minimumsMet": 0, "minimumsTotal": 0, "shortfallByEmployee": {} } } if status not in ["OPTIMAL", "FEASIBLE"]: # Mark all shifts as unfilled for infeasible solutions explanation = "Model is infeasible" if infeasible_subset: explanation = infeasible_subset.get("explanation", explanation) for shift in open_shifts: result["unfilledShifts"].append({ "shiftId": shift["shiftId"], "reason": "infeasible", "conflictingConstraints": [explanation] }) return result # Build employee lookup employees_by_id = {emp["userId"]: emp for emp in problem_data["employees"]} # Track hours per employee hours_by_employee: Dict[int, float] = {emp["userId"]: emp["currentPeriodHours"] for emp in problem_data["employees"]} # Process assignments filled_shift_ids = set() for shift_id, employee_id in assignments.items(): shift = next((s for s in problem_data["shifts"] if s["shiftId"] == shift_id), None) if not shift: continue employee = employees_by_id.get(employee_id) if not employee: continue filled_shift_ids.add(shift_id) hours_by_employee[employee_id] += shift["durationHours"] # Calculate labor cost labor_cost = shift["durationHours"] * employee["hourlyRate"] # Calculate hours deviation hours_after = hours_by_employee[employee_id] hours_deviation = hours_after - employee["hoursRequested"] # Check overtime risk overtime_risk = hours_after > 40.0 or shift["durationHours"] > 8.0 # Count alternatives (employees who could work this shift) alternatives_count = sum( 1 for emp in problem_data["employees"] if emp["userId"] != employee_id and employee_meets_keyholder_requirement(emp, shift) and hours_by_employee[emp["userId"]] + shift["durationHours"] <= emp["hoursMax"] ) result["assignments"].append({ "shiftId": shift_id, "employeeId": employee_id, "factors": { "roleQualified": employee_meets_keyholder_requirement(employee, shift), "availabilityMatch": True, # If assigned, it must match availability "hoursAfter": round(hours_after, 2), "hoursMax": employee["hoursMax"], "hoursDeviation": round(hours_deviation, 2), "overtimeRisk": overtime_risk, "alternativesCount": alternatives_count, "laborCost": round(labor_cost, 2), "priorityScores": {} # Would be populated during solving } }) result["scorecard"]["totalLaborCost"] += labor_cost # Find unfilled shifts and analyze WHY each is unfilled locked_shifts = problem_data.get("lockedShifts", []) locked_dates_by_emp = {} for ls in locked_shifts: ls_emp_id = ls.get("employeeId") ls_date = ls.get("date") if ls_emp_id is not None and ls_date is not None: if ls_emp_id not in locked_dates_by_emp: locked_dates_by_emp[ls_emp_id] = set() locked_dates_by_emp[ls_emp_id].add(ls_date) for shift in open_shifts: if shift["shiftId"] not in filled_shift_ids: # Analyze why this shift couldn't be filled blocking_reasons = analyze_unfilled_shift( shift, problem_data["employees"], hours_by_employee, locked_dates_by_emp, filled_shift_ids, assignments, problem_data["shifts"] ) reason = blocking_reasons[0] if blocking_reasons else "no_qualified_employee" result["unfilledShifts"].append({ "shiftId": shift["shiftId"], "reason": reason, "conflictingConstraints": blocking_reasons }) # Calculate fairness CV (coefficient of variation) if problem_data["employees"]: hours_list = [hours_by_employee[emp["userId"]] for emp in problem_data["employees"]] mean_hours = sum(hours_list) / len(hours_list) if mean_hours > 0: variance = sum((h - mean_hours) ** 2 for h in hours_list) / len(hours_list) std_dev = variance ** 0.5 result["scorecard"]["fairnessCV"] = round(std_dev / mean_hours, 4) # Calculate overtime for employee_id, total_hours in hours_by_employee.items(): if total_hours > 40.0: overtime = total_hours - 40.0 result["scorecard"]["totalOvertimeHours"] += overtime result["scorecard"]["overtimeByEmployee"][str(employee_id)] = round(overtime, 2) # Calculate minimum-hours metrics: how many employees with a hoursMin target # reached their minimum, and by how much the others fell short. Skip # employees with hoursRequested = 0 — they've opted out of being scheduled. for emp in problem_data["employees"]: hours_min = emp.get("hoursMin") or 0.0 hours_requested = emp.get("hoursRequested") or 0.0 if hours_min <= 0 or hours_requested <= 0: continue result["scorecard"]["minimumsTotal"] += 1 total_hours = hours_by_employee.get(emp["userId"], 0.0) if total_hours + 1e-6 >= hours_min: result["scorecard"]["minimumsMet"] += 1 else: result["scorecard"]["shortfallByEmployee"][str(emp["userId"])] = round( hours_min - total_hours, 2 ) result["scorecard"]["totalLaborCost"] = round(result["scorecard"]["totalLaborCost"], 2) result["scorecard"]["totalOvertimeHours"] = round(result["scorecard"]["totalOvertimeHours"], 2) result["scorecard"]["coverageFilled"] = len(filled_shift_ids) # Post-solve constraint validation violations = validate_constraints(problem_data, assignments) total_violations = sum(violations.values()) result["scorecard"]["constraintViolations"] = total_violations # Build constraint report for employee in problem_data["employees"]: emp_id = employee["userId"] total_hours = hours_by_employee[emp_id] hours_max = employee["hoursMax"] if total_hours >= hours_max: result["constraintReport"]["bindingConstraints"].append({ "type": "hoursMax", "employeeId": emp_id, "value": round(total_hours, 2), "limit": hours_max }) elif total_hours > 0: result["constraintReport"]["slackConstraints"].append({ "type": "hoursMax", "employeeId": emp_id, "value": round(total_hours, 2), "limit": hours_max, "slack": round(hours_max - total_hours, 2) }) return result