339 lines
13 KiB
Python
339 lines
13 KiB
Python
# ============================================================
|
||
# NSGA-II 多目标求解器黄金测试（矩阵 85，round-40 方向 U）
|
||
# 覆盖：非支配排序/拥挤度/交叉变异可行性/确定性/Pareto 解集性质/
|
||
# 参数模板 manifest/rank_scenarios 消费/大算例 GA 基线边界与性能
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from random import Random
|
||
|
||
import pytest
|
||
|
||
from server.agent_core.algolib import AlgorithmManifest
|
||
from server.aps_domain.scenario_selection import (
|
||
nsga2_solutions_to_cards,
|
||
rank_scenarios,
|
||
)
|
||
from server.engines import NSGA2Engine, get_engine
|
||
from server.engines.base import EngineParams
|
||
from server.engines.ga_engine import optimize_genetic_assignment
|
||
from server.engines.nsga2_engine import (
|
||
_base_individual,
|
||
_build_jobs,
|
||
_order_crossover,
|
||
_partially_mapped_crossover,
|
||
_resource_migration,
|
||
_swap_mutation,
|
||
build_nsga2_manifest,
|
||
crowding_distance,
|
||
fast_non_dominated_sort,
|
||
nsga2_defaults,
|
||
solve_nsga2,
|
||
)
|
||
from server.engines.rule_engine import RuleEngine
|
||
from server.state.seed import seed_world
|
||
from server.timeutil import add_minutes, fmt_date, today0
|
||
|
||
|
||
def _params(engine_type: str = "NSGA2", time_limit: float = 5.0) -> EngineParams:
|
||
return EngineParams(
|
||
engineType=engine_type,
|
||
strategyTemplate="COMPREHENSIVE",
|
||
planningHorizonDays=14,
|
||
startDate=fmt_date(add_minutes(today0(), 24 * 60)),
|
||
timeLimitSeconds=time_limit,
|
||
)
|
||
|
||
|
||
def _world_entries(world=None):
|
||
world = world if world is not None else seed_world()
|
||
entries, _, _ = RuleEngine().collect_and_order(world, _params("RULE"))
|
||
return world, entries
|
||
|
||
|
||
def _obj_tuple(solution: dict) -> tuple[float, int, float]:
|
||
objs = solution["objectives"]
|
||
return (float(objs["totalTardiness"]), int(objs["conflictCount"]), -float(objs["loadBalance"]))
|
||
|
||
|
||
def _dominates_left(left: dict, right: dict) -> bool:
|
||
a, b = _obj_tuple(left), _obj_tuple(right)
|
||
return all(av <= bv for av, bv in zip(a, b)) and any(av < bv for av, bv in zip(a, b))
|
||
|
||
|
||
def _assert_feasible_sequence(sequence, jobs) -> None:
|
||
counts = [0] * len(jobs)
|
||
for job_id in sequence:
|
||
counts[job_id] += 1
|
||
assert counts == [job.op_count for job in jobs]
|
||
# 同单工序沿工艺路线保序：出现次数即路由序（occurrence 隐含），染色体恒为合法工序序列
|
||
|
||
|
||
def _random_feasible_sequence(jobs, rng):
|
||
order = list(range(len(jobs)))
|
||
rng.shuffle(order)
|
||
return tuple(j for j in order for _ in range(jobs[j].op_count))
|
||
|
||
|
||
# ---------------- 非支配排序 / 拥挤度 ----------------
|
||
def test_fast_non_dominated_sort_returns_expected_fronts():
|
||
objectives = [
|
||
(1.0, 1.0, 0.0), # 0: 与 1/3 互不支配
|
||
(2.0, 0.0, 0.0), # 1: 与 0/3 互不支配
|
||
(3.0, 3.0, 0.0), # 2: 被 0 支配
|
||
(0.0, 2.0, 0.0), # 3: 与 0/1 互不支配
|
||
(2.0, 2.0, 0.0), # 4: 被 0/1/3 支配
|
||
]
|
||
fronts = fast_non_dominated_sort(objectives)
|
||
assert [sorted(f) for f in fronts] == [[0, 1, 3], [4], [2]]
|
||
assert fast_non_dominated_sort([]) == [[]]
|
||
|
||
|
||
def test_crowding_distance_boundary_infinite_and_interior_sum():
|
||
objectives = [
|
||
(0.0, 4.0, 0.0), # A 边界（obj1 最小）
|
||
(1.0, 2.0, 0.0), # B
|
||
(2.0, 1.0, 0.0), # C
|
||
(4.0, 0.0, 0.0), # D 边界（obj1 最大）
|
||
]
|
||
dist = crowding_distance(objectives, [0, 1, 2, 3])
|
||
assert math.isinf(dist[0]) and math.isinf(dist[3])
|
||
assert dist[1] == pytest.approx(1.25)
|
||
assert dist[2] == pytest.approx(1.25)
|
||
|
||
|
||
# ---------------- 交叉 / 变异可行性 ----------------
|
||
def test_order_crossover_and_pmx_preserve_operation_feasibility():
|
||
world, entries = _world_entries()
|
||
jobs = _build_jobs(world, entries, _params())
|
||
rng = Random(7)
|
||
for _ in range(120):
|
||
a = _random_feasible_sequence(jobs, rng)
|
||
b = _random_feasible_sequence(jobs, rng)
|
||
_assert_feasible_sequence(_order_crossover(a, b, rng), jobs)
|
||
_assert_feasible_sequence(_partially_mapped_crossover(a, b, rng), jobs)
|
||
mutated = _swap_mutation(a, rng)
|
||
_assert_feasible_sequence(mutated, jobs)
|
||
assert len(mutated) == len(a)
|
||
|
||
|
||
def test_pmx_child_preserves_job_occurrence_counts():
|
||
a = (0, 0, 1, 2, 1, 2)
|
||
b = (2, 1, 0, 2, 1, 0)
|
||
rng = Random(3)
|
||
child = _partially_mapped_crossover(a, b, rng)
|
||
assert sorted(child) == sorted(a)
|
||
assert [child.count(0), child.count(1), child.count(2)] == [2, 2, 2]
|
||
|
||
|
||
def test_resource_migration_switches_to_valid_alternate_line():
|
||
world, entries = _world_entries()
|
||
jobs = _build_jobs(world, entries, _params())
|
||
base = _base_individual(jobs)
|
||
assert any(len(job.line_options) > 1 for job in jobs), "演示算例应存在多产线订单"
|
||
seen_switch = False
|
||
for trial in range(300):
|
||
rng = Random(1000 + trial)
|
||
migrated = _resource_migration(base.lines, jobs, rng)
|
||
assert len(migrated) == len(base.lines)
|
||
for idx, line in enumerate(migrated):
|
||
assert line in jobs[idx].line_options
|
||
if migrated != base.lines:
|
||
seen_switch = True
|
||
assert seen_switch
|
||
|
||
|
||
# ---------------- 参数模板 ----------------
|
||
def test_nsga2_defaults_template_and_manifest_registration():
|
||
for key in (
|
||
"populationSize", "generations", "crossoverRate", "mutationRate",
|
||
"lineMutationRate", "crossover", "tournamentSize", "seed",
|
||
):
|
||
assert key in nsga2_defaults
|
||
assert nsga2_defaults["crossover"] in ("OX", "PMX")
|
||
assert 0.0 <= nsga2_defaults["crossoverRate"] <= 1.0
|
||
assert 0.0 <= nsga2_defaults["mutationRate"] <= 1.0
|
||
assert isinstance(nsga2_defaults["seed"], int)
|
||
|
||
manifest = build_nsga2_manifest()
|
||
model = AlgorithmManifest(**manifest)
|
||
assert model.validate_manifest() == []
|
||
assert manifest["parameters"] == nsga2_defaults
|
||
assert model.entrypoint == "server.engines.nsga2_engine:solve_nsga2"
|
||
assert model.deterministic is True
|
||
assert model.random_seed == nsga2_defaults["seed"]
|
||
assert model.category == "C"
|
||
assert "tests/golden/test_nsga2_engine.py" in model.golden_tests
|
||
|
||
|
||
# ---------------- 求解器：确定性 / Pareto 性质 ----------------
|
||
def test_solve_nsga2_deterministic_and_pareto_properties():
|
||
params = _params(time_limit=8.0)
|
||
outputs = []
|
||
for _ in range(2):
|
||
world, entries = _world_entries()
|
||
pareto, meta = solve_nsga2(world, params, entries=entries, seed=42)
|
||
outputs.append((pareto, meta))
|
||
(pareto_a, meta_a), (pareto_b, meta_b) = outputs
|
||
|
||
signature = lambda out: [
|
||
(s["objectives"], s["jobOrder"], s["lines"]) for s in out
|
||
]
|
||
assert signature(pareto_a) == signature(pareto_b)
|
||
assert meta_a["generations"] == meta_b["generations"]
|
||
assert meta_a["generations"] == meta_a["generationBudget"]
|
||
assert meta_a["backend"] == "NSGA-II"
|
||
assert meta_a["paretoSize"] == len(pareto_a)
|
||
|
||
assert len(pareto_a) >= 2, "Pareto 前沿应非空且有多解（多目标）"
|
||
for i, left in enumerate(pareto_a):
|
||
for j, right in enumerate(pareto_a):
|
||
if i != j:
|
||
assert not _dominates_left(left, right), "Pareto 前沿内不应互相支配"
|
||
kpi = left["kpi"]
|
||
assert kpi["totalTardiness"] >= 0
|
||
assert kpi["conflictCount"] >= 0
|
||
assert 0.0 <= kpi["loadBalance"] <= 1.0
|
||
assert 0.0 <= kpi["avgUtilization"] <= 1.0
|
||
assert left["hardFeasible"] is True
|
||
assert sorted(left["jobOrder"]) == list(range(len(entries)))
|
||
base = meta_a["baselineObjectives"]
|
||
assert base["conflictCount"] >= 0 and base["loadBalance"] >= 0
|
||
|
||
|
||
def test_solve_nsga2_time_budget_deterministic_and_bounded():
|
||
params = _params(time_limit=0.05)
|
||
outputs = []
|
||
for _ in range(2):
|
||
world, entries = _world_entries()
|
||
pareto, meta = solve_nsga2(world, params, entries=entries, seed=1)
|
||
outputs.append((pareto, meta))
|
||
(pareto_a, meta_a), (pareto_b, meta_b) = outputs
|
||
assert meta_a["generations"] == meta_b["generations"]
|
||
assert 1 <= meta_a["generations"] <= 5
|
||
assert meta_a["generationBudget"] == 5
|
||
assert [s["objectives"] for s in pareto_a] == [s["objectives"] for s in pareto_b]
|
||
|
||
|
||
# ---------------- scenario_selection 衔接 ----------------
|
||
def test_rank_scenarios_consumes_nsga2_pareto_candidates():
|
||
world, entries = _world_entries()
|
||
pareto, _ = solve_nsga2(world, _params(), entries=entries, seed=7)
|
||
cards = nsga2_solutions_to_cards(pareto)
|
||
assert len(cards) == len(pareto)
|
||
ranked = rank_scenarios(cards)
|
||
assert len(ranked) == len(cards)
|
||
assert all("normalizedKpi" in card and "weightedScore" in card for card in ranked)
|
||
assert any(card["isPareto"] and card["selectionStatus"] == "PARETO" for card in ranked)
|
||
recommended = [card for card in ranked if card["isRecommended"]]
|
||
assert len(recommended) == 1
|
||
assert recommended[0]["rank"] == 1
|
||
assert recommended[0]["isPareto"] is True
|
||
assert recommended[0]["source"]["algo"] == "NSGA-II"
|
||
|
||
# 混入被支配候选：rank_scenarios 应如实标为 DOMINATED（消费路径生效）
|
||
dominated = {
|
||
"scenarioId": "junk",
|
||
"strategy": "JUNK",
|
||
"hardFeasible": True,
|
||
"robustness": 0.1,
|
||
"kpi": {
|
||
"totalTardiness": 1e12,
|
||
"conflictCount": 10 ** 9,
|
||
"totalCost": 1e12,
|
||
"avgUtilization": 0.0,
|
||
"loadBalance": 0.0,
|
||
"totalChangeoverMin": 0.0,
|
||
},
|
||
}
|
||
mixed = rank_scenarios([*cards, dominated])
|
||
assert next(card for card in mixed if card["strategy"] == "JUNK")["selectionStatus"] == "DOMINATED"
|
||
|
||
|
||
# ---------------- 大算例质量/性能基线 ----------------
|
||
def test_medium_instance_pareto_not_worse_than_single_objective_ga_baseline():
|
||
world, entries = _world_entries()
|
||
big_entries = [dict(entry) for _ in range(5) for entry in entries] # 35 订单（30-50 区间）
|
||
assert len(big_entries) >= 30
|
||
params = _params(time_limit=30.0)
|
||
|
||
# 单目标 GA 基线（确定性种子 42）
|
||
ga_ordered, ga_meta = optimize_genetic_assignment(world, big_entries, params)
|
||
used: set[int] = set()
|
||
ga_order: list[int] = []
|
||
ga_lines: list[int] = []
|
||
for target in ga_ordered:
|
||
probe = {k: v for k, v in target.items() if k != "forcedLineId"}
|
||
for idx, src in enumerate(big_entries):
|
||
if idx not in used and src == probe:
|
||
used.add(idx)
|
||
ga_order.append(idx)
|
||
forced = target.get("forcedLineId")
|
||
ga_lines.append(int(forced) if forced is not None and int(forced) >= 0 else -1)
|
||
break
|
||
else:
|
||
raise AssertionError("GA 基线回映射失败")
|
||
assert sorted(ga_order) == list(range(len(big_entries)))
|
||
|
||
pareto, meta = solve_nsga2(
|
||
world, params, entries=big_entries, seed=42,
|
||
seed_solutions=[(tuple(ga_order), tuple(ga_lines))],
|
||
nsga2_params={"populationSize": 40, "generations": 50},
|
||
)
|
||
assert len(pareto) >= 1
|
||
# 性能上限：固定代数内完成（不依赖超时中断）
|
||
assert meta["generations"] == meta["generationBudget"] == 50
|
||
assert meta["wallTimeSec"] < 15.0
|
||
# 解集性质：Pareto 前沿互不支配
|
||
for i, left in enumerate(pareto):
|
||
for j, right in enumerate(pareto):
|
||
if i != j:
|
||
assert not _dominates_left(left, right)
|
||
|
||
# 边界约束：存在前沿成员在 总延迟 与 冲突 两个目标上均不劣于 GA 基线
|
||
seed_obj = meta["seedObjectives"][0]
|
||
assert seed_obj["conflictCount"] >= 0
|
||
assert any(
|
||
s["kpi"]["totalTardiness"] <= seed_obj["totalTardiness"]
|
||
and s["kpi"]["conflictCount"] <= seed_obj["conflictCount"]
|
||
for s in pareto
|
||
), "Pareto 前沿应存在不劣于单目标 GA 基线的成员（精英保留保证）"
|
||
assert ga_meta["objective"] >= 0
|
||
|
||
|
||
# ---------------- 引擎集成 ----------------
|
||
def test_get_engine_nsga2_is_real_engine_and_materializes_schedule():
|
||
engine = get_engine("NSGA2")
|
||
assert isinstance(engine, NSGA2Engine)
|
||
assert engine.name == "NSGA2"
|
||
assert engine.supports_anytime is 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
|
||
|
||
worlds = [seed_world(), seed_world()]
|
||
results = []
|
||
metas = []
|
||
for world in worlds:
|
||
result = engine.solve(world, _params(time_limit=6.0), _next_id_factory())
|
||
meta = world["scheduleVersions"][0]["solverMeta"]
|
||
results.append(result)
|
||
metas.append(meta)
|
||
assert results[0].solveStatus == "FEASIBLE"
|
||
assert results[0].poCount == 7
|
||
assert results[0].woCount == 30
|
||
assert len(worlds[0]["productionOrders"]) == 7
|
||
# 确定性种子支持：两次求解元信息一致
|
||
assert metas[0]["backend"] == "NSGA-II"
|
||
assert metas[0]["engineType"] == "NSGA2"
|
||
assert metas[0]["paretoSize"] == metas[1]["paretoSize"]
|
||
assert metas[0]["generations"] == metas[1]["generations"]
|
||
assert results[0].totalTardiness == results[1].totalTardiness
|