91 lines
3.2 KiB
Python
91 lines
3.2 KiB
Python
# ============================================================
|
|
# SC-03 遗传算法引擎黄金测试
|
|
# ============================================================
|
|
from __future__ import annotations
|
|
|
|
from server.engines import GeneticAlgorithmEngine, RuleEngine, get_engine
|
|
from server.engines.base import EngineParams
|
|
from server.engines.ga_engine import optimize_genetic_assignment
|
|
from server.state.seed import seed_world
|
|
from server.timeutil import add_minutes, fmt_date, parse_dt, today0
|
|
|
|
|
|
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):
|
|
params = EngineParams(
|
|
engineType="GA",
|
|
strategyTemplate="COMPREHENSIVE",
|
|
planningHorizonDays=14,
|
|
startDate=fmt_date(add_minutes(today0(), 24 * 60)),
|
|
timeLimitSeconds=0.5,
|
|
)
|
|
return get_engine("GA").solve(world, params, _next_id_factory())
|
|
|
|
|
|
def test_get_engine_ga_is_real_solver():
|
|
engine = get_engine("GA")
|
|
assert isinstance(engine, GeneticAlgorithmEngine)
|
|
assert engine.name == "GA"
|
|
assert engine.supports_anytime is True
|
|
|
|
|
|
def test_ga_solver_meta_and_counts():
|
|
world = seed_world()
|
|
result = _run(world)
|
|
meta = world["scheduleVersions"][0]["solverMeta"]
|
|
assert result.engineType == "GA"
|
|
assert result.solveStatus == "FEASIBLE"
|
|
assert meta["backend"] == "Genetic Algorithm"
|
|
assert meta["pipeline"] == "GA->shift-slot"
|
|
assert meta["population"] >= 12
|
|
assert meta["generations"] >= 1
|
|
assert result.poCount == 7
|
|
assert result.woCount == 30
|
|
|
|
|
|
def test_ga_is_deterministic_and_preserves_hard_resource_constraint():
|
|
worlds = [seed_world(), seed_world()]
|
|
results = [_run(world) for world in worlds]
|
|
assert results[0].totalTardiness == results[1].totalTardiness
|
|
assert [po["salesOrderId"] for po in worlds[0]["productionOrders"]] == [
|
|
po["salesOrderId"] for po in worlds[1]["productionOrders"]
|
|
]
|
|
|
|
by_workstation: dict[int, list[tuple]] = {}
|
|
for work_order in worlds[0]["workOrders"]:
|
|
by_workstation.setdefault(work_order["workstationId"], []).append(
|
|
(parse_dt(work_order["plannedStartTime"]), parse_dt(work_order["plannedEndTime"]))
|
|
)
|
|
for intervals in by_workstation.values():
|
|
intervals.sort()
|
|
for (_, previous_end), (next_start, _) in zip(intervals, intervals[1:]):
|
|
assert previous_end <= next_start
|
|
|
|
|
|
def test_ga_time_budget_has_deterministic_generation_count_and_objective():
|
|
world = seed_world()
|
|
params = EngineParams(engineType="GA", timeLimitSeconds=0.05)
|
|
entries, _, _ = RuleEngine().collect_and_order(world, params)
|
|
large_entries = [dict(entry) for _ in range(8) for entry in entries]
|
|
|
|
runs = [optimize_genetic_assignment(world, large_entries, params) for _ in range(4)]
|
|
signatures = [
|
|
(
|
|
tuple((entry["so"]["id"], entry.get("forcedLineId")) for entry in ordered),
|
|
meta["objective"],
|
|
meta["generations"],
|
|
)
|
|
for ordered, meta in runs
|
|
]
|
|
assert all(signature == signatures[0] for signature in signatures[1:])
|
|
assert signatures[0][2] == 5
|