WTOAAM/dataAnalysisBehavior/behavior/baseAnalyst.py

# baseAnalyst.py

import concurrent.futures
import logging
from abc import ABC, abstractmethod
from logging import Logger
from typing import Any, Tuple
from urllib.parse import quote

import numpy as np
import pandas as pd
import traceback
from algorithmContract.confBusiness import *
from algorithmContract.configAnalysis import ConfigAnalysis
from algorithmContract.contract import Contract
from behavior.dataMarker import DataMarker
from common.appConfig import GetBusinessDbUtil
from common.commonBusiness import CommonBusiness
from numpy.typing import NDArray
from sqlalchemy.sql import text
from utils.directoryUtil import DirectoryUtil as dir
from utils.minioUtil.threadSafeMinioClient import ThreadSafeMinioClient
from utils.rdbmsUtil.databaseUtil import DatabaseUtil


class BaseAnalyst(ABC):
    def __init__(self, logger: Logger, dbUtil: dict[str, DatabaseUtil], minioUtil: ThreadSafeMinioClient, conf: Contract,
                 powerFarmInfo: pd.DataFrame, turbineInfo: pd.DataFrame, turbineModelInfo: pd.DataFrame, dataTransfer: pd.DataFrame,
                 weatherStationInfo: pd.DataFrame, dataFrameContractOfTurbine: pd.DataFrame):
        self.logger = logger
        self.dbUtil = dbUtil
        self.minioUtil = minioUtil
        self.conf = conf
        # 定义风速区间
        self.binsWindSpeed: NDArray[np.floating[Any]] = np.arange(0, 26, 0.5)

        self.dataMarker = DataMarker()
        self.common = CommonBusiness()

        self.customFilterStatusOfTurbine = "valueTurbineStatus"
        self.customFilterPitchAngle = "valuePitchAngle"
        self.customFilterWindSpeed = "valueWindSpeed"
        self.customFilterActivePower = "valueActivePower"
        self.customFilterGeneratorSpeed = "valueGeneratorSpeed"

        # 加载所有新能源场站信息
        self.powerFarmInfo = powerFarmInfo
        self.currPowerFarmInfo = self.common.getPowerFarm(
            self.conf.dataContract.dataFilter.powerFarmID, self.powerFarmInfo)

        # 加载所有风电机组信息
        self.turbineInfo = turbineInfo

        # 加载数据转换信息
        # self.dataTransfer = dataTransfer

        # 加载机型信息
        self.turbineModelInfo = turbineModelInfo

        # 加载所有测风塔信息
        self.weatherStationInfo = weatherStationInfo

        # 加载所有新能源场站，及所属风电机组机型的合同功率曲线
        self.dataFrameContractOfTurbine = dataFrameContractOfTurbine

        # 图表 轴系设置
        # 常值
        self.RatedPowerOfTurbine = self.turbineInfo[Field_RatedPower].iloc[0]

        # 图表 轴系设置
        # 轴系 发电机转速
        iGeneratorSpeed = "IgeneratorSpeed"
        dGeneratorSpeed = "DgeneratorSpeed"
        if self.turbineModelInfo[Field_MotionType] .iloc[0] == 2:
            # 直驱 发电机转速
            self.axisStepGeneratorSpeed = conf.dataContract.graphSets[dGeneratorSpeed].step if not self.common.isNone(
                conf.dataContract.graphSets[dGeneratorSpeed]) and not self.common.isNone(
                conf.dataContract.graphSets[dGeneratorSpeed].step) else 1
            self.axisLowerLimitGeneratorSpeed = conf.dataContract.graphSets[dGeneratorSpeed].min if not self.common.isNone(
                conf.dataContract.graphSets[dGeneratorSpeed].min) else 5
            self.axisUpperLimitGeneratorSpeed = conf.dataContract.graphSets[dGeneratorSpeed].max if not self.common.isNone(
                conf.dataContract.graphSets[dGeneratorSpeed].max) else 25
        else:
            # 非直驱 发电机转速
            self.axisStepGeneratorSpeed = conf.dataContract.graphSets[iGeneratorSpeed].step if not self.common.isNone(
                conf.dataContract.graphSets[iGeneratorSpeed]) and not self.common.isNone(
                conf.dataContract.graphSets[iGeneratorSpeed].step) else 200
            self.axisLowerLimitGeneratorSpeed = conf.dataContract.graphSets[iGeneratorSpeed].min if not self.common.isNone(
                conf.dataContract.graphSets[iGeneratorSpeed].min) else 1000
            self.axisUpperLimitGeneratorSpeed = conf.dataContract.graphSets[iGeneratorSpeed].max if not self.common.isNone(
                conf.dataContract.graphSets[iGeneratorSpeed].max) else 2000
