""" 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 ortools.sat.python import cp_model from datetime import datetime, timedelta 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 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: Role qualification if emp["role"] > shift["minRoleId"]: violations["c3_role_unqualified"] += 1 # C4: Opening/closing requires role <= 3 min_role_id = shift.get("minRoleId", 5) if min_role_id <= 3 and emp["role"] > 3: violations["c4_opening_closing"] += 1 # C5: Availability dow = shift["dayOfWeek"] shift_start = parse_time(shift["startTime"]) shift_end = parse_time(shift["endTime"]) is_overnight = is_overnight_shift(shift["startTime"], shift["endTime"]) avail_list = emp.get("availability", []) if is_overnight: start_day_avail = [a for a in avail_list if a["dayOfWeek"] == dow] next_day = (dow % 7) + 1 next_day_avail = [a for a in avail_list if a["dayOfWeek"] == next_day] start_ok = any( parse_time(a["startTime"]) <= shift_start and parse_time(a["endTime"]) >= shift_start for a in start_day_avail ) end_ok = any( parse_time(a["startTime"]) <= shift_end and parse_time(a["endTime"]) >= shift_end for a in next_day_avail ) if not start_ok or not end_ok: violations["c5_availability"] += 1 else: day_avail = [a for a in avail_list if a["dayOfWeek"] == dow] has_coverage = any( parse_time(a["startTime"]) <= shift_start and parse_time(a["endTime"]) >= shift_end for a in day_avail ) if not has_coverage: violations["c5_availability"] += 1 # C6: Time-off time_off_dates = {to["date"] for to in emp.get("timeOff", [])} if shift["date"] in 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 (simplified check) 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:]: # Check overlap date1 = datetime.strptime(s1["date"], "%Y-%m-%d") date2 = datetime.strptime(s2["date"], "%Y-%m-%d") start1 = parse_time(s1["startTime"]) end1 = parse_time(s1["endTime"]) start2 = parse_time(s2["startTime"]) end2 = parse_time(s2["endTime"]) is_overnight1 = is_overnight_shift(s1["startTime"], s1["endTime"]) is_overnight2 = is_overnight_shift(s2["startTime"], s2["endTime"]) overlaps = False if s1["date"] == s2["date"]: if is_overnight1 or is_overnight2: overlaps = True else: overlaps = (start1 < end2 and start2 < end1) elif (date2 - date1).days == 1 and is_overnight1: overlaps = (end1 > start2) elif (date1 - date2).days == 1 and is_overnight2: overlaps = (end2 > start1) if overlaps: 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"] shift_start = parse_time(shift["startTime"]) shift_end = parse_time(shift["endTime"]) min_role_id = shift.get("minRoleId", 5) duration = shift.get("durationHours", 0) is_overnight = is_overnight_shift(shift["startTime"], shift["endTime"]) reasons = [] all_blocked = True for emp in employees: emp_id = emp["userId"] emp_name = emp.get("name", f"Employee #{emp_id}") emp_role = emp["role"] blocker = None # Check role qualification if emp_role > min_role_id: blocker = f"{emp_name}: role {emp_role} doesn't meet required role <= {min_role_id}" # 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: if is_overnight: start_day_avail = [a for a in avail_list if a["dayOfWeek"] == dow] next_day = (dow % 7) + 1 next_day_avail = [a for a in avail_list if a["dayOfWeek"] == next_day] start_ok = any( parse_time(a["startTime"]) <= shift_start for a in start_day_avail ) end_ok = any( parse_time(a["endTime"]) >= shift_end for a in next_day_avail ) if not start_ok or not end_ok: day_names = {1: "Mon", 2: "Tue", 3: "Wed", 4: "Thu", 5: "Fri", 6: "Sat", 7: "Sun"} blocker = f"{emp_name}: not available on {day_names.get(dow, f'day {dow}')} (overnight shift)" else: day_avail = [a for a in avail_list if a["dayOfWeek"] == dow] has_coverage = any( parse_time(a["startTime"]) <= shift_start and parse_time(a["endTime"]) >= shift_end for a in day_avail ) if not has_coverage: day_names = {1: "Mon", 2: "Tue", 3: "Wed", 4: "Thu", 5: "Fri", 6: "Sat", 7: "Sun"} if day_avail: windows = ", ".join( f"{a['startTime']}-{a['endTime']}" for a in day_avail ) blocker = f"{emp_name}: availability on {day_names.get(dow, f'day {dow}')} ({windows}) doesn't cover shift" else: blocker = f"{emp_name}: no availability on {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") # 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(problem_data["shifts"]), "constraintViolations": 0, "totalOvertimeHours": 0.0, "overtimeByEmployee": {} } } 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 problem_data["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 emp["role"] <= shift["minRoleId"] and hours_by_employee[emp["userId"]] + shift["durationHours"] <= emp["hoursMax"] ) result["assignments"].append({ "shiftId": shift_id, "employeeId": employee_id, "factors": { "roleQualified": employee["role"] <= shift["minRoleId"], "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 problem_data["shifts"]: if shift["shiftId"] not in filled_shift_ids and not shift.get("isLocked", False): # 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) 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