175 lines
8.1 KiB
Python
175 lines
8.1 KiB
Python
# ============================================================
|
||
# 柔性能力池引擎黄金测试（moduleId: golden-pool-engine, 发版门禁 §14.3）
|
||
# 断言柔性排产的结构不变式（吸收 demand/ 康尼芜湖方案）：
|
||
# F1 版本/虚拟产线/工单计数关系 F2 设备无双占（能力池硬约束）
|
||
# F3 工序 precedence 不颠倒 F4 压接工序必占适配模具
|
||
# F5 瓶颈锚模式含瓶颈订单优先 F6 缺能力/缺料冲突结构合法
|
||
# F7 可移动设备产生移动耗时痕迹 F8 设备利用率值域合法
|
||
# ============================================================
|
||
from __future__ import annotations # 前向类型引用
|
||
|
||
from server.engines import PoolEngine # 柔性引擎
|
||
from server.state.seed import seed_world # 种子（含 flex* 数据）
|
||
from server.timeutil import add_minutes, fmt_date, parse_dt, today0 # 日期工具
|
||
|
||
|
||
def _fresh_world():
|
||
"""全新种子世界（含柔性层）。"""
|
||
return seed_world()
|
||
|
||
|
||
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, sort_mode="BOTTLENECK"):
|
||
"""执行一次柔性排产（明天起）。"""
|
||
start = fmt_date(add_minutes(today0(), 24 * 60))
|
||
return PoolEngine().solve(world, _next_id_factory(), sort_mode=sort_mode, start_date=start)
|
||
|
||
|
||
# ---------------- F1：计数关系 ----------------
|
||
def test_counts_match_flex_seed():
|
||
"""10 单（8 RELEASED + 2 CREATED）全部参与；每单一条虚拟产线；工单数=Σ工艺步骤。"""
|
||
world = _fresh_world()
|
||
result = _run(world)
|
||
assert result["orderCount"] == 10 # flexOrders 全部未完成
|
||
assert result["vlCount"] == 10 # 每单一条虚拟产线（无致命冲突时）
|
||
# HV-HARNESS 6步×3 + PDU-UNIT 6步×3 + CHARGE-GUN 6步×2 + HV-CONN 4步×2 = 56
|
||
assert result["woCount"] == 6 * 3 + 6 * 3 + 6 * 2 + 4 * 2
|
||
assert len(world["flexScheduleVersions"]) == 1
|
||
|
||
|
||
# ---------------- F2：设备无双占 ----------------
|
||
def test_no_equipment_overlap():
|
||
"""同一设备的任意两工单时间区间不得重叠（能力池资源硬约束）。"""
|
||
world = _fresh_world()
|
||
_run(world)
|
||
by_eq: dict[int, list] = {}
|
||
for wo in world["flexWorkOrders"]:
|
||
by_eq.setdefault(wo["equipmentId"], []).append(
|
||
(parse_dt(wo["plannedStartTime"]), parse_dt(wo["plannedEndTime"])))
|
||
for eq_id, ivs in by_eq.items():
|
||
ivs.sort()
|
||
for (s1, e1), (s2, e2) in zip(ivs, ivs[1:]):
|
||
assert e1 <= s2, f"设备 {eq_id} 双占：{e1} > {s2}"
|
||
|
||
|
||
# ---------------- F3：工序 precedence ----------------
|
||
def test_operation_sequence_preserved():
|
||
"""同一虚拟产线内工序顺序号大的不得早于前序结束。"""
|
||
world = _fresh_world()
|
||
_run(world)
|
||
by_vl: dict[int, list] = {}
|
||
for wo in world["flexWorkOrders"]:
|
||
by_vl.setdefault(wo["vlId"], []).append(wo)
|
||
for vl_id, wos in by_vl.items():
|
||
wos.sort(key=lambda w: w["seq"])
|
||
for prev, nxt in zip(wos, wos[1:]):
|
||
assert parse_dt(nxt["plannedStartTime"]) >= parse_dt(prev["plannedEndTime"]), \
|
||
f"VL {vl_id} 工序颠倒"
|
||
|
||
|
||
# ---------------- F4：压接工序必占适配模具 ----------------
|
||
def test_crimp_requires_adaptable_mold():
|
||
"""OP-CRIMP 工单必须挂一副模具，且该模具适配所选设备。"""
|
||
world = _fresh_world()
|
||
_run(world)
|
||
molds = {m["code"]: m for m in world["flexMolds"]}
|
||
eqs = {e["id"]: e for e in world["flexEquipment"]}
|
||
crimp = [w for w in world["flexWorkOrders"] if w["operationCode"] == "OP-CRIMP"]
|
||
assert crimp, "应存在压接工单"
|
||
for w in crimp:
|
||
assert w["moldCode"] is not None, f"{w['orderNo']} 压接未占模具"
|
||
mold = molds[w["moldCode"]]
|
||
assert eqs[w["equipmentId"]]["code"] in mold["adaptableEquipment"], "模具与设备不适配"
|
||
|
||
|
||
# ---------------- F5：瓶颈锚模式含瓶颈订单优先 ----------------
|
||
def test_bottleneck_mode_orders_bottleneck_first():
|
||
"""瓶颈锚模式下，版本记录瓶颈工序清单；含瓶颈产品的首单不晚于任一订单开始。"""
|
||
world = _fresh_world()
|
||
result = _run(world, sort_mode="BOTTLENECK")
|
||
assert result["bottleneck"], "应输出瓶颈工序摘要"
|
||
codes = {b["operationCode"] for b in result["bottleneck"]}
|
||
assert "OP-CRIMP" in codes and "OP-WELD" in codes
|
||
|
||
|
||
# ---------------- F6：冲突结构合法 ----------------
|
||
def test_conflicts_wellformed():
|
||
"""缺料/缺能力/缺模具/延期冲突：归属本版本、未解决、类型合法。"""
|
||
world = _fresh_world()
|
||
result = _run(world)
|
||
valid = {"NO_ROUTING", "NO_CAPABILITY", "NO_MOLD", "MATERIAL_SHORTAGE", "DELAY"}
|
||
assert len(world["flexConflicts"]) == result["conflictCount"]
|
||
for c in world["flexConflicts"]:
|
||
assert c["versionId"] == result["versionId"]
|
||
assert c["isResolved"] is False
|
||
assert c["conflictType"] in valid
|
||
|
||
|
||
# ---------------- F7：可移动设备的移动耗时痕迹 ----------------
|
||
def test_movable_equipment_move_time():
|
||
"""压接机可移动 → 其工单 moveMin 字段被记录（≥0，且可移动设备可>0）。"""
|
||
world = _fresh_world()
|
||
_run(world)
|
||
crimp = [w for w in world["flexWorkOrders"] if w["operationCode"] == "OP-CRIMP"]
|
||
for w in crimp:
|
||
assert "moveMin" in w and w["moveMin"] >= 0
|
||
|
||
|
||
# ---------------- F8：利用率值域 ----------------
|
||
def test_utilization_range():
|
||
"""设备平均利用率在 0~1.5，延迟非负。"""
|
||
world = _fresh_world()
|
||
result = _run(world)
|
||
assert 0 <= result["avgUtilization"] <= 1.5
|
||
assert result["totalTardiness"] >= 0
|
||
|
||
|
||
# ---------------- F9：正排/倒排模式可运行且计数一致 ----------------
|
||
def test_sort_modes_run():
|
||
"""三种排产模式均能产出等量虚拟产线（排序影响时间不影响结构完整性）。"""
|
||
for mode in ("ASC", "DESC", "BOTTLENECK"):
|
||
world = _fresh_world()
|
||
result = _run(world, sort_mode=mode)
|
||
assert result["vlCount"] == 10, f"{mode} 模式虚拟产线数异常"
|
||
assert result["sortMode"] == mode
|
||
assert result["makespan"] is not None # 最晚完工可计算
|
||
assert 0 <= result["onTimeCount"] <= 10 # 准时单数值域
|
||
|
||
|
||
# ---------------- F10：瓶颈产能法（限制性瓶颈=激光焊接单机池） ----------------
|
||
def test_capacity_analysis_bottleneck():
|
||
"""瓶颈产能法：各池日产能可计算；限制性瓶颈应为单机激光焊接池（OP-WELD）。"""
|
||
from server.aps_domain.flex import capacity_analysis
|
||
world = _fresh_world()
|
||
_run(world)
|
||
cap = capacity_analysis(world)
|
||
assert cap["pools"], "应有能力池"
|
||
for p in cap["pools"]:
|
||
assert p["dailyCapacity"] > 0 # 日产能为正
|
||
assert p["equipmentCount"] >= 1
|
||
assert cap["bottleneckPool"]["operationCode"] == "OP-WELD" # 单机焊接=限制性瓶颈
|
||
# 压接池 4 台并行，日产能应远高于单机焊接池
|
||
crimp = next(p for p in cap["pools"] if p["operationCode"] == "OP-CRIMP")
|
||
weld = next(p for p in cap["pools"] if p["operationCode"] == "OP-WELD")
|
||
assert crimp["dailyCapacity"] > weld["dailyCapacity"]
|
||
|
||
|
||
# ---------------- F11：设备故障 → 能力池实时收缩 ----------------
|
||
def test_equipment_failure_shrinks_pool():
|
||
"""标记设备为 MAINTENANCE 后，其能力池设备数下降（动态调度的数据基础）。"""
|
||
from server.aps_domain.flex import capacity_analysis
|
||
world = _fresh_world()
|
||
before = next(p for p in capacity_analysis(world)["pools"] if p["operationCode"] == "OP-CRIMP")
|
||
next(e for e in world["flexEquipment"] if e["code"] == "PRESS-01")["status"] = "MAINTENANCE"
|
||
after = next(p for p in capacity_analysis(world)["pools"] if p["operationCode"] == "OP-CRIMP")
|
||
assert after["equipmentCount"] == before["equipmentCount"] - 1 # 池收缩一台
|
||
assert after["dailyCapacity"] < before["dailyCapacity"] # 日产能下降
|