WTOAAM/dataAnalysisBusiness/algorithm/pitchPowerAnalyst.py

import os
import pandas as pd
import numpy as np
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import seaborn as sns
import matplotlib.pyplot as plt
from matplotlib.ticker import MultipleLocator
from behavior.analyst import Analyst
from utils.directoryUtil import DirectoryUtil as dir
from algorithmContract.confBusiness import *


class PitchPowerAnalyst(Analyst):
    """
    风电机组变桨-功率分析
    """

    def typeAnalyst(self):
        return "pitch_power"

    def turbinesAnalysis(self, dataFrameMerge, outputAnalysisDir, confData: ConfBusiness):
        self.plot_power_pitch_angle(
            dataFrameMerge, outputAnalysisDir, confData)
        self.drawScatterGraph(dataFrameMerge, outputAnalysisDir, confData)

    def plot_power_pitch_angle(self, dataFrameMerge, outputAnalysisDir, confData: ConfBusiness):
        x_name = 'power'
        y_name = 'pitch_angle'
        # 按设备名分组数据
        grouped = dataFrameMerge.groupby(Field_NameOfTurbine)
        print("self.ratedPower {}".format(confData.rated_power))
        # 遍历每个设备并绘制散点图
        for name, group in grouped:
            # sns.lmplot函数参数scatter_kws： 设置为{"s": 5}时，会出现颜色丢失问题；
            g = sns.lmplot(x=confData.field_power, y=confData.field_pitch_angle1, data=group,
                           fit_reg=False, scatter_kws={"s": 5, "color": "b"}, legend=False, height=6, aspect=1.2)
            # g = sns.lmplot(x=confData.field_power, y=confData.field_pitch_angle1, data=group,
            #                fit_reg=False, scatter_kws={"s": 5}, legend=False, height=6, aspect=1.2)

            # 设置x轴和y轴的刻度
            for ax in g.axes.flat:
                ax.xaxis.set_major_locator(MultipleLocator(confData.graphSets["activePower"]["step"] if not self.common.isNone(
                    confData.graphSets["activePower"]) and not self.common.isNone(
                    confData.graphSets["activePower"]["step"]) else 250))
                ax.set_xlim(confData.graphSets["activePower"]["min"] if not self.common.isNone(
                    confData.graphSets["activePower"]["min"]) else 0, confData.graphSets["activePower"]["max"] if not self.common.isNone(confData.graphSets["activePower"]["max"]) else confData.rated_power*1.2)

                ax.yaxis.set_major_locator(MultipleLocator(confData.graphSets["pitchAngle"]["step"] if not self.common.isNone(
                    confData.graphSets["pitchAngle"]["step"]) else 2))
                ax.set_ylim(confData.graphSets["pitchAngle"]["min"] if not self.common.isNone(
                    confData.graphSets["pitchAngle"]["min"]) else -2, confData.graphSets["pitchAngle"]["max"] if not self.common.isNone(confData.graphSets["pitchAngle"]["max"]) else 28)

                ax.set_xlabel(x_name)
                ax.set_ylabel(y_name)

            # 设置x轴刻度值旋转角度为45度
            plt.tick_params(axis='x', rotation=45)
            # 调整布局和设置标题
            plt.tight_layout()
            plt.title(f'{Field_NameOfTurbine}={name}')

            # 保存图像并关闭绘图窗口
            output_file = os.path.join(outputAnalysisDir, f"{name}.png")
            plt.savefig(output_file, bbox_inches='tight', dpi=120)
            plt.close()

    def drawScatterGraph(self, dataFrame: pd.DataFrame,  outputAnalysisDir, confData: ConfBusiness):
        """  
        绘制变桨-功率分布图并保存为文件。  

        参数:  
        dataFrameMerge (pd.DataFrame): 包含数据的DataFrame，需要包含设备名、风速和功率列。
        outputAnalysisDir (str): 分析输出目录。  
        confData (ConfBusiness): 配置   
        """
        # 按设备名分组数据
        colorsList = ['#1f77b4', '#ff7f0e', '#2ca02c', '#d62728', '#9467bd',
                      '#8c564b', '#e377c2', '#7f7f7f', '#bcbd22', '#17becf', '#aec7e8', '#ffbb78']
        grouped = dataFrame.groupby(Field_NameOfTurbine)

        # 遍历每个设备的数据
        for name, group in grouped:
            # 创建颜色映射，将每个年月映射到一个唯一的颜色
            unique_months = group[Field_YearMonth].unique()
            colors = [
                colorsList[i % 12] for i in range(len(unique_months))]
            color_map = dict(zip(unique_months, colors))

            # 使用go.Scatter3d创建3D散点图
            trace = go.Scatter3d(
                x=group[confData.field_pitch_angle1],
                y=group[Field_YearMonth],
                z=group[confData.field_power],
                mode='markers',
                marker=dict(
                    color=[color_map[month]
                           for month in group[Field_YearMonth]],
                    size=1.5,
                    line=dict(
                        color='rgba(0, 0, 0, 0)',  # 设置边框颜色为透明，以去掉白色边框
                        width=0  # 设置边框宽度为0，进一步确保没有边框
                    ),
                    opacity=0.8  # 调整散点的透明度，增加透视效果
                )
            )

            # 创建图形
            fig = go.Figure(data=[trace])

            # 更新图形的布局
            fig.update_layout(
                title={
                    "text": f'Monthly pitch to power 3D scatter plot {name}',
                    "x": 0.5
                },
                scene=dict(
                    xaxis=dict(
                        title='Pitch Angle',
                        dtick=confData.graphSets["pitchAngle"]["step"] if not self.common.isNone(
                            confData.graphSets["pitchAngle"]["step"]) else 2,  # 设置y轴刻度间隔为0.1
                        range=[confData.graphSets["pitchAngle"]["min"] if not self.common.isNone(
                            confData.graphSets["pitchAngle"]["min"]) else -2, confData.graphSets["pitchAngle"]["max"] if not self.common.isNone(confData.graphSets["pitchAngle"]["max"]) else 28],  # 设置y轴的范围从0到1
                        showgrid=True,  # 显示网格线
                    ),
                    yaxis=dict(
                        title='Time',
                        tickmode='array',
                        tickvals=unique_months,
                        ticktext=unique_months,
                        showgrid=True,  # 显示网格线
                        categoryorder='category ascending'
                    ),
                    zaxis=dict(
                        title='Power',
                        dtick=confData.graphSets["activePower"]["step"] if not self.common.isNone(
                            confData.graphSets["activePower"]) and not self.common.isNone(
                            confData.graphSets["activePower"]["step"]) else 250,
                        range=[confData.graphSets["activePower"]["min"] if not self.common.isNone(
                            confData.graphSets["activePower"]["min"]) else 0, confData.graphSets["activePower"]["max"] if not self.common.isNone(confData.graphSets["activePower"]["max"]) else confData.rated_power*1.2],
                    )
                ),
                scene_camera=dict(
                    up=dict(x=0, y=0, z=1),  # 保持相机向上方向不变
                    center=dict(x=0, y=0, z=0),  # 保持相机中心位置不变
                    eye=dict(x=-1.8, y=-1.8, z=1.2)  # 调整eye属性以实现水平旋转180°
                ),
                margin=dict(t=50, b=10)  # t为顶部（top）间距，b为底部（bottom）间距
            )

            # 保存图像
            outputFileHtml = os.path.join(
                outputAnalysisDir, "{}.html".format(name))

            fig.write_html(outputFileHtml)