115 lines
3.9 KiB
Python
115 lines
3.9 KiB
Python
|
"""Optimize scheduling engine integrated with the APS V2 closed loop.
|
||
|
|
|
||
|
|
The engine owns algorithm selection and provenance. APS still owns the world,
|
||
|
|
admission, candidate validation, version materialization, and audit trail.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from typing import Any, Callable
|
||
|
|
|
||
|
|
from server.contracts import ScheduleResult
|
||
|
|
from server.engines.base import EngineParams, ISchedulingEngine
|
||
|
|
from server.engines.pool_engine import PoolEngine
|
||
|
|
|
||
|
|
|
||
|
|
DISPATCH_RULES = frozenset({"EDD", "SPT", "PRIORITY", "FIFO", "LPT", "CR", "ATC"})
|
||
|
|
|
||
|
|
|
||
|
|
def normalize_dispatch_rule(value: str | None) -> str:
|
||
|
|
rule = str(value or "EDD").strip().upper().replace("-", "_")
|
||
|
|
aliases = {
|
||
|
|
"DELIVERY_FIRST": "EDD",
|
||
|
|
"EARLIEST_DUE_DATE": "EDD",
|
||
|
|
"FIRST_IN_FIRST_OUT": "FIFO",
|
||
|
|
"APPARENT_TARDINESS_COST": "ATC",
|
||
|
|
}
|
||
|
|
rule = aliases.get(rule, rule)
|
||
|
|
return rule if rule in DISPATCH_RULES else "EDD"
|
||
|
|
|
||
|
|
|
||
|
|
class OptimizeEngine(ISchedulingEngine):
|
||
|
|
"""Python-native Optimize entry point for APS scheduling.
|
||
|
|
|
||
|
|
The first integration reuses PoolEngine's already validated flex
|
||
|
|
materializer. Its ordering policy is supplied by ``dispatch_rule`` so the
|
||
|
|
seven optimize rules share APS calendars, teams, tooling, and rollback
|
||
|
|
semantics while the V2 runtime remains the authority for validation.
|
||
|
|
"""
|
||
|
|
|
||
|
|
name = "OPTIMIZE"
|
||
|
|
supports_anytime = False
|
||
|
|
|
||
|
|
def solve(
|
||
|
|
self,
|
||
|
|
world: dict[str, Any],
|
||
|
|
params: EngineParams,
|
||
|
|
next_id: Callable[[str], int],
|
||
|
|
) -> ScheduleResult:
|
||
|
|
rule = normalize_dispatch_rule(params.strategyTemplate)
|
||
|
|
solved = PoolEngine().solve(
|
||
|
|
world,
|
||
|
|
next_id,
|
||
|
|
sort_mode="ASC",
|
||
|
|
order_ids=params.orderIds or None,
|
||
|
|
start_date=params.startDate,
|
||
|
|
name=params.name,
|
||
|
|
window=None,
|
||
|
|
enforce_teams=params.constraints.get("personnel") if params.constraints else None,
|
||
|
|
dispatch_rule=rule,
|
||
|
|
)
|
||
|
|
return _summary_to_result(solved, rule)
|
||
|
|
|
||
|
|
def solve_flex(
|
||
|
|
self,
|
||
|
|
world: dict[str, Any],
|
||
|
|
next_id: Callable[[str], int],
|
||
|
|
*,
|
||
|
|
dispatch_rule: str | None = None,
|
||
|
|
order_ids: list[int] | None = None,
|
||
|
|
start_date: str | None = None,
|
||
|
|
name: str | None = None,
|
||
|
|
window: str | None = None,
|
||
|
|
enforce_teams: bool | None = None,
|
||
|
|
) -> dict[str, Any]:
|
||
|
|
"""Materialize an Optimize candidate for the closed-loop V2 adapter."""
|
||
|
|
|
||
|
|
rule = normalize_dispatch_rule(dispatch_rule)
|
||
|
|
solved = PoolEngine().solve(
|
||
|
|
world,
|
||
|
|
next_id,
|
||
|
|
sort_mode="ASC",
|
||
|
|
order_ids=order_ids,
|
||
|
|
start_date=start_date,
|
||
|
|
name=name,
|
||
|
|
window=window,
|
||
|
|
enforce_teams=enforce_teams,
|
||
|
|
dispatch_rule=rule,
|
||
|
|
)
|
||
|
|
solved.update({
|
||
|
|
"engineType": "OPTIMIZE",
|
||
|
|
"algorithmId": f"optimize.{rule.lower()}",
|
||
|
|
"algorithmVersion": "1.0.0",
|
||
|
|
"solverId": "optimize-dispatch",
|
||
|
|
"solverVersion": "1.0.0",
|
||
|
|
"dispatchRule": rule,
|
||
|
|
})
|
||
|
|
return solved
|
||
|
|
|
||
|
|
|
||
|
|
def _summary_to_result(solved: dict[str, Any], rule: str) -> ScheduleResult:
|
||
|
|
return ScheduleResult(
|
||
|
|
versionId=int(solved["versionId"]),
|
||
|
|
versionNo=str(solved["versionNo"]),
|
||
|
|
engineType="OPTIMIZE",
|
||
|
|
strategy=rule,
|
||
|
|
status="DRAFT",
|
||
|
|
orderCount=int(solved.get("orderCount") or 0),
|
||
|
|
poCount=int(solved.get("vlCount") or 0),
|
||
|
|
woCount=int(solved.get("woCount") or 0),
|
||
|
|
conflictCount=int(solved.get("conflictCount") or 0),
|
||
|
|
totalTardiness=float(solved.get("totalTardiness") or 0),
|
||
|
|
avgUtilization=float(solved.get("avgUtilization") or 0),
|
||
|
|
evidenceRefs=[f"algorithm:optimize.{rule.lower()}", f"run:{solved['versionId']}"],
|
||
|
|
solveStatus="FEASIBLE" if not solved.get("conflictCount") else "PARTIAL",
|
||
|
|
)
|