File size: 3,231 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 |
from datetime import datetime, date
from . import domain
# Conversion functions from domain to API models
def employee_to_model(employee: domain.Employee) -> domain.EmployeeModel:
return domain.EmployeeModel(
name=employee.name,
skills=list(employee.skills),
unavailable_dates=[d.isoformat() for d in employee.unavailable_dates],
undesired_dates=[d.isoformat() for d in employee.undesired_dates],
desired_dates=[d.isoformat() for d in employee.desired_dates],
)
def shift_to_model(shift: domain.Shift) -> domain.ShiftModel:
return domain.ShiftModel(
id=shift.id,
start=shift.start.isoformat(),
end=shift.end.isoformat(),
location=shift.location,
required_skill=shift.required_skill,
employee=employee_to_model(shift.employee) if shift.employee else None,
)
def schedule_to_model(
schedule: domain.EmployeeSchedule,
) -> domain.EmployeeScheduleModel:
return domain.EmployeeScheduleModel(
employees=[employee_to_model(e) for e in schedule.employees],
shifts=[shift_to_model(s) for s in schedule.shifts],
score=str(schedule.score) if schedule.score else None,
solver_status=schedule.solver_status.name if schedule.solver_status else None,
)
# Conversion functions from API models to domain
def model_to_employee(model: domain.EmployeeModel) -> domain.Employee:
return domain.Employee(
name=model.name,
skills=set(model.skills),
unavailable_dates={date.fromisoformat(d) for d in model.unavailable_dates},
undesired_dates={date.fromisoformat(d) for d in model.undesired_dates},
desired_dates={date.fromisoformat(d) for d in model.desired_dates},
)
def model_to_shift(model: domain.ShiftModel, employee_lookup: dict) -> domain.Shift:
# Handle employee reference
employee = None
if model.employee:
if isinstance(model.employee, str):
employee = employee_lookup[model.employee]
else:
employee = model_to_employee(model.employee)
return domain.Shift(
id=model.id,
start=datetime.fromisoformat(model.start),
end=datetime.fromisoformat(model.end),
location=model.location,
required_skill=model.required_skill,
employee=employee,
)
def model_to_schedule(model: domain.EmployeeScheduleModel) -> domain.EmployeeSchedule:
# Convert employees first
employees = [model_to_employee(e) for e in model.employees]
# Create lookup dictionary for employee references
employee_lookup = {e.name: e for e in employees}
# Convert shifts with employee lookups
shifts = [model_to_shift(s, employee_lookup) for s in model.shifts]
# Handle score
score = None
if model.score:
from solverforge_legacy.solver.score import HardSoftDecimalScore
score = HardSoftDecimalScore.parse(model.score)
# Handle solver status
solver_status = domain.SolverStatus.NOT_SOLVING
if model.solver_status:
solver_status = domain.SolverStatus[model.solver_status]
return domain.EmployeeSchedule(
employees=employees, shifts=shifts, score=score, solver_status=solver_status
)
|