Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions launch.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
{
"version": "0.2.0",
"configurations": [
{
"name": "Python: Attach to Docker debugpy",
"type": "debugpy",
"request": "attach",
"connect": {
"host": "localhost",
"port": 5678
},
"pathMappings": [
{
"localRoot": "${workspaceFolder}",
"remoteRoot": "/app"
}
],
"justMyCode": false
}
]
}
45 changes: 45 additions & 0 deletions src/scheduling/solver/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,22 @@
from pydantic import Field

from scheduling.domain import SchedulingBaseModel
from scheduling.solver.cp_sat.constraints.availabilities_constraint import AvailabilitiesConstraint
from scheduling.solver.cp_sat.constraints.free_day_after_night_shift_phase import FreeDayAfterNightShiftPhase
from scheduling.solver.cp_sat.constraints.hierarchy_of_intermediate_shifts import HierarchyOfIntermediateShifts
from scheduling.solver.cp_sat.constraints.minimum_staffing import MinimumStaffing
from scheduling.solver.cp_sat.constraints.one_assignment_per_day import OneAssignmentPerDay
from scheduling.solver.cp_sat.constraints.rounds_in_early_shift import RoundsInEarlyShift
from scheduling.solver.cp_sat.constraints.target_working_time import TargetWorkingTime
from scheduling.solver.cp_sat.objectives.every_second_weekend_free import EverySecondWeekendFree
from scheduling.solver.cp_sat.objectives.fair_preferences import FairPreferencesObjective
from scheduling.solver.cp_sat.objectives.free_day_after_night_shift_phase import FreeDaysAfterNightShiftPhase
from scheduling.solver.cp_sat.objectives.free_days_near_weekend import FreeDaysNearWeekend
from scheduling.solver.cp_sat.objectives.minimize_consecutive_night_shifts import MinimizeConsecutiveNightShifts
from scheduling.solver.cp_sat.objectives.minimize_overtime import MinimizeOvertime
from scheduling.solver.cp_sat.objectives.not_too_many_consecutive_days import NotTooManyConsecutiveDays
from scheduling.solver.cp_sat.objectives.preferred_block_length import PreferredBlockLength
from scheduling.solver.cp_sat.objectives.rotate_shits_foward import RotateShiftsForward
from scheduling.solver.cp_sat.objectives.temporary_balance_generated_assignments import (
TemporaryBalanceGeneratedAssignments,
)
Expand Down Expand Up @@ -34,11 +49,41 @@ def create_base_solver_config() -> SolverConfig:
return SolverConfig(
constraints={
MinimumStaffing.id: ConstraintConfig(enabled=True),
FreeDayAfterNightShiftPhase.id: ConstraintConfig(enabled=True),
RoundsInEarlyShift.id: ConstraintConfig(enabled=True),
AvailabilitiesConstraint.id: ConstraintConfig(enabled=True),
HierarchyOfIntermediateShifts.id: ConstraintConfig(enabled=True),
OneAssignmentPerDay.id: ConstraintConfig(enabled=True),
TargetWorkingTime.id: ConstraintConfig(enabled=True),
},
objectives={
TemporaryBalanceGeneratedAssignments.id: ObjectiveConfig(
enabled=True,
weight=1,
),
MinimizeOvertime.id: ObjectiveConfig(
enabled=True,
weight=100,
),
NotTooManyConsecutiveDays.id: ObjectiveConfig(
enabled=True,
weight=1,
),
PreferredBlockLength.id: ObjectiveConfig(
enabled=True,
weight=1,
),
RotateShiftsForward.id: ObjectiveConfig(
enabled=True,
weight=1,
),
EverySecondWeekendFree.id: ObjectiveConfig(enabled=True, weight=1),
FairPreferencesObjective.id: ObjectiveConfig(
enabled=True,
weight=1,
),
FreeDaysAfterNightShiftPhase.id: ObjectiveConfig(enabled=True, weight=1),
FreeDaysNearWeekend.id: ObjectiveConfig(enabled=True, weight=1),
MinimizeConsecutiveNightShifts.id: ObjectiveConfig(enabled=True, weight=1),
},
)
38 changes: 36 additions & 2 deletions src/scheduling/solver/cp_sat/builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,17 +5,51 @@
from scheduling.domain import SchedulingDataset
from scheduling.solver.config import SolverConfig, create_base_solver_config
from scheduling.solver.cp_sat.constraint import Constraint
from scheduling.solver.cp_sat.constraints.availabilities_constraint import AvailabilitiesConstraint
from scheduling.solver.cp_sat.constraints.free_day_after_night_shift_phase import FreeDayAfterNightShiftPhase
from scheduling.solver.cp_sat.constraints.hierarchy_of_intermediate_shifts import HierarchyOfIntermediateShifts
from scheduling.solver.cp_sat.constraints.minimum_staffing import MinimumStaffing
from scheduling.solver.cp_sat.constraints.one_assignment_per_day import OneAssignmentPerDay
from scheduling.solver.cp_sat.constraints.rounds_in_early_shift import RoundsInEarlyShift
from scheduling.solver.cp_sat.constraints.target_working_time import TargetWorkingTime
from scheduling.solver.cp_sat.context import SolverContext, create_context
from scheduling.solver.cp_sat.objective import Objective, WeightedPenalty, minimize_weighted_penalties
from scheduling.solver.cp_sat.objectives.every_second_weekend_free import EverySecondWeekendFree
from scheduling.solver.cp_sat.objectives.fair_preferences import FairPreferencesObjective
from scheduling.solver.cp_sat.objectives.free_day_after_night_shift_phase import FreeDaysAfterNightShiftPhase
from scheduling.solver.cp_sat.objectives.free_days_near_weekend import FreeDaysNearWeekend
from scheduling.solver.cp_sat.objectives.minimize_consecutive_night_shifts import MinimizeConsecutiveNightShifts
from scheduling.solver.cp_sat.objectives.minimize_overtime import MinimizeOvertime
from scheduling.solver.cp_sat.objectives.not_too_many_consecutive_days import NotTooManyConsecutiveDays
from scheduling.solver.cp_sat.objectives.preferred_block_length import PreferredBlockLength
from scheduling.solver.cp_sat.objectives.rotate_shits_foward import RotateShiftsForward
from scheduling.solver.cp_sat.objectives.temporary_balance_generated_assignments import (
TemporaryBalanceGeneratedAssignments,
)
from scheduling.solver.cp_sat.variables import create_assignment_variables