# --------以下为原写法，未区分驱动方式直接选择了非直驱--------
        # # 直驱 发电机转速
        # self.axisStepGeneratorSpeedWithDirect = conf.dataContract.graphSets[dGeneratorSpeed].step if not self.common.isNone(
        #     conf.dataContract.graphSets[dGeneratorSpeed]) and not self.common.isNone(
        #     conf.dataContract.graphSets[dGeneratorSpeed].step) else 5
        # self.axisLowerLimitGeneratorSpeedWithDirect = conf.dataContract.graphSets[dGeneratorSpeed].min if not self.common.isNone(
        #     conf.dataContract.graphSets[dGeneratorSpeed].min) else 0
        # self.axisUpperLimitGeneratorSpeedWithDirect = conf.dataContract.graphSets[dGeneratorSpeed].max if not self.common.isNone(
        #     conf.dataContract.graphSets[dGeneratorSpeed].max) else 30
        #     # 非直驱 发电机转速
        # self.axisStepGeneratorSpeed = conf.dataContract.graphSets[iGeneratorSpeed].step if not self.common.isNone(
        #     conf.dataContract.graphSets[iGeneratorSpeed]) and not self.common.isNone(
        #     conf.dataContract.graphSets[iGeneratorSpeed].step) else 200
        # self.axisLowerLimitGeneratorSpeed = conf.dataContract.graphSets[iGeneratorSpeed].min if not self.common.isNone(
        #     conf.dataContract.graphSets[iGeneratorSpeed].min) else 1000
        # self.axisUpperLimitGeneratorSpeed = conf.dataContract.graphSets[iGeneratorSpeed].max if not self.common.isNone(
        #     conf.dataContract.graphSets[iGeneratorSpeed].max) else 2000

        # 轴系 发电机转矩
        iGgeneratorTorque = "IgeneratorTorque"
        dGgeneratorTorque = "DgeneratorTorque"
        # if turbineModelInfo[Field_MotionType].iloc[0]==2 :
        #             # 直驱 发电机转矩
        #     self.axisStepGeneratorTorque = conf.dataContract.graphSets[dGgeneratorTorque].step if not self.common.isNone(
        #         conf.dataContract.graphSets[dGgeneratorTorque]) and not self.common.isNone(
        #         conf.dataContract.graphSets[dGgeneratorTorque].step) else 10000
        #     self.axisLowerLimitGeneratorTorque = conf.dataContract.graphSets[dGgeneratorTorque].min if not self.common.isNone(
        #         conf.dataContract.graphSets[dGgeneratorTorque].min) else 0
        #     self.axisUpperLimitGeneratorTorque = conf.dataContract.graphSets[dGgeneratorTorque].max if not self.common.isNone(
        #         conf.dataContract.graphSets[dGgeneratorTorque].max) else 100000
        # else:
        #     # 非直驱 发电机转矩
        #     self.axisStepGeneratorTorque = conf.dataContract.graphSets[iGgeneratorTorque].step if not self.common.isNone(
        #         conf.dataContract.graphSets[iGgeneratorTorque]) and not self.common.isNone(
        #         conf.dataContract.graphSets[iGgeneratorTorque].step) else 2000
        #     self.axisLowerLimitGeneratorTorque = conf.dataContract.graphSets[iGgeneratorTorque].min if not self.common.isNone(
        #         conf.dataContract.graphSets[iGgeneratorTorque].min) else 0
        #     self.axisUpperLimitGeneratorTorque = conf.dataContract.graphSets[iGgeneratorTorque].max if not self.common.isNone(
        #         conf.dataContract.graphSets[iGgeneratorTorque].max) else 12000
