WTOAAM/dataAnalysisBusiness/algorithm/pitchTSRCpAnalyst.py

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


class PitchTSRCpAnalyst(AnalystWithGoodPoint):

    def typeAnalyst(self):
        return "pitch_tsr_cp"

    def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes):
        dictionary=self.processTurbineData(turbineCodes,conf,[Field_DeviceCode,Field_Time,Field_PitchAngel1,Field_WindSpeed,Field_ActiverPower,Field_RotorSpeed,Field_GeneratorSpeed])
        dataFrameMerge=self.userDataFrame(dictionary,conf.dataContract.configAnalysis,self)
        turbineInfos = self.common.getTurbineInfos(conf.dataContract.dataFilter.powerFarmID, turbineCodes, self.turbineInfo)
        return self.windRoseAnalysis(dataFrameMerge,turbineInfos, outputAnalysisDir, conf)


    def windRoseAnalysis(self, dataFrameMerge: pd.DataFrame, turbineModelInfo: pd.Series,outputAnalysisDir, conf: Contract):
        """
        绘制3D曲面图。

        参数:
        df: pandas.DataFrame, 必须包含Field_PitchAngel1, Field_TSR, 和 Field_Cp这三个字段。

        返回:
        一个Plotly图形对象。
        """
        # 检查所需列是否存在
        required_columns = {Field_PitchAngel1, Field_TSR, Field_Cp}
        if not required_columns.issubset(dataFrameMerge.columns):
            raise ValueError(f"DataFrame缺少必要的列。需要的列有: {required_columns},Available: {dataFrameMerge.columns}")

        # # 按设备名分组数据
        # grouped = dataFrameMerge.groupby([Field_NameOfTurbine, Field_CodeOfTurbine])

        result_rows = []

        ## 分图
        dataFrameMerge['PitchAngleBin'] = pd.cut(dataFrameMerge[Field_PitchAngel1], bins=[-1, 1, 3, 5, 7, 9], right=True, labels=['[-1,1]', '(1,3]', '(3,5]', '(5,7]', '(7,9]'])
        # turbines = dataFrameMerge[Field_NameOfTurbine].unique()
        turbines = dataFrameMerge.groupby([Field_NameOfTurbine, Field_CodeOfTurbine]).groups.keys()

        for turbine_name, turbine_code in turbines:
            fig = make_subplots(rows=1, cols=1)
            # 选择当前风力发电机的数据
            turbine_data = dataFrameMerge[dataFrameMerge[Field_NameOfTurbine] == turbine_name]
            # 按角区间分组
            grouped = turbine_data.groupby('PitchAngleBin')
            # 为每个区间添加一个数据迹
            for pitch_bin, group in grouped:
                fig.add_trace(go.Scatter(
                    x=group[Field_TSR],
                    y=group[Field_Cp],
                    mode='markers',
                    name=pitch_bin,
                    marker=dict(
                        size=3,  # Smaller size, default is usually around 10
                        opacity=0.8  # Reduced opacity, default is 1
                    )
                    # name=f'Pitch: {name}'
                ))

            # 更新布局
            fig.update_layout(
                title=f'风能利用系数 vs. 叶尖速比 : {turbine_name}',
                xaxis_title='叶尖速比',
                yaxis_title='风能利用系数',
                legend_title='桨距角',
                legend=dict(
                    # orientation="h",
                    itemsizing="constant",  # Use constant size for legend items
                    itemwidth=80  # Set the width of legend items to 50 pixels
                ),
                xaxis=dict(
                    # dtick=2,
                    range=[5,20]),
                yaxis=dict(
                    dtick=self.axisStepCp,
                    range=[self.axisLowerLimitCp,self.axisUpperLimitCp]
                )
                # legend=dict(
                #     orientation="h",
                #     xanchor="center",
                #     x=0.5,
                #     y=-0.2
                # )
            )
            # 确保从 Series 中提取的是具体的值
            engineTypeCode = turbineModelInfo.get(Field_MillTypeCode, "")
            if isinstance(engineTypeCode, pd.Series):
                engineTypeCode = engineTypeCode.iloc[0]

            engineTypeName = turbineModelInfo.get(Field_MachineTypeCode, "")
            if isinstance(engineTypeName, pd.Series):
                engineTypeName = engineTypeName.iloc[0]
            # 构建最终的JSON对象
            json_output = {
                "analysisTypeCode": "变桨和叶尖速比及风能利用系数分析",
                "engineCode":  engineTypeCode,
                "engineTypeName": engineTypeName,
                "xaixs": "叶尖速比",
                "yaixs": "风能利用系数",
                "data": [{
                        "engineName": turbine_name,
                        "engineCode": turbine_code,
                        "title":f' 风能利用系数 vs. 叶尖速比-{turbine_name}',
                        # "xData": group[Field_TSR].tolist(),
                        "xData":[None if math.isinf(val) else val for val in group[Field_TSR].tolist()],
                        # "yData":group[Field_Cp].tolist(),
                        "yData":[None if math.isinf(val) else val for val in group[Field_Cp].tolist()],
                        "colorbar": pitch_bin,
                        "colorbartitle": "桨距角"
                        }]
            }


            # 保存html
            # filePathOfPng = os.path.join(outputAnalysisDir, f"{turbine_name}.png")
            # fig.write_image(filePathOfPng)
            # filePathOfHtml = os.path.join(outputAnalysisDir, f"{turbine_name}.html")
            # fig.write_html(filePathOfHtml)


            # 将JSON对象保存到文件
            output_json_path = os.path.join(outputAnalysisDir, f"pitch_TSR_cp{turbine_name}.json")
            with open(output_json_path, 'w', encoding='utf-8') as f:
                import json
                json.dump(json_output, f, ensure_ascii=False, indent=4)

            # 如果需要返回DataFrame，可以包含文件路径
            result_rows.append({
                Field_Return_TypeAnalyst: self.typeAnalyst(),
                Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
                Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
                Field_CodeOfTurbine: turbine_code,
                Field_Return_FilePath: output_json_path,
                Field_Return_IsSaveDatabase: True
            })
            # result_rows.append({
            #     Field_Return_TypeAnalyst: self.typeAnalyst(),
            #     Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
            #     Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
            #     Field_CodeOfTurbine: turbine_code,
            #     Field_Return_FilePath: filePathOfPng,
            #     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: turbine_code,
            #     Field_Return_FilePath: filePathOfHtml,
            #     Field_Return_IsSaveDatabase: True
            # })


        result_df = pd.DataFrame(result_rows)

        return result_df