225 lines
8.1 KiB
Python
225 lines
8.1 KiB
Python
# ============================================================
|
||
# 遗传算法排产引擎（moduleId: engines-ga, SC-03，可重生）
|
||
# 多线订单排序 + 产线分配；最终班次占槽复用 RuleEngine 的确定性物化。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
from random import Random
|
||
from time import monotonic
|
||
from typing import Any, Callable
|
||
|
||
from server.contracts import ScheduleResult
|
||
from server.engines.base import EngineParams
|
||
from server.engines.cp_engine import _due_minutes, _job_duration_min
|
||
from server.engines.queries import (
|
||
find_product_lines,
|
||
find_routing_steps,
|
||
find_workstation_for_operation,
|
||
)
|
||
from server.engines.rule_engine import RuleEngine
|
||
from server.timeutil import add_minutes, parse_dt, today0
|
||
|
||
World = dict[str, Any]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class _Candidate:
|
||
order: tuple[int, ...]
|
||
lines: tuple[int, ...]
|
||
score: float
|
||
|
||
|
||
def _problem_data(
|
||
world: World,
|
||
entries: list[dict],
|
||
params: EngineParams,
|
||
) -> tuple[list[list[int]], dict[tuple[int, int], int], list[int], list[int]]:
|
||
from server.aps_domain.params import level_weight
|
||
|
||
base_start = (
|
||
parse_dt(params.startDate + " 08:00")
|
||
if params.startDate
|
||
else add_minutes(today0(), 24 * 60)
|
||
)
|
||
if params.freezeWindowHours is not None and float(params.freezeWindowHours) > 0:
|
||
base_start = add_minutes(base_start, int(float(params.freezeWindowHours) * 60))
|
||
due_buffer = (
|
||
1.0
|
||
if params.deliveryBufferRatio is None
|
||
else max(0.5, min(1.0, float(params.deliveryBufferRatio)))
|
||
)
|
||
|
||
options: list[list[int]] = []
|
||
durations: dict[tuple[int, int], int] = {}
|
||
dues: list[int] = []
|
||
weights: list[int] = []
|
||
for idx, entry in enumerate(entries):
|
||
line_ids: list[int] = []
|
||
routing_steps = find_routing_steps(world, entry["item"]["productId"])
|
||
for link in find_product_lines(world, entry["item"]["productId"]):
|
||
line_id = int(link["lineId"])
|
||
if any(
|
||
find_workstation_for_operation(world, line_id, step["operationId"]) is None
|
||
for step in routing_steps
|
||
):
|
||
continue
|
||
line = next(line for line in world["lines"] if int(line["id"]) == line_id)
|
||
line_ids.append(line_id)
|
||
durations[(idx, line_id)] = _job_duration_min(world, entry["item"], line)
|
||
options.append(line_ids or [-1])
|
||
if not line_ids:
|
||
durations[(idx, -1)] = 1
|
||
dues.append(_due_minutes(entry["so"], base_start, due_buffer))
|
||
weight = int(round(level_weight(world, entry["so"].get("customerLevel")) * 100))
|
||
if entry["so"].get("isRush"):
|
||
weight += 200
|
||
if entry["so"].get("isForecast"):
|
||
weight = max(1, weight // 2)
|
||
weights.append(max(1, weight))
|
||
return options, durations, dues, weights
|
||
|
||
|
||
def _score(
|
||
order: tuple[int, ...],
|
||
lines: tuple[int, ...],
|
||
durations: dict[tuple[int, int], int],
|
||
dues: list[int],
|
||
weights: list[int],
|
||
) -> float:
|
||
cursor: dict[int, int] = {}
|
||
tardiness = 0
|
||
for idx in order:
|
||
line_id = lines[idx]
|
||
end = cursor.get(line_id, 0) + durations[(idx, line_id)]
|
||
cursor[line_id] = end
|
||
tardiness += max(0, end - dues[idx]) * weights[idx]
|
||
makespan = max(cursor.values(), default=0)
|
||
return float(tardiness * 1000 + makespan)
|
||
|
||
|
||
def _order_crossover(a: tuple[int, ...], b: tuple[int, ...], rng: Random) -> tuple[int, ...]:
|
||
if len(a) < 2:
|
||
return a
|
||
left, right = sorted(rng.sample(range(len(a)), 2))
|
||
child: list[int | None] = [None] * len(a)
|
||
child[left : right + 1] = a[left : right + 1]
|
||
remaining = [gene for gene in b if gene not in child]
|
||
pos = 0
|
||
for idx in list(range(right + 1, len(a))) + list(range(0, left)):
|
||
child[idx] = remaining[pos]
|
||
pos += 1
|
||
return tuple(int(gene) for gene in child if gene is not None)
|
||
|
||
|
||
def optimize_genetic_assignment(
|
||
world: World,
|
||
entries: list[dict],
|
||
params: EngineParams,
|
||
) -> tuple[list[dict], dict[str, Any]]:
|
||
"""确定性 GA：联合优化订单顺序和可选产线，返回 RuleEngine 可物化的条目。"""
|
||
started = monotonic()
|
||
count = len(entries)
|
||
meta: dict[str, Any] = {
|
||
"backend": "Genetic Algorithm",
|
||
"pipeline": "GA->shift-slot",
|
||
"placement": "shift-slot",
|
||
"seed": 42,
|
||
}
|
||
if count == 0:
|
||
meta.update({"status": "TRIVIAL", "wallTimeSec": 0.0, "objective": 0.0,
|
||
"population": 0, "generations": 0, "gap": None})
|
||
return entries, meta
|
||
|
||
rng = Random(42)
|
||
options, durations, dues, weights = _problem_data(world, entries, params)
|
||
population_size = max(12, min(48, count * 6))
|
||
max_generations = max(20, min(120, count * 15))
|
||
time_limit = max(0.05, float(params.timeLimitSeconds or 3.0))
|
||
generation_budget = max(1, min(max_generations, int(round(time_limit * 100))))
|
||
|
||
base_order = tuple(range(count))
|
||
base_lines = tuple(lines[0] for lines in options)
|
||
|
||
def make_candidate(order: tuple[int, ...], lines: tuple[int, ...]) -> _Candidate:
|
||
return _Candidate(order, lines, _score(order, lines, durations, dues, weights))
|
||
|
||
baseline = make_candidate(base_order, base_lines)
|
||
population = [baseline]
|
||
while len(population) < population_size:
|
||
order = list(base_order)
|
||
rng.shuffle(order)
|
||
lines = tuple(rng.choice(options[idx]) for idx in range(count))
|
||
population.append(make_candidate(tuple(order), lines))
|
||
|
||
completed_generations = 0
|
||
for generation in range(generation_budget):
|
||
population.sort(key=lambda candidate: candidate.score)
|
||
next_population = population[: max(2, population_size // 8)]
|
||
while len(next_population) < population_size:
|
||
contenders = rng.sample(population[: max(4, population_size // 2)], 4)
|
||
parent_a, parent_b = sorted(contenders, key=lambda candidate: candidate.score)[:2]
|
||
order = list(_order_crossover(parent_a.order, parent_b.order, rng))
|
||
lines = [
|
||
parent_a.lines[idx] if rng.random() < 0.5 else parent_b.lines[idx]
|
||
for idx in range(count)
|
||
]
|
||
if count > 1 and rng.random() < 0.35:
|
||
first, second = rng.sample(range(count), 2)
|
||
order[first], order[second] = order[second], order[first]
|
||
if rng.random() < 0.4:
|
||
gene = rng.randrange(count)
|
||
lines[gene] = rng.choice(options[gene])
|
||
next_population.append(make_candidate(tuple(order), tuple(lines)))
|
||
population = next_population
|
||
completed_generations = generation + 1
|
||
|
||
best = min(population, key=lambda candidate: candidate.score)
|
||
output: list[dict] = []
|
||
for idx in best.order:
|
||
entry = dict(entries[idx])
|
||
if best.lines[idx] >= 0:
|
||
entry["forcedLineId"] = best.lines[idx]
|
||
output.append(entry)
|
||
|
||
meta.update({
|
||
"status": "FEASIBLE",
|
||
"wallTimeSec": round(monotonic() - started, 4),
|
||
"timeLimitSec": time_limit,
|
||
"objective": best.score,
|
||
"baselineObjective": baseline.score,
|
||
"population": population_size,
|
||
"generations": completed_generations,
|
||
"generationBudget": generation_budget,
|
||
"gap": None,
|
||
})
|
||
return output, meta
|
||
|
||
|
||
class GeneticAlgorithmEngine(RuleEngine):
|
||
"""SC-03：GA 搜索顺序与产线，班次和硬约束由 RuleEngine 统一物化。"""
|
||
|
||
name = "GA"
|
||
supports_anytime = True
|
||
|
||
def __init__(self) -> None:
|
||
super().__init__(requested_type="GA")
|
||
|
||
def solve(
|
||
self,
|
||
world: World,
|
||
params: EngineParams,
|
||
next_id: Callable[[str], int],
|
||
) -> ScheduleResult:
|
||
entries, campaign_meta, source_count = self.collect_and_order(world, params)
|
||
ordered, solver_meta = optimize_genetic_assignment(world, entries, params)
|
||
return self.materialize_schedule(
|
||
world,
|
||
params,
|
||
next_id,
|
||
ordered,
|
||
campaign_meta,
|
||
source_count,
|
||
solver_meta=solver_meta,
|
||
)
|