from logging import Logger import pandas as pd from common.commonBusiness import CommonBusiness from algorithmContract.const import * from algorithmContract.confBusiness import * from utils.rdbmsUtil.databaseUtil import DatabaseUtil from sqlalchemy.orm import Session from sqlalchemy.sql import text class DALAnalyst: def __init__(self, logger: Logger, dbUtil: dict[str, DatabaseUtil]) -> None: self.logger = logger self.dbUtil = dbUtil def loadPowerFarmInfos(self, powerFarmID: str): """ 获取场站基础信息 """ dbUtil: DatabaseUtil = self.dbUtil[DATABASE_BusinessFoundationDb] with dbUtil.session_scope() as session: # 执行原生 SQL 查询 result = session.execute(text( f"SELECT field_code,company_code,field_name,density,state,engine_number,rated_capacity_number,province_id,province_name,city_id,city_name,longitude,latitude,elevation_height,power_contract_url FROM wind_field where del_state=0 and field_code='{powerFarmID}'")).fetchall() # 获取查询结果的列名 columns = ['field_code', 'company_code', 'field_name', 'density', 'state', 'engine_number', 'rated_capacity_number', 'province_id', 'province_name', 'city_id', 'city_name', 'longitude', 'latitude', 'elevation_height', 'power_contract_url'] # 将查询结果转换为 DataFrame dataFrame = pd.DataFrame(result, columns=columns) return dataFrame def loadTurbineInfos(self, powerFarmID: str): """ 获取风电机组基础信息 """ dbUtil: DatabaseUtil = self.dbUtil[DATABASE_BusinessFoundationDb] with dbUtil.session_scope() as session: # 执行原生 SQL 查询 result = session.execute(text( f"SELECT field_code,engine_code,engine_name,mill_type_code,rated_capacity,elevation_height,hub_height,state,longitude,latitude,sightcing FROM wind_engine_group where del_state=0 and field_code='{powerFarmID}'")).fetchall() # 获取查询结果的列名 columns = ['field_code', 'engine_code', 'engine_name', Field_MillTypeCode, 'rated_capacity', 'elevation_height', 'hub_height', 'state', 'longitude', 'latitude', 'sightcing'] # 将查询结果转换为 DataFrame dataFrame = pd.DataFrame(result, columns=columns) return dataFrame def loadDataTransfer(self, powerFarmID: str, dataBatchNum: str): """ 获取数据操作信息 """ dbUtil: DatabaseUtil = self.dbUtil[DATABASE_BusinessFoundationDb] with dbUtil.session_scope() as session: # 执行原生 SQL 查询 result = session.execute(text( f"SELECT field_code, batch_code, engine_count, transfer_type, transfer_addr, time_granularity FROM data_transfer where field_code='{powerFarmID}' and batch_code='{dataBatchNum}' ")).fetchall() # 获取查询结果的列名 columns = ['field_code', 'batch_code', 'engine_count', 'transfer_type', 'transfer_addr', 'time_granularity'] # 将查询结果转换为 DataFrame dataFrame = pd.DataFrame(result, columns=columns) return dataFrame def loadTurbineModelInfos(self, turbineModels: list): """ 获取型号基础信息 """ if len(turbineModels) <= 0: return pd.DataFrame() turbineModelStr = ", ".join( f"'{model}'" for model in turbineModels) # 使用%s作为占位符，稍后可以替换为实际值 dbUtil: DatabaseUtil = self.dbUtil[DATABASE_BusinessFoundationDb] with dbUtil.session_scope() as session: # 执行原生 SQL 查询 result = session.execute(text(f"SELECT mill_type_code,machine_type_code,manufacturer_name,manufacturer_code,brand,tower_height,vane_long,curved_motion_type,combination,power_criterion_url,rotor_diameter,rotational_speed_ratio,rated_wind_speed,rated_cut_in_windspeed,rated_cut_out_windspeed FROM wind_engine_mill where del_state=0 and state=1 and mill_type_code in ({turbineModelStr})" )).fetchall() # 获取查询结果的列名 columns = [Field_MillTypeCode, Field_MachineTypeCode, Field_ManufacturerName, Field_ManufacturerCode, Field_Brand, Field_HubHeight, Field_VaneLong, Field_MotionType, Field_Combination, Field_PowerCriterionURL, Field_RotorDiameter, Field_RSR, Field_RatedWindSpeed, Field_CutInWS, Field_CutOutWS] # 将查询结果转换为 DataFrame dataFrame = pd.DataFrame(result, columns=columns) return dataFrame def loadWeatherStationInfos(self, powerFarmID: str): """ 获取气象站（测风塔）基础信息 """ dbUtil: DatabaseUtil = self.dbUtil[DATABASE_BusinessFoundationDb] with dbUtil.session_scope() as session: # 执行原生 SQL 查询 result = session.execute(text( f"select y.field_code ,x.anemometer_code ,x.anemometer_name ,x.longitude ,x.latitude from anemometer_tower as x inner join anemometer_tower_relation as y on x.anemometer_code =y.tower_code where x.del_state=0 and x.state=1 and y.field_code='{powerFarmID}'")).fetchall() # 获取查询结果的列名 columns = ['field_code', 'anemometer_code', 'anemometer_name', 'longitude', 'latitude'] # 将查询结果转换为 DataFrame dataFrame = pd.DataFrame(result, columns=columns) return dataFrame def processContractData(self, common: CommonBusiness, powerFarmID: str, airDensity: float, turbineModelInfo: pd.DataFrame): """ 获取合同功率曲线数据 """ dataFrameMerge = pd.DataFrame() turbineModels = turbineModelInfo[Field_MillTypeCode] turbineModelStr = ", ".join(f"'{model}'" for model in turbineModels) dbUtil: DatabaseUtil = self.dbUtil[DATABASE_BusinessFoundationDb] with dbUtil.session_scope() as session: # 执行原生 SQL 查询 result = session.execute(text( f"SELECT field_code, mill_type_code,active_power,wind_speed as wind_velocity FROM power_word_relation_contract where field_code='{powerFarmID}' and mill_type_code in ({turbineModelStr})")).fetchall() # 获取查询结果的列名 columns = [Field_PowerFarmCode, Field_MillTypeCode, Field_ActiverPower, Field_WindSpeed] # 将查询结果转换为 DataFrame contractPowerCurves = pd.DataFrame(result, columns=columns) grouped = contractPowerCurves.groupby(Field_MillTypeCode) for name, group in grouped: print("current turbine model :", name) model = turbineModelInfo[turbineModelInfo[Field_MillTypeCode] == name] if len(model) <= 0: continue dataFrame = common.calculateCp2( group, airDensity, model[Field_RotorDiameter].iloc[0], Field_WindSpeed, Field_ActiverPower) dataFrameMerge = pd.concat( [dataFrameMerge, dataFrame], axis=0, sort=False) return dataFrameMerge