112 lines
4.7 KiB
Python
112 lines
4.7 KiB
Python
# f_net1125.py ← 边关系已算完，只算任务网络
|
||
from __future__ import annotations
|
||
import numpy as np
|
||
from typing import Dict, List, Tuple
|
||
|
||
NodeID = str
|
||
TaskScore = float # 网络效能得分 [0,1]
|
||
|
||
|
||
class TaskNetworkEvaluator:
|
||
"""
|
||
输入：
|
||
nodes - 节点数据（含 function_vector / performance / temporal）
|
||
relations - 已算好的全部边关系
|
||
格式：relations[("IS","DEF001","CMD001")] = 0.87
|
||
"""
|
||
def __init__(self,
|
||
nodes: Dict[NodeID, Dict],
|
||
relations: Dict[Tuple[str, NodeID, NodeID], float],
|
||
weights: Dict[str, Dict[str, float]] = None):
|
||
self.nodes = nodes
|
||
self.rels = relations
|
||
self.w = weights
|
||
|
||
# ---------- 工具 ----------
|
||
def _avg(self, scores: List[float]) -> float:
|
||
return float(np.mean(scores)) if scores else 0.0
|
||
|
||
def _collect_func(self, key: str) -> List[float]:
|
||
"""所有节点指定功能值"""
|
||
return [n["function_vector"][key] for n in self.nodes.values()]
|
||
|
||
def _collect_rel(self, tag: str) -> List[float]:
|
||
"""预存关系强度列表"""
|
||
return [v for k, v in self.rels.items() if k[0] == tag]
|
||
|
||
def _collect_resp(self) -> List[float]:
|
||
"""响应时间得分"""
|
||
return [np.exp(-0.1 * n["temporal"]["response_time"]) for n in self.nodes.values()]
|
||
|
||
# ---------- 2.1 综合防御 ----------
|
||
def eval_defense(self) -> TaskScore:
|
||
w = self.w["defense"]
|
||
f_ic = self._avg(self._collect_func("f_IC"))
|
||
f_ia = self._avg(self._collect_func("f_IA"))
|
||
l_is = self._avg(self._collect_rel("IS"))
|
||
t_resp = self._avg(self._collect_resp())
|
||
l_cc = self._avg(self._collect_rel("CC"))
|
||
return w["w1"] * f_ic + w["w2"] * f_ia + w["w3"] * l_is + w["w4"] * t_resp + w["w5"] * l_cc
|
||
|
||
# ---------- 2.2 火力打击 ----------
|
||
def eval_fire(self) -> TaskScore:
|
||
w = self.w["fire"]
|
||
f_cs = self._avg(self._collect_func("f_CS"))
|
||
f_ia = self._avg(self._collect_func("f_IA"))
|
||
l_co = self._avg(self._collect_rel("CO"))
|
||
perf_cs = self._avg([n["performance"]["core_performance"] * n["function_vector"]["f_CS"]
|
||
for n in self.nodes.values()])
|
||
l_is = self._avg(self._collect_rel("IS"))
|
||
return w["w1"] * f_cs + w["w2"] * f_ia + w["w3"] * l_co + w["w4"] * perf_cs + w["w5"] * l_is
|
||
|
||
# ---------- 2.3 后勤保障 ----------
|
||
def eval_logistics(self) -> TaskScore:
|
||
w = self.w["logistics"]
|
||
f_cps = self._avg(self._collect_func("f_CPS"))
|
||
f_dp = self._avg(self._collect_func("f_DP"))
|
||
l_pd = self._avg(self._collect_rel("PD"))
|
||
p_rel = self._avg([n["performance"]["mtbf"] / 1000 for n in self.nodes.values()])
|
||
f_it = self._avg(self._collect_func("f_IT"))
|
||
return w["w1"] * f_cps + w["w2"] * f_dp + w["w3"] * l_pd + w["w4"] * p_rel + w["w5"] * f_it
|
||
|
||
# ---------- 2.4 医疗救援 ----------
|
||
def eval_medical(self) -> TaskScore:
|
||
w = self.w["medical"]
|
||
f_cps = self._avg(self._collect_func("f_CPS"))
|
||
t_resp = self._avg(self._collect_resp())
|
||
f_it = self._avg(self._collect_func("f_IT"))
|
||
l_sf = self._avg(self._collect_rel("SF"))
|
||
fresh = self._avg([np.exp(-0.1 * n["temporal"]["data_age"]) for n in self.nodes.values()])
|
||
return w["w1"] * f_cps + w["w2"] * t_resp + w["w3"] * f_it + w["w4"] * l_sf + w["w5"] * fresh
|
||
|
||
# ---------- 2.5 紧急疏散 ----------
|
||
def eval_evacuation(self) -> TaskScore:
|
||
w = self.w["evacuation"]
|
||
f_cc = self._avg(self._collect_func("f_CC"))
|
||
f_it = self._avg(self._collect_func("f_IT"))
|
||
l_cc = self._avg(self._collect_rel("CC"))
|
||
f_cps = self._avg(self._collect_func("f_CPS"))
|
||
t_resp = self._avg(self._collect_resp())
|
||
return w["w1"] * f_cc + w["w2"] * f_it + w["w3"] * l_cc + w["w4"] * f_cps + w["w5"] * t_resp
|
||
|
||
# ---------- 一键评估 ----------
|
||
def eval_all(self) -> Dict[str, TaskScore]:
|
||
results = {
|
||
"defense": self.eval_defense(),
|
||
"fire": self.eval_fire(),
|
||
"logistics": self.eval_logistics(),
|
||
"medical": self.eval_medical(),
|
||
"evacuation": self.eval_evacuation(),
|
||
}
|
||
|
||
# 计算五个维度的分数
|
||
dimensions = {
|
||
"function": self._avg(self._collect_func("f_IC")),
|
||
"relationship": self._avg(self._collect_rel("IS")),
|
||
"performance": self._avg([n["performance"]["core_performance"] for n in self.nodes.values()]),
|
||
"temporal": self._avg(self._collect_resp()),
|
||
"interaction": self._avg(self._collect_rel("SF"))
|
||
}
|
||
|
||
results["dimensions"] = dimensions
|
||
return results |