190 lines
7.3 KiB
Python
190 lines
7.3 KiB
Python
from __future__ import annotations
|
||
|
||
from collections.abc import Iterable
|
||
from copy import deepcopy
|
||
from dataclasses import dataclass
|
||
from math import isfinite
|
||
from typing import Any
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MetricSpec:
|
||
key: str
|
||
direction: str
|
||
default_weight: float = 0.0
|
||
|
||
|
||
PARETO_METRICS: tuple[MetricSpec, ...] = (
|
||
MetricSpec("totalTardiness", "min", 0.40),
|
||
MetricSpec("conflictCount", "min", 0.0),
|
||
MetricSpec("totalCost", "min", 0.30),
|
||
MetricSpec("avgUtilization", "max", 0.20),
|
||
MetricSpec("loadBalance", "max", 0.10),
|
||
MetricSpec("totalChangeoverMin", "min", 0.0),
|
||
)
|
||
|
||
|
||
def _number(value: Any) -> float:
|
||
try:
|
||
number = float(value)
|
||
except (TypeError, ValueError):
|
||
return 0.0
|
||
return number if isfinite(number) else 0.0
|
||
|
||
|
||
def load_balance_score(loads: Iterable[float]) -> float:
|
||
"""Return a bounded 0..1 balance score based on the coefficient of variation."""
|
||
values = [max(0.0, _number(value)) for value in loads]
|
||
if not values or max(values, default=0.0) <= 1e-12:
|
||
return 1.0
|
||
mean = sum(values) / len(values)
|
||
variance = sum((value - mean) ** 2 for value in values) / len(values)
|
||
coefficient = variance ** 0.5 / mean if mean > 1e-12 else 0.0
|
||
return round(1.0 / (1.0 + coefficient), 6)
|
||
|
||
|
||
def _dominates(left: dict[str, Any], right: dict[str, Any], metrics: tuple[MetricSpec, ...]) -> bool:
|
||
no_worse = True
|
||
strictly_better = False
|
||
for metric in metrics:
|
||
lval = _number((left.get("kpi") or {}).get(metric.key))
|
||
rval = _number((right.get("kpi") or {}).get(metric.key))
|
||
if metric.direction == "max":
|
||
no_worse = no_worse and lval >= rval
|
||
strictly_better = strictly_better or lval > rval
|
||
else:
|
||
no_worse = no_worse and lval <= rval
|
||
strictly_better = strictly_better or lval < rval
|
||
if not no_worse:
|
||
return False
|
||
return strictly_better
|
||
|
||
|
||
def pareto_front_indices(
|
||
cards: list[dict[str, Any]],
|
||
*,
|
||
metrics: tuple[MetricSpec, ...] = PARETO_METRICS,
|
||
eligible_indices: Iterable[int] | None = None,
|
||
) -> set[int]:
|
||
candidates = list(eligible_indices if eligible_indices is not None else range(len(cards)))
|
||
front: set[int] = set()
|
||
for idx in candidates:
|
||
if not any(
|
||
other != idx and _dominates(cards[other], cards[idx], metrics)
|
||
for other in candidates
|
||
):
|
||
front.add(idx)
|
||
return front
|
||
|
||
|
||
def _normalized_values(
|
||
cards: list[dict[str, Any]],
|
||
indices: list[int],
|
||
metrics: tuple[MetricSpec, ...],
|
||
) -> dict[int, dict[str, float]]:
|
||
normalized = {idx: {} for idx in indices}
|
||
for metric in metrics:
|
||
values = [_number((cards[idx].get("kpi") or {}).get(metric.key)) for idx in indices]
|
||
low, high = min(values), max(values)
|
||
span = high - low
|
||
for idx, value in zip(indices, values):
|
||
if span <= 1e-12:
|
||
score = 1.0
|
||
elif metric.direction == "max":
|
||
score = (value - low) / span
|
||
else:
|
||
score = (high - value) / span
|
||
normalized[idx][metric.key] = round(score, 6)
|
||
return normalized
|
||
|
||
|
||
def rank_scenarios(
|
||
cards: list[dict[str, Any]],
|
||
*,
|
||
weights: dict[str, float] | None = None,
|
||
metrics: tuple[MetricSpec, ...] = PARETO_METRICS,
|
||
) -> list[dict[str, Any]]:
|
||
"""Hard-filter, Pareto-filter, then rank the front using normalized weighted KPIs."""
|
||
ranked = deepcopy(cards)
|
||
if not ranked:
|
||
return ranked
|
||
|
||
hard_eligible = [idx for idx, card in enumerate(ranked) if card.get("hardFeasible", True)]
|
||
front = pareto_front_indices(ranked, metrics=metrics, eligible_indices=hard_eligible)
|
||
normalized = _normalized_values(ranked, sorted(front), metrics) if front else {}
|
||
|
||
requested = weights or {metric.key: metric.default_weight for metric in metrics}
|
||
active = {
|
||
metric.key: max(0.0, _number(requested.get(metric.key, metric.default_weight)))
|
||
for metric in metrics
|
||
}
|
||
total_weight = sum(active.values())
|
||
if total_weight <= 1e-12:
|
||
active = {metric.key: metric.default_weight for metric in metrics}
|
||
total_weight = sum(active.values())
|
||
active = {key: round(value / total_weight, 6) for key, value in active.items()}
|
||
|
||
front_scores: list[tuple[float, float, str, int]] = []
|
||
for idx, card in enumerate(ranked):
|
||
card["normalizedKpi"] = normalized.get(idx, {})
|
||
card["rankingWeights"] = active
|
||
card["isPareto"] = idx in front
|
||
card["isRecommended"] = False
|
||
card["rank"] = None
|
||
card["weightedScore"] = None
|
||
if idx not in hard_eligible:
|
||
card["selectionStatus"] = "HARD_REJECTED"
|
||
elif idx not in front:
|
||
card["selectionStatus"] = "DOMINATED"
|
||
else:
|
||
score = sum(
|
||
active.get(metric.key, 0.0) * normalized[idx].get(metric.key, 0.0)
|
||
for metric in metrics
|
||
)
|
||
card["weightedScore"] = round(score, 6)
|
||
card["selectionStatus"] = "PARETO"
|
||
tie_key = str(card.get("strategy") or card.get("scenarioId") or idx)
|
||
robustness = max(0.0, min(1.0, _number(card.get("robustness"))))
|
||
front_scores.append((-score, -robustness, tie_key, idx))
|
||
|
||
front_scores.sort()
|
||
for rank, (_, _, _, idx) in enumerate(front_scores, start=1):
|
||
ranked[idx]["rank"] = rank
|
||
if front_scores:
|
||
ranked[front_scores[0][3]]["isRecommended"] = True
|
||
return ranked
|
||
|
||
# ---------------- NSGA-II 鍊欓€夊崱閫傞厤锛堢煩闃?85锛宺ound-40 鏂瑰悜 U锛?----------------
|
||
def nsga2_solutions_to_cards(
|
||
solutions: list[dict[str, Any]],
|
||
*,
|
||
prefix: str = "NSGA2",
|
||
) -> list[dict[str, Any]]:
|
||
"""鎶?NSGA-II 姹傝В鍣紙engines.nsga2_engine.solve_nsga2锛変骇鍑虹殑 Pareto 瑙? 閫傞厤涓?rank_scenarios 鍙秷璐圭殑鍊欓€夊崱銆?
|
||
瑙ｅ崱瀛楁濂戠害锛歴olution 鑷冲皯鍚?strategy/hardFeasible/robustness/kpi锛? kpi 閿笌 PARETO_METRICS 瀵归綈锛坱otalTardiness/conflictCount/totalCost/
|
||
avgUtilization/loadBalance/totalChangeoverMin锛夛紱jobOrder/lines 淇濈暀鍦? source 渚涚墿鍖栧洖鏀撅紙濡?NSGA2Engine 鐨勬帹鑽愯В閫夋嫨锛夈€? """
|
||
cards: list[dict[str, Any]] = []
|
||
for idx, solution in enumerate(solutions):
|
||
kpi = solution.get("kpi") or {}
|
||
cards.append({
|
||
"scenarioId": f"{prefix.lower()}-{idx:02d}",
|
||
"strategy": solution.get("strategy") or f"{prefix}-{idx:02d}",
|
||
"hardFeasible": bool(solution.get("hardFeasible", True)),
|
||
"robustness": float(solution.get("robustness") or 0.5),
|
||
"kpi": {
|
||
"totalTardiness": kpi.get("totalTardiness", 0.0),
|
||
"conflictCount": kpi.get("conflictCount", 0),
|
||
"totalCost": kpi.get("totalCost", 0.0),
|
||
"avgUtilization": kpi.get("avgUtilization", 0.0),
|
||
"loadBalance": kpi.get("loadBalance", 0.0),
|
||
"totalChangeoverMin": kpi.get("totalChangeoverMin", 0.0),
|
||
},
|
||
"source": {
|
||
"algo": "NSGA-II",
|
||
"solutionId": solution.get("solutionId"),
|
||
"order": solution.get("jobOrder") or [],
|
||
"lines": solution.get("lines") or [],
|
||
},
|
||
})
|
||
return cards
|