Backend api

src/main.py

@@ -12,7 +12,11 @@
 from scheduling.timeoffice.facts import TIMEOFFICE_FACTS
 from scheduling.timeoffice.reading.container import TimeOfficeReaders
 from scheduling.timeoffice.service import TimeOfficeService
+from scheduling.timeoffice.writing.demand import TimeOfficeDemandWriter
+from scheduling.timeoffice.writing.objective_weights import TimeOfficeWeightsWriter
+from scheduling.timeoffice.writing.roster import TimeOfficeAvailabilityWriter
 from scheduling.timeoffice.writing.solution import TimeOfficeSolutionWriter
+from scheduling.timeoffice.writing.wishes import TimeOfficeWishWriter
 
 
 def _parse_date(value: str) -> date:
@@ -47,6 +51,12 @@ def _solve(
         engine=engine,
         readers=TimeOfficeReaders.create(facts=facts),
         solution_writer=TimeOfficeSolutionWriter(),
+        wish_writer=TimeOfficeWishWriter(
+            target_planning_status_id=facts.target_planning_status_id,
+        ),
+        demand_writer=TimeOfficeDemandWriter(),
+        objective_weights_writer=TimeOfficeWeightsWriter(),
+        availability_writer=TimeOfficeAvailabilityWriter(),
     )
     solver = SolverService(
         settings=settings,

src/scheduling/api/app.py

@@ -8,7 +8,11 @@
 from scheduling.api.dependencies import ApiRuntime
 from scheduling.api.solve.job_store import InMemorySolveJobStore
 from scheduling.api.solve.router import solve_router
-from scheduling.api.web.router import web_router
+from scheduling.api.web.employee_router import employee_router
+from scheduling.api.web.minimal_staff_router import minimal_staff_router
+from scheduling.api.web.schedule_router import schedule_router
+from scheduling.api.web.weights_router import weights_router
+from scheduling.api.web.wishes_availabilities_router import wishes_and_availabilities_router
 from scheduling.logging import configure_logging
 from scheduling.settings import get_settings
 from scheduling.solver.cp_sat.builder import create_cp_sat_model_builder
@@ -17,7 +21,11 @@
 from scheduling.timeoffice.facts import TIMEOFFICE_FACTS
 from scheduling.timeoffice.reading.container import TimeOfficeReaders
 from scheduling.timeoffice.service import TimeOfficeService
+from scheduling.timeoffice.writing.demand import TimeOfficeDemandWriter
+from scheduling.timeoffice.writing.objective_weights import TimeOfficeWeightsWriter
+from scheduling.timeoffice.writing.roster import TimeOfficeAvailabilityWriter
 from scheduling.timeoffice.writing.solution import TimeOfficeSolutionWriter
+from scheduling.timeoffice.writing.wishes import TimeOfficeWishWriter
 
 settings = get_settings()
 configure_logging(level=settings.log_level)
@@ -38,6 +46,12 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
             engine=engine,
             readers=TimeOfficeReaders.create(facts=facts),
             solution_writer=TimeOfficeSolutionWriter(),
+            wish_writer=TimeOfficeWishWriter(
+                target_planning_status_id=facts.target_planning_status_id,
+            ),
+            demand_writer=TimeOfficeDemandWriter(),
+            objective_weights_writer=TimeOfficeWeightsWriter(),
+            availability_writer=TimeOfficeAvailabilityWriter(),
         ),
         solver_service=SolverService(
             settings=settings,
@@ -54,8 +68,12 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
 
 
 app = FastAPI(title="Staff Scheduling API", lifespan=lifespan)
+app.include_router(employee_router)
+app.include_router(weights_router)
+app.include_router(wishes_and_availabilities_router)
+app.include_router(minimal_staff_router)
+app.include_router(schedule_router)
 app.include_router(solve_router)
-app.include_router(web_router)
 
 
 @app.get("/status")

src/scheduling/api/solve/router.py

@@ -77,8 +77,20 @@ def kernel(command: SolveCommand) -> Solution:
         logger.info("Solving scheduling dataset for solve job: job_id=%s", job.job_id)
         solution = solver.solve(dataset)
 
-        logger.info("Running TimeOffice writeback dry-run for solve job: job_id=%s", job.job_id)
-        timeoffice.write_solution_dry_run(solution)
+        start_date = command.planning_month.start.isoformat()
+        end_date = command.planning_month.end.isoformat()
+        for unit in command.planning_unit_ids:
+            solution_name = f"solution_{unit}_{start_date}-{end_date}_wdefault"
+            logger.info(
+                "Writing legacy solution for solve job: job_id=%s solution_name=%s",
+                job.job_id,
+                solution_name,
+            )
+            timeoffice.write_solution_legacy_format(
+                dataset=dataset,
+                solution=solution,
+                solution_name=solution_name,
+            )
 
         return solution
 

src/scheduling/api/timeoffice/schemas.py

@@ -0,0 +1,16 @@
+from typing import Any
+
+from pydantic import Field
+
+from scheduling.domain import PlanningMonth, SchedulingBaseModel
+
+
+class DBRequest(SchedulingBaseModel):
+    planning_unit_id: int
+    year: int = Field(ge=2000, le=2100)
+    month: int = Field(ge=1, le=12)
+
+    def planning_month(self) -> PlanningMonth:
+        return PlanningMonth(year=self.year, month=self.month)
+
+    solution_data: dict[str, Any] | None = None

src/scheduling/api/web/router.py → src/scheduling/api/web/employee_router.py

@@ -11,10 +11,10 @@
 logger = logging.getLogger(__name__)
 
 
-web_router = APIRouter()
+employee_router = APIRouter()
 
 
-@web_router.get("/employees")
+@employee_router.get("/employees")
 async def get_employees(
     planning_unit: int,
     from_date: date,

src/scheduling/api/web/minimal_staff_router.py

@@ -0,0 +1,185 @@
+import logging
+from datetime import date, timedelta
+from typing import Annotated
+
+from fastapi import APIRouter, Depends
+
+from scheduling.api.dependencies import get_timeoffice_service
+from scheduling.api.web.schemas import SuccessResponse, UpdateMinimalStaffRequest
+from scheduling.domain import DemandRequirement, PlanningMonth, StaffLevel
+from scheduling.timeoffice.facts import TIMEOFFICE_FACTS
+from scheduling.timeoffice.service import TimeOfficeService
+
+logger = logging.getLogger(__name__)
+
+
+minimal_staff_router = APIRouter()
+
+DAYS_OF_WEEK = ("Mo", "Di", "Mi", "Do", "Fr", "Sa", "So")
+
+FRONTEND_STAFF_LEVEL_BY_DOMAIN = {
+    StaffLevel.TRAINEE: "Azubi",
+    StaffLevel.PROFESSIONAL: "Fachkraft",
+    StaffLevel.ASSISTANT: "Hilfskraft",
+}
+
+DOMAIN_STAFF_LEVEL_BY_FRONTEND = {
+    "Azubi": StaffLevel.TRAINEE,
+    "Fachkraft": StaffLevel.PROFESSIONAL,
+    "Hilfskraft": StaffLevel.ASSISTANT,
+}
+
+FRONTEND_STAFF_LEVELS = ("Azubi", "Fachkraft", "Hilfskraft")
+
+FRONTEND_SHIFT_CODES = ("F", "S", "N", "Z")
+
+WEEKDAY_TO_ID = {
+    "Mo": 1,
+    "Di": 2,
+    "Mi": 3,
+    "Do": 4,
+    "Fr": 5,
+    "Sa": 6,
+    "So": 7,
+}
+
+
+@minimal_staff_router.get("/minimal-staff")
+async def get_minimal_staff(
+    planning_unit: int,
+    from_date: date,
+    timeoffice: Annotated[TimeOfficeService, Depends(get_timeoffice_service)],
+) -> dict[str, dict[str, dict[str, int]]]:
+    month = PlanningMonth(year=from_date.year, month=from_date.month)
+
+    dataset = timeoffice.fetch_dataset(
+        planning_unit_ids=(planning_unit,),
+        planning_month=month,
+    )
+
+    return _minimal_staff_to_frontend(dataset.demand_requirements)
+
+
+def _minimal_staff_to_frontend(
+    demand_requirements: tuple[DemandRequirement, ...],
+) -> dict[str, dict[str, dict[str, int]]]:
+    output = _empty_minimal_staff_response()
+
+    for requirement in demand_requirements:
+        staff_level = FRONTEND_STAFF_LEVEL_BY_DOMAIN[requirement.staff_level]
+        day = DAYS_OF_WEEK[requirement.date.isoweekday() - 1]
+        shift_code = _shift_code_from_shift_id(requirement.shift_id)
+
+        output[staff_level][day][shift_code] = requirement.required_count
+
+    return output
+
+
+def _empty_minimal_staff_response() -> dict[str, dict[str, dict[str, int]]]:
+    return {
+        staff_level: {day: dict.fromkeys(FRONTEND_SHIFT_CODES, 0) for day in DAYS_OF_WEEK}
+        for staff_level in FRONTEND_STAFF_LEVELS
+    }
+
+
+def _shift_code_from_shift_id(shift_id: int) -> str:
+    shift_fact = TIMEOFFICE_FACTS.reference_shift_facts_by_id.get(shift_id)
+
+    if shift_fact is None:
+        raise ValueError(f"Unknown reference shift_id={shift_id}.")
+
+    if shift_fact.expected_code not in FRONTEND_SHIFT_CODES:
+        raise ValueError(f"Unsupported minimal staff shift code={shift_fact.expected_code!r} for shift_id={shift_id}.")
+
+    return shift_fact.expected_code
+
+
+@minimal_staff_router.put("/minimal-staff")
+async def put_minimal_staff(
+    planning_unit: int,
+    from_date: date,
+    request: UpdateMinimalStaffRequest,
+    timeoffice: Annotated[TimeOfficeService, Depends(get_timeoffice_service)],
+) -> SuccessResponse:
+    planning_month = PlanningMonth(year=from_date.year, month=from_date.month)
+
+    demand_requirements = _minimal_staff_request_to_domain(
+        planning_unit=planning_unit,
+        planning_month=planning_month,
+        minimal_staff=request.data,
+    )
+
+    timeoffice.replace_minimal_staffing(
+        planning_unit_id=planning_unit,
+        demand_requirements=demand_requirements,
+    )
+
+    return SuccessResponse()
+
+
+def _minimal_staff_request_to_domain(
+    *,
+    planning_unit: int,
+    planning_month: PlanningMonth,
+    minimal_staff: dict[str, dict[str, dict[str, int]]],
+) -> tuple[DemandRequirement, ...]:
+    requirements: list[DemandRequirement] = []
+
+    for frontend_staff_level, demand_by_day in minimal_staff.items():
+        staff_level = DOMAIN_STAFF_LEVEL_BY_FRONTEND[frontend_staff_level]
+
+        for frontend_day, demand_by_shift in demand_by_day.items():
+            weekday_iso = WEEKDAY_TO_ID[frontend_day]
+
+            for shift_code, minimum_count in demand_by_shift.items():
+                if minimum_count < 0:
+                    raise ValueError("Minimal staff count must be >=0")
+
+                if minimum_count == 0:
+                    continue
+
+                shift_id = _shift_id_from_shift_code(shift_code)
+
+                for requirement_date in _dates_for_weekday(
+                    planning_month=planning_month,
+                    weekday_iso=weekday_iso,
+                ):
+                    requirements.append(
+                        DemandRequirement(
+                            planning_unit_id=planning_unit,
+                            date=requirement_date,
+                            shift_id=shift_id,
+                            staff_level=staff_level,
+                            required_count=minimum_count,
+                        )
+                    )
+
+    return tuple(requirements)
+
+
+def _shift_id_from_shift_code(shift_code: str) -> int:
+    if shift_code not in FRONTEND_SHIFT_CODES:
+        raise ValueError(f"Shift Code is not defined in frontend: shift_code={shift_code}.")
+
+    for shift_id, shift_fact in TIMEOFFICE_FACTS.reference_shift_facts_by_id.items():
+        if shift_fact.expected_code == shift_code:
+            return shift_id
+
+    raise ValueError(f"Unknown shift_code={shift_code}.")
+
+
+def _dates_for_weekday(
+    *,
+    planning_month: PlanningMonth,
+    weekday_iso: int,
+) -> tuple[date, ...]:
+    dates: list[date] = []
+
+    current_date = planning_month.start
+    while current_date <= planning_month.end:
+        if current_date.isoweekday() == weekday_iso:
+            dates.append(current_date)
+
+        current_date += timedelta(days=1)
+
+    return tuple(dates)

src/scheduling/api/web/schedule_router.py

@@ -0,0 +1,38 @@
+import json
+import logging
+from datetime import date
+from pathlib import Path
+from typing import Any
+
+from fastapi import APIRouter
+
+logger = logging.getLogger(__name__)
+
+PROCESSED_SOLUTIONS_DIR = Path("processed_solutions")
+
+schedule_router = APIRouter()
+
+# TODO: change solve router to use the lagacy solution writer already present in timeoffice
+# TODO: adapt the get_schedule acordingly,
+# TODO: use _get_solution_legacy to fetch the data so, when the db works, we only need to relace that one function
+
+
+@schedule_router.get("/schedules/{schedule_id}")
+async def get_schedule(
+    planning_unit: int,
+    from_date: date,
+    schedule_id: str,
+) -> Any:
+    """Return a schedule solution for a planning unit and month."""
+    return _get_solution_legacy(schedule_id)
+
+
+def _get_solution_legacy(schedule_id: str) -> Any:
+    """Read a processed solution from the legacy JSON files, place holder until database read write."""
+    solution_path = PROCESSED_SOLUTIONS_DIR / f"{schedule_id}_processed.json"
+
+    if not solution_path.is_file():
+        return {"solution": None}
+
+    with solution_path.open(encoding="utf-8") as f:
+        return json.load(f)