import numpy as np from typing import Dict, List, Any from scipy.spatial.distance import euclidean class NetworkUtils: """无需 aux_data,全部现场从节点属性读取""" """ 角色匹配矩阵暂时没加 """ # ----------- 通用兜底 ----------- _DEFAULT = { "protocol": 0.80, "format": 0.85, "security": 0.90, "role": 0.70, "area": 0.50, "org": 0.70, "area_1": 1, "area_2": 0.7, "history": 0.8 } # ----------- 空间 ----------- @staticmethod def calculate_distance(pos1, pos2): return euclidean(pos1, pos2) @staticmethod def calculate_range_match(R_i: float, R_j: float, d_ij: float) -> float: r_min = min(R_i, R_j) if d_ij <= r_min: return 1.0 return r_min / d_ij if d_ij > 0 else 0.0 @staticmethod def calculate_time_window_overlap(node_i, node_j): w_i = node_i["temporal"].get("time_window", 24) w_j = node_j["temporal"].get("time_window", 24) overlap = min(w_i, w_j) return overlap / max(w_i, w_j, 1) # ----------- 协议兼容性:Jaccard ----------- @staticmethod def get_protocol_compatibility(node_i, node_j, _dummy=None): """节点属性里放 'protocol_list'""" p_i = set(node_i.get("protocol_list", [])) p_j = set(node_j.get("protocol_list", [])) if not p_i or not p_j: return NetworkUtils._DEFAULT["protocol"] intersection = len(p_i & p_j) union = len(p_i | p_j) return intersection / union if union else NetworkUtils._DEFAULT["protocol"] # ----------- 安全等级:1-5 映射 0-1 ----------- @staticmethod def get_security_compatibility(node_i, node_j, _dummy=None): lv_i = node_i.get("security_level", 3) lv_j = node_j.get("security_level", 3) gap = abs(lv_i - lv_j) return max(0, 1 - gap / 5) # ----------- 组织隶属:同单位给高分 ----------- @staticmethod def get_organization_relation(node_i, node_j, _dummy=None): unit_i = node_i.get("org_unit", "") unit_j = node_j.get("org_unit", "") return 1.0 if unit_i == unit_j and unit_i else NetworkUtils._DEFAULT["org"] # ----------- 历史交互:直接读属性 ----------- @staticmethod def get_interaction_history(node_i, node_j, _dummy=None): his_i = node_i.get("history_success", NetworkUtils._DEFAULT["history"]) # print("111111111111111111111",node_i.get("history_success")) his_j = node_j.get("history_success", NetworkUtils._DEFAULT["history"]) return (his_i + his_j) / 2 # ----------- 区域关联:同 code 高分 ----------- @staticmethod def get_area_relation(node_i, node_j, _dummy=None): code_i = node_i.get("nation") code_j = node_j.get("nation") if code_i == code_j: return NetworkUtils._DEFAULT["area_1"] else: return NetworkUtils._DEFAULT["area_2"] # ----------- 接口/角色/格式:若节点属性扩展了再读,否则给默认 ----------- @staticmethod def get_role_match(node_i, node_j, _dummy=None): return node_i.get("role_match", NetworkUtils._DEFAULT["role"]) @staticmethod def get_interface_standard(node_i, node_j, _dummy=None) -> float: """接口标准兼容性 = Jaccard(i_list, j_list)""" i_set = set(node_i.get("interface_list", [])) j_set = set(node_j.get("interface_list", [])) if not i_set or not j_set: return NetworkUtils._DEFAULT["protocol"] # 默认0.8 inter = len(i_set & j_set) union = len(i_set | j_set) return inter / union if union else NetworkUtils._DEFAULT["protocol"] @staticmethod def get_format_compatibility(node_i, node_j, _dummy=None) -> float: """数据格式兼容性 = Jaccard(i_format, j_format)""" i_set = set(node_i.get("format_list", [])) j_set = set(node_j.get("format_list", [])) if not i_set or not j_set: return NetworkUtils._DEFAULT["format"] # 默认0.85 inter = len(i_set & j_set) union = len(i_set | j_set) return inter / union if union else NetworkUtils._DEFAULT["format"]