WTOAAM/dataAnalysisBusiness/algorithm/generatorSpeedTorqueAnalyst.py

import os
import pandas as pd
import numpy as np
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import plotly.express as px
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 GeneratorSpeedTorqueAnalyst(Analyst):
    """
    风电机组发电机转速-转矩分析
    """

    def typeAnalyst(self):
        return "speed_torque"

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

    def create_and_save_plots(self, dataFrame: pd.DataFrame, outputAnalysisDir, confData: ConfBusiness):
        # 检查所需列是否存在
        required_columns = {confData.field_gen_speed, Field_GeneratorTorque}
        if not required_columns.issubset(dataFrame.columns):
            raise ValueError(f"DataFrame缺少必要的列。需要的列有: {required_columns}")

        x_name = 'generator_speed'
        y_name = 'generator_torque'
        maxTurque = dataFrame[Field_GeneratorTorque].max()
        grouped = dataFrame.groupby(Field_NameOfTurbine)
        for name, group in grouped:
            groupNew = group.copy()

            if not self.common.isNone(confData.value_gen_speed_multiple):
                groupNew[confData.field_gen_speed] = group[confData.field_gen_speed] * \
                    confData.value_gen_speed_multiple
            # sns.lmplot函数参数scatter_kws： 设置为{"s": 5}时，会出现颜色丢失问题；改为={"s": 5, "color": "b"}后，则造成图形风格不统一问题；
            g = sns.lmplot(x=confData.field_gen_speed, y=Field_GeneratorTorque, data=groupNew, fit_reg=False, scatter_kws={
                           "s": 5, "color": "b"}, legend=False, height=6, aspect=1.2)
            # g = sns.lmplot(x=fieldGeneratorSpeed, y=Field_GeneratorTorque, data=group, fit_reg=False, scatter_kws={
            #                "s": 5}, legend=False, height=6, aspect=1.2)

            for ax in g.axes.flat:
                # 创建每100个单位一个刻度的定位器
                loc = MultipleLocator(confData.graphSets["generatorSpeed"]["step"] if not self.common.isNone(
                    confData.graphSets["generatorSpeed"]) and not self.common.isNone(
                    confData.graphSets["generatorSpeed"]["step"]) else 200)
                ax.xaxis.set_major_locator(loc)  # 将定位器应用到x轴上
                ax.set_xlim(confData.graphSets["generatorSpeed"]["min"] if not self.common.isNone(
                    confData.graphSets["generatorSpeed"]["min"]) else 1000, confData.graphSets["generatorSpeed"]["max"] if not self.common.isNone(confData.graphSets["generatorSpeed"]["max"]) else 2000)

                yloc = MultipleLocator(confData.graphSets["generatorTorque"]["step"] if not self.common.isNone(
                    confData.graphSets["generatorTorque"]["step"]) else 200)
                ax.yaxis.set_major_locator(yloc)  # 将定位器应用到y轴上
                ax.set_ylim(confData.graphSets["generatorTorque"]["min"] if not self.common.isNone(
                            confData.graphSets["generatorTorque"]["min"]) else 0, confData.graphSets["generatorTorque"]["max"] if not self.common.isNone(confData.graphSets["generatorTorque"]["max"]) else 2000)

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

            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): 配置   
        """
        dataFrame = dataFrame[(dataFrame[Field_GeneratorTorque] > 0)]
        # 按设备名分组数据
        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 % len(colorsList)] for i in range(len(unique_months))]
            color_map = dict(zip(unique_months, colors))

            # 使用go.Scatter3d创建3D散点图
            trace = go.Scatter3d(
                x=group[confData.field_gen_speed],
                y=group[Field_YearMonth],
                z=group[Field_GeneratorTorque],
                mode='markers',
                marker=dict(
                    color=[color_map[month]
                           for month in group[Field_YearMonth]],
                    size=2,
                    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 generator speed torque 3D scatter plot {name}',
                    "x": 0.5
                },
                scene=dict(
                    xaxis=dict(
                        title='Generator Speed',
                        dtick=confData.graphSets["generatorSpeed"]["step"] if not self.common.isNone(
                            confData.graphSets["generatorSpeed"]["step"]) else 200,  # 设置y轴刻度间隔为0.1
                        range=[confData.graphSets["generatorSpeed"]["min"] if not self.common.isNone(
                            confData.graphSets["generatorSpeed"]["min"]) else 1000, confData.graphSets["generatorSpeed"]["max"] if not self.common.isNone(confData.graphSets["generatorSpeed"]["max"]) else 2000],  # 设置y轴的范围从0到1
                        showgrid=True,  # 显示网格线
                    ),
                    yaxis=dict(
                        title='Time',
                        tickmode='array',
                        tickvals=unique_months,
                        ticktext=unique_months,
                        # dtick=500,  # 设置y轴刻度间隔
                        # range=[0,
                        #        group[Field_GeneratorTorque].max()],  # 设置y轴的范围
                        showgrid=True,  # 显示网格线
                        # categoryorder='category ascending'
                    ),
                    zaxis=dict(
                        title='Generator Torque',
                        dtick=confData.graphSets["generatorTorque"]["step"] if not self.common.isNone(
                            confData.graphSets["generatorTorque"]["step"]) else 200,  # 设置y轴刻度间隔为0.1
                        range=[confData.graphSets["generatorTorque"]["min"] if not self.common.isNone(
                            confData.graphSets["generatorTorque"]["min"]) else 0, confData.graphSets["generatorTorque"]["max"] if not self.common.isNone(confData.graphSets["generatorTorque"]["max"]) else 2000],  # 设置y轴的范围从0到1
                    )
                ),
                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)

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

        参数:  
        dataFrameMerge (pd.DataFrame): 包含数据的DataFrame，需要包含设备名、风速和功率列。
        outputAnalysisDir (str): 分析输出目录。  
        confData (ConfBusiness): 配置   
        """
        dataFrame = dataFrame[(dataFrame[confData.field_power] > 0)].sort_values(
            by=Field_NameOfTurbine)

        # 创建3D散点图
        fig = px.scatter_3d(dataFrame,
                            x=confData.field_gen_speed,
                            y=Field_NameOfTurbine,
                            z=Field_GeneratorTorque,
                            color=Field_NameOfTurbine,
                            labels={confData.field_gen_speed: 'Generator Speed',
                                    Field_NameOfTurbine: 'Turbine', Field_GeneratorTorque: 'Generator Torque'},
                            )

        # 设置固定散点大小
        fig.update_traces(marker=dict(size=1.5))

        # 更新图形的布局
        fig.update_layout(
            title={
                "text": 'Turbine generator speed Turque 3D scatter plot',
                "x": 0.5
            },
            scene=dict(
                xaxis=dict(
                    title='Generator Speed',
                    dtick=confData.graphSets["generatorSpeed"]["step"] if not self.common.isNone(
                        confData.graphSets["generatorSpeed"]["step"]) else 200,  # 设置y轴刻度间隔为0.1
                    range=[confData.graphSets["generatorSpeed"]["min"] if not self.common.isNone(
                        confData.graphSets["generatorSpeed"]["min"]) else 1000, confData.graphSets["generatorSpeed"]["max"] if not self.common.isNone(confData.graphSets["generatorSpeed"]["max"]) else 2000],  # 设置y轴的范围从0到1
                    showgrid=True,  # 显示网格线
                ),
                yaxis=dict(
                    title='Turbine',
                    showgrid=True,  # 显示网格线
                ),
                zaxis=dict(
                    title='Generator Turque',
                    dtick=confData.graphSets["generatorTorque"]["step"] if not self.common.isNone(
                        confData.graphSets["generatorTorque"]["step"]) else 200,  # 设置y轴刻度间隔为0.1
                    range=[confData.graphSets["generatorTorque"]["min"] if not self.common.isNone(
                        confData.graphSets["generatorTorque"]["min"]) else 0, confData.graphSets["generatorTorque"]["max"] if not self.common.isNone(confData.graphSets["generatorTorque"]["max"]) else 2000],  # 设置y轴的范围从0到1
                )
            ),
            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°
            ),
            # 设置图例标题
            legend_title_text='Turbine',
            margin=dict(t=50, b=10)  # t为顶部（top）间距，b为底部（bottom）间距
        )

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

        fig.write_html(outputFileHtml)