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