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
|