import numpy as np import pandas as pd from sklearn.neighbors import BallTree from typing import Dict, List from functools import lru_cache import hashlib import json import redis class HealthAssessor: def __init__(self): self.subsystem_config = { # 发电机 'generator': { # 双馈 'dfig': { 'fixed': ['generator_winding1_temperature', 'generator_winding2_temperature', 'generator_winding3_temperature', 'generatordrive_end_bearing_temperature', 'generatornon_drive_end_bearing_temperature'], }, # 直驱 'direct': { 'fixed': ['generator_winding1_temperature', 'generator_winding2_temperature', 'generator_winding3_temperature', 'main_bearing_temperature'], } }, # 机舱系统 'nacelle': { 'fixed': ['front_back_vibration_of_the_cabin', 'side_to_side_vibration_of_the_cabin', 'cabin_position', 'cabin_temperature'], }, # 电网环境 'grid': { 'fixed': ['reactive_power', 'active_power', 'grid_a_phase_current', 'grid_b_phase_current', 'grid_c_phase_current'], }, # 传动系统 'drive_train': { 'fixed': ['main_bearing_temperature'], 'keywords': [ {'include': ['gearbox', 'temperature'], 'exclude': [], 'min_count': 2}, ] } } # 嵌入源代码的MSET实现 self.mset = self._create_mset_core() def _create_mset_core(self): """创建MSET核心计算模块""" class MSETCore: def __init__(self): self.matrixD = None self.normalDataBallTree = None self.healthyResidual = None def calcSimilarity(self, x, y): """优化后的相似度计算""" diff = np.array(x) - np.array(y) return 1 / (1 + np.sqrt(np.sum(diff ** 2))) def genDLMatrix(self, trainDataset, dataSize4D=15, dataSize4L=5): """优化矩阵生成过程""" m, n = trainDataset.shape # 快速选择极值点 min_indices = np.argmin(trainDataset, axis=0) max_indices = np.argmax(trainDataset, axis=0) unique_indices = np.unique(np.concatenate([min_indices, max_indices])) self.matrixD = trainDataset[unique_indices].copy() # 快速填充剩余点 remaining_indices = np.setdiff1d(np.arange(m), unique_indices) np.random.shuffle(remaining_indices) needed = max(0, dataSize4D - len(unique_indices)) if needed > 0: self.matrixD = np.vstack([self.matrixD, trainDataset[remaining_indices[:needed]]]) # 使用与源代码一致的BallTree参数 self.normalDataBallTree = BallTree( self.matrixD, leaf_size=40, metric=lambda i, j: 1 - self.calcSimilarity(i, j) # 自定义相似度 ) # 使用所有数据计算残差 self.healthyResidual = self.calcResidualByLocallyWeightedLR(trainDataset) return 0 def calcResidualByLocallyWeightedLR(self, newStates): """优化残差计算""" if len(newStates.shape) == 1: newStates = newStates.reshape(-1, 1) dist, iList = self.normalDataBallTree.query( newStates, k=min(10, len(self.matrixD)), return_distance=True ) weights = 1 / (dist + 1e-5) weights /= weights.sum(axis=1)[:, np.newaxis] est_X = np.sum(weights[:, :, np.newaxis] * self.matrixD[iList[0]], axis=1) return est_X - newStates def calcSPRT(self, newsStates, feature_weight, alpha=0.1, beta=0.1, decisionGroup=1): """优化SPRT计算""" stateResidual = self.calcResidualByLocallyWeightedLR(newsStates) weightedStateResidual = np.dot(stateResidual, feature_weight) weightedHealthyResidual = np.dot(self.healthyResidual, feature_weight) mu0 = np.mean(weightedHealthyResidual) sigma0 = np.std(weightedHealthyResidual) # 向量化计算 n = len(newsStates) if n < decisionGroup: return [50] # 中性值 rolling_mean = np.convolve(weightedStateResidual, np.ones(decisionGroup) / decisionGroup, 'valid') si = (rolling_mean - mu0) * (rolling_mean + mu0 - 2 * mu0) / (2 * sigma0 ** 2) lowThres = np.log(beta / (1 - alpha)) highThres = np.log((1 - beta) / alpha) si = np.clip(si, lowThres, highThres) si = np.where(si > 0, si / highThres, si / lowThres) flag = 100 - si * 100 # 填充不足的部分 if len(flag) < n: flag = np.pad(flag, (0, n - len(flag)), mode='edge') return flag.tolist() def CRITIC_prepare(self, data, flag=1): """标准化处理""" data = data.astype(float) numeric_cols = data.select_dtypes(include=[np.number]).columns negative_cols = [col for col in numeric_cols if any(kw in col for kw in ['temperature'])] positive_cols = list(set(numeric_cols) - set(negative_cols)) # 负向标准化 if negative_cols: max_val = data[negative_cols].max() min_val = data[negative_cols].min() data[negative_cols] = (max_val - data[negative_cols]) / (max_val - min_val).replace(0, 1e-5) # 正向标准化 if positive_cols: max_val = data[positive_cols].max() min_val = data[positive_cols].min() data[positive_cols] = (data[positive_cols] - min_val) / (max_val - min_val).replace(0, 1e-5) return data def CRITIC(self, data): """CRITIC权重计算(支持单特征)""" try: # 处理单特征情况 if len(data.columns) == 1: return pd.Series([1.0], index=data.columns) data_norm = self.CRITIC_prepare(data.copy()) std = data_norm.std(ddof=0).clip(lower=0.01) # 计算相关系数矩阵(添加异常处理) try: corr = np.abs(np.corrcoef(data_norm.T)) np.fill_diagonal(corr, 0) conflict = np.sum(1 - corr, axis=1) except: # 如果计算相关系数失败，使用等权重 return pd.Series(np.ones(len(data.columns))/len(data.columns)) info = std * conflict weights = info / info.sum() return pd.Series(weights, index=data.columns) except Exception as e: print(f"CRITIC计算失败: {str(e)}") return pd.Series(np.ones(len(data.columns))/len(data.columns)) def ahp(self, matrix): """AHP权重计算""" eigenvalue, eigenvector = np.linalg.eig(matrix) max_idx = np.argmax(eigenvalue) weight = eigenvector[:, max_idx].real return weight / weight.sum(), eigenvalue[max_idx].real return MSETCore() def assess_turbine(self, engine_code, data, mill_type, wind_turbine_name): # 生成缓存键 cache_key = f"assessment:{engine_code}:{data.shape[0]}:{hashlib.sha256(pd.util.hash_pandas_object(data).values.tobytes()).hexdigest()}" try: # 尝试从缓存获取 cached_result = self.cache_client.get(cache_key) if cached_result: return json.loads(cached_result) """评估单个风机""" results = { "engine_code": engine_code, "wind_turbine_name": wind_turbine_name, "mill_type": mill_type, "total_health_score": None, "subsystems": {}, "assessed_subsystems": [] } # 各子系统评估 subsystems_to_assess = [ ('generator', self.subsystem_config['generator'][mill_type], 1), ('nacelle', self.subsystem_config['nacelle'], 1), ('grid', self.subsystem_config['grid'], 1), ('drive_train', self.subsystem_config['drive_train'] if mill_type == 'dfig' else None, 1) ] for subsystem, config, min_features in subsystems_to_assess: if config is None: continue features = self._get_subsystem_features(config, data) # 功能1：无论特征数量是否足够都输出结果 if len(features) >= min_features: assessment = self._assess_subsystem(data[features]) else: assessment = { 'health_score': -1, # 特征不足时输出'-' 'weights': {}, 'message': f'Insufficient features (required {min_features}, got {len(features)})' } print('结果打印',assessment) # 功能3：删除features内容 if 'features' in assessment: del assessment['features'] # 最终清理：确保没有NaN值 for sys, result in results["subsystems"].items(): if isinstance(result['health_score'], float) and np.isnan(result['health_score']): result['health_score'] = -1 result['message'] = (result.get('message') or '') + '; NaN detected' if isinstance(results["total_health_score"], float) and np.isnan(results["total_health_score"]): results["total_health_score"] = -1 results["subsystems"][subsystem] = assessment # 计算整机健康度（使用新字段名） if results["subsystems"]: # 只计算健康值为数字的子系统 valid_subsystems = [ k for k, v in results["subsystems"].items() if isinstance(v['health_score'], (int, float)) and v['health_score'] >= 0 ] if valid_subsystems: weights = self._get_subsystem_weights(valid_subsystems) health_scores = [results["subsystems"][sys]['health_score'] for sys in valid_subsystems] results["total_health_score"] = float(np.dot(health_scores, weights)) results["assessed_subsystems"] = valid_subsystems return results except Exception as e: print('评估出错') def _get_all_possible_features(self, mill_type, available_columns): """ 获取所有可能的特征列（基于实际存在的列） 参数: mill_type: 机型类型 available_columns: 数据库实际存在的列名列表 """ features = [] available_columns_lower = [col.lower() for col in available_columns] # 不区分大小写匹配 # for subsys_name, subsys_config in assessor.subsystem_config.items(): for subsys_name, subsys_config in self.subsystem_config.items(): # 处理子系统配置 if subsys_name == 'generator': config = subsys_config.get(mill_type, {}) elif subsys_name == 'drive_train' and mill_type != 'dfig': continue else: config = subsys_config # 处理固定特征 if 'fixed' in config: for f in config['fixed']: if f in available_columns: features.append(f) # 处理关键词特征 if 'keywords' in config: for rule in config['keywords']: matched = [] include_kws = [kw.lower() for kw in rule['include']] exclude_kws = [ex.lower() for ex in rule.get('exclude', [])] for col in available_columns: col_lower = col.lower() # 检查包含关键词 include_ok = all(kw in col_lower for kw in include_kws) # 检查排除关键词 exclude_ok = not any(ex in col_lower for ex in exclude_kws) if include_ok and exclude_ok: matched.append(col) if len(matched) >= rule.get('min_count', 1): features.extend(matched) return list(set(features)) # 去重 def _get_subsystem_features(self, config: Dict, data: pd.DataFrame) -> List[str]: """最终版特征获取方法""" available_features = [] # 固定特征检查（要求至少10%非空） if 'fixed' in config: for f in config['fixed']: if f in data.columns and data[f].notna().mean() > 0.1: available_features.append(f) print(f"匹配到的固定特征: {available_features}") # 关键词特征检查 if 'keywords' in config: for rule in config['keywords']: matched = [ col for col in data.columns if all(kw.lower() in col.lower() for kw in rule['include']) and not any(ex.lower() in col.lower() for ex in rule.get('exclude', [])) and data[col].notna().mean() > 0.1 # 数据有效性检查 ] if len(matched) >= rule.get('min_count', 1): available_features.extend(matched) print(f"匹配到的关键词特征: {available_features}") return list(set(available_features)) def _assess_subsystem(self, data: pd.DataFrame) -> Dict: """评估子系统(支持单特征)""" # 数据清洗 clean_data = data.dropna() if len(clean_data) < 10: # 降低最小样本量要求(原为20) return {'health_score': -1, 'weights': {}, 'features': list(data.columns), 'message': 'Insufficient data'} try: # 标准化 normalized_data = self.mset.CRITIC_prepare(clean_data) # 计算权重 - 处理单特征情况 if len(normalized_data.columns) == 1: weights = pd.Series([1.0], index=normalized_data.columns) else: weights = self.mset.CRITIC(normalized_data) # MSET评估 health_score = self._run_mset_assessment(normalized_data.values, weights.values) bins = [0, 10, 20, 30, 40, 50, 60, 70, 80] adjust_values = [87, 77, 67, 57, 47, 37, 27, 17, 7] def adjust_score(score): for i in range(len(bins)): if score < bins[i]: return score + adjust_values[i-1] return score # adjusted_score = adjust_score(health_score) # if adjusted_score >= 100: adjusted_score = 92.8 return { 'health_score': float(health_score), 'weights': weights.to_dict(), 'features': list(data.columns) } except Exception as e: return {'health_score': -1, 'weights': {}, 'features': list(data.columns), 'message': str(e)} @lru_cache(maxsize=10) def _get_mset_model(self, train_data: tuple): """缓存MSET模型""" # 注意：由于lru_cache需要可哈希参数，这里使用元组 arr = np.array(train_data) model = self._create_mset_core() model.genDLMatrix(arr) return model def _run_mset_assessment(self, data: np.ndarray, weights: np.ndarray) -> float: """执行MSET评估""" # 检查权重有效性 if np.isnan(weights).any() or np.isinf(weights).any(): weights = np.ones_like(weights) / len(weights) # 重置为等权重 # 分割训练集和测试集 split_idx = len(data) // 2 train_data = data[:split_idx] test_data = data[split_idx:] # 使用缓存模型 try: model = self._get_mset_model(tuple(map(tuple, train_data))) flags = model.calcSPRT(test_data, weights) # 过滤NaN值并计算均值 valid_flags = [x for x in flags if not np.isnan(x)] if not valid_flags: return 50.0 # 默认中性值 return float(np.mean(valid_flags)) except Exception as e: print(f"MSET评估失败: {str(e)}") return 50.0 # 默认中性值 def _get_subsystem_weights(self, subsystems: List[str]) -> np.ndarray: """生成等权重的子系统权重向量""" n = len(subsystems) if n == 0: return np.array([]) # 直接返回等权重向量 return np.ones(n) / n