Spaces:
Runtime error
Runtime error
File size: 5,386 Bytes
50f82a1 | 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 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 | """
Converters for bidirectional transformation between domain objects and API models.
"""
from datetime import date
from typing import Dict, List, Optional
from solverforge_legacy.solver import SolverStatus
from solverforge_legacy.solver.score import HardSoftScore
from . import domain
# ************************************************************************
# Helper functions
# ************************************************************************
def date_to_iso(d: Optional[date]) -> Optional[str]:
"""Convert a date to ISO format string."""
return d.isoformat() if d else None
def iso_to_date(s: Optional[str]) -> Optional[date]:
"""Convert an ISO format string to a date."""
return date.fromisoformat(s) if s else None
# ************************************************************************
# Domain -> Model conversions
# ************************************************************************
def work_calendar_to_model(wc: domain.WorkCalendar) -> domain.WorkCalendarModel:
"""Convert a WorkCalendar domain object to its API model."""
return domain.WorkCalendarModel(
id=wc.id,
from_date=date_to_iso(wc.from_date),
to_date=date_to_iso(wc.to_date),
)
def crew_to_model(crew: domain.Crew) -> domain.CrewModel:
"""Convert a Crew domain object to its API model."""
return domain.CrewModel(id=crew.id, name=crew.name)
def job_to_model(job: domain.Job) -> domain.JobModel:
"""Convert a Job domain object to its API model."""
return domain.JobModel(
id=job.id,
name=job.name,
duration_in_days=job.duration_in_days,
min_start_date=date_to_iso(job.min_start_date),
max_end_date=date_to_iso(job.max_end_date),
ideal_end_date=date_to_iso(job.ideal_end_date),
tags=list(job.tags),
crew=crew_to_model(job.crew) if job.crew else None,
start_date=date_to_iso(job.start_date),
end_date=date_to_iso(job.get_end_date()),
)
def schedule_to_model(schedule: domain.MaintenanceSchedule) -> domain.MaintenanceScheduleModel:
"""Convert a MaintenanceSchedule domain object to its API model."""
return domain.MaintenanceScheduleModel(
work_calendar=work_calendar_to_model(schedule.work_calendar),
crews=[crew_to_model(c) for c in schedule.crews],
jobs=[job_to_model(j) for j in schedule.jobs],
start_date_range=[date_to_iso(d) for d in schedule.start_date_range],
score=str(schedule.score) if schedule.score else None,
solver_status=schedule.solver_status.name if schedule.solver_status else None,
)
# ************************************************************************
# Model -> Domain conversions
# ************************************************************************
def model_to_work_calendar(model: domain.WorkCalendarModel) -> domain.WorkCalendar:
"""Convert a WorkCalendarModel to its domain object."""
return domain.WorkCalendar(
id=model.id,
from_date=iso_to_date(model.from_date),
to_date=iso_to_date(model.to_date),
)
def model_to_crew(model: domain.CrewModel) -> domain.Crew:
"""Convert a CrewModel to its domain object."""
return domain.Crew(id=model.id, name=model.name)
def model_to_schedule(model: domain.MaintenanceScheduleModel) -> domain.MaintenanceSchedule:
"""
Convert a MaintenanceScheduleModel to its domain object.
Handles reference resolution for crew assignments.
"""
# Create work calendar
work_calendar = model_to_work_calendar(model.work_calendar)
# Create crews and lookup
crews: List[domain.Crew] = [model_to_crew(c) for c in model.crews]
crew_lookup: Dict[str, domain.Crew] = {c.id: c for c in crews}
# Create jobs with crew references resolved
jobs: List[domain.Job] = []
for job_model in model.jobs:
# Resolve crew reference
crew = None
if job_model.crew:
if isinstance(job_model.crew, str):
crew = crew_lookup.get(job_model.crew)
elif isinstance(job_model.crew, domain.CrewModel):
crew = crew_lookup.get(job_model.crew.id)
job = domain.Job(
id=job_model.id,
name=job_model.name,
duration_in_days=job_model.duration_in_days,
min_start_date=iso_to_date(job_model.min_start_date),
max_end_date=iso_to_date(job_model.max_end_date),
ideal_end_date=iso_to_date(job_model.ideal_end_date),
tags=set(job_model.tags) if job_model.tags else set(),
crew=crew,
start_date=iso_to_date(job_model.start_date) if job_model.start_date else None,
)
jobs.append(job)
# Parse start_date_range
start_date_range = (
[iso_to_date(d) for d in model.start_date_range]
if model.start_date_range
else []
)
# Parse score
score = None
if model.score:
score = HardSoftScore.parse(model.score)
# Parse solver status
solver_status = SolverStatus.NOT_SOLVING
if model.solver_status:
solver_status = SolverStatus[model.solver_status]
return domain.MaintenanceSchedule(
work_calendar=work_calendar,
crews=crews,
jobs=jobs,
start_date_range=start_date_range,
score=score,
solver_status=solver_status,
)
|