Solveur : le malus de chaînage n'est pas limité au même poste

Correction : le malus ajouté précédemment ne se déclenchait que si le
bénévole avait tenu le créneau précédent du *même* poste. L'intention était
plus générale — limiter les enchaînements de shifts d'un bénévole, quel que
soit le poste, pas seulement faire tourner un poste en particulier.

- `SolverShift.predecessor_shift_id: str | None` → `predecessor_shift_ids:
  frozenset[str]` : l'ensemble des créneaux (n'importe quel poste) du même
  jour dont la fin coïncide avec le début de celui-ci. Peut contenir
  plusieurs ids (plusieurs postes se terminant à la même heure).
- Loader : le calcul du prédécesseur se fait maintenant sur tous les
  créneaux de l'édition regroupés par jour (comparaison des heures de fin),
  plus par poste.
- Miroir côté mock (`buildPredecessorMap`, `runSolver`).
- Nouveau test d'intégration (`test_api_solver.py`) : deux postes différents,
  confirme que le chaînage s'applique bien d'un poste à l'autre.
- CLAUDE.md à jour.

Vérifié : 185 tests backend (dont le nouveau), ruff, alembic check ;
typecheck/lint/build/smoke frontend.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017uUsRg7XzopkCPWimwg9ZM
This commit is contained in:
AdamThomas
2026-09-15 21:20:41 +02:00
co-authored by Claude Sonnet 5
parent 64e2f86647
commit b97b95c606
7 changed files with 127 additions and 64 deletions
+13 -9
View File
@@ -7,9 +7,10 @@ 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é, repos,
chaînage sur le même poste, 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.
chaînage entre créneaux consécutifs, 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.
Pas de plafond de cumul journalier côté solveur : `DAILY_HOURS_EXCEEDED`
reste vérifiée par la validation, à la confirmation.
"""
@@ -44,8 +45,9 @@ _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
# Le bénévole a déjà tenu le créneau précédent de ce même poste (même jour) :
# favorise la rotation plutôt que d'enchaîner toujours la même personne.
# Le bénévole a déjà tenu le créneau immédiatement précédent (même jour,
# n'importe quel poste) : limite les enchaînements de shifts plutôt que de
# faire tourner un poste en particulier.
_CHAIN_PENALTY = 40
@@ -70,11 +72,13 @@ def _has_insufficient_rest(volunteer: SolverVolunteer, shift: SolverShift) -> bo
def _chained_from_predecessor(volunteer: SolverVolunteer, shift: SolverShift) -> bool:
"""Le bénévole a déjà (affectation décidée) le créneau précédent de ce
même poste, le même jour."""
if shift.predecessor_shift_id is None:
"""Le bénévole a déjà (affectation décidée) l'un des créneaux qui
précèdent immédiatement celui-ci, le même jour — n'importe quel poste :
il s'agit de limiter les enchaînements de shifts, pas de faire tourner un
poste en particulier."""
if not shift.predecessor_shift_ids:
return False
return any(s.id == shift.predecessor_shift_id for s in volunteer.existing_shifts)
return any(s.id in shift.predecessor_shift_ids for s in volunteer.existing_shifts)
def _score(volunteer: SolverVolunteer, shift: SolverShift) -> int:
+18 -15
View File
@@ -40,21 +40,24 @@ def build_solver_input(*, session: Session, edition: Edition) -> SolverInput:
)
shifts_by_id = {s.id: s for s in shifts}
# Prédécesseur (même poste, même jour, créneau juste avant) : sert au
# malus de chaînage (voir `engine._score`). Calculé sur tous les créneaux
# du poste, pas seulement ceux à compléter.
predecessor_id: dict = {}
by_position: dict = defaultdict(list)
# Prédécesseur(s) : créneaux (n'importe quel poste) du même jour dont la
# fin coïncide avec le début de celui-ci — sert au malus de chaînage
# (voir `engine._score`), pour limiter les enchaînements de shifts plutôt
# que cibler la rotation d'un poste en particulier. Calculé sur tous les
# créneaux de l'édition, pas seulement ceux à compléter.
predecessor_ids: dict = {}
by_day: dict = defaultdict(list)
for s in shifts:
by_position[s.position_id].append(s)
for position_shifts in by_position.values():
by_day: dict = defaultdict(list)
for s in position_shifts:
by_day[s.starts_at.date()].append(s)
for day_shifts in by_day.values():
day_shifts.sort(key=lambda s: s.starts_at)
for previous, current in zip(day_shifts, day_shifts[1:], strict=False):
predecessor_id[current.id] = str(previous.id)
by_day[s.starts_at.date()].append(s)
for day_shifts in by_day.values():
ids_by_end: dict = defaultdict(list)
for s in day_shifts:
ids_by_end[s.ends_at].append(str(s.id))
distinct_ends = sorted(ids_by_end)
for s in day_shifts:
earlier_ends = [e for e in distinct_ends if e <= s.starts_at]
if earlier_ends:
predecessor_ids[s.id] = frozenset(ids_by_end[earlier_ends[-1]])
assigned_count: dict = defaultdict(int)
shifts_by_volunteer: dict = defaultdict(list)
@@ -73,7 +76,7 @@ def build_solver_input(*, session: Session, edition: Edition) -> SolverInput:
min_staff=s.min_staff,
ideal_staff=s.ideal_staff,
already_assigned=assigned_count.get(s.id, 0),
predecessor_shift_id=predecessor_id.get(s.id),
predecessor_shift_ids=predecessor_ids.get(s.id, frozenset()),
)
for s in shifts
if assigned_count.get(s.id, 0) < s.ideal_staff
+6 -4
View File
@@ -22,9 +22,11 @@ class SolverShift:
min_staff: int
ideal_staff: int
already_assigned: int
# Créneau du même poste, même jour, juste avant celui-ci (`None` s'il n'y
# en a pas) — sert au malus de chaînage (voir `engine._score`).
predecessor_shift_id: str | None = None
# Créneau(x) — n'importe quel poste — dont la fin coïncide avec le début
# de celui-ci, le même jour (vide s'il n'y en a pas) : sert au malus de
# chaînage (voir `engine._score`), pour limiter les enchaînements de
# shifts plutôt que cibler la rotation d'un poste en particulier.
predecessor_shift_ids: frozenset[str] = frozenset()
@property
def shift_info(self) -> ShiftInfo:
@@ -41,7 +43,7 @@ class SolverVolunteer:
# lui ferait manquer le passage.
wished_artist_windows: tuple[ShiftInfo, ...] = ()
# Créneaux déjà occupés (affectations existantes, décisions figées) :
# comptent pour chevauchement, repos, chaînage sur le même poste.
# comptent pour chevauchement, repos, chaînage.
existing_shifts: tuple[ShiftInfo, ...] = ()
# Total déjà affecté sur l'édition (minutes), pour répartir la charge.
existing_minutes: int = 0
+31
View File
@@ -71,6 +71,37 @@ def test_avoids_volunteer_whose_wished_artist_would_be_missed(client, edition, s
assert suggestion["volunteer_ids"] == [str(indiff.id)]
def test_chain_penalty_applies_across_different_positions(client, edition, session):
from app.models import Assignment, PositionCategory
# Deux postes différents, même jour, repos suffisant (8 h pile) entre les
# deux créneaux : seul le chaînage (n'importe quel poste) doit
# différencier les deux bénévoles pour le second créneau.
bar = make_position(session, edition, name="Bar")
entrance = make_position(session, edition, name="Entrée", category=PositionCategory.WELCOME)
morning_shift = make_shift(
session, edition, bar,
start=dt_at(10, 8), end=dt_at(10, 12), min_staff=1, ideal_staff=1,
)
evening_shift = make_shift(
session, edition, entrance,
start=dt_at(10, 20), end=dt_at(10, 23), min_staff=1, ideal_staff=1,
)
chained = make_volunteer(session, edition, email="chained@ex.fr")
session.add(
Assignment(edition_id=edition.id, volunteer_id=chained.id, shift_id=morning_shift.id)
)
session.commit()
fresh = make_volunteer(session, edition, email="fresh@ex.fr")
resp = client.post(_url(edition, "/suggestions"))
suggestion = next(
s for s in resp.json()["suggestions"] if s["shift_id"] == str(evening_shift.id)
)
assert suggestion["volunteer_ids"] == [str(fresh.id)]
def test_unknown_edition_is_404(client):
assert client.post(f"/editions/{uuid.uuid4()}/solver/suggestions").status_code == 404
+31 -12
View File
@@ -20,7 +20,7 @@ def _shift(
min_staff=1,
ideal_staff=1,
already_assigned=0,
predecessor_shift_id: str | None = None,
predecessor_shift_ids=frozenset(),
) -> SolverShift:
return SolverShift(
id=sid,
@@ -30,7 +30,7 @@ def _shift(
min_staff=min_staff,
ideal_staff=ideal_staff,
already_assigned=already_assigned,
predecessor_shift_id=predecessor_shift_id,
predecessor_shift_ids=frozenset(predecessor_shift_ids),
)
@@ -169,10 +169,27 @@ def test_daily_cumulative_hours_no_longer_penalized():
assert set(_by_shift(solution, "s1")) == {"long_day", "fresh"}
def test_prefers_volunteer_without_the_previous_shift_on_the_same_position():
# Même poste, repos suffisant (8 h pile) entre les deux créneaux : seul
# le chaînage doit différencier les deux bénévoles.
shift = _shift(start=_dt(10, 20), end=_dt(10, 23), predecessor_shift_id="prev")
def test_prefers_volunteer_without_the_immediately_preceding_shift():
# Repos suffisant (8 h pile) entre les deux créneaux : seul le chaînage
# doit différencier les deux bénévoles.
shift = _shift(start=_dt(10, 20), end=_dt(10, 23), predecessor_shift_ids={"prev"})
chained = _volunteer(
"chained",
availability=[_full_day(10)],
existing_shifts=[ShiftInfo(id="prev", starts_at=_dt(10, 8), ends_at=_dt(10, 12))],
)
fresh = _volunteer("fresh", availability=[_full_day(10)])
solution = solve(SolverInput(shifts=(shift,), volunteers=(chained, fresh)))
assert _by_shift(solution, "s1") == ("fresh",)
def test_chain_penalty_applies_across_different_positions():
# Le prédécesseur n'est pas limité au même poste : le bénévole enchaîne
# ici depuis un autre poste, ça compte quand même — l'idée est de limiter
# les enchaînements de shifts, pas de faire tourner un poste précis.
shift = _shift(
position_id="bar", start=_dt(10, 20), end=_dt(10, 23), predecessor_shift_ids={"prev"}
)
chained = _volunteer(
"chained",
availability=[_full_day(10)],
@@ -186,7 +203,7 @@ def test_prefers_volunteer_without_the_previous_shift_on_the_same_position():
def test_still_suggests_the_only_candidate_despite_chaining():
# Le chaînage reste un avertissement, pas un blocage : personne d'autre
# pour ce créneau urgent, le solveur le propose quand même.
shift = _shift(start=_dt(10, 20), end=_dt(10, 23), predecessor_shift_id="prev")
shift = _shift(start=_dt(10, 20), end=_dt(10, 23), predecessor_shift_ids={"prev"})
chained = _volunteer(
"chained",
availability=[_full_day(10)],
@@ -197,11 +214,13 @@ def test_still_suggests_the_only_candidate_despite_chaining():
def test_an_unrelated_existing_shift_does_not_trigger_the_chain_penalty():
# Le bénévole a bien un créneau existant, mais ce n'est pas le
# prédécesseur désigné de ce créneau (ex. un autre poste) : pas de malus
# de chaînage (le prédécesseur est identifié par id, pas par proximité).
# Les deux candidats doivent donc être à égalité.
shift = _shift(ideal_staff=2, start=_dt(10, 20), end=_dt(10, 23), predecessor_shift_id="prev")
# Le bénévole a bien un créneau existant, mais ce n'est pas l'un des
# prédécesseurs désignés de ce créneau : pas de malus de chaînage (le
# prédécesseur est identifié par id, pas par proximité). Les deux
# candidats doivent donc être à égalité.
shift = _shift(
ideal_staff=2, start=_dt(10, 20), end=_dt(10, 23), predecessor_shift_ids={"prev"}
)
unrelated = ShiftInfo(id="not-the-predecessor", starts_at=_dt(10, 8), ends_at=_dt(10, 12))
almost_chained = _volunteer(
"almost_chained", availability=[_full_day(10)], existing_shifts=[unrelated]