Feat/standard shift constraints

python/python-core/src/main/python/domain/__init__.py

@@ -30,6 +30,7 @@
 from ._value_range import *
 from ._variable_listener import *
 from typing import TYPE_CHECKING as _TYPE_CHECKING
+from .shift_scheduling import Timeslot, Skill, Employee, Shift
 
 if _TYPE_CHECKING:
     class CountableValueRange:
@@ -43,5 +44,5 @@ def __getattr__(name: str):
 
 if not _TYPE_CHECKING:
     exported = [name for name in globals().keys() if not name.startswith('_')]
-    exported += ['CountableValueRange']
+    exported += ['CountableValueRange', 'Timeslot', 'Skill', 'Employee', 'Shift']
     __all__ = exported

python/python-core/src/main/python/domain/shift_scheduling.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import datetime
+from dataclasses import dataclass, field
+from typing import List
+
+from timefold.solver.domain import problem_fact, planning_entity, planning_variable
+
+
+@dataclass
+class Timeslot:
+    id: str
+    start_datetime: datetime.datetime
+    end_datetime: datetime.datetime
+
+    def overlaps(self, other_timeslot: Timeslot) -> bool:
+        return max(self.start_datetime, other_timeslot.start_datetime) < min(self.end_datetime, other_timeslot.end_datetime)
+
+
+@dataclass(frozen=True)
+class Skill:
+    name: str
+
+    def __hash__(self):
+        return hash(self.name)
+
+    def __eq__(self, other):
+        if not isinstance(other, Skill):
+            return False
+        return self.name == other.name
+
+
+@problem_fact
+@dataclass
+class Employee:
+    id: str
+    name: str
+    skills: List[Skill] = field(default_factory=list)
+    unavailable_timeslots: List[Timeslot] = field(default_factory=list)
+    cannot_work_with: List[str] = field(default_factory=list)
+    classification: str = "unknown"
+    is_senior: bool = False
+
+
+@planning_entity
+@dataclass
+class Shift:
+    id: str
+    timeslot: Timeslot
+    required_skill: Skill
+    employee: Employee = field(default=None)
+
+    @planning_variable(Employee)
+    def get_employee(self) -> Employee | None:
+        return self.employee
+
+    def set_employee(self, employee: Employee | None) -> None:
+        self.employee = employee

python/python-core/src/main/python/score/__init__.py

@@ -27,3 +27,24 @@
 from ._score import *
 from ._score_analysis import *
 from ._score_director import *
+from .standard_shift_constraints import define_standard_shift_constraints
+
+__all__ = [
+    # from ._annotations
+    'constraint_provider',
+    # from ._constraint_factory
+    'ConstraintFactory',
+    # from ._constraint_builder
+    'ConstraintBuilder', # Note: Or specific builders if preferred
+    # from ._constraint_stream
+    'ConstraintStream', # Or specific stream types
+    # from ._joiners
+    'Joiners',
+    # from ._score
+    'HardSoftScore', 'SimpleScore', 'HardMediumSoftScore', # Add other score types as needed
+    # from ._incremental_score_calculator
+    'IncrementalScoreCalculator',
+    # from current file
+    'define_standard_shift_constraints'
+    # Potentially others from _score_director, _score_analysis depending on typical public API
+]

python/python-core/src/main/python/score/standard_shift_constraints.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from timefold.solver.score import ConstraintFactory, HardSoftScore, Joiners
+
+from ..domain.shift_scheduling import Shift, Employee, Skill
+
+
+def no_overlapping_shifts(constraint_factory: ConstraintFactory):
+    return (constraint_factory.for_each_unique_pair(
+        Shift,
+        Joiners.equal(lambda shift: shift.employee),
+        Joiners.filtering(lambda shift1, shift2: shift1.timeslot.overlaps(shift2.timeslot))
+    )
+            .penalize(HardSoftScore.ONE_HARD)
+            .as_constraint("no_overlapping_shifts"))
+
+
+def employee_availability(constraint_factory: ConstraintFactory):
+    return (constraint_factory.for_each(Shift)
+            .filter(lambda shift: shift.employee is not None and
+                                 any(shift.timeslot.overlaps(unavailable_slot)
+                                     for unavailable_slot in shift.employee.unavailable_timeslots))
+            .penalize(HardSoftScore.ONE_HARD)
+            .as_constraint("employee_availability"))
+
+
+def skill_requirement(constraint_factory: ConstraintFactory):
+    return (constraint_factory.for_each(Shift)
+            .filter(lambda shift: (shift.employee is not None and
+                                   shift.required_skill is not None and
+                                   shift.required_skill not in shift.employee.skills))
+            .penalize(HardSoftScore.ONE_HARD)
+            .as_constraint("skill_requirement"))
+
+
+def filter_cannot_work_together(shift1: Shift, shift2: Shift) -> bool:
+    if shift1.employee is None or shift2.employee is None:
+        return False
+    # If employees are the same, this is not a "cannot work together" scenario,
+    # but potentially an "overlapping shift" for the same employee, handled by another constraint.
+    if shift1.employee.id == shift2.employee.id:
+        return False
+    if not shift1.timeslot.overlaps(shift2.timeslot):
+        return False
+    if shift1.employee.id in shift2.employee.cannot_work_with:
+        return True
+    if shift2.employee.id in shift1.employee.cannot_work_with:
+        return True
+    return False
+
+
+def cannot_work_together(constraint_factory: ConstraintFactory):
+    return (constraint_factory.for_each_unique_pair(
+        Shift,
+        Joiners.filtering(filter_cannot_work_together)
+    )
+            .penalize(HardSoftScore.ONE_HARD)
+            .as_constraint("cannot_work_together"))
+
+
+def define_standard_shift_constraints(constraint_factory: ConstraintFactory):
+    return [
+        no_overlapping_shifts(constraint_factory),
+        employee_availability(constraint_factory),
+        skill_requirement(constraint_factory),
+        cannot_work_together(constraint_factory),
+        seniority_requirement(constraint_factory)
+    ]
+
+
+def seniority_requirement(constraint_factory: ConstraintFactory):
+    return (constraint_factory.for_each(Shift)
+            .filter(lambda s_junior: (s_junior.employee is not None and
+                                      s_junior.employee.classification in ["nurse", "paramedic"] and
+                                      not s_junior.employee.is_senior))
+            .if_not_exists(Shift,
+                           Joiners.filtering(lambda s_junior, s_other: s_junior.id != s_other.id),
+                           Joiners.filtering(lambda s_junior, s_other: (s_other.employee is not None and
+                                                                       s_other.employee.classification in ["nurse", "paramedic"])),
+                           Joiners.filtering(lambda s_junior, s_other: s_junior.timeslot.overlaps(s_other.timeslot))
+                           )
+            .penalize(HardSoftScore.ONE_HARD)
+            .as_constraint("seniority_requirement"))