aps-agent/tests/golden/test_cp_operation_level.py

# ============================================================
# CP-SAT 工序级模型黄金测试（方向 D）
# 覆盖：C1 工艺先后序 / C2 同工位不重叠 / C10 换型计入 / C11 冻结窗固定
# ============================================================
from __future__ import annotations

import itertools
from copy import deepcopy

from server.engines import get_engine
from server.engines.base import EngineParams
from server.state.seed import seed_world
from server.timeutil import add_minutes, fmt_date, today0

DEFAULT_CONSTRAINTS = {
    "materialKit": True, "equipment": True, "personnel": True, "changeover": True,
    "capacity": True, "dueDate": True, "tooling": True, "freeze": True,
}


def _next_id_factory():
    counters: dict[str, int] = {}

    def next_id(kind: str) -> int:
        counters[kind] = counters.get(kind, 0) + 1
        return counters[kind]

    return next_id


def _run(world, engine="CP", time_limit=8.0, **params_kw):
    start = fmt_date(add_minutes(today0(), 24 * 60))
    base = {
        "orderIds": [], "engineType": engine, "strategyTemplate": "COMPREHENSIVE",
        "planningHorizonDays": 14, "startDate": start, "timeLimitSeconds": time_limit,
    }
    base.update(params_kw)
    return get_engine(engine).solve(world, EngineParams(**base), _next_id_factory())


def _meta(world):
    return world["scheduleVersions"][-1].get("solverMeta") or {}


def _new_slots(world):
    """CP 模型新排的工序槽位（排除冻结障碍）。"""
    return [s for s in (_meta(world).get("operationSlots") or []) if not s.get("isFrozen")]


def _steps_of(world, product_id):
    routing = next(r for r in world["routings"]
                   if r["productId"] == product_id and r["isDefault"])
    return sorted([s for s in world["routingSteps"] if s["routingId"] == routing["id"]],
                  key=lambda s: s["sequenceNo"])


def _force_single_line(world, line_id=1):
    """所有成品只能去一条产线，保证同线邻接排程（换型测试用）。"""
    pids = sorted({m["id"] for m in world["materials"] if m["type"] == "FINISHED_PRODUCT"})
    world["lineProducts"] = [
        {"id": 100 + i, "lineId": line_id, "productId": pid,
         "standardCapacity": 1000, "priority": 1, "setupTime": 30}
        for i, pid in enumerate(pids)
    ]


def test_operation_level_precedence():
    """C1：同订单后道工序 start >= 前道 end + 转移/等待。"""
    world = seed_world()
    so = next(s for s in world["salesOrders"] if s["items"][0]["productId"] == 1)
    result = _run(world, orderIds=[so["id"]])
    assert result.engineType == "CP"
    slots = _new_slots(world)
    assert slots, "期望至少一个工序槽位"
    steps = _steps_of(world, 1)
    assert len(slots) == len(steps), "工序数应与工艺路线一致"
    by_seq = sorted(slots, key=lambda s: s["sequenceNo"])
    step_by_op = {st["operationId"]: st for st in steps}
    for a, b in itertools.pairwise(by_seq):
        gap = int(step_by_op[a["operationId"]]["transferTime"]) + int(step_by_op[a["operationId"]]["waitTime"])
        assert b["startMin"] >= a["endMin"] + gap, (
            f"工序 {a['sequenceNo']}->{b['sequenceNo']} 违反先后序"
            f"（{b['startMin']} < {a['endMin']} + {gap}）")
    # 首工序顺序号应为 1
    assert by_seq[0]["sequenceNo"] == 1


def test_operation_level_no_workstation_overlap():
    """C2：同一工位/设备上的工序（跨订单）不重叠。"""
    world = seed_world()
    sos = [s for s in world["salesOrders"] if s["items"][0]["productId"] == 1][:2]
    _run(world, orderIds=[s["id"] for s in sos])
    by_ws: dict[int, list[tuple[int, int]]] = {}
    for s in _new_slots(world):
        by_ws.setdefault(s["workstationId"], []).append((s["startMin"], s["endMin"]))
    assert by_ws, "期望存在工位占用"
    assert len({s["orderIndex"] for s in _new_slots(world)}) == 2
    for ws_id, ivs in by_ws.items():
        ivs.sort()
        for (s1, e1), (s2, e2) in itertools.pairwise(ivs):
            assert e1 <= s2, f"工位 {ws_id} 出现工序重叠（{s1}-{e1} vs {s2}-{e2}）"


