WTOAAM/dataAnalysisBehavior/common/commonBusiness.py

import pandas as pd
import numpy as np
from algorithmContract.confBusiness import *
from algorithmContract.contract import Contract


class CommonBusiness:
    def getFloat32(self):
        return "float32"

    def isNone(self, value):
        result: bool = False
        try:
            result = value is None
        except:
            result = False
        return result

    def calculateTSR(self, dataFrame: pd.DataFrame, vaneLong: float):
        """
        使用叶轮/主轴转速、风速、叶轮直径，计算叶尖速比(TSR)
        """
        # Calculate tsr
        if Field_WindSpeed in dataFrame.columns \
           and Field_RotorSpeed in dataFrame.columns:
            dataFrame[Field_RotorSpeed] = pd.to_numeric(
                dataFrame[Field_RotorSpeed], errors='coerce')
            dataFrame[Field_WindSpeed] = pd.to_numeric(
                dataFrame[Field_WindSpeed], errors='coerce')

            dataFrame[Field_TSR] = (dataFrame[Field_RotorSpeed] * 0.104667 *
                                    (vaneLong/2)) / dataFrame[Field_WindSpeed]

    def calculateCp2(self, dataFrame: pd.DataFrame, airDensity, rotorDiameter, fieldWindSpeed, fieldActivePower):
        """
        使用有功功率、风速、空气密度、叶轮直径，计算风能利用系数(Cp)
        """
        # Calculate cp
        if not self.isNone(airDensity) and not self.isNone(rotorDiameter) \
           and not self.isNone(fieldWindSpeed) and fieldWindSpeed in dataFrame.columns \
           and not self.isNone(fieldActivePower) and fieldActivePower in dataFrame.columns:
            dataFrame[fieldActivePower] = pd.to_numeric(
                dataFrame[fieldActivePower], errors='coerce')
            dataFrame[fieldWindSpeed] = pd.to_numeric(
                dataFrame[fieldWindSpeed], errors='coerce')

            dataFrame[Field_PowerFloor] = dataFrame[Field_ActiverPower].apply(
                lambda x: int(x / 10) *
                10 if pd.notnull(x) else np.nan  # 保留NaN值
            )

            rotor_diameter = pd.to_numeric(
                rotorDiameter, errors='coerce')
            air_density = pd.to_numeric(airDensity, errors='coerce')

            dataFrame[Field_Cp] = dataFrame[fieldActivePower] * 1000 / \
                (0.5 * np.pi * air_density *
                 (rotor_diameter ** 2) / 4 * dataFrame[fieldWindSpeed] ** 3)

        return dataFrame

    def calculateCp(self, dataFrame: pd.DataFrame, powerFarmInfo: pd.DataFrame, turbineModelInfo: pd.DataFrame):
        """
        使用有功功率、风速、空气密度、叶轮直径，计算风能利用系数(Cp)
        """
        self.calculateCp2(dataFrame, powerFarmInfo[Field_AirDensity].iloc[0],
                          turbineModelInfo[Field_RotorDiameter].iloc[0], Field_WindSpeed, Field_ActiverPower)

    def getPowerFarm(self, powerFarmCode: str, powerFarmInfo: pd.DataFrame) -> pd.Series:
        """
        根据场站代码获取对应场站信息
        """
        selectRow = powerFarmInfo[powerFarmInfo[Field_PowerFarmCode]
                                  == powerFarmCode].iloc[0]

        if selectRow.empty:
            raise CustomError(104, powerFarmCode)

        return selectRow

    def getTurbineInfo(self, powerFarmCode: str, turbineCode: str, turbineInfos: pd.DataFrame) -> pd.Series:
        """
        获取对应机组信息
        """
        currTurbineInfo = turbineInfos[(turbineInfos[Field_PowerFarmCode] == powerFarmCode) & (
            turbineInfos[Field_CodeOfTurbine] == turbineCode)]
        if currTurbineInfo.empty:
            raise CustomError(115, turbineCode)

        return currTurbineInfo.iloc[0]
    
    def getTurbineInfos(self, powerFarmCode: str, turbineCodes: list, turbineInfos: pd.DataFrame) -> pd.DataFrame:
        """
        获取对应机组信息
        """
        currTurbineInfo = turbineInfos[(turbineInfos[Field_PowerFarmCode] == powerFarmCode) & (
            turbineInfos[Field_CodeOfTurbine].isin(turbineCodes) )]
        if currTurbineInfo.empty:
            raise CustomError(115, turbineCodes)

        return currTurbineInfo

    def getTurbineModelByTurbine(self, turbineInfo: pd.Series, turbineModelInfos: pd.DataFrame) -> pd.Series:
        """
        根据机组获取对应机型信息
        """
        turbineModelCode = turbineInfo[Field_MillTypeCode]

        if self.isNone(turbineModelCode):
            raise CustomError(114, turbineInfo[Field_CodeOfTurbine])

        selectRow = turbineModelInfos[turbineModelInfos[Field_MillTypeCode]
                                      == turbineModelCode].iloc[0]

        if selectRow.empty:
            raise CustomError(114, turbineInfo[Field_CodeOfTurbine])

        return selectRow
    
    def getTurbineModelByCode(self, turbineModelCode: str, turbineModelInfos: pd.DataFrame) -> pd.Series:
        """
        根据机组获取对应机型信息
        """
        selectRow = turbineModelInfos[turbineModelInfos[Field_MillTypeCode]
                                      == turbineModelCode].iloc[0]

        if selectRow.empty:
            raise CustomError(116, turbineModelCode)

        return selectRow

    def calculateCpOfSingleTurbine(self, dataFrame: pd.DataFrame, airDensity: float, rotorDiameter: float):
        """
        基于风电机组SCADA数据计算风能利用系数Cp
        """
        if dataFrame.empty:
            return

        if self.isNone(airDensity):
            raise CustomError(110)

        if self.isNone(rotorDiameter):
            raise CustomError(111)

        dataFrame[Field_Cp] = dataFrame[Field_ActiverPower] * 1000 / \
            (0.5 * np.pi * airDensity *
             (rotorDiameter ** 2) / 4 * dataFrame[Field_WindSpeed] ** 3)

        return dataFrame