aps-agent/tests/golden/test_cp_cumulative.py

# ============================================================
# CP-SAT Cumulative 班组/工装聚合容量 + gap/时限/性能基线（方向 V，矩阵 84）
# 覆盖：
#   - C12 Cumulative 聚合容量：同一班组（teamId）/工装（toolingId）在时间上
#     并行的工序占用总和 ≤ 可用容量（model.AddCumulative，demand=1/工序）；
#     硬约束开关 C12（params.personnel/tooling + 世界剖面 C12_team/C12_tooling）
#     控制启用；开启时并行超容被约束消解，关闭时行为与既有一致。
#   - solverMeta.cumulative：enabled / capacity / resources（含 peakConcurrent）
#     / unwired（引用了资源但主数据无可用容量 → 诚实报告，不静默建约束）。
#   - gap：保留 obj-vs-bound 计算并输出 objective/bestBound/gap；
#     gap=None 语义：OR-Tools 仅可行性时（UNKNOWN/INFEASIBLE，无 bound）不输出
#     objective/bestBound，gap 置 None 并保留启发式序 fallback（代码与测试文档化）。
#   - 时限：timeLimitSeconds 生效；确定性超时用例返回 FEASIBLE/OPTIMAL +
#     solverMeta.status/timeLimitSec/wallTimeSec。
#   - 性能基线：固定算例（含班组/工装聚合容量冲突）断言 <30s 壁钟、可解
#     （OPTIMAL 或 FEASIBLE + 无硬违反）、Cumulative 冲突被消除或如实报告。
# ============================================================
from __future__ import annotations

import copy

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

