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
|