CP_SAT_CONSTRAINTS: tuple[Constraint, ...] = (MinimumStaffing(),)
CP_SAT_CONSTRAINTS: tuple[Constraint, ...] = (
MinimumStaffing(),
FreeDayAfterNightShiftPhase(),
RoundsInEarlyShift(),
AvailabilitiesConstraint(),
HierarchyOfIntermediateShifts(),
OneAssignmentPerDay(),
TargetWorkingTime(),
)

CP_SAT_OBJECTIVES: tuple[Objective, ...] = (TemporaryBalanceGeneratedAssignments(),)
CP_SAT_OBJECTIVES: tuple[Objective, ...] = (
TemporaryBalanceGeneratedAssignments(),
MinimizeOvertime(),
NotTooManyConsecutiveDays(),
PreferredBlockLength(),
RotateShiftsForward(),
EverySecondWeekendFree(),
FairPreferencesObjective(),
FreeDaysAfterNightShiftPhase(),
MinimizeConsecutiveNightShifts(),
FreeDaysNearWeekend(),
)


@dataclass(frozen=True, slots=True)
Expand Down
154 changes: 154 additions & 0 deletions src/scheduling/solver/cp_sat/constraints/availabilities_constraint.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,154 @@
import datetime
from collections import defaultdict
from collections.abc import Mapping
from typing import Any, ClassVar

from scheduling.solver.audit import AuditFinding, AuditSeverity
from scheduling.solver.cp_sat.context import AuditContext, SolverContext
from scheduling.solver.diagnostics import SolverDiagnostic
from scheduling.solver.index import is_night_shift

# Angenommen, Ihre Enum/Models sind importierbar, andernfalls als Typ-Hinweise nutzen
# from scheduling.domain.availability import Availability, AvailabilityType


class AvailabilitiesConstraint:
"""Ensure employees do not work on dates or shifts they are unavailable for.

Also prevents night shifts from spilling over into full-day absences (e.g., vacations).
"""

id: ClassVar[str] = "employee_availabilities"
required: ClassVar[bool] = True

def add_to_model(
self,
ctx: SolverContext,
params: Mapping[str, Any],
) -> tuple[SolverDiagnostic, ...]:
del params

blocked_days, allowed_shifts_for_day = _parse_availabilities(ctx)

for key, variable in ctx.assignment_variables.items():
employee_id, _, date, shift_id, _ = key

