""" Constraint builder for CP-SAT scheduling model. Implements the 12 hard constraints from Spec 038. """ from typing import Dict, List, Any, Tuple from ortools.sat.python import cp_model from datetime import datetime, timedelta from zoneinfo import ZoneInfo 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 shift_interval(shift: Dict[str, Any]) -> Tuple[datetime, datetime]: """Return the real dated half-open interval represented by a shift. ``durationHours`` disambiguates equal wall times (a valid 24-hour shift in the PHP API) and shifts lasting more than one calendar day. When an explicit ``endDate`` is provided it is authoritative. """ start_utc_value = shift.get("startDateTimeUtc") end_utc_value = shift.get("endDateTimeUtc") if start_utc_value and end_utc_value: start = datetime.fromisoformat(str(start_utc_value).replace("Z", "+00:00")) end = datetime.fromisoformat(str(end_utc_value).replace("Z", "+00:00")) if end <= start: raise ValueError("endDateTimeUtc must be after startDateTimeUtc") return start, end start_date = datetime.strptime(shift["date"], "%Y-%m-%d") start_minutes = parse_time(shift["startTime"]) end_minutes = parse_time(shift["endTime"]) start = start_date + timedelta(minutes=start_minutes) end_date_value = shift.get("endDate") if end_date_value: end_date = datetime.strptime(str(end_date_value)[:10], "%Y-%m-%d") return start, end_date + timedelta(minutes=end_minutes) raw_wall_minutes = end_minutes - start_minutes duration_value = shift.get("durationHours") if duration_value is not None and float(duration_value) > 0: duration_minutes = int(round(float(duration_value) * 60)) day_offset = max(0, int(round((duration_minutes - raw_wall_minutes) / (24 * 60)))) else: day_offset = 1 if raw_wall_minutes <= 0 else 0 end = start_date + timedelta(days=day_offset, minutes=end_minutes) if end <= start: end += timedelta(days=1) return start, end def shift_local_interval(shift: Dict[str, Any]) -> Tuple[datetime, datetime]: """Return the shift interval converted to its configured store timezone.""" start, end = shift_interval(shift) timezone_name = shift.get("storeTimezone") if timezone_name and start.tzinfo is not None and end.tzinfo is not None: timezone = ZoneInfo(str(timezone_name)) return start.astimezone(timezone), end.astimezone(timezone) return start, end def availability_covers_shift(shift: Dict[str, Any], availability: List[Dict[str, Any]]) -> bool: """Require continuous availability for every local-day segment of a shift.""" start, end = shift_local_interval(shift) def instant_value(value: datetime) -> float: return value.timestamp() day_start = datetime.combine( start.date(), datetime.min.time(), tzinfo=start.tzinfo, ) while instant_value(day_start) < instant_value(end): next_day = day_start + timedelta(days=1) segment_start = max(start, day_start, key=instant_value) segment_end = min(end, next_day, key=instant_value) if instant_value(segment_start) < instant_value(segment_end): start_minute = int((segment_start - day_start).total_seconds() // 60) end_minute = int((segment_end - day_start).total_seconds() // 60) # Existing minute-granularity inputs use 23:59 as end-of-day. required_end = (24 * 60 - 1) if end_minute == 24 * 60 else end_minute required_duration = int( (instant_value(segment_end) - instant_value(segment_start)) // 60 ) day_of_week = day_start.isoweekday() def window_covers_segment(window: Dict[str, Any]) -> bool: window_start = parse_time(str(window.get("startTime", ""))) window_end = parse_time(str(window.get("endTime", ""))) # A 00:00-23:59 window is the minute-granularity representation # of the entire local day, including a repeated DST hour. window_duration = ( 24 * 60 if window_start == 0 and window_end == 24 * 60 - 1 else window_end - window_start ) return ( int(window.get("dayOfWeek", -1)) == day_of_week and window_start <= start_minute and window_end >= required_end and window_duration >= required_duration ) if not any(window_covers_segment(window) for window in availability): return False day_start = next_day return True def shifts_overlap(first: Dict[str, Any], second: Dict[str, Any]) -> bool: """Return whether two dated shifts overlap, including across midnight.""" first_start, first_end = shift_interval(first) second_start, second_end = shift_interval(second) return first_start < second_end and second_start < first_end def build_constraints( model: cp_model.CpModel, problem_data: Dict[str, Any], x: Dict[Tuple[int, int], cp_model.IntVar] ) -> Dict[str, List[Any]]: """ Build all 12 hard constraints for the scheduling model. Args: model: CP-SAT model instance problem_data: Problem input data x: Decision variables x[employee_id, shift_id] Returns: Dict of constraint metadata for debugging/reporting """ constraints_metadata = { "c1_one_per_shift": [], "c2_locked_shifts": [], "c3_role_qualification": [], "c4_opening_closing": [], "c5_availability": [], "c6_time_off": [], "c7_hours_max": [], "c8_one_per_day": [], "c9_no_overlap": [], "c10_max_shifts_per_week": [], "c11_hours_max_hard": [] } 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_assignments = {ls["shiftId"]: ls["employeeId"] for ls in locked_shifts} locked_shift_ids = set(locked_assignments) # C1: Each shift assigned to at most 1 employee for shift in shifts: shift_id = shift["shiftId"] if shift_id in locked_assignments: # Locked shifts handled separately continue employee_vars = [x.get((emp["userId"], shift_id)) for emp in employees] employee_vars = [v for v in employee_vars if v is not None] if employee_vars: model.Add(sum(employee_vars) <= 1) constraints_metadata["c1_one_per_shift"].append(shift_id) # C2: Locked shifts are immutable for locked_shift in locked_shifts: shift_id = locked_shift["shiftId"] employee_id = locked_shift["employeeId"] # Force this assignment if (employee_id, shift_id) in x: model.Add(x[(employee_id, shift_id)] == 1) constraints_metadata["c2_locked_shifts"].append((shift_id, employee_id)) # Prevent other employees from taking this shift for emp in employees: if emp["userId"] != employee_id and (emp["userId"], shift_id) in x: model.Add(x[(emp["userId"], shift_id)] == 0) # C3/C4: Keyholder qualification. Explicit keyholder requirements and # opener/closer duties are modeled independently so diagnostics identify # the actual business rule that blocked an assignment. for shift in shifts: if shift["shiftId"] in locked_shift_ids or shift.get("isLocked"): continue shift_id = shift["shiftId"] requires_keyholder = shift.get("requiresKeyholder", False) is_opening_or_closing = shift.get("isOpener", False) or shift.get("isCloser", False) # Also check legacy minRoleId for backward compatibility during transition min_role_id = shift.get("minRoleId") needs_keyholder = requires_keyholder or (min_role_id is not None and min_role_id <= 3) if needs_keyholder or is_opening_or_closing: for emp in employees: emp_id = emp["userId"] is_keyholder = emp.get("isKeyholder", False) # Legacy fallback: role <= 3 counts as keyholder emp_role = emp.get("role", 5) emp_is_keyholder = is_keyholder or emp_role <= 3 if not emp_is_keyholder: if (emp_id, shift_id) in x: model.Add(x[(emp_id, shift_id)] == 0) if needs_keyholder: constraints_metadata.setdefault("c3_keyholder_qualification", []).append( (emp_id, shift_id, "not_keyholder") ) if is_opening_or_closing: constraints_metadata["c4_opening_closing"].append( (emp_id, shift_id, "not_keyholder") ) # C5: Availability windows for emp in employees: emp_id = emp["userId"] avail_list = emp.get("availability", []) for shift in shifts: shift_id = shift["shiftId"] if shift_id in locked_shift_ids or shift.get("isLocked"): continue if not availability_covers_shift(shift, avail_list): if (emp_id, shift_id) in x: model.Add(x[(emp_id, shift_id)] == 0) shift_start, shift_end = shift_interval(shift) reason = "overnight_mismatch" if shift_start.date() != shift_end.date() else "outside_window" constraints_metadata["c5_availability"].append((emp_id, shift_id, reason)) # C6: Time-off respected for emp in employees: emp_id = emp["userId"] time_off_dates = {to["date"] for to in emp.get("timeOff", [])} for shift in shifts: shift_id = shift["shiftId"] if shift_id in locked_shift_ids or shift.get("isLocked"): continue 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) conflicting_dates = sorted(touched_dates & time_off_dates) if conflicting_dates and (emp_id, shift_id) in x: model.Add(x[(emp_id, shift_id)] == 0) for conflict_date in conflicting_dates: constraints_metadata["c6_time_off"].append( (emp_id, shift_id, conflict_date) ) # C7 & C11: hoursMax never exceeded (hard cap) for emp in employees: emp_id = emp["userId"] hours_max = emp["hoursMax"] current_hours = emp["currentPeriodHours"] # Convert to centihours (integer) hours_max_centi = int(hours_max * 100) current_hours_centi = int(current_hours * 100) # Sum of assigned shift hours assigned_hours = [] for shift in shifts: shift_id = shift["shiftId"] if shift_id in locked_shift_ids or shift.get("isLocked"): continue 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) model.Add(total_hours <= hours_max_centi) constraints_metadata["c7_hours_max"].append((emp_id, hours_max)) # C8: One shift per employee per day # Build lookup of which employees already have locked shifts on which dates locked_dates_by_emp = {} # emp_id -> set of dates with locked shifts 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) shifts_by_date = {} for shift in shifts: date = shift["date"] if date not in shifts_by_date: shifts_by_date[date] = [] shifts_by_date[date].append(shift) for emp in employees: emp_id = emp["userId"] emp_locked_dates = locked_dates_by_emp.get(emp_id, set()) for date, date_shifts in shifts_by_date.items(): shift_vars = [ x.get((emp_id, s["shiftId"])) for s in date_shifts if s["shiftId"] not in locked_shift_ids and not s.get("isLocked") ] shift_vars = [v for v in shift_vars if v is not None] if date in emp_locked_dates: # Employee already has a locked shift on this date — # prevent ANY open shift assignment on this day for var in shift_vars: model.Add(var == 0) constraints_metadata["c8_one_per_day"].append((emp_id, date, "locked")) elif len(shift_vars) > 1: model.Add(sum(shift_vars) <= 1) constraints_metadata["c8_one_per_day"].append((emp_id, date, len(shift_vars))) # C9: No overlapping shifts, including conflicts with immutable shifts. for emp in employees: emp_id = emp["userId"] emp_shifts = [ (s["shiftId"], s) for s in shifts if s["shiftId"] not in locked_shift_ids and not s.get("isLocked") ] for i, (sid1, s1) in enumerate(emp_shifts): for sid2, s2 in emp_shifts[i+1:]: if shifts_overlap(s1, s2): if (emp_id, sid1) in x and (emp_id, sid2) in x: model.Add(x[(emp_id, sid1)] + x[(emp_id, sid2)] <= 1) constraints_metadata["c9_no_overlap"].append((emp_id, sid1, sid2)) employee_locked_shifts = [ locked for locked in locked_shifts if locked.get("employeeId") == emp_id ] for shift_id, shift in emp_shifts: for locked in employee_locked_shifts: if shifts_overlap(shift, locked) and (emp_id, shift_id) in x: model.Add(x[(emp_id, shift_id)] == 0) constraints_metadata["c9_no_overlap"].append( (emp_id, shift_id, locked["shiftId"]) ) # C10: Max 5 shifts per employee per week for emp in employees: emp_id = emp["userId"] shifts_this_week = emp.get("shiftsThisWeek", 0) # Count assigned shifts shift_vars = [ x.get((emp_id, s["shiftId"])) for s in shifts if s["shiftId"] not in locked_shift_ids and not s.get("isLocked") ] shift_vars = [v for v in shift_vars if v is not None] if shift_vars: model.Add(sum(shift_vars) + shifts_this_week <= 5) constraints_metadata["c10_max_shifts_per_week"].append((emp_id, shifts_this_week)) # C12: Position eligibility # If a shift has a positionId, only employees with that position in # their allowedPositionIds can be assigned. Shifts with positionId=None # (no position) can be worked by any employee. constraints_metadata["c12_position_eligibility"] = [] for emp in employees: emp_id = emp["userId"] allowed_pos_ids = set(emp.get("allowedPositionIds", [])) for shift in shifts: shift_id = shift["shiftId"] if shift_id in locked_shift_ids or shift.get("isLocked"): continue shift_pos_id = shift.get("positionId") # Only enforce if the shift HAS a position assigned if shift_pos_id is not None and shift_pos_id not in allowed_pos_ids: if (emp_id, shift_id) in x: model.Add(x[(emp_id, shift_id)] == 0) constraints_metadata["c12_position_eligibility"].append( (emp_id, shift_id, shift_pos_id, list(allowed_pos_ids)) ) return constraints_metadata