360 lines
16 KiB
Python
360 lines
16 KiB
Python
# ============================================================
|
||
# 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（冲突未消解）"
|