# ============================================================ # CP-SAT 工序级模型黄金测试（方向 D） # 覆盖：C1 工艺先后序 / C2 同工位不重叠 / C10 换型计入 / C11 冻结窗固定 # ============================================================ from __future__ import annotations import itertools 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 = [(w["id"], w["plannedStartTime"], w["plannedEndTime"]) for w in frozen_wos] _run(world, engine="CP", freezeWindowHours=24.0) meta = _meta(world) assert meta.get("frozenCount", 0) >= 1, "应识别窗内冻结工单" # 冻结工单时间保持不变（不重排） for wo, (wid, st, en) in zip(frozen_wos, frozen_before): assert wo["plannedStartTime"] == st and wo["plannedEndTime"] == en frozen_ivs = [(s["workstationId"], s["startMin"], s["endMin"]) for s in (meta.get("operationSlots") or []) if s.get("isFrozen")] assert frozen_ivs 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")