WTOAAM/dataAnalysisBusiness/algorithm/cpWindSpeedAnalyst.py

import os

import pandas as pd
import plotly.graph_objects as go
from algorithmContract.confBusiness import *
from algorithmContract.contract import Contract
from behavior.analystWithGoodPoint import AnalystWithGoodPoint


class CpWindSpeedAnalyst(AnalystWithGoodPoint):
    """
    风电机组风能利用系数分析
    """

    def typeAnalyst(self):
        return "cp_windspeed"

    def dataReprocess(self, dataFrameTurbines: pd.DataFrame) -> pd.DataFrame:
        dataFrame = dataFrameTurbines.groupby([Field_CodeOfTurbine, Field_WindSpeedFloor]).agg(
            cp=(Field_Cp, 'median'),
            cp_max=(Field_Cp, 'max'),
            cp_min=(Field_Cp, 'min'),
        ).reset_index()

        dataFrame.columns = [Field_CodeOfTurbine, Field_WindSpeedFloor,
                             Field_Cp, Field_CpMax, Field_CpMin]
        dataFrame = dataFrame.sort_values(
            by=[Field_CodeOfTurbine, Field_WindSpeedFloor], ascending=[True, True])

        return dataFrame

    def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes):
        dictionary = self.processTurbineData(turbineCodes, conf, [
                                             Field_DeviceCode, Field_Time, Field_WindSpeed, Field_ActiverPower])
        dataFrameOfTurbines = self.userDataFrame(
            dictionary, conf.dataContract.configAnalysis, self)
        # 检查所需列是否存在
        required_columns = {Field_WindSpeedFloor, Field_Cp}
        if not required_columns.issubset(dataFrameOfTurbines.columns):
            raise ValueError(f"DataFrame缺少必要的列。需要的列有: {required_columns}")

        turbrineInfos = self.common.getTurbineInfos(
            conf.dataContract.dataFilter.powerFarmID, turbineCodes, self.turbineInfo)

        groupedOfTurbineModel = turbrineInfos.groupby(Field_MillTypeCode)

        returnDatas = []
        for turbineModelCode, group in groupedOfTurbineModel:
            currTurbineCodes = group[Field_CodeOfTurbine].unique().tolist()
            currTurbineModeInfo = self.common.getTurbineModelByCode(
                turbineModelCode, self.turbineModelInfo)

            currDataFrameOfTurbines = dataFrameOfTurbines[dataFrameOfTurbines[Field_CodeOfTurbine].isin(
                currTurbineCodes)]

            dataFrame = self.dataReprocess(currDataFrameOfTurbines)

            returnData = self.buildChart(
                dataFrame, outputAnalysisDir, conf, currTurbineModeInfo)
            returnDatas.append(returnData)

        returnResult = pd.concat(returnDatas, ignore_index=True)

        return returnResult

    def buildChart(self, dataFrameOfTurbines: pd.DataFrame, outputAnalysisDir, conf: Contract, turbineModelInfo: pd.Series):
        # Create the main Cp distribution plot using Plotly
        fig = go.Figure()
        # colors = px.colors.sequential.Turbo

        for turbineCode in dataFrameOfTurbines[Field_CodeOfTurbine].unique():
            group = dataFrameOfTurbines[dataFrameOfTurbines[Field_CodeOfTurbine] == turbineCode]
            currTurbineInfo = self.common.getTurbineInfo(
                conf.dataContract.dataFilter.powerFarmID, turbineCode, self.turbineInfo)
            fig.add_trace(go.Scatter(x=group[Field_WindSpeedFloor], y=group[Field_Cp],
                                     mode='lines',
                                     # line=dict(color=colors[idx % len(colors)]),
                                     name=currTurbineInfo[Field_NameOfTurbine]))

        fig.update_layout(title={'text': f'风能利用系数分布-{turbineModelInfo[Field_MachineTypeCode]}', 'x': 0.5},
                          xaxis_title='风速', yaxis_title='风能利用系数',
                          legend=dict(
            orientation="h",  # Horizontal orientation
            xanchor="center",  # Anchor the legend to the center
            x=0.5,  # Position legend at the center of the x-axis
            y=-0.2,  # Position legend below the x-axis
            # itemsizing='constant',  # Keep the size of the legend entries constant
            # itemwidth=50
        ),
            xaxis=dict(range=[0, 26], tickmode='linear',
                       dtick=1, tickangle=-45),
            yaxis=dict(
            dtick=self.axisStepCp,
            range=[self.axisLowerLimitCp,
                   self.axisUpperLimitCp]
        )
        )
        # 保存HTML
        htmlFileName = f"{self.powerFarmInfo[Field_PowerFarmName].iloc[0]}-{turbineModelInfo[Field_MillTypeCode]}-Cp-Distribution.html"
        htmlFilePath = os.path.join(outputAnalysisDir, htmlFileName)
        fig.write_html(htmlFilePath)

        result_rows = []
        result_rows.append({
            Field_Return_TypeAnalyst: self.typeAnalyst(),
            Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
            Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
            Field_CodeOfTurbine: Const_Output_Total,
            Field_Return_FilePath: htmlFilePath,
            Field_Return_IsSaveDatabase: True
        })

        # Generate individual turbine plots
        for turbineCode, group in dataFrameOfTurbines.groupby(Field_CodeOfTurbine):
            currTurbineInfo = self.common.getTurbineInfo(
                conf.dataContract.dataFilter.powerFarmID, turbineCode, self.turbineInfo)
            fig = go.Figure()
            for turbineCode in dataFrameOfTurbines[Field_CodeOfTurbine].unique():
                tempDataFrame = dataFrameOfTurbines[dataFrameOfTurbines[Field_CodeOfTurbine] == turbineCode]
                fig.add_trace(go.Scatter(x=tempDataFrame[Field_WindSpeedFloor],
                                         y=tempDataFrame[Field_Cp],
                                         mode='lines',
                                         line=dict(color='lightgray', width=1),
                                         showlegend=False))
            fig.add_trace(go.Scatter(x=group[Field_WindSpeedFloor], y=group[Field_Cp], mode='lines', line=dict(
                color='darkblue'), showlegend=False))
            fig.update_layout(title=f'风机: {currTurbineInfo[Field_NameOfTurbine]}',
                              xaxis_title='风速', yaxis_title='风能利用系数',
                              xaxis=dict(
                                  range=[0, 26], tickmode='linear', dtick=1, tickangle=-45),
                              yaxis=dict(
                                  dtick=self.axisStepCp,
                                  range=[self.axisLowerLimitCp,
                                         self.axisUpperLimitCp]
                              )
                              )
            # 保存图像
            pngFileName = f"{currTurbineInfo[Field_NameOfTurbine]}.png"
            pngFilePath = os.path.join(outputAnalysisDir, pngFileName)
            fig.write_image(pngFilePath, scale=3)

            # 保存HTML
            htmlFileName = f"{currTurbineInfo[Field_NameOfTurbine]}.html"
            htmlFilePath = os.path.join(outputAnalysisDir, htmlFileName)
            fig.write_html(htmlFilePath)

            result_rows.append({
                Field_Return_TypeAnalyst: self.typeAnalyst(),
                Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
                Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
                Field_CodeOfTurbine: currTurbineInfo[Field_CodeOfTurbine],
                Field_Return_FilePath: pngFilePath,
                Field_Return_IsSaveDatabase: False
            })

            result_rows.append({
                Field_Return_TypeAnalyst: self.typeAnalyst(),
                Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
                Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
                Field_CodeOfTurbine: currTurbineInfo[Field_CodeOfTurbine],
                Field_Return_FilePath: htmlFilePath,
                Field_Return_IsSaveDatabase: True
            })

        result_df = pd.DataFrame(result_rows)

        return result_df