File size: 3,281 Bytes
e510416 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 |
from solverforge_legacy.solver import SolverStatus
from solverforge_legacy.solver.domain import (
planning_entity,
planning_solution,
PlanningId,
PlanningVariable,
PlanningEntityCollectionProperty,
ProblemFactCollectionProperty,
ValueRangeProvider,
PlanningScore,
)
from solverforge_legacy.solver.score import HardSoftDecimalScore
from datetime import datetime, date
from typing import Annotated, List, Optional, Union
from dataclasses import dataclass, field
from .json_serialization import JsonDomainBase
from pydantic import Field
@dataclass
class Employee:
name: Annotated[str, PlanningId]
skills: set[str] = field(default_factory=set)
unavailable_dates: set[date] = field(default_factory=set)
undesired_dates: set[date] = field(default_factory=set)
desired_dates: set[date] = field(default_factory=set)
@planning_entity
@dataclass
class Shift:
id: Annotated[str, PlanningId]
start: datetime
end: datetime
location: str
required_skill: str
employee: Annotated[Employee | None, PlanningVariable] = None
def has_required_skill(self) -> bool:
"""Check if assigned employee has the required skill."""
if self.employee is None:
return False
return self.required_skill in self.employee.skills
def is_overlapping_with_date(self, dt: date) -> bool:
"""Check if shift overlaps with a specific date."""
return self.start.date() == dt or self.end.date() == dt
def get_overlapping_duration_in_minutes(self, dt: date) -> int:
"""Calculate overlap duration in minutes for a specific date."""
start_date_time = datetime.combine(dt, datetime.min.time())
end_date_time = datetime.combine(dt, datetime.max.time())
# Calculate overlap between date range and shift range
max_start_time = max(start_date_time, self.start)
min_end_time = min(end_date_time, self.end)
minutes = (min_end_time - max_start_time).total_seconds() / 60
return int(max(0, minutes))
@planning_solution
@dataclass
class EmployeeSchedule:
employees: Annotated[
list[Employee], ProblemFactCollectionProperty, ValueRangeProvider
]
shifts: Annotated[list[Shift], PlanningEntityCollectionProperty]
score: Annotated[HardSoftDecimalScore | None, PlanningScore] = None
solver_status: SolverStatus = SolverStatus.NOT_SOLVING
# Pydantic REST models for API (used for deserialization and context)
class EmployeeModel(JsonDomainBase):
name: str
skills: List[str] = Field(default_factory=list)
unavailable_dates: List[str] = Field(default_factory=list, alias="unavailableDates")
undesired_dates: List[str] = Field(default_factory=list, alias="undesiredDates")
desired_dates: List[str] = Field(default_factory=list, alias="desiredDates")
class ShiftModel(JsonDomainBase):
id: str
start: str # ISO datetime string
end: str # ISO datetime string
location: str
required_skill: str = Field(..., alias="requiredSkill")
employee: Union[str, EmployeeModel, None] = None
class EmployeeScheduleModel(JsonDomainBase):
employees: List[EmployeeModel]
shifts: List[ShiftModel]
score: Optional[str] = None
solver_status: Optional[str] = None
|