101 lines
4.7 KiB
Python
101 lines
4.7 KiB
Python
|
"""Round 89 slice 1: V2 原生 CP-SAT 求解器（SchedulingProblemV2 -> SchedulingSolutionV2）。
|
|||
|
|
|
|||
|
|
本文件只验证「问题 -> 解」这一段：CP-SAT 自己决定资源分配和时序，解必须通过
|
|||
|
|
APS 的独立 V2 校验，且总拖期不劣于同口径的 EDD 基线。物化到 flex* 行仍由 APS
|
|||
|
|
既有通道负责（slice 2）。
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import math
|
|||
|
|
from datetime import date, timedelta
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
import pytest
|
|||
|
|
|
|||
|
|
try: # 环境缺 OR-Tools，或 venv 内 NumPy 基线与该机 CPU 不兼容（当前 sd-server 即如此）
|
|||
|
|
from ortools.sat.python import cp_model # noqa: F401
|
|||
|
|
except Exception as exc: # pragma: no cover - 环境分支
|
|||
|
|
pytest.skip(f"OR-Tools 在此环境不可用：{type(exc).__name__}: {exc}", allow_module_level=True)
|
|||
|
|
|
|||
|
|
from server.aps_domain.closed_loop_problem import build_closed_loop_problem # noqa: E402
|
|||
|
|
from server.aps_domain.closed_loop_runtime import closed_loop_to_problem_v2 # noqa: E402
|
|||
|
|
from server.aps_domain.scheduling_problem_v2 import SchedulingProblemV2, SolveStatus # noqa: E402
|
|||
|
|
from server.aps_domain.scheduling_validator import validate_solution # noqa: E402
|
|||
|
|
from server.engines.optimize_cpsat import solve_problem_v2 # noqa: E402
|
|||
|
|
from server.state.packs import load_pack # noqa: E402
|
|||
|
|
|
|||
|
|
PACK_PATH = Path(__file__).resolve().parents[2] / "server" / "data" / "packs" / "optimize-simulation-v1.json"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _pack_problem() -> tuple[dict, SchedulingProblemV2]:
|
|||
|
|
world = load_pack(str(PACK_PATH))
|
|||
|
|
closed_loop = build_closed_loop_problem(world, business_date=date.today().isoformat(), strict=True)
|
|||
|
|
return world, closed_loop_to_problem_v2(world, closed_loop)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _edd_baseline_tardiness(problem: SchedulingProblemV2) -> float:
|
|||
|
|
"""同口径基线：完全按交期排序、每台设备串行占用的贪心解总拖期（分钟）。"""
|
|||
|
|
|
|||
|
|
base = problem.planningStart
|
|||
|
|
order_index = {requirement.requirementId: idx for idx, requirement in enumerate(
|
|||
|
|
sorted(problem.requirements, key=lambda row: row.requiredAt))}
|
|||
|
|
activities_by_requirement: dict[str, list] = {}
|
|||
|
|
for activity in problem.activities:
|
|||
|
|
activities_by_requirement.setdefault(activity.requirementId, []).append(activity)
|
|||
|
|
|
|||
|
|
cursor: dict[str, timedelta] = {}
|
|||
|
|
end_of: dict[str, timedelta] = {}
|
|||
|
|
total = 0.0
|
|||
|
|
for requirement_id in sorted(activities_by_requirement, key=lambda rid: order_index[rid]):
|
|||
|
|
requirement = next(row for row in problem.requirements if row.requirementId == requirement_id)
|
|||
|
|
for activity in sorted(activities_by_requirement[requirement_id], key=lambda row: row.sequence):
|
|||
|
|
duration = timedelta(minutes=math.ceil(activity.durationMin))
|
|||
|
|
earliest = max(
|
|||
|
|
[timedelta(0)]
|
|||
|
|
+ [end_of[predecessor] for predecessor in activity.predecessorActivityIds if predecessor in end_of]
|
|||
|
|
)
|
|||
|
|
resource_id = sorted(activity.eligibleResourceIds)[0]
|
|||
|
|
start = max(earliest, cursor.get(resource_id, timedelta(0)))
|
|||
|
|
end = start + duration
|
|||
|
|
cursor[resource_id] = end
|
|||
|
|
end_of[activity.activityId] = end
|
|||
|
|
completion = max(
|
|||
|
|
(end_of[activity.activityId] for activity in activities_by_requirement[requirement_id]),
|
|||
|
|
default=timedelta(0),
|
|||
|
|
)
|
|||
|
|
total += max(0.0, (base + completion - requirement.requiredAt).total_seconds() / 60.0)
|
|||
|
|
return total
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_native_cpsat_schedules_every_pack_activity_and_passes_v2_validator():
|
|||
|
|
world, problem = _pack_problem()
|
|||
|
|
|
|||
|
|
outcome = solve_problem_v2(problem, time_limit_seconds=15.0)
|
|||
|
|
solution = outcome.solution
|
|||
|
|
|
|||
|
|
assert solution.solveStatus in (SolveStatus.OPTIMAL, SolveStatus.FEASIBLE)
|
|||
|
|
assert len(solution.activities) == len(problem.activities) == 72
|
|||
|
|
assert solution.unscheduledRequirements == ()
|
|||
|
|
assert outcome.meta["solverId"] == "optimize-cpsat"
|
|||
|
|
|
|||
|
|
activity_by_id = {activity.activityId: activity for activity in problem.activities}
|
|||
|
|
for row in solution.activities:
|
|||
|
|
activity = activity_by_id[row.activityId]
|
|||
|
|
assert row.resourceId in activity.eligibleResourceIds # 只能用声明的合格设备
|
|||
|
|
assert row.start >= problem.planningStart
|
|||
|
|
span = (row.end - row.start).total_seconds() / 60.0
|
|||
|
|
assert activity.durationMin <= span <= activity.durationMin + 1 # 工时按分钟向上取整
|
|||
|
|
|
|||
|
|
report = validate_solution(problem, solution, world=world)
|
|||
|
|
assert report.valid is True
|
|||
|
|
assert not report.hardViolations
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_native_cpsat_is_not_worse_than_the_edd_baseline_in_the_same_metric():
|
|||
|
|
_, problem = _pack_problem()
|
|||
|
|
|
|||
|
|
outcome = solve_problem_v2(problem, time_limit_seconds=15.0)
|
|||
|
|
baseline = _edd_baseline_tardiness(problem)
|
|||
|
|
|
|||
|
|
assert outcome.solution.objectiveValues["totalTardiness"] <= baseline + 0.001
|