295 lines
12 KiB
Python
295 lines
12 KiB
Python
# ============================================================
|
||
# 康尼柔性场景黄金测试：交期三区间 / 三模式 / 故障 / 模具寿命
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from server.aps_domain.flex import (
|
||
apply_equipment_fault, compare_sort_modes, insert_rush_order, simulate_due,
|
||
)
|
||
from server.engines import PoolEngine
|
||
from server.state.seed import seed_world
|
||
|
||
|
||
class _MemStore:
|
||
def __init__(self, data):
|
||
self.data = data
|
||
|
||
def next_id(self, kind: str) -> int:
|
||
key = f"_c_{kind}"
|
||
self.data[key] = self.data.get(key, 100) + 1
|
||
return self.data[key]
|
||
|
||
def save(self):
|
||
pass
|
||
|
||
|
||
def test_simulate_due_has_three_bands_and_gaps():
|
||
store = _MemStore(seed_world())
|
||
r = simulate_due(store, "PDU-UNIT", 500)
|
||
assert r["optimisticFinish"]
|
||
assert r["expectedFinish"]
|
||
assert r["pessimisticFinish"]
|
||
assert r["gaps"]
|
||
assert any(g.get("kind") for g in r["gaps"])
|
||
|
||
|
||
def test_compare_modes_five_rows():
|
||
import copy
|
||
|
||
world = seed_world()
|
||
eligible = [o for o in world["flexOrders"] if o["status"] in ("RELEASED", "CREATED")]
|
||
assert len(eligible) == 10
|
||
before = copy.deepcopy(world)
|
||
store = _MemStore(world)
|
||
cmp = compare_sort_modes(store, compress_due=True)
|
||
assert len(cmp["rows"]) == 5
|
||
modes = {r["sortMode"] for r in cmp["rows"]}
|
||
assert modes == {"ASC", "DESC", "BOTTLENECK", "KITTING_FIRST", "SKILL_FIRST"}
|
||
labels = {r["label"] for r in cmp["rows"]}
|
||
assert labels == {"正排", "倒排", "瓶颈锚", "齐套优先", "技能优先"}
|
||
for row in cmp["rows"]:
|
||
assert {"sortMode", "label", "totalTardiness", "onTimeCount", "orderCount",
|
||
"makespan", "avgUtilization", "conflictCount", "vlCount"} <= set(row)
|
||
assert row["orderCount"] == 10
|
||
assert store.data == before
|
||
|
||
quality = cmp["dataQuality"]
|
||
assert quality["eligibleOrderCount"] == 10
|
||
assert quality["kitStatusKnownCount"] == 0
|
||
assert quality["skillLevelKnownCount"] == 0
|
||
rows = {row["sortMode"]: row for row in cmp["rows"]}
|
||
assert rows["KITTING_FIRST"]["strategyStatus"] == "DEGRADED"
|
||
assert rows["KITTING_FIRST"]["strategyEvidence"][0]["code"] == "MISSING_KIT_STATUS"
|
||
assert rows["SKILL_FIRST"]["strategyStatus"] == "DEGRADED"
|
||
assert rows["SKILL_FIRST"]["strategyEvidence"][0]["code"] == "MISSING_SKILL_LEVEL"
|
||
assert "当前数据证据不足" in cmp["hint"]
|
||
|
||
|
||
def test_capacity_forecast_horizons():
|
||
from server.aps_domain.flex import capacity_analysis, run_flex_schedule
|
||
store = _MemStore(seed_world())
|
||
run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
|
||
cap = capacity_analysis(store.data)
|
||
assert cap["horizons"]
|
||
assert {h["key"] for h in cap["horizons"]} == {"h4", "d1", "w1"}
|
||
assert cap["pools"]
|
||
p0 = cap["pools"][0]
|
||
assert "forecast" in p0
|
||
assert set(p0["forecast"].keys()) >= {"h4", "d1", "w1"}
|
||
assert 0 <= p0["forecast"]["d1"]["utilization"]
|
||
assert p0["forecast"]["d1"]["alert"] in ("ok", "warn", "crit")
|
||
|
||
|
||
def test_mold_life_consumed_and_can_lock():
|
||
world = seed_world()
|
||
mold = next(m for m in world["flexMolds"] if m["code"] == "MOLD-HV-01")
|
||
before = mold["lifeUsed"]
|
||
mold["lifeTotal"] = before + 50 # 排一单就会到限
|
||
counters: dict[str, int] = {}
|
||
|
||
def nid(k: str) -> int:
|
||
counters[k] = counters.get(k, 0) + 1
|
||
return counters[k]
|
||
|
||
PoolEngine().solve(world, nid, sort_mode="BOTTLENECK")
|
||
assert mold["lifeUsed"] > before
|
||
assert mold["status"] == "LOCKED" or mold["lifeUsed"] >= mold["lifeTotal"]
|
||
|
||
|
||
def test_fault_weld_can_raise_no_capability():
|
||
store = _MemStore(seed_world())
|
||
# 先正常排一轮，再焊机故障
|
||
from server.aps_domain.flex import run_flex_schedule
|
||
run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
|
||
r = apply_equipment_fault(store, "WELD-01", status="MAINTENANCE", reschedule=True, actor="test")
|
||
assert r["equipment"]["afterStatus"] == "MAINTENANCE"
|
||
# 焊接单点：无备机 → L4 缩池重排
|
||
assert r.get("responseLevel") == "L4"
|
||
types = {c.get("conflictType") for c in r.get("conflicts") or []}
|
||
assert "NO_CAPABILITY" in types or r["schedule"]["conflictCount"] >= 0
|
||
|
||
|
||
def test_fault_press_prefers_l1_swap():
|
||
store = _MemStore(seed_world())
|
||
from server.aps_domain.flex import run_flex_schedule
|
||
run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
|
||
before_ver = len(store.data["flexScheduleVersions"])
|
||
r = apply_equipment_fault(store, "PRESS-01", status="MAINTENANCE", reschedule=True, actor="test")
|
||
assert r["equipment"]["afterStatus"] == "MAINTENANCE"
|
||
# 压接池有冗余：优先 L1，不新建版本
|
||
if r.get("responseLevel") == "L1":
|
||
assert r.get("swap", {}).get("swapped")
|
||
assert r.get("schedule") is None
|
||
assert len(store.data["flexScheduleVersions"]) == before_ver
|
||
eq = next(e for e in store.data["flexEquipment"] if e["code"] == "PRESS-01")
|
||
assert eq["status"] == "MAINTENANCE"
|
||
else:
|
||
# 极端拥挤时退化为 L4 亦可接受
|
||
assert r.get("responseLevel") == "L4"
|
||
assert r.get("schedule")
|
||
|
||
|
||
def test_rush_insert_order():
|
||
store = _MemStore(seed_world())
|
||
before = len(store.data["flexOrders"])
|
||
r = insert_rush_order(store, "PDU-UNIT", 80, actor="test")
|
||
assert len(store.data["flexOrders"]) == before + 1
|
||
assert r["order"]["orderNo"].startswith("RUSH-")
|
||
|
||
|
||
def test_pool_engine_kitting_and_skill_sort_modes():
|
||
import copy
|
||
|
||
base = seed_world()
|
||
orders = [o for o in base.get("flexOrders", []) if o["status"] in ("RELEASED", "CREATED")]
|
||
assert orders
|
||
for i, o in enumerate(orders):
|
||
o["kitStatus"] = "未完成" if i % 2 else "已完成"
|
||
o["requiredSkillLevel"] = "L2"
|
||
orders[1]["requiredSkillLevel"] = "L4"
|
||
|
||
w1 = copy.deepcopy(base)
|
||
r1 = PoolEngine().solve(w1, _MemStore(w1).next_id, sort_mode="KITTING_FIRST")
|
||
assert r1["orderCount"] == len(orders)
|
||
ready = {o["orderNo"] for o in orders if o["kitStatus"] == "已完成"}
|
||
assert w1["flexVirtualLines"][0]["orderNo"] in ready
|
||
|
||
w2 = copy.deepcopy(base)
|
||
r2 = PoolEngine().solve(w2, _MemStore(w2).next_id, sort_mode="SKILL_FIRST")
|
||
assert r2["orderCount"] == len(orders)
|
||
high = {o["orderNo"] for o in orders if o["requiredSkillLevel"] == "L4"}
|
||
assert w2["flexVirtualLines"][0]["orderNo"] in high
|
||
|
||
|
||
def test_compare_modes_uses_kitting_and_skill_when_evidence_complete():
|
||
world = seed_world()
|
||
orders = [o for o in world["flexOrders"] if o["status"] in ("RELEASED", "CREATED")]
|
||
assert len(orders) == 10
|
||
for i, order in enumerate(orders):
|
||
order["kitStatus"] = "已完成" if i % 2 == 0 else "未完成"
|
||
order["requiredSkillLevel"] = "L4" if i % 3 == 0 else "L2"
|
||
|
||
cmp = compare_sort_modes(_MemStore(world), compress_due=False)
|
||
rows = {row["sortMode"]: row for row in cmp["rows"]}
|
||
assert rows["KITTING_FIRST"]["strategyStatus"] == "READY"
|
||
assert rows["KITTING_FIRST"]["strategyEvidence"] == []
|
||
assert rows["SKILL_FIRST"]["strategyStatus"] == "READY"
|
||
assert rows["SKILL_FIRST"]["strategyEvidence"] == []
|
||
|
||
|
||
def test_compare_modes_returns_missing_capability_conflict_evidence():
|
||
world = seed_world()
|
||
for equipment in world["flexEquipment"]:
|
||
equipment["capabilities"] = [
|
||
code for code in equipment.get("capabilities", []) if code != "OP-WELD"
|
||
]
|
||
|
||
cmp = compare_sort_modes(_MemStore(world), compress_due=False)
|
||
assert "OP-WELD" in cmp["dataQuality"]["uncoveredOperationCodes"]
|
||
assert any(w["code"] == "UNCOVERED_OPERATION_CAPABILITY" for w in cmp["warnings"])
|
||
for row in cmp["rows"]:
|
||
assert row["resultStatus"] == "PARTIAL"
|
||
assert "NO_CAPABILITY" in row["conflictTypes"]
|
||
assert any(c["conflictType"] == "NO_CAPABILITY" for c in row["conflictEvidence"])
|
||
|
||
|
||
def test_compare_modes_marks_shared_placeholder_capability_as_partial():
|
||
world = seed_world()
|
||
required_ops = sorted({
|
||
routing["operationCode"] for routing in world["flexRoutings"]
|
||
if routing.get("operationCode")
|
||
})
|
||
for equipment in world["flexEquipment"]:
|
||
equipment["capabilities"] = []
|
||
world["flexEquipment"][0]["capabilities"] = required_ops
|
||
|
||
cmp = compare_sort_modes(_MemStore(world), compress_due=False)
|
||
quality = cmp["dataQuality"]
|
||
assert quality["capabilityMappedEquipmentCount"] == 1
|
||
assert quality["unmappedEquipmentCount"] == quality["equipmentCount"] - 1
|
||
assert quality["uncoveredOperationCodes"] == []
|
||
assert quality["status"] == "PARTIAL"
|
||
assert any(w["code"] == "PARTIAL_EQUIPMENT_CAPABILITY" for w in cmp["warnings"])
|
||
|
||
|
||
def test_compare_modes_preserves_kangni_resource_gap_counts():
|
||
import copy
|
||
|
||
world = seed_world()
|
||
required_ops = sorted({
|
||
routing["operationCode"] for routing in world["flexRoutings"]
|
||
if routing.get("operationCode")
|
||
})
|
||
shared_capabilities = required_ops + [
|
||
f"KNI-OP-{i:02d}" for i in range(1, 27 - len(required_ops))
|
||
]
|
||
shared = copy.deepcopy(world["flexEquipment"][0])
|
||
shared.update({
|
||
"id": 1000,
|
||
"code": "EQ-SHARED-CG-01",
|
||
"name": "共享占位设备",
|
||
"capabilities": shared_capabilities,
|
||
"opStdTime": {code: 1 for code in shared_capabilities},
|
||
"adaptableMolds": [],
|
||
"status": "RUNNING",
|
||
})
|
||
world["flexEquipment"] = [shared] + [{
|
||
"id": 1000 + i,
|
||
"code": f"EQ-KNI-{i:02d}",
|
||
"name": f"待映射设备{i:02d}",
|
||
"capabilities": [],
|
||
"opStdTime": {},
|
||
"adaptableMolds": [],
|
||
"availabilityRate": 1,
|
||
"status": "RUNNING",
|
||
} for i in range(1, 41)]
|
||
world["flexMolds"] = [{
|
||
"id": i,
|
||
"code": f"MOLD-KNI-{i:02d}",
|
||
"name": f"待映射模具{i:02d}",
|
||
"operationCode": None,
|
||
"adaptableEquipment": [],
|
||
"lifeTotal": None,
|
||
"lifeUsed": 0,
|
||
"status": "AVAILABLE",
|
||
} for i in range(1, 41)]
|
||
|
||
cmp = compare_sort_modes(_MemStore(world), compress_due=False)
|
||
quality = cmp["dataQuality"]
|
||
assert quality["equipmentCount"] == 41
|
||
assert quality["capabilityMappedEquipmentCount"] == 1
|
||
assert quality["capabilityMappedEquipmentCodes"] == ["EQ-SHARED-CG-01"]
|
||
assert quality["unmappedEquipmentCount"] == 40
|
||
assert quality["moldCount"] == 40
|
||
assert quality["fullyMappedMoldCount"] == 0
|
||
assert quality["status"] == "PARTIAL"
|
||
assert quality["productionReady"] is False
|
||
warning_codes = {warning["code"] for warning in cmp["warnings"]}
|
||
assert {"PARTIAL_EQUIPMENT_CAPABILITY", "INCOMPLETE_MOLD_MAPPING"} <= warning_codes
|
||
assert "设备能力仅覆盖 1/41 台设备" in cmp["hint"]
|
||
assert "模具完整映射仅覆盖 0/40 套" in cmp["hint"]
|
||
|
||
|
||
def test_compare_modes_treats_unknown_kit_and_uniform_skill_as_degraded_evidence():
|
||
world = seed_world()
|
||
orders = [o for o in world["flexOrders"] if o["status"] in ("RELEASED", "CREATED")]
|
||
assert len(orders) == 10
|
||
for order in orders:
|
||
order["kitStatus"] = "未知"
|
||
order["requiredSkillLevel"] = "L3"
|
||
orders[0]["kitStatus"] = "已完成"
|
||
|
||
cmp = compare_sort_modes(_MemStore(world), compress_due=False)
|
||
quality = cmp["dataQuality"]
|
||
assert quality["kitStatusKnownCount"] == 1
|
||
assert quality["kitStatusDistinctCount"] == 1
|
||
assert quality["skillLevelKnownCount"] == 10
|
||
assert quality["skillLevelDistinctCount"] == 1
|
||
rows = {row["sortMode"]: row for row in cmp["rows"]}
|
||
assert rows["KITTING_FIRST"]["strategyStatus"] == "PARTIAL"
|
||
assert rows["KITTING_FIRST"]["strategyEvidence"][0]["code"] == "PARTIAL_KIT_STATUS"
|
||
assert rows["SKILL_FIRST"]["strategyStatus"] == "DEGRADED"
|
||
assert rows["SKILL_FIRST"]["strategyEvidence"][0]["code"] == "NO_SKILL_LEVEL_VARIATION"
|
||
warning_codes = {warning["code"] for warning in cmp["warnings"]}
|
||
assert {"MISSING_KIT_STATUS", "NO_SKILL_LEVEL_VARIATION"} <= warning_codes
|