def test_operation_level_changeover_applied():
    """C10：不同产品族同线邻接时，换型矩阵 setup 计入工序序列。"""
    world = seed_world()
    _force_single_line(world, 1)
    # 各取一单：CTRL-STD（产品1）与 CTRL-HF（产品3），保证跨族
    sos = [next(s for s in world["salesOrders"] if s["items"][0]["productId"] == pid)
           for pid in (1, 3)]
    _run(world, orderIds=[s["id"] for s in sos])
    meta = _meta(world)
    first_ops = sorted([s for s in _new_slots(world) if s["sequenceNo"] == 1],
                       key=lambda s: s["startMin"])
    assert len(first_ops) == 2
    gap = first_ops[1]["startMin"] - first_ops[0]["endMin"]
    # 种子矩阵：CTRL-STD→CTRL-HF=45，CTRL-HF→CTRL-STD=40
    assert gap >= 40, f"相邻首工序间隙 {gap} 未计入换型 setup"
    assert meta.get("totalChangeoverMin", 0) >= 40
    assert sum((s.get("changeoverMin") or 0) for s in _new_slots(world)) >= 40
    # 同线：两条订单确实落在同一产线
    lines = {s["lineId"] for s in _new_slots(world)}
    assert len(lines) == 1


def test_operation_level_changeover_disabled():
    """C10 关闭：换型 setup 不生效，但同工位仍不重叠。"""
    world = seed_world()
    _force_single_line(world, 1)
    sos = [next(s for s in world["salesOrders"] if s["items"][0]["productId"] == pid)
           for pid in (1, 3)]
    constraints = dict(DEFAULT_CONSTRAINTS)
    constraints["changeover"] = False
    _run(world, orderIds=[s["id"] for s in sos], constraints=constraints)
    meta = _meta(world)
    assert meta.get("totalChangeoverMin", 0) == 0
    first_ops = [s for s in _new_slots(world) if s["sequenceNo"] == 1]
    assert len(first_ops) == 2
    first_ops.sort(key=lambda s: s["startMin"])
    assert first_ops[0]["endMin"] <= first_ops[1]["startMin"], "同工位首工序仍不得重叠"


def test_operation_level_freeze_window_fixes_scheduled_work():
    """C11：冻结窗内已排工单固定为障碍，新排工序不早于窗且不与冻结区间重叠。"""
    world = seed_world()
    _run(world, engine="RULE")  # 先生成一批已排工单
    wos = [w for w in world["workOrders"] if w.get("productionOrderId") is not None]
    assert wos
    frozen_wos = wos[:3]
    for wo in frozen_wos:
        wo["isFrozen"] = True
    frozen_before = {int(wo["id"]): deepcopy(wo) for wo in frozen_wos}
    production_versions = {
        int(row["id"]): row.get("schedulingVersionId")
        for row in world["productionOrders"]
    }

    _run(world, engine="CP", freezeWindowHours=24.0)
    meta = _meta(world)
    assert meta.get("frozenCount", 0) >= 1, "应识别窗内冻结工单"

    # 冻结工单完整业务对象保持不变（不仅是不重排时间）。
    for wo in frozen_wos:
        assert wo == frozen_before[int(wo["id"])]

    frozen_slots = [
        slot for slot in (meta.get("operationSlots") or []) if slot.get("isFrozen")
    ]
    frozen_ivs = [
        (slot["workstationId"], slot["startMin"], slot["endMin"])
        for slot in frozen_slots
    ]
    assert frozen_ivs
    for slot in frozen_slots:
        source = frozen_before[int(slot["sourceWorkOrderId"])]
        assert slot["sourceSchedulingVersionId"] == (
            source.get("schedulingVersionId")
            or production_versions[int(source["productionOrderId"])]
        )
        assert slot["orderNo"] == source["orderNo"]
        assert slot["lineId"] == source["lineId"]
        assert slot["workstationId"] == source["workstationId"]
    freeze_min = 24 * 60
    for s in _new_slots(world):
        assert s["startMin"] >= freeze_min, f"新排工序早于冻结窗末端：{s}"
        for ws_id, fs, fe in frozen_ivs:
            if ws_id == s["workstationId"]:
                assert s["endMin"] <= fs or s["startMin"] >= fe, (
                    f"工序 {s['orderNo']} 与冻结工单重叠（{fs}-{fe}）")


def test_operation_level_solver_meta_observable():
    """solverMeta 携带工序级模型信息与槽位明细。"""
    world = seed_world()
    result = _run(world)
    meta = _meta(world)
    assert meta.get("backend") == "OR-Tools CP-SAT"
    assert "operation-level" in str(meta.get("model") or "")
    assert isinstance(meta.get("operationSlots"), list)
    assert len(meta.get("operationSlots") or []) >= 4
    assert result.solveStatus in ("OPTIMAL", "FEASIBLE")