"""Solveur de suggestions (OR-Tools CP-SAT) — cœur du produit, second moteur. Ne place jamais personne : il propose, pour chaque créneau encore sous son effectif idéal, une liste ordonnée de bénévoles candidats. C'est la coordination qui affecte (dans l'app, via l'écran habituel — le solveur se contente de faire remonter ces candidats en tête de liste). Contrainte dure : aucun chevauchement de créneaux pour un même bénévole (règle `SHIFT_OVERLAP`, cf. `app.validation`). Tout le reste (disponibilité, cumul de 8 h, repos, vœu de poste, vœu d'artiste, équité de charge) est un terme de l'objectif, pas une contrainte — comme le moteur de validation, qui reste seul juge au moment où la coordination confirme une affectation. """ import datetime as dt from ortools.sat.python import cp_model from app.solver.models import ( ShiftSuggestion, SolverInput, SolverShift, SolverSolution, SolverVolunteer, ) from app.validation.engine import uncovered_minutes from app.validation.models import DayPartWindow, ShiftInfo _EIGHT_HOURS_MIN = 8 * 60 # Poids de l'objectif (unités arbitraires, cf. docstring de `_score`). # L'écart à un effectif minimal non atteint prime sur tout le reste : le # solveur doit toujours proposer quelqu'un pour un créneau sous-staffé, quitte # à ce que la disponibilité soit imparfaite (avertissement, pas un blocage — # la validation reste seule juge à la confirmation). Au-delà du minimum, un # candidat mal assorti peut légitimement ne pas être suggéré du tout. _URGENT_BASE = 1000 # créneau encore sous son effectif minimal _NORMAL_BASE = 20 # créneau entre effectif minimal et idéal _WISH_BONUS = 30 _AVAIL_PENALTY_PER_HALF_HOUR = 5 _DAILY_OVERAGE_PENALTY_PER_MIN = 2 _REST_PENALTY = 60 _FAIRNESS_PENALTY_PER_HOUR = 2 # Le créneau lui ferait manquer le passage d'un artiste qu'il veut voir. _ARTIST_CONFLICT_PENALTY = 80 def _overlaps(a: ShiftInfo, b: ShiftInfo) -> bool: return a.starts_at < b.ends_at and b.starts_at < a.ends_at def _day_boundary(day_parts: tuple[DayPartWindow, ...]) -> dt.time: if not day_parts: return dt.time(0, 0) return min(day_parts, key=lambda part: part.order).start_time def _day_key(moment: dt.datetime, boundary: dt.time) -> dt.date: if moment.time() >= boundary: return moment.date() return moment.date() - dt.timedelta(days=1) def _daily_overage_minutes( volunteer: SolverVolunteer, shift: SolverShift, boundary: dt.time ) -> int: """Minutes qui dépasseraient les 8 h de la journée du créneau, en ne comptant que les affectations déjà décidées (pas les autres suggestions du même calcul — ce cas limite reste sous la vigilance du moteur de validation au moment de la confirmation).""" key = _day_key(shift.starts_at, boundary) total = shift.ends_at - shift.starts_at for other in volunteer.existing_shifts: if _day_key(other.starts_at, boundary) == key: total += other.ends_at - other.starts_at minutes = int(total.total_seconds() // 60) return max(0, minutes - _EIGHT_HOURS_MIN) def _has_insufficient_rest(volunteer: SolverVolunteer, shift: SolverShift) -> bool: candidate = shift.shift_info for other in volunteer.existing_shifts: if _overlaps(candidate, other): continue if other.ends_at <= candidate.starts_at: gap = candidate.starts_at - other.ends_at elif candidate.ends_at <= other.starts_at: gap = other.starts_at - candidate.ends_at else: continue if gap < dt.timedelta(minutes=_EIGHT_HOURS_MIN): return True return False def _score( volunteer: SolverVolunteer, shift: SolverShift, day_boundary: dt.time, ) -> int: """Score d'affinité (bénévole, créneau) : plus haut = meilleur candidat. Purement additif (aucune interaction entre deux affectations proposées), ce qui garde le modèle CP-SAT linéaire simple. Cumul de 8 h et repos ne sont vérifiés que contre les affectations déjà décidées, pas entre deux suggestions du même calcul : la validation finale (à la confirmation) reste seule juge en cas de cas limite entre deux suggestions. """ score = _URGENT_BASE if shift.already_assigned < shift.min_staff else _NORMAL_BASE if shift.position_id in volunteer.wished_position_ids: score += _WISH_BONUS missed = uncovered_minutes(shift.shift_info, volunteer.availability) score -= (missed // 30) * _AVAIL_PENALTY_PER_HALF_HOUR score -= _DAILY_OVERAGE_PENALTY_PER_MIN * _daily_overage_minutes( volunteer, shift, day_boundary ) if _has_insufficient_rest(volunteer, shift): score -= _REST_PENALTY if any( _overlaps(shift.shift_info, passage) for passage in volunteer.wished_artist_windows ): score -= _ARTIST_CONFLICT_PENALTY score -= (_FAIRNESS_PENALTY_PER_HOUR * volunteer.existing_minutes) // 60 return score def solve( data: SolverInput, *, time_limit_seconds: float = 8.0, seed: int = 42, ) -> SolverSolution: """Calcule les suggestions. Fonction pure : ne touche jamais la base.""" boundary = _day_boundary(data.day_parts) model = cp_model.CpModel() variables: dict[tuple[int, int], cp_model.IntVar] = {} scores: dict[tuple[int, int], int] = {} for si, shift in enumerate(data.shifts): gap = shift.ideal_staff - shift.already_assigned if gap <= 0: continue for vi, volunteer in enumerate(data.volunteers): if any(_overlaps(shift.shift_info, other) for other in volunteer.existing_shifts): continue key = (vi, si) variables[key] = model.NewBoolVar(f"x_{vi}_{si}") scores[key] = _score(volunteer, shift, boundary) if not variables: return SolverSolution() # Effectif idéal comme plafond dur — le solveur ne sur-staffe jamais. for si, shift in enumerate(data.shifts): gap = shift.ideal_staff - shift.already_assigned n_volunteers = len(data.volunteers) shift_vars = [variables[vi, si] for vi in range(n_volunteers) if (vi, si) in variables] if shift_vars and gap > 0: model.Add(sum(shift_vars) <= gap) # Aucun chevauchement entre deux créneaux nouvellement suggérés au même # bénévole (règle `SHIFT_OVERLAP`, bloquante). for vi in range(len(data.volunteers)): my_shifts = [si for si in range(len(data.shifts)) if (vi, si) in variables] for a in range(len(my_shifts)): for b in range(a + 1, len(my_shifts)): si_a, si_b = my_shifts[a], my_shifts[b] if _overlaps(data.shifts[si_a].shift_info, data.shifts[si_b].shift_info): model.Add(variables[vi, si_a] + variables[vi, si_b] <= 1) model.Maximize(sum(score * variables[key] for key, score in scores.items())) solver = cp_model.CpSolver() solver.parameters.max_time_in_seconds = time_limit_seconds solver.parameters.num_search_workers = 1 solver.parameters.random_seed = seed status = solver.Solve(model) chosen: dict[int, list[int]] = {} if status in (cp_model.OPTIMAL, cp_model.FEASIBLE): for (vi, si), var in variables.items(): if solver.Value(var): chosen.setdefault(si, []).append(vi) suggestions: list[ShiftSuggestion] = [] for si, shift in enumerate(data.shifts): picked = sorted(chosen.get(si, []), key=lambda vi: -scores[(vi, si)]) if picked: suggestions.append( ShiftSuggestion( shift_id=shift.id, volunteer_ids=tuple(data.volunteers[vi].id for vi in picked), ) ) return SolverSolution(suggestions=tuple(suggestions))