import pymysql import threading from typing import List, Dict, Any, Optional, Tuple import logging from datetime import datetime from collections import Counter import statistics import math # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class ConnectionPool: """MySQL数据库连接池""" def __init__(self, host, port, user, password, database, charset='utf8mb4', max_connections=2, mem_quota=4 << 30): self.host = host self.port = port self.user = user self.password = password self.database = database self.charset = charset self.max_connections = max_connections self.mem_quota = mem_quota self._lock = threading.Lock() self._connections = [] self._in_use = set() def _create_connection(self): """创建新连接并设置内存配额""" try: conn = pymysql.connect( host=self.host, port=self.port, user=self.user, password=self.password, database=self.database, charset=self.charset, cursorclass=pymysql.cursors.DictCursor ) # 设置会话内存配额 with conn.cursor() as cursor: cursor.execute(f"SET SESSION tidb_mem_quota_query = {self.mem_quota}") conn.commit() logger.debug(f"创建新数据库连接，设置内存配额为 {self.mem_quota}") return conn except Exception as e: logger.error(f"创建数据库连接失败: {e}") raise def get_connection(self): """从连接池获取连接""" with self._lock: # 如果有空闲连接，返回一个 for conn in self._connections: if conn not in self._in_use: self._in_use.add(conn) logger.debug(f"从连接池获取现有连接") return conn # 如果没有空闲连接但可以创建新连接 if len(self._connections) < self.max_connections: conn = self._create_connection() self._connections.append(conn) self._in_use.add(conn) logger.debug(f"创建新连接，当前连接数: {len(self._connections)}") return conn # 连接池已满，等待 logger.warning("连接池已满，等待可用连接...") import time time.sleep(1) return self.get_connection() def release_connection(self, conn): """释放连接回连接池""" with self._lock: if conn in self._in_use: self._in_use.remove(conn) logger.debug(f"释放连接回连接池") def close_all(self): """关闭所有连接""" with self._lock: for conn in self._connections: try: conn.close() except: pass self._connections.clear() self._in_use.clear() logger.info("已关闭所有数据库连接") class DatabaseConfig: """数据库配置类""" def __init__(self, host='192.168.50.234', port=4000, user='root', password='123456', database='wind_data', charset='utf8mb4', max_connections=2, mem_quota=4 << 30): self.host = host self.port = port self.user = user self.password = password self.database = database self.charset = charset self.max_connections = max_connections self.mem_quota = mem_quota class ManufacturerInfo: """主机厂商信息库，用于合理性校验""" # 各厂商典型传动比范围（基于公开数据） TRANSMISSION_RATIOS = { # 直驱风机：传动比接近1 "金风科技": {"type": "直驱", "ratio_range": (0.95, 1.05)}, "远景能源": {"type": "直驱", "ratio_range": (0.95, 1.05)}, "明阳智能": {"type": "半直驱", "ratio_range": (30.0, 50.0)}, "运达股份": {"type": "双馈", "ratio_range": (70.0, 100.0)}, "上海电气": {"type": "双馈", "ratio_range": (70.0, 100.0)}, "东方电气": {"type": "双馈", "ratio_range": (70.0, 100.0)}, "华锐风电": {"type": "双馈", "ratio_range": (70.0, 100.0)}, "联合动力": {"type": "双馈", "ratio_range": (70.0, 100.0)}, "海装风电": {"type": "双馈", "ratio_range": (70.0, 100.0)}, "Vestas": {"type": "双馈", "ratio_range": (70.0, 100.0)}, "GE": {"type": "直驱", "ratio_range": (0.95, 1.05)}, "Siemens": {"type": "直驱", "ratio_range": (0.95, 1.05)}, } # 各容量等级典型叶轮转速范围（rpm） ROTOR_SPEED_RANGES = { 1500: (15.0, 20.0), # 1.5MW 2000: (14.0, 18.0), # 2.0MW 2500: (13.0, 17.0), # 2.5MW 3000: (12.0, 16.0), # 3.0MW 3500: (11.0, 15.0), # 3.5MW 4000: (10.0, 14.0), # 4.0MW 5000: (9.0, 13.0), # 5.0MW 6000: (8.0, 12.0), # 6.0MW } # 各类型风机典型发电机转速范围（rpm） GENERATOR_SPEED_RANGES = { "直驱": (10.0, 30.0), # 直驱：低速发电机 "双馈": (1000.0, 1800.0), # 双馈：接近同步转速 "半直驱": (300.0, 800.0), # 半直驱：中速发电机 } @classmethod def get_manufacturer_info(cls, manufacturer: str) -> Dict: """获取厂商信息""" manufacturer_lower = str(manufacturer).strip().lower() if manufacturer else "" for key, value in cls.TRANSMISSION_RATIOS.items(): if key.lower() in manufacturer_lower: return value return {"type": "未知", "ratio_range": (0.5, 200.0)} @classmethod def get_rotor_speed_range(cls, rated_capacity: int) -> Tuple[float, float]: """根据额定容量获取合理的叶轮转速范围""" if not rated_capacity: return (8.0, 25.0) # 找到最接近的容量等级 capacities = sorted(cls.ROTOR_SPEED_RANGES.keys()) closest_capacity = min(capacities, key=lambda x: abs(x - rated_capacity)) return cls.ROTOR_SPEED_RANGES.get(closest_capacity, (8.0, 25.0)) @classmethod def get_generator_speed_range(cls, turbine_type: str) -> Tuple[float, float]: """根据风机类型获取合理的发电机转速范围""" return cls.GENERATOR_SPEED_RANGES.get(turbine_type, (0.0, 2000.0)) class SCADADataProcessor: """SCADA数据处理类，用于计算额定转速和传动比""" @staticmethod def calculate_robust_mode(values: List[float], decimal_places: int = 1) -> float: """ 计算稳健的众数（使用核密度估计） Args: values: 数值列表 decimal_places: 保留的小数位数 Returns: 众数值 """ if not values: return 0.0 # 1. 数据清洗：移除异常值（使用IQR方法） if len(values) >= 10: q1, q3 = np.percentile(values, [25, 75]) iqr = q3 - q1 lower_bound = q1 - 1.5 * iqr upper_bound = q3 + 1.5 * iqr filtered_values = [v for v in values if lower_bound <= v <= upper_bound] if filtered_values: values = filtered_values # 2. 对值进行四舍五入处理 rounded_values = [round(v, decimal_places) for v in values] # 3. 使用Counter统计频率 counter = Counter(rounded_values) if not counter: return 0.0 # 4. 找到最高频率 max_count = max(counter.values()) # 5. 获取所有具有最高频率的值 modes = [value for value, count in counter.items() if count == max_count] # 6. 如果有多个众数，返回平均值 if len(modes) > 1: mode_value = sum(modes) / len(modes) logger.debug(f"多个众数: {modes}, 使用平均值: {mode_value}") else: mode_value = modes[0] logger.debug(f"众数统计: 值={mode_value}, 频次={max_count}, 总数据点={len(values)}") return mode_value @staticmethod def calculate_rated_parameters_with_validation( rotor_speeds: List[float], gen_speeds: List[float], manufacturer: str, rated_capacity: int ) -> Tuple[float, float, float, str]: """ 计算额定参数并进行合理性校验 Args: rotor_speeds: 叶轮转速列表 gen_speeds: 发电机转速列表 manufacturer: 制造商 rated_capacity: 额定容量(kW) Returns: Tuple[float, float, float, str]: (rated_rotor_spd, rated_gen_spd, transmission_ratio, turbine_type) """ if not rotor_speeds or not gen_speeds: logger.warning(f"数据不足: 转子转速点={len(rotor_speeds)}, 发电机转速点={len(gen_speeds)}") return 0.0, 0.0, 0.0, "未知" # 获取厂商信息 manu_info = ManufacturerInfo.get_manufacturer_info(manufacturer) expected_type = manu_info["type"] expected_ratio_range = manu_info["ratio_range"] # 获取合理的转速范围 rotor_range = ManufacturerInfo.get_rotor_speed_range(rated_capacity) gen_range = ManufacturerInfo.get_generator_speed_range(expected_type) logger.debug(f"厂商: {manufacturer}, 预期类型: {expected_type}, " f"传动比范围: {expected_ratio_range}, " f"叶轮转速范围: {rotor_range}, 发电机转速范围: {gen_range}") # 第一阶段：初始计算 # 叶轮转速：保留1位小数 rated_rotor_spd = round(SCADADataProcessor.calculate_robust_mode(rotor_speeds, 1), 1) # 发电机转速：先计算众数，然后取整 gen_mode = SCADADataProcessor.calculate_robust_mode(gen_speeds, 1) rated_gen_spd = int(round(gen_mode)) # 计算传动比：保留3位小数 if rated_rotor_spd > 0: transmission_ratio = round(rated_gen_spd / rated_rotor_spd, 3) else: transmission_ratio = 0.0 # 第二阶段：合理性校验与调整 adjustments = [] # 1. 校验叶轮转速 if not (rotor_range[0] <= rated_rotor_spd <= rotor_range[1]): adjustments.append(f"叶轮转速超出范围: {rated_rotor_spd:.1f} rpm, 合理范围: {rotor_range}") # 使用中位数并四舍五入到1位小数 rated_rotor_spd = round(float(statistics.median(rotor_speeds)), 1) rated_rotor_spd = max(rotor_range[0], min(rotor_range[1], rated_rotor_spd)) # 2. 校验发电机转速 if not (gen_range[0] <= rated_gen_spd <= gen_range[1]): adjustments.append(f"发电机转速超出范围: {rated_gen_spd} rpm, 合理范围: {gen_range}") # 使用中位数并取整 gen_median = float(statistics.median(gen_speeds)) rated_gen_spd = int(round(gen_median)) rated_gen_spd = int(max(gen_range[0], min(gen_range[1], rated_gen_spd))) # 3. 重新计算传动比 if rated_rotor_spd > 0: transmission_ratio = round(rated_gen_spd / rated_rotor_spd, 3) else: transmission_ratio = 0.0 # 4. 校验传动比 if not (expected_ratio_range[0] <= transmission_ratio <= expected_ratio_range[1]): adjustments.append(f"传动比超出范围: {transmission_ratio:.3f}, 合理范围: {expected_ratio_range}") # 如果传动比异常，基于预期类型调整 if expected_type == "直驱": # 直驱：传动比应接近1 transmission_ratio = 1.000 rated_gen_spd = int(round(rated_rotor_spd)) elif expected_type == "双馈": # 双馈：典型传动比90:1 transmission_ratio = 90.000 rated_gen_spd = int(round(rated_rotor_spd * transmission_ratio)) elif expected_type == "半直驱": # 半直驱：典型传动比40:1 transmission_ratio = 40.000 rated_gen_spd = int(round(rated_rotor_spd * transmission_ratio)) # 第三阶段：最终校验与类型判断 # 基于最终传动比判断类型 if transmission_ratio <= 1.2: turbine_type = "直驱" elif 1.2 < transmission_ratio <= 30: turbine_type = "半直驱" elif 30 < transmission_ratio <= 120: turbine_type = "双馈" else: turbine_type = "未知" # 如果计算的类型与预期不符，使用预期类型 if expected_type != "未知" and turbine_type != expected_type: adjustments.append(f"类型不符: 计算={turbine_type}, 预期={expected_type}") turbine_type = expected_type # 记录调整信息 if adjustments: logger.info(f"参数调整: {manufacturer} {rated_capacity}kW - " + "; ".join(adjustments)) # 最终格式化 # 叶轮转速：保留1位小数 rated_rotor_spd = round(rated_rotor_spd, 1) # 发电机转速：整数 rated_gen_spd = int(rated_gen_spd) # 传动比：保留3位小数 transmission_ratio = round(transmission_ratio, 3) logger.debug(f"最终结果: 转子转速={rated_rotor_spd:.1f} rpm, " f"发电机转速={rated_gen_spd} rpm, " f"传动比={transmission_ratio:.3f}, " f"类型={turbine_type}") return rated_rotor_spd, rated_gen_spd, transmission_ratio, turbine_type @staticmethod def calculate_rated_speeds_and_ratio( data: List[Dict[str, Any]], manufacturer: str = "", rated_capacity: int = 0 ) -> Tuple[float, float, float, str]: """ 计算额定叶轮转速、额定发电机转速、传动比和风机类型 Args: data: SCADA数据列表 manufacturer: 制造商（用于合理性校验） rated_capacity: 额定容量（用于合理性校验） Returns: Tuple[float, float, float, str]: (rated_rotor_spd, rated_gen_spd, transmission_ratio, turbine_type) """ if not data: return 0.0, 0.0, 0.0, "未知" # 提取数据 rotor_speeds = [] gen_speeds = [] for record in data: rotor_spd = record.get('rotor_spd') gen_spd = record.get('gen_spd') if rotor_spd is not None and gen_spd is not None: try: rotor_val = float(rotor_spd) gen_val = float(gen_spd) # 基本数据清洗 if 0 < rotor_val < 50 and 0 < gen_val < 2500: rotor_speeds.append(rotor_val) gen_speeds.append(gen_val) except (ValueError, TypeError): continue if not rotor_speeds or not gen_speeds: logger.warning(f"有效数据不足: 转子转速点={len(rotor_speeds)}, 发电机转速点={len(gen_speeds)}") return 0.0, 0.0, 0.0, "未知" # 计算额定参数 return SCADADataProcessor.calculate_rated_parameters_with_validation( rotor_speeds, gen_speeds, manufacturer, rated_capacity ) # SQL语句定义（保持不变） CREATE_MODEL_TURBINE_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS info_model_turbine ( id INT AUTO_INCREMENT PRIMARY KEY, no_model VARCHAR(255) NOT NULL COMMENT '机型唯一标识', model VARCHAR(100) COMMENT '机型', manufacturer VARCHAR(100) COMMENT '制造商', rated_capacity INT COMMENT '额定容量(kW)', cut_in_wind_speed DECIMAL(5, 2) COMMENT '切入风速(m/s)', cut_out_wind_speed DECIMAL(5, 2) COMMENT '切出风速(m/s)', rotor_diameter INT COMMENT '叶轮直径(m)', hub_height DECIMAL(10, 2) COMMENT '轮毂高度(m)', turbine_count INT DEFAULT 0 COMMENT '该机型风机数量', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY idx_no_model (no_model), INDEX idx_model (model), INDEX idx_manufacturer (manufacturer), INDEX idx_rated_capacity (rated_capacity) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='风机机型信息表'; """ ALTER_MODEL_TURBINE_TABLE_SQL = """ ALTER TABLE info_model_turbine ADD COLUMN IF NOT EXISTS rated_rotor_spd DECIMAL(10, 1) COMMENT '额定叶轮转速(rpm)', ADD COLUMN IF NOT EXISTS rated_gen_spd INT COMMENT '额定发电机转速(rpm)', ADD COLUMN IF NOT EXISTS transmission_ratio DECIMAL(10, 3) COMMENT '传动比', ADD COLUMN IF NOT EXISTS turbine_type VARCHAR(20) COMMENT '风机类型(直驱/双馈/半直驱)', ADD COLUMN IF NOT EXISTS calculation_time TIMESTAMP COMMENT '参数计算时间', ADD COLUMN IF NOT EXISTS data_points INT DEFAULT 0 COMMENT '用于计算的数据点数', ADD COLUMN IF NOT EXISTS calculation_status VARCHAR(20) DEFAULT 'pending' COMMENT '计算状态(pending/success/error/adjusted)', ADD COLUMN IF NOT EXISTS validation_info TEXT COMMENT '合理性校验信息'; """ UPDATE_MODEL_PARAMETERS_SQL = """ UPDATE info_model_turbine SET rated_rotor_spd = %s, rated_gen_spd = %s, transmission_ratio = %s, turbine_type = %s, calculation_time = %s, data_points = %s, calculation_status = %s, validation_info = %s, updated_at = CURRENT_TIMESTAMP WHERE no_model = %s """ class ModelTurbineManager: """风机机型信息管理器""" def __init__(self, db_config: DatabaseConfig): self.db_config = db_config self.connection_pool = None self._initialize_connection_pool() def _initialize_connection_pool(self): """初始化数据库连接池""" self.connection_pool = ConnectionPool( host=self.db_config.host, port=self.db_config.port, user=self.db_config.user, password=self.db_config.password, database=self.db_config.database, charset=self.db_config.charset, max_connections=self.db_config.max_connections, mem_quota=self.db_config.mem_quota ) logger.info(f"数据库连接池初始化完成，最大连接数: {self.db_config.max_connections}") # ... 其他方法保持不变 ... def calculate_and_update_parameters_for_model(self, no_model: str) -> Dict[str, Any]: """ 计算并更新指定机型的额定参数 Args: no_model: 机型标识 Returns: 计算结果的字典 """ try: # 先获取机型基本信息 model_info_sql = """ SELECT manufacturer, rated_capacity FROM info_model_turbine WHERE no_model = %s """ model_info = self.execute_query(model_info_sql, (no_model,)) if not model_info: logger.warning(f"机型 {no_model}: 未找到基本信息") return { "no_model": no_model, "success": False, "reason": "未找到机型基本信息", "data_points": 0 } manufacturer = model_info[0].get('manufacturer', '') rated_capacity = model_info[0].get('rated_capacity', 0) # 获取该机型的SCADA数据 scada_data = self.get_scada_data_for_no_model(no_model) if not scada_data: logger.warning(f"机型 {no_model}: 没有可用于计算的SCADA数据") return { "no_model": no_model, "success": False, "reason": "无SCADA数据", "data_points": 0 } # 计算额定参数（带合理性校验） rated_rotor_spd, rated_gen_spd, transmission_ratio, turbine_type = \ SCADADataProcessor.calculate_rated_speeds_and_ratio( scada_data, manufacturer, rated_capacity ) # 判断计算状态 if rated_rotor_spd <= 0 or transmission_ratio <= 0: calculation_status = "error" validation_info = "参数无效" success = False else: # 检查是否需要调整 manu_info = ManufacturerInfo.get_manufacturer_info(manufacturer) expected_ratio_range = manu_info["ratio_range"] rotor_range = ManufacturerInfo.get_rotor_speed_range(rated_capacity) gen_range = ManufacturerInfo.get_generator_speed_range(turbine_type) # 判断是否经过调整 adjustments = [] if not (rotor_range[0] <= rated_rotor_spd <= rotor_range[1]): adjustments.append(f"叶轮转速调整: {rated_rotor_spd:.1f} rpm") if not (gen_range[0] <= rated_gen_spd <= gen_range[1]): adjustments.append(f"发电机转速调整: {rated_gen_spd} rpm") if not (expected_ratio_range[0] <= transmission_ratio <= expected_ratio_range[1]): adjustments.append(f"传动比调整: {transmission_ratio:.3f}") if adjustments: calculation_status = "adjusted" validation_info = "; ".join(adjustments) else: calculation_status = "success" validation_info = "参数合理" success = True # 更新数据库 update_result = self.execute_update( UPDATE_MODEL_PARAMETERS_SQL, ( rated_rotor_spd, rated_gen_spd, transmission_ratio, turbine_type, datetime.now(), len(scada_data), calculation_status, validation_info, no_model ) ) if update_result > 0: log_msg = (f"机型 {no_model}: {calculation_status} - " f"叶轮转速={rated_rotor_spd:.1f} rpm, " f"发电机转速={rated_gen_spd} rpm, " f"传动比={transmission_ratio:.3f}, " f"类型={turbine_type}, " f"数据点={len(scada_data)}") if calculation_status == "adjusted": logger.info(log_msg + f", 调整: {validation_info}") else: logger.info(log_msg) return { "no_model": no_model, "success": success, "status": calculation_status, "rated_rotor_spd": rated_rotor_spd, "rated_gen_spd": rated_gen_spd, "transmission_ratio": transmission_ratio, "turbine_type": turbine_type, "data_points": len(scada_data), "validation_info": validation_info } else: logger.warning(f"机型 {no_model}: 数据库更新失败") return { "no_model": no_model, "success": False, "reason": "数据库更新失败", "status": "error", "rated_rotor_spd": rated_rotor_spd, "rated_gen_spd": rated_gen_spd, "transmission_ratio": transmission_ratio, "turbine_type": turbine_type, "data_points": len(scada_data) } except Exception as e: logger.error(f"计算机型 {no_model} 参数时出错: {e}") return { "no_model": no_model, "success": False, "reason": str(e), "status": "error", "data_points": 0 } # ... 其他方法保持不变 ... def main(): """主函数""" # 数据库配置 db_config = DatabaseConfig( # host="106.120.102.238", # port=44000, host="192.168.50.234", port=4000, user='root', password='123456', database='wind_data', charset='utf8mb4', max_connections=2, mem_quota=4 << 30 # 4GB ) # 创建管理器实例 manager = ModelTurbineManager(db_config) # 运行机型数据提取和参数计算流程 success = manager.run_model_extraction_pipeline( recreate_table=False, # 是否重新创建表 calculate_params=True # 是否计算额定参数 ) if success: logger.info("风机机型数据提取和参数计算成功完成！") else: logger.error("风机机型数据提取和参数计算失败！") if __name__ == "__main__": # 导入numpy用于统计计算 try: import numpy as np except ImportError: logger.warning("未找到numpy库，使用简单统计方法") # 定义简单的替代函数 def percentile(data, percentiles): """简单的百分位数计算""" if not data: return [0, 0] sorted_data = sorted(data) n = len(sorted_data) return [ sorted_data[int((n-1) * p / 100)] for p in percentiles ] # 临时创建numpy模块 class NumpyStub: @staticmethod def percentile(data, percentiles): return percentile(data, percentiles) np = NumpyStub() main()