# 世界约束剖面默认全开（C12_team/C12_tooling 默认 enabled=True）
DEFAULT_CONSTRAINTS = {
    "materialKit": True, "equipment": True, "personnel": True, "changeover": True,
    "capacity": False, "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, constraints=None, order_ids=None):
    start = fmt_date(add_minutes(today0(), 24 * 60))
    base = {
        "orderIds": order_ids or [], "engineType": engine, "strategyTemplate": "COMPREHENSIVE",
        "planningHorizonDays": 14, "startDate": start, "timeLimitSeconds": time_limit,
        "constraints": constraints or dict(DEFAULT_CONSTRAINTS),
    }
    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 _restrict_two_lines(world):
    """产品1 → 产线1、产品3 → 产线3：让两份订单落在不同工位（跨工位共享资源）。"""
    keep = [(row["productId"], row["lineId"])
            for row in world["lineProducts"]
            if (row["productId"] == 1 and row["lineId"] == 1)
            or (row["productId"] == 3 and row["lineId"] == 3)]
    world["lineProducts"] = [
        row for row in world["lineProducts"]
        if (row["productId"], row["lineId"]) in keep
    ]


def _wire_cumulative_world(team=True, tooling=True, team_cap=1, tooling_cap=1):
    """两份订单（产品1→线1，产品3→线3）：

    - WS001（线1 SMT）与 WS010（线3 SMT）共享 teamId=1（班组）
    - WS002（线1 DIP）与 WS011（线3 DIP）共享 toolingId=1（工装）
    默认容量 1：两个跨工位并行工序必须被 Cumulative 约束串行化。
    """
    world = seed_world()
    _restrict_two_lines(world)
    for ws in world["workstations"]:
        if ws["code"] in ("WS001", "WS010"):   # 两条产线的 SMT 工位
            ws["teamId"] = 1
        if ws["code"] in ("WS002", "WS011"):   # 两条产线的 DIP 工位
            ws["toolingId"] = 1
    if team:
        teams = [t for t in world["teams"] if t["id"] == 1]
        if teams:
            teams[0]["memberCount"] = team_cap
        else:
            world["teams"].append({"id": 1, "code": "T001", "name": "SMT班组", "memberCount": team_cap})
    if tooling:
        world["toolings"] = [{
            "id": 1, "code": "M001", "name": "DIP治具", "availableCount": tooling_cap,
        }]
    return world


def _two_order_ids(world):
    pid1 = next(s["id"] for s in world["salesOrders"] if s["items"][0]["productId"] == 1)
    pid3 = next(s["id"] for s in world["salesOrders"] if s["items"][0]["productId"] == 3)
    return [pid1, pid3]


def _peak_overlap(intervals: list[tuple[int, int]]) -> int:
    """区间集 [start, end) 最大同时重叠数（端点相切不算重叠）。"""
    events: list[tuple[int, int]] = []
    for s, e in intervals:
        events.append((s, 1))
        events.append((e, -1))
    events.sort(key=lambda ev: (ev[0], ev[1]))
    cur = peak = 0
    for _, d in events:
        cur += d
        peak = max(peak, cur)
    return peak


def _independent_peak_by_resource(world, slots) -> dict[tuple[str, int], int]:
    """从物化槽位独立重算每班组/工装的并行峰值（不依赖 solverMeta 的 peakConcurrent）。"""
    by_ws = {ws["id"]: ws for ws in world["workstations"]}
    ivs: dict[tuple[str, int], list[tuple[int, int]]] = {}
    for s in slots:
        ws = by_ws.get(s.get("workstationId"))
        if not ws:
            continue
        for kind, key in (("team", "teamId"), ("tooling", "toolingId")):
            rid = ws.get(key)
            if rid is not None:
                ivs.setdefault((kind, int(rid)), []).append((s["startMin"], s["endMin"]))
    return {k: _peak_overlap(v) for k, v in ivs.items()}


# ---------------- C12 Cumulative 聚合容量 ----------------

def test_team_cumulative_serializes_parallel_capacity_conflict():
    """C12_team：跨工位共享班组且容量=1 → 并行 SMT 工序被串行化（约束消解）。"""
    world = _wire_cumulative_world(team=True, tooling=False, team_cap=1)
    result = _run(world, order_ids=_two_order_ids(world))
    meta = _meta(world)
    assert result.solveStatus in ("OPTIMAL", "FEASIBLE")

    cum = meta.get("cumulative") or {}
    assert cum["enabled"]["team"] is True
    team_res = [r for r in cum.get("resources", []) if r["kind"] == "team"]
    assert len(team_res) == 1, f"应接线 1 个班组资源，实际 {team_res}"
    assert team_res[0]["id"] == 1
    assert team_res[0]["capacity"] == 1
    assert team_res[0]["intervalCount"] == 2
    assert team_res[0]["peakConcurrent"] == 1, "峰值并行不得超过班组容量"

    smt = sorted([s for s in _new_slots(world) if s["sequenceNo"] == 1],
                 key=lambda s: s["startMin"])
    assert len(smt) == 2
    # 两个 SMT 工序位于不同工位（WS001/WS010），仍不得时间重叠
    assert smt[0]["endMin"] <= smt[1]["startMin"], (
        f"跨工位共享班组工序未串行化：{smt[0]} vs {smt[1]}")


def test_tooling_cumulative_serializes_parallel_capacity_conflict():
    """C12_tooling：跨工位共享工装且可用数量=1 → 并行 DIP 工序被串行化。"""
    world = _wire_cumulative_world(team=False, tooling=True, tooling_cap=1)
    result = _run(world, order_ids=_two_order_ids(world))
    meta = _meta(world)
    assert result.solveStatus in ("OPTIMAL", "FEASIBLE")

    cum = meta.get("cumulative") or {}
    tool_res = [r for r in cum.get("resources", []) if r["kind"] == "tooling"]
    assert len(tool_res) == 1
    assert tool_res[0]["id"] == 1
    assert tool_res[0]["capacity"] == 1
    assert tool_res[0]["intervalCount"] == 2
    assert tool_res[0]["peakConcurrent"] == 1

    dip = sorted([s for s in _new_slots(world) if s["sequenceNo"] == 2],
                 key=lambda s: s["startMin"])
    assert len(dip) == 2
    assert dip[0]["endMin"] <= dip[1]["startMin"], (
        f"跨工位共享工装工序未串行化：{dip[0]} vs {dip[1]}")


def test_cumulative_disabled_preserves_existing_parallel_behavior():
    """C12 关闭（personnel/tooling=False）：不建 Cumulative 约束，行为与既有一致
    （跨工位并行工序可自由重叠）。"""
    world = _wire_cumulative_world(team=True, tooling=True, team_cap=1, tooling_cap=1)
    flags = dict(DEFAULT_CONSTRAINTS)
    flags["personnel"] = False
    flags["tooling"] = False
    result = _run(world, order_ids=_two_order_ids(world), constraints=flags)
    meta = _meta(world)
    assert result.solveStatus in ("OPTIMAL", "FEASIBLE")

    cum = meta.get("cumulative") or {}
    assert cum["enabled"] == {"team": False, "tooling": False}
    assert cum.get("resources") == [], "关闭后不得接线任何累计资源"

    smt = sorted([s for s in _new_slots(world) if s["sequenceNo"] == 1],
                 key=lambda s: s["startMin"])
    assert len(smt) == 2
    assert smt[0]["startMin"] == smt[1]["startMin"] == 0, (
        "关闭后两份订单应保持既有的并行最早开始行为")
    assert smt[0]["endMin"] > smt[1]["startMin"], "关闭后并行重叠应保留"


def test_cumulative_unwired_reference_reported():
    """引用了班组/工装但主数据无可用容量 → unwired 如实报告，不静默建约束。"""
    world = seed_world()
    _restrict_two_lines(world)
    for ws in world["workstations"]:
        if ws["code"] in ("WS001", "WS010"):
            ws["teamId"] = 99          # 不存在于 world["teams"]
    result = _run(world, order_ids=_two_order_ids(world))
    meta = _meta(world)
    assert result.solveStatus in ("OPTIMAL", "FEASIBLE")
    cum = meta.get("cumulative") or {}
    assert cum.get("resources") == []
    assert {"kind": "team", "id": 99} in cum.get("unwired", [])


def test_team_and_tooling_cumulative_combined_and_independent_peaks():
    """班组+工装同时接线：solverMeta 峰值与从槽位独立重算一致且 ≤ 容量。"""
    world = _wire_cumulative_world(team=True, tooling=True, team_cap=2, tooling_cap=2)
    result = _run(world, order_ids=_two_order_ids(world))
    meta = _meta(world)
    assert result.solveStatus in ("OPTIMAL", "FEASIBLE")

    cum = meta.get("cumulative") or {}
    kinds = {r["kind"] for r in cum.get("resources", [])}
    assert kinds == {"team", "tooling"}

    independent = _independent_peak_by_resource(world, _new_slots(world))
    for r in cum.get("resources", []):
        key = (r["kind"], r["id"])
        assert r["peakConcurrent"] <= r["capacity"], (
            f"{r['kind']}#{r['id']} 峰值超容量：{r['peakConcurrent']} > {r['capacity']}")
        assert independent[key] == r["peakConcurrent"], (
            f"solverMeta 峰值与槽位独立重算不一致：{key}")
        assert independent[key] <= r["capacity"]


# ---------------- gap / 时限 / 性能基线 ----------------

def _hard_world(n_orders=12, qty=1200, team_cap=2, tooling_cap=2):
    """固定算例：n 份订单放大数量 + 班组/工装聚合容量冲突（容量 2 < 并行需求）。"""
    world = seed_world()
    for so in world["salesOrders"]:
        for it in so["items"]:
            it["quantity"] = qty
    base = list(world["salesOrders"])
    mid = max(s["id"] for s in world["salesOrders"])
    for i in range(n_orders - len(base)):
        src = copy.deepcopy(base[i % len(base)])
        mid += 1
        src["id"] = mid
        src["orderNo"] = f"HRD-{mid}"
        src["items"] = [dict(src["items"][0], quantity=qty)]
        world["salesOrders"].append(src)
    for ws in world["workstations"]:
        if ws["code"] in ("WS001", "WS006", "WS010"):
            ws["teamId"] = 1
        if ws["code"] in ("WS002", "WS007", "WS011"):
            ws["toolingId"] = 1
    for t in world["teams"]:
        if t["id"] == 1:
            t["memberCount"] = team_cap
    world["toolings"] = [{
        "id": 1, "code": "M001", "name": "DIP治具", "availableCount": tooling_cap,
    }]
    return world


def test_gap_best_bound_metadata_optimal_and_feasible():
    """gap：OPTIMAL → gap=0 且 bestBound==objective；FEASIBLE → 数值 gap 与
    |obj-bound|/obj 一致且 bestBound 输出。"""
    # 小算例：必然 OPTIMAL（gap=0，bound 闭合）
    world = seed_world()
    _run(world, order_ids=[world["salesOrders"][0]["id"]], time_limit=8.0)
    meta = _meta(world)
    assert meta["status"] == "OPTIMAL"
    assert meta["gap"] == 0.0
    assert float(meta["bestBound"]) == float(meta["objective"])

    # 硬算例 + 短时限：FEASIBLE（有解但未证明最优）→ 数值 gap + bestBound
    world2 = _hard_world(n_orders=10, qty=1200)
    _run(world2, time_limit=0.5)
    meta2 = _meta(world2)
    assert meta2["status"] == "FEASIBLE"
    assert meta2["timeLimitSec"] == 0.5
    obj, bound, gap = float(meta2["objective"]), float(meta2["bestBound"]), float(meta2["gap"])
    assert obj > 0
    assert 0 <= bound <= obj
    assert 0.0 <= gap <= 1.0
    assert abs(gap - round(abs(obj - bound) / obj, 6)) < 1e-6, (
        "gap 必须是 obj-vs-bound 相对差")

    # gap=None 语义文档化：仅可行性（UNKNOWN/INFEASIBLE，无 bound）时置 None，
    # 见 cp_engine.py 非可行分支注释与模块 docstring。


def test_time_limit_deterministic_feasible_timeout():
    """时限：0.5s 确定性超时 → 返回当前可行解 + solverMeta.status=FEASIBLE/OPTIMAL，
    且 wallTimeSec 受控（不吞掉时限，也不远超时限）。"""
    world = _hard_world()
    result = _run(world, time_limit=0.5)
    meta = _meta(world)
    assert meta["timeLimitSec"] == 0.5
    assert meta["status"] in ("FEASIBLE", "OPTIMAL")
    assert result.solveStatus in ("FEASIBLE", "OPTIMAL")
    assert meta["wallTimeSec"] < 5.0, f"0.5s 时限用例壁钟 {meta['wallTimeSec']}s 超界"
    assert meta["objective"] is not None
    assert meta["gap"] is not None
    assert "bestBound" in meta
    # 确定性：同一固定算例再跑一次，anytime 求解器契约稳定（状态/时限元数据
    # 一致；并行搜索下可行解目标允许机器/调度差异，不做位级相同断言）
    world2 = _hard_world()
    _run(world2, time_limit=0.5)
    meta2 = _meta(world2)
    assert meta2["status"] in ("FEASIBLE", "OPTIMAL")
    assert meta2["timeLimitSec"] == 0.5
    assert meta2["wallTimeSec"] < 5.0
    assert meta2["objective"] is not None and meta2["gap"] is not None
    assert "bestBound" in meta2


def test_performance_baseline_wall_time_solvable_no_hard_violation():
    """性能基线：固定算例（含班组/工装聚合容量冲突）<30s 壁钟、可解
    （OPTIMAL 或 FEASIBLE + 无硬违反）、Cumulative 冲突被消除或如实报告。"""
    world = _hard_world(n_orders=12, qty=1200)
    result = _run(world, time_limit=8.0)
    meta = _meta(world)
    assert meta["wallTimeSec"] < 30.0, f"性能基线超 30s 壁钟：{meta['wallTimeSec']}s"
    assert meta["status"] in ("OPTIMAL", "FEASIBLE")
    assert result.solveStatus in ("OPTIMAL", "FEASIBLE")

    # 无硬违反：solverMeta 峰值与槽位独立重算均 ≤ 容量（Cumulative 冲突已消解）
    cum = meta.get("cumulative") or {}
    assert {r["kind"] for r in cum.get("resources", [])} == {"team", "tooling"}
    independent = _independent_peak_by_resource(world, _new_slots(world))
    for r in cum.get("resources", []):
        key = (r["kind"], r["id"])
        assert r["peakConcurrent"] <= r["capacity"], (
            f"{r['kind']}#{r['id']} 峰值超容量：{r['peakConcurrent']} > {r['capacity']}")
        assert independent[key] <= r["capacity"]
        assert independent[key] == r["peakConcurrent"]

    # 明确冲突工序（SMT 共享班组 / DIP 共享工装）：任意时刻并行占用 ≤ 容量
    # （容量=2 时允许 2 道并行，但不得超容——独立重算已校验）
    smt = sorted([s for s in _new_slots(world) if s["sequenceNo"] == 1],
                 key=lambda s: s["startMin"])
    assert smt, "期望存在 SMT 工序"
    team_ivs = []
    for s in smt:
        ws = next(w for w in world["workstations"] if w["id"] == s["workstationId"])
        assert ws.get("teamId") == 1, f"SMT 工序应落在共享班组工位：{s}"
        team_ivs.append((s["startMin"], s["endMin"]))
    assert _peak_overlap(team_ivs) <= 2, "SMT 班组并行占用超过容量 2（冲突未消解）"