# Regel 1: Voller Abwesenheitstag (Urlaub, Training, etc.)
if (employee_id, date) in blocked_days:
ctx.model.add(variable == 0).with_name(
f"avail_blocked_full__emp_{employee_id}_date_{date:%Y%m%d}_shift_{shift_id}"
)
continue

# Regel 2: Partielle Verfügbarkeit (AVAILABLE_ONLY)
if (employee_id, date) in allowed_shifts_for_day:
if shift_id not in allowed_shifts_for_day[(employee_id, date)]:
ctx.model.add(variable == 0).with_name(
f"avail_blocked_partial__emp_{employee_id}_date_{date:%Y%m%d}_shift_{shift_id}"
)
continue

# Regel 3: Spillover-Prävention (Nachtschicht vor einem vollen Abwesenheitstag)
if is_night_shift(ctx.index.shifts_by_id[shift_id]):
tomorrow = date + datetime.timedelta(days=1)
if (employee_id, tomorrow) in blocked_days:
ctx.model.add(variable == 0).with_name(
f"avail_blocked_spillover__emp_{employee_id}_date_{date:%Y%m%d}_shift_{shift_id}"
)

return ()

def audit(
self,
ctx: AuditContext,
params: Mapping[str, Any],
) -> tuple[AuditFinding, ...]:
del params

findings: list[AuditFinding] = []
blocked_days, allowed_shifts_for_day = _parse_availabilities(ctx)

for assignment in ctx.assignments:
# Type-Guards (Nur relevante Zuweisungen prüfen)
if assignment.employee_id is None or assignment.shift_id is None or assignment.date is None: # type: ignore
continue

emp_id = assignment.employee_id
shift_id = assignment.shift_id
date = assignment.date

# Audit 1: Zuweisung an einem voll blockierten Tag
if (emp_id, date) in blocked_days:
findings.append(
AuditFinding(
code="employee_availabilities.violation_full_day",
severity=AuditSeverity.ERROR,
source_id=self.id,
message=(
f"Employee is assigned to a shift on a fully blocked day. "
f"employee_id={emp_id} date={date.isoformat()} shift_id={shift_id}."
),
date=date,
)
)

# Audit 2: Zuweisung einer unzulässigen Schicht (AVAILABLE_ONLY)
elif (emp_id, date) in allowed_shifts_for_day and shift_id not in allowed_shifts_for_day[(emp_id, date)]:
findings.append(
AuditFinding(
code="employee_availabilities.violation_partial_day",
severity=AuditSeverity.ERROR,
source_id=self.id,
message=(
f"Employee is assigned to a restricted shift type. "
f"employee_id={emp_id} date={date.isoformat()} shift_id={shift_id}."
),
date=date,
)
)

# Audit 3: Unzulässiger Spillover (Nachtschicht am Vortag)
elif is_night_shift(ctx.index.shifts_by_id[shift_id]):
tomorrow = date + datetime.timedelta(days=1)
if (emp_id, tomorrow) in blocked_days:
findings.append(
AuditFinding(
code="employee_availabilities.violation_spillover",
severity=AuditSeverity.ERROR,
source_id=self.id,
message=(
f"Employee is assigned to a night shift spilling over into a blocked day. "
f"employee_id={emp_id} date={date.isoformat()} shift_id={shift_id}."
),
date=tomorrow, # Das Datum des Regelbruchs ist der blockierte Tag
)
)

return tuple(findings)


# --- Helper Functions ---


def _parse_availabilities(
ctx: SolverContext | AuditContext,
) -> tuple[set[tuple[int, datetime.date]], dict[tuple[int, datetime.date], set[int]]]:
"""
Parses the new Availability Pydantic models into O(1) lookup structures.
"""
blocked_days: set[tuple[int, datetime.date]] = set()
allowed_shifts: defaultdict[tuple[int, datetime.date], set[int]] = defaultdict(set)

# Annahme: Availabilities sind Teil des Contexts oder Index
# (Passen Sie den Attributnamen ctx.availabilities entsprechend Ihrer Struktur an)
availabilities = ctx.dataset.availability

for avail in availabilities:
# Hier nutzen wir das Enum als String, um Abhängigkeiten gering zu halten
if str(avail.availability_type) == "available_only":
if avail.shift_ids:
allowed_shifts[(avail.employee_id, avail.date)].update(avail.shift_ids)
else:
blocked_days.add((avail.employee_id, avail.date))

return blocked_days, dict(allowed_shifts)
Loading
Loading