# --------以下为原写法，未区分驱动方式直接选择了非直驱--------
        # 直驱 发电机转矩
        self.axisStepGeneratorTorqueWithDirect = conf.dataContract.graphSets[dGgeneratorTorque].step if not self.common.isNone(
            conf.dataContract.graphSets[dGgeneratorTorque]) and not self.common.isNone(
            conf.dataContract.graphSets[dGgeneratorTorque].step) else 10000
        self.axisLowerLimitGeneratorTorqueWithDirect = conf.dataContract.graphSets[dGgeneratorTorque].min if not self.common.isNone(
            conf.dataContract.graphSets[dGgeneratorTorque].min) else 0
        self.axisUpperLimitGeneratorTorqueWithDirect = conf.dataContract.graphSets[dGgeneratorTorque].max if not self.common.isNone(
            conf.dataContract.graphSets[dGgeneratorTorque].max) else 100000
        # 非直驱 发电机转矩
        self.axisStepGeneratorTorque = conf.dataContract.graphSets[iGgeneratorTorque].step if not self.common.isNone(
            conf.dataContract.graphSets[iGgeneratorTorque]) and not self.common.isNone(
            conf.dataContract.graphSets[iGgeneratorTorque].step) else 2000
        self.axisLowerLimitGeneratorTorque = conf.dataContract.graphSets[iGgeneratorTorque].min if not self.common.isNone(
            conf.dataContract.graphSets[iGgeneratorTorque].min) else 0
        self.axisUpperLimitGeneratorTorque = conf.dataContract.graphSets[iGgeneratorTorque].max if not self.common.isNone(
            conf.dataContract.graphSets[iGgeneratorTorque].max) else 12000

        # 轴系 有功功率
        activePower = "activePower"
        self.axisStepActivePower = conf.dataContract.graphSets[activePower].step if not self.common.isNone(
            conf.dataContract.graphSets[activePower]) and not self.common.isNone(
            conf.dataContract.graphSets[activePower].step) else 250
        self.axisLowerLimitActivePower = conf.dataContract.graphSets[activePower].min if not self.common.isNone(
            conf.dataContract.graphSets[activePower].min) else 0
        self.axisUpperLimitActivePower = conf.dataContract.graphSets[activePower].max if not self.common.isNone(
            conf.dataContract.graphSets[activePower].max) else self.RatedPowerOfTurbine*1.2

        # 轴系 桨距角
        pitchAngle = "pitchAngle"
        self.axisStepPitchAngle = conf.dataContract.graphSets[pitchAngle].step if not self.common.isNone(
            conf.dataContract.graphSets[pitchAngle]) and not self.common.isNone(
            conf.dataContract.graphSets[pitchAngle].step) else 2
        self.axisLowerLimitPitchAngle = conf.dataContract.graphSets[pitchAngle].min if not self.common.isNone(
            conf.dataContract.graphSets[pitchAngle].min) else -2
        self.axisUpperLimitPitchAngle = conf.dataContract.graphSets[pitchAngle].max if not self.common.isNone(
            conf.dataContract.graphSets[pitchAngle].max) else 28

        # 轴系 风能利用系数
        cp = "cp"
        self.axisStepCp = conf.dataContract.graphSets[cp].step if not self.common.isNone(
            conf.dataContract.graphSets[cp]) and not self.common.isNone(
            conf.dataContract.graphSets[cp].step) else 0.1
        self.axisLowerLimitCp = conf.dataContract.graphSets[cp].min if not self.common.isNone(
            conf.dataContract.graphSets[cp].min) else 0
        self.axisUpperLimitCp = conf.dataContract.graphSets[cp].max if not self.common.isNone(
            conf.dataContract.graphSets[cp].max) else 1

        # 轴系 叶尖速比
        tsr = "tsr"
        self.axisStepTSR = conf.dataContract.graphSets[tsr].step if not self.common.isNone(
            conf.dataContract.graphSets[tsr]) and not self.common.isNone(
            conf.dataContract.graphSets[tsr].step) else 5
        self.axisLowerLimitTSR = conf.dataContract.graphSets[tsr].min if not self.common.isNone(
            conf.dataContract.graphSets[tsr].min) else 0
        self.axisUpperLimitTSR = conf.dataContract.graphSets[tsr].max if not self.common.isNone(
            conf.dataContract.graphSets[tsr].max) else 20

    @abstractmethod
    def typeAnalyst(self):
        pass

    def getOutputAnalysisDir(self):
        """
        获取当前分析的输出目录
        """
        outputAnalysisDir = f"output/{self.conf.dataContract.dataFilter.powerFarmID}/{self.conf.dataContract.dataFilter.dataBatchNum}/{self.typeAnalyst()}/{self.conf.dataContract.autoOrManual}"

        if not dir.check_directory_exists(outputAnalysisDir):
            dir.create_directory(outputAnalysisDir)

        return outputAnalysisDir

    def filterCommon(self, dataFrame: pd.DataFrame, conf: Contract):
        # Filter rows where turbine state
        if Field_StatusOfTurbine in dataFrame.columns and self.customFilterStatusOfTurbine in conf.dataContract.dataFilter.customFilter.keys() and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterStatusOfTurbine]):
            stateTurbine = conf.dataContract.dataFilter.customFilter[self.customFilterStatusOfTurbine]
            dataFrame = dataFrame[dataFrame[Field_StatusOfTurbine].isin(
                stateTurbine)]

        # Filter rows where wind speed
        if Field_WindSpeed in dataFrame.columns and self.customFilterWindSpeed in conf.dataContract.dataFilter.customFilter.keys() and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterWindSpeed]) and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterWindSpeed].min):
            windSpeedMin = conf.dataContract.dataFilter.customFilter[self.customFilterWindSpeed].min
            dataFrame = dataFrame[(
                dataFrame[Field_WindSpeed] >= windSpeedMin)]

        if Field_WindSpeed in dataFrame.columns and self.customFilterWindSpeed in conf.dataContract.dataFilter.customFilter.keys() and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterWindSpeed]) and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterWindSpeed].max):
            windSpeedMax = conf.dataContract.dataFilter.customFilter[self.customFilterWindSpeed].max
            dataFrame = dataFrame[(
                dataFrame[Field_WindSpeed] < windSpeedMax)]

        # Filter rows where pitch
        if Field_PitchAngel1 in dataFrame.columns and self.customFilterPitchAngle in conf.dataContract.dataFilter.customFilter.keys() and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterPitchAngle]) and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterPitchAngle].min):
            anglePitchMin = conf.dataContract.dataFilter.customFilter[
                self.customFilterPitchAngle].min
            dataFrame = dataFrame[(
                dataFrame[Field_PitchAngel1] >= anglePitchMin)]

        if Field_PitchAngel1 in dataFrame.columns and self.customFilterPitchAngle in conf.dataContract.dataFilter.customFilter.keys() and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterPitchAngle]) and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterPitchAngle].max):
            anglePitchMax = conf.dataContract.dataFilter.customFilter[
                self.customFilterPitchAngle].max
            dataFrame = dataFrame[(
                dataFrame[Field_PitchAngel1] < anglePitchMax)]

        # Filter rows where power
        if Field_ActiverPower in dataFrame.columns and self.customFilterActivePower in conf.dataContract.dataFilter.customFilter.keys() and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterActivePower]) and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterActivePower].min):
            activePowerMin = conf.dataContract.dataFilter.customFilter[
                self.customFilterActivePower].min
            dataFrame = dataFrame[(
                dataFrame[Field_ActiverPower] >= activePowerMin)]

        if Field_ActiverPower in dataFrame.columns and self.customFilterActivePower in conf.dataContract.dataFilter.customFilter.keys() and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterActivePower]) and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterActivePower].max):
            activePowerMax = conf.dataContract.dataFilter.customFilter[
                self.customFilterActivePower].max
            dataFrame = dataFrame[(
                dataFrame[Field_ActiverPower] < activePowerMax)]

        # Filter rows where generator speed
        if Field_GeneratorSpeed in dataFrame.columns and self.customFilterGeneratorSpeed in conf.dataContract.dataFilter.customFilter.keys() and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterGeneratorSpeed]) and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterGeneratorSpeed].min):
            speedGeneratorMin = conf.dataContract.dataFilter.customFilter[
                self.customFilterGeneratorSpeed].min
            dataFrame = dataFrame[(
                dataFrame[Field_GeneratorSpeed] >= speedGeneratorMin)]

        if Field_GeneratorSpeed in dataFrame.columns and self.customFilterGeneratorSpeed in conf.dataContract.dataFilter.customFilter.keys() and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterGeneratorSpeed]) and not self.common.isNone(conf.dataContract.dataFilter.customFilter[self.customFilterGeneratorSpeed].max):
            speedGeneratorMax = conf.dataContract.dataFilter.customFilter[
                self.customFilterGeneratorSpeed].max
            dataFrame = dataFrame[(
                dataFrame[Field_GeneratorSpeed] < speedGeneratorMax)]

        # if not self.common.isNone(confData.field_activePowerSet) and confData.field_activePowerSet in dataFrame.columns:
        #     dataFrame = dataFrame[dataFrame[confData.field_activePowerSet]
        #                           == confData.rated_power]

        # if not self.common.isNone(confData.field_activePowerAvailable) and confData.field_activePowerAvailable in dataFrame.columns \
        #       and self.node_activePowerAvailable in confData.filter and not self.common.isNone(confData.filter[self.node_activePowerAvailable]):
        #     state = confData.filter[self.node_activePowerAvailable]
        #     dataFrame = dataFrame[dataFrame[confData.field_activePowerAvailable].isin(
        #         state)]

        return dataFrame

    def filterCustomForTurbine(self, dataFrame: pd.DataFrame, conf: Contract):
        return self.filterCommon(dataFrame, conf)

    def analysisOfTurbine(self,
                          outputAnalysisDir,
                          outputFilePath,
                          conf: Contract,
                          turbineCode):
        return self.turbineAnalysis(outputAnalysisDir,
                                    outputFilePath, conf, turbineCode)

    def turbineAnalysis(self,
                        outputAnalysisDir,
                        outputFilePath,
                        conf: Contract,
                        turbineCode):
        return pd.DataFrame()

    def filterCustomForTurbines(self, dataFrame: pd.DataFrame, conf: Contract):
        return self.filterCommon(dataFrame, conf)

    def analysisOfTurbines(self,  outputAnalysisDir, conf: Contract, turbineCodes):
        # self.logger.info(
        #     f"typeAnalyst: {self.typeAnalyst()}  method: analysisOfTurbines , turbineCodes : {turbineCodes}")
        return self.turbinesAnalysis(outputAnalysisDir, conf, turbineCodes)

    def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes):
        return pd.DataFrame()

    def userDataFrame(self, dictionary: dict, configs: list[ConfigAnalysis], analyst) -> pd.DataFrame:
        timeGranularity = next(
            (config.scada for config in configs if config.className.lower() == type(analyst).__name__.lower()), None)
        return pd.DataFrame() if dictionary[timeGranularity].empty else dictionary[timeGranularity]

    def calculateAngleIncluded(self, array1, array2):
        """
        计算两个相同长度角度数组中两两对应角度值的偏差。
        结果限制在-90°到+90°之间，并保留两位小数。

        参数:
        array1 (list): 第一个角度数组
        array2 (list): 第二个角度数组

        返回:
        list: 两两对应角度的偏差列表
        """
        deviations = []
        for angle1, angle2 in zip(array1, array2):
            # 计算原始偏差
            deviation = angle1-angle2

            # 调整偏差，使其位于-180°到+180°范围内
            if deviation == 0.0:
                deviation = 0.0
            else:
                deviation = (deviation + 180) % 360 - 180

            # 将偏差限制在-90°到+90°范围内
            if deviation > 90:
                deviation -= 180
            elif deviation < -90:
                deviation += 180

            # 保留两位小数
            deviations.append(round(deviation, 2))

        return deviations

    def recalculationOfIncludedAngle(self, dataFrame: pd.DataFrame, fieldAngleIncluded, fieldWindDirect, fieldNacellePos):
        """
        依据机舱位置（角度）、风向计算两者夹角
        """
        if fieldNacellePos in dataFrame.columns and fieldWindDirect in dataFrame.columns and not dataFrame[fieldNacellePos].isna().all() and not dataFrame[fieldWindDirect].isna().all():
            # 计算 angle_included_list
            angle_included_list = self.calculateAngleIncluded(
                dataFrame[fieldWindDirect], dataFrame[fieldNacellePos])

            # 检查 angle_included_list 的长度是否与 dataFrame 一致
            if len(angle_included_list) != len(dataFrame):
                raise ValueError(
                    "The length of the calculated angle_included_list does not match the length of the DataFrame")

            # 创建一个新的列来保存计算结果
            dataFrame['Calculated_AngleIncluded'] = angle_included_list

            # 使用新列填充原列的缺失值
            dataFrame[Field_AngleIncluded] = dataFrame[Field_AngleIncluded].fillna(
                dataFrame['Calculated_AngleIncluded'])

            # 删除临时列
            dataFrame.drop(columns=['Calculated_AngleIncluded'], inplace=True)

    def recalculationOfFieldPowerFloor(self, dataFrame: pd.DataFrame, fieldActivePower):
        """
        功率计算
        """
        if fieldActivePower in dataFrame.columns:
            dataFrame[Field_PowerFloor] = dataFrame[fieldActivePower].apply(
                lambda x: int(
                    x / 10) * 10 if pd.notnull(x) or pd.notna(x) else np.nan  # 保留NaN值
            )

    def recalculationOfFieldWindSpeedFloor(self, dataFrame: pd.DataFrame, fieldWindSpeed):
        """
        风速计算
        """
        if fieldWindSpeed in dataFrame.columns:
            dataFrame[Field_WindSpeedFloor] = dataFrame[fieldWindSpeed].apply(
                lambda x: int(x/1) +
                0.5 if pd.notnull(x) or pd.notna(x) else np.nan
            )

    def recalculationOfGeneratorSpeed(self, turbineModelInfo: pd.Series, dataFrame: pd.DataFrame, fieldRotorSpeed, fieldGeneratorSpeed, rotationalSpeedRatio: float):
        """
        风电机组发电机转速再计算，公式：转速比=发电机转速/叶轮或主轴转速
        """

        if fieldGeneratorSpeed in dataFrame.columns and fieldRotorSpeed in dataFrame.columns:

            if turbineModelInfo[Field_MotionType] == 2:
                dataFrame[fieldGeneratorSpeed] = dataFrame[fieldGeneratorSpeed].fillna(
                    dataFrame[fieldRotorSpeed])
                # dataFrame[fieldGeneratorSpeed] =dataFrame[fieldRotorSpeed]
            else:
                dataFrame[fieldGeneratorSpeed] = dataFrame[fieldGeneratorSpeed].fillna(
                    dataFrame[fieldRotorSpeed]*rotationalSpeedRatio)
                # dataFrame[fieldGeneratorSpeed] = dataFrame[fieldRotorSpeed]*rotationalSpeedRatio

    def recalculationOfRotorSpeed(self, turbineModelInfo: pd.Series, dataFrame: pd.DataFrame, fieldRotorSpeed, fieldGeneratorSpeed, rotationalSpeedRatio: float):
        """
        风电机组发电机转速再计算，公式：转速比=发电机转速/叶轮或主轴转速
        """
        if fieldRotorSpeed in dataFrame.columns and fieldGeneratorSpeed in dataFrame.columns:
            if turbineModelInfo[Field_MotionType] == 2:
                dataFrame[fieldRotorSpeed] = dataFrame[fieldRotorSpeed].fillna(
                    dataFrame[fieldGeneratorSpeed])
            else:
                dataFrame[fieldRotorSpeed] = dataFrame[fieldRotorSpeed].fillna(
                    dataFrame[fieldGeneratorSpeed] / rotationalSpeedRatio)

    def recalculationOfRotorTorque(self, dataFrame: pd.DataFrame, fieldGeneratorTorque, fieldActivePower, fieldGeneratorSpeed):
        """
        风电机组发电机转矩计算，P的单位换成KW转矩计算公式：
        P*1000= pi/30*T*n
        30000/pi*P=T*n
        30000/3.1415926*P=T*n
        9549.297*p=T*n
        其中：n为发电机转速，p为有功功率，T为转矩
        """
        if fieldGeneratorTorque in dataFrame.columns and fieldActivePower in dataFrame.columns and fieldGeneratorSpeed in dataFrame.columns:
            dataFrame[Field_GeneratorTorque] = dataFrame[Field_GeneratorTorque].fillna(
                9549.297 * dataFrame[fieldActivePower]/dataFrame[fieldGeneratorSpeed])

    def recalculation(self, turbineModelInfo: pd.Series, dataFrame: pd.DataFrame):
        """
        再计算数据测点
        参数：
        dataFrame 原始数据
        conf  配置数据
        """
        self.recalculationOfFieldPowerFloor(dataFrame, Field_ActiverPower)

        self.recalculationOfFieldWindSpeedFloor(dataFrame, Field_WindSpeed)

        self.recalculationOfGeneratorSpeed(
            turbineModelInfo, dataFrame, Field_RotorSpeed, Field_GeneratorSpeed, self.turbineModelInfo[Field_RSR].iloc[0])

        self.recalculationOfRotorSpeed(turbineModelInfo, dataFrame, Field_RotorSpeed,
                                       Field_GeneratorSpeed, self.turbineModelInfo[Field_RSR].iloc[0])

        self.recalculationOfRotorTorque(
            dataFrame, Field_GeneratorTorque, Field_ActiverPower, Field_GeneratorSpeed)

        self.recalculationOfIncludedAngle(
            dataFrame, Field_AngleIncluded, Field_WindDirection, Field_NacPos)

        self.common.calculateTSR(
            dataFrame, self.turbineModelInfo[Field_RotorDiameter].iloc[0])

        self.common.calculateCpOfSingleTurbine(
            dataFrame, self.currPowerFarmInfo[Field_AirDensity], turbineModelInfo[Field_RotorDiameter])

    def processDateTimeForAll(self, dataFrame: pd.DataFrame):
        dataFrame[Field_UnixYearMonth] = pd.to_datetime(
            dataFrame[Field_Time], format="%Y-%m").apply(lambda x: x.timestamp())

        dataFrame[Field_YearMonth] = dataFrame[Field_Time].dt.strftime('%Y-%m')

        dataFrame[Field_YearMonthDay] = dataFrame[Field_Time].dt.strftime(
            '%Y-%m-%d')

        return dataFrame

    def processDateTime(self, dataFrame: pd.DataFrame, fieldTime: str = None):

        if Field_Time not in dataFrame.columns:
            self.logger.info(
                "Field_Time is not in dataFrame columns, Do not handle time processing")
            return dataFrame

        if fieldTime is None:
            return self.processDateTimeForAll(dataFrame)

        timeSeries = pd.Series
        if fieldTime == Field_UnixYearMonth:
            timeSeries = pd.to_datetime(
                dataFrame[Field_Time], format="%Y-%m").apply(lambda x: x.timestamp())

        if fieldTime == Field_YearMonth:
            timeSeries = dataFrame[Field_Time].dt.strftime('%Y-%m')

        if fieldTime == Field_YearMonthDay:
            timeSeries = dataFrame[Field_Time].dt.strftime('%Y-%m-%d')

        dataFrame[fieldTime] = timeSeries

        return dataFrame

    def loadData(self, powerFarmID: str, timeGranularity: str, turbineCode: str, select: str, condition: str) -> pd.DataFrame:
        selectStr = ", ".join(f"{field}" for field in select)
        businessDB: DatabaseUtil = GetBusinessDbUtil()
        

        # query_text = f"""SELECT wind_turbine_number,time_stamp, active_power,wind_velocity,cabin_position,true_wind_direction, rotor_speed, generator_speed,actual_torque,yaw_error1,outside_cabin_temperature,cabin_temperature,main_bearing_temperature,gearboxmedium_speed_shaftbearing_temperature,gearbox_low_speed_shaft_bearing_temperature,gearbox_high_speed_shaft_bearing_temperature,generatordrive_end_bearing_temperature,generatornon_drive_end_bearing_temperature,generator_winding1_temperature,pitch_angle_blade_1,pitch_angle_blade_2,pitch_angle_blade_3,front_back_vibration_of_the_cabin,side_to_side_vibration_of_the_cabin
        #                    FROM `{dataBatchNum}_{timeGranularity}`
        #                    WHERE wind_turbine_number IN ({turbineCode}) AND {condition}"""
        query_text = f"""SELECT {selectStr}
                            FROM `{powerFarmID}_{timeGranularity}`
                            WHERE wind_turbine_number IN ({turbineCode}) AND {condition}"""
              
        # select = [Field_DeviceCode, Field_Time, Field_ActiverPower, Field_WindSpeed, Field_NacPos, Field_WindDirection, Field_RotorSpeed, Field_GeneratorSpeed, Field_GeneratorTorque, Field_AngleIncluded, Field_EnvTemp, Field_NacTemp, Field_MainBearTemp, Field_GbMsBearTemp, Field_GbLsBearTemp,
        #       Field_GbHsBearTemp, Field_GeneratorDE, Field_GeneratorNDE, Field_GenWiTemp1, Field_PitchAngel1, Field_PitchAngel2, Field_PitchAngel3, Field_NacFbVib, Field_NacLrVib]

        chunks = []
        with businessDB.session_scope() as session:
            # query_result = session.execute(text(query_text)).fetchall()
            
            # dataFrame = pd.DataFrame(query_result, columns=select)

        # return dataFrame

            conn=session.connection()
            result=conn.execution_options(stream_results=True).execute(text(query_text))
             # 每次取 50000 行，可根据实际情况调整
            while True:
                rows = result.fetchmany(50000)
                if not rows:
                    break
                chunks.append(pd.DataFrame(rows, columns=select))
        
        if not chunks:
            return pd.DataFrame()
        
        return pd.concat(chunks, ignore_index=True)

    def dataProcess(self, powerFarmID: str, dataBatchNum: str, timeGranularity: str, turbineCode: str, select: str, customCondition) -> Tuple[pd.DataFrame, str, str]:
        self.logger.info(
            f"load data -> Time Granulary： {timeGranularity} Power Farm: {powerFarmID} Batch: {dataBatchNum}  Turbine: {turbineCode} .")
        # Get current turbine info
        currTurbineInfo = self.common.getTurbineInfo(
            self.conf.dataContract.dataFilter.powerFarmID, turbineCode, self.turbineInfo)
        # Get turbine model info
        currTurbineModelInfo = self.common.getTurbineModelByTurbine(
            currTurbineInfo, self.turbineModelInfo)
        # select DB data
        dataFrameOfTurbine = self.loadData(
            powerFarmID, timeGranularity, f"'{turbineCode}'", select, customCondition)

        if dataFrameOfTurbine.empty:
            self.logger.warning(
                f"{turbineCode} Time Granulary： {timeGranularity} scada data is empty.")
            return pd.DataFrame(), dataBatchNum, turbineCode
        else:
            if timeGranularity in ['fault', 'warn']:
                return dataFrameOfTurbine, dataBatchNum, turbineCode
            else:
                # Add property to dataFrame
                self.addPropertyToDataFrame(
                    dataFrameOfTurbine, currTurbineInfo, currTurbineModelInfo)
                # Add engine_name to dataFrame
                dataFrameOfTurbine[Field_NameOfTurbine] = currTurbineInfo.loc[Field_NameOfTurbine]
                # Additional data processing steps
                self.processDateTime(dataFrameOfTurbine)
                # Recalculation
                self.recalculation(currTurbineModelInfo, dataFrameOfTurbine)
                # Filter data
                dataFrameOfTurbine = self.filterCommon(
                    dataFrameOfTurbine, self.conf)

                return dataFrameOfTurbine, dataBatchNum, turbineCode

    def selectTimeCondition(self, conf: Contract, conditions: list[str]):
        """
        时间过滤条件组装
        从json配置中获取时间过滤条件进行组装
        """
        # 时间过滤条件
        if conf.dataContract.dataFilter.beginTime:
            conditions.append(
                f"time_stamp >= '{conf.dataContract.dataFilter.beginTime}'")

        if conf.dataContract.dataFilter.endTime:
            conditions.append(
                f"time_stamp <= '{conf.dataContract.dataFilter.endTime}'")

        # 排除月份
        if conf.dataContract.dataFilter.excludingMonths:
            excluding_months = ", ".join(
                f"'{month}'" for month in conf.dataContract.dataFilter.excludingMonths)
            excluding_condition = f"DATE_FORMAT(time_stamp, '%Y-%m') NOT IN ({excluding_months})"
            conditions.append(excluding_condition)

    def selectLabCondition(self, conditions: list[str]):
        """
        lab 过滤条件组装
        根据lab获取相应的数据
        逻辑实现在相应的Behavior子类中，子类没有实现则无需过滤
        """
        return

    def selectAllCondition(self, conf: Contract):
        conditions = []
        # 时间过滤条件
        self.selectTimeCondition(conf, conditions)
        # lab过滤
        self.selectLabCondition(conditions)
        return " AND ".join(conditions) if conditions else "1=1"

    def selectFaultTimeCondition(self, conf: Contract, conditions: list[str]):
        """
        时间过滤条件组装
        从json配置中获取时间过滤条件进行组装
        """
        # 时间过滤条件
        if conf.dataContract.dataFilter.beginTime:
            conditions.append(
                f"begin_time >= '{conf.dataContract.dataFilter.beginTime}'")

        if conf.dataContract.dataFilter.endTime:
            conditions.append(
                f"end_time <= '{conf.dataContract.dataFilter.endTime}'")

        # 排除月份
        if conf.dataContract.dataFilter.excludingMonths:
            excluding_months = ", ".join(
                f"'{month}'" for month in conf.dataContract.dataFilter.excludingMonths)
            excluding_condition = f"DATE_FORMAT(time_stamp, '%Y-%m') NOT IN ({excluding_months})"
            conditions.append(excluding_condition)
    # 故障数据过滤条件

    def selectAllFaultCondition(self, conf: Contract):
        conditions = []
        # 时间过滤条件
        self.selectFaultTimeCondition(conf, conditions)
        # lab过滤
        self.selectLabCondition(conditions)
        return " AND ".join(conditions) if conditions else "1=1"

    def processTurbineData(self, turbines, conf: Contract, select: str):
        try:
            # add "where" condition
            select_conditions = self.selectAllCondition(conf)

            scadaTimeGranularities = self.getTimeGranularitys(conf)
            self.logger.info(
                f"typeAnalyst: {self.typeAnalyst()}  method: processTurbineData , scadaTimeGranularities : {scadaTimeGranularities}  current class : {self.__class__.__name__}" )
            dictionary = dict()
            for timeGranularity in scadaTimeGranularities:
                dataFrames = []
                dataFrameOfTurbines = pd.DataFrame()
                if timeGranularity in ['fault', 'warn']:
                    select_conditions = self.selectAllFaultCondition(conf)
                maxWorkers = 3
                with concurrent.futures.ThreadPoolExecutor(max_workers=maxWorkers) as executor:
                    futures = [
                        executor.submit(self.dataProcess, conf.dataContract.dataFilter.powerFarmID,
                                        conf.dataContract.dataFilter.dataBatchNum, timeGranularity, turbine, select, select_conditions)
                        for turbine in turbines
                    ]

                    for future in concurrent.futures.as_completed(futures):
                        try:
                            dataFrameOfTurbine, dataBatchNum, turbineCode = future.result()

                            dataFrames.append(dataFrameOfTurbine)
                            self.logger.info(
                                f"typeAnalyst: {self.typeAnalyst()}  data frame append,dataBatchNum: {dataBatchNum} timeGranularity: {timeGranularity}  turbineCode: {turbineCode}")
                        except Exception as exc:
                            raise exc

                dataFrameOfTurbines = pd.concat(dataFrames, ignore_index=True)

                self.logger.info(
                    f"typeAnalyst: {self.typeAnalyst()} data frame concat dataBatchNum: {dataBatchNum}  timeGranularity: {timeGranularity} finish")
                if dataFrameOfTurbines.empty:
                    # excption = CustomError(102)
                    # self.logger.warning(
                    #     f"{excption.message} typeAnalyst: {self.typeAnalyst()}  Power Farm: {conf.dataContract.dataFilter.powerFarmID} Batch : {conf.dataContract.dataFilter.dataBatchNum} Time Granularity : {timeGranularity}")
                    # raise excption

                    # 改为仅记录警告，不抛出异常
                    self.logger.warning(
                        f"Data is empty after processing. typeAnalyst: {self.typeAnalyst()}  Power Farm: {conf.dataContract.dataFilter.powerFarmID} Batch : {conf.dataContract.dataFilter.dataBatchNum} Time Granularity : {timeGranularity}")
                    dictionary[timeGranularity] = pd.DataFrame()
                else:
                    self.logger.info(
                        f"typeAnalyst: {self.typeAnalyst()} data frame concat dataBatchNum: {dataBatchNum}  timeGranularity: {timeGranularity} dataFrameOfTurbines : {dataFrameOfTurbines}")
                    if Field_DeviceCode in dataFrameOfTurbines.columns and Field_CodeOfTurbine not in dataFrameOfTurbines.columns:
                        dataFrameOfTurbines = dataFrameOfTurbines.rename(
                            columns={Field_DeviceCode: Field_CodeOfTurbine})

                    dictionary[timeGranularity] = dataFrameOfTurbines
            return dictionary
        except Exception as e:
            self.logger.error(
                f"Error processing turbine data:{traceback.format_exc()}")
            raise

    def addPropertyToDataFrame(self, dataFrameOfTurbine: pd.DataFrame, currTurbineInfo: pd.Series, currTurbineModelInfo: pd.Series):
        """
        用来添加额外当前风机属性
        在business中相应的分析员中有实现，没有实现就无额外参数添加
        """
        return

    def getTimeGranularitys(self, conf: str):
        """
            根据配置获取数据源类型
        """
        configAnalysisDF = pd.DataFrame(
            [config.to_dict() for config in conf.dataContract.configAnalysis])
        configAnalysisDF = configAnalysisDF[(configAnalysisDF["className"] == self.__class__.__name__)]

        return configAnalysisDF["scada"].unique()

    def escape_special_characters(self, original_string: str):
        """
            ---废弃---
            特殊字符url编码处理
            "/" 符号单独处理
        """
        retrun_string = quote(original_string)
        if "/" in retrun_string:
            retrun_string = retrun_string.replace("/", "%2F")
        return retrun_string