WTOAAM/dataAnalysisBusiness/algorithm/pitchPowerAnalyst.py

import os
from datetime import datetime

import numpy as np
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from algorithmContract.confBusiness import *
from algorithmContract.contract import Contract
from behavior.analystWithGoodBadPoint import AnalystWithGoodBadPoint


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

    def typeAnalyst(self):
        return "pitch_power"

    def selectColumns(self):
        return [Field_DeviceCode, Field_Time,Field_WindSpeed, Field_ActiverPower, Field_PitchAngel1]

    def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes):
        dictionary = self.processTurbineData(turbineCodes, conf,self.selectColumns())
        dataFrame = self.userDataFrame(dictionary,conf.dataContract.configAnalysis,self)
        result_df1 = self.plot_power_pitch_angle(dataFrame, outputAnalysisDir, conf)
        result_df2 = self.drawScatterGraph(dataFrame, outputAnalysisDir, conf)
        result_df = pd.concat([result_df1, result_df2], ignore_index=True)

        return result_df

    def plot_power_pitch_angle(self, dataFrame:pd.DataFrame, outputAnalysisDir:str, conf: Contract):
        # 按设备名分组数据

        dataFrameMerge = dataFrame[(dataFrame[Field_ActiverPower] > 0)].sort_values(by=Field_YearMonth)
        grouped = dataFrameMerge.groupby([Field_NameOfTurbine, Field_CodeOfTurbine])
                # 定义固定的颜色映射列表
        fixed_colors = [
                "#3E409C",
                "#476CB9",
                "#3586BF",
                "#4FA4B5",
                "#52A3AE",
                "#60C5A3",
                "#85D0AE",
                "#A8DCA2",
                "#CFEE9E",
                "#E4F39E",
                "#EEF9A7",
                "#FBFFBE",
                "#FDF1A9",
                "#FFE286",
                "#FFC475",
                "#FCB06C",
                "#F78F4F",
                "#F96F4A",
                "#E4574C",
                "#CA3756",
                "#AF254F"
        ]

        # 将 fixed_colors 转换为 Plotly 的 colorscale 格式
        fixed_colorscale = [
            [i / (len(fixed_colors) - 1), color] for i, color in enumerate(fixed_colors)
        ]
        # 遍历每个设备并绘制散点图
        result_rows1 = []
        for name, group in grouped:
            # 创建图形
            fig = go.Figure()

            # 添加散点图
            fig.add_trace(go.Scatter(
                x=group[Field_ActiverPower],
                y=group[Field_PitchAngel1],
                mode='markers',
                # marker=dict(color='blue', size=3.5)
                marker=dict(
                    color=group[Field_UnixYearMonth],
                    colorscale=fixed_colorscale,
                    size=3,
                    opacity=0.7,
                    colorbar=dict(
                        tickvals=np.linspace(
                            group[Field_UnixYearMonth].min(), group[Field_UnixYearMonth].max(), 6),
                        ticktext=[datetime.fromtimestamp(ts).strftime('%Y-%m') for ts in np.linspace(
                            group[Field_UnixYearMonth].min(), group[Field_UnixYearMonth].max(), 6)],
                        thickness=18,
                        len=1,  # 设置颜色条的长度，使其占据整个图的高度
                        outlinecolor='rgba(255,255,255,0)'
                    ),
                    showscale=True
                ),
                showlegend=False
            ))

            # 设置图形布局
            fig.update_layout(
                title=f'机组: {name[0]}',
                xaxis=dict(
                    title='功率',
                    range=[self.axisLowerLimitActivePower,
                           self.axisUpperLimitActivePower],
                    dtick=self.axisStepActivePower,
                    tickangle=-45  # 设置x轴刻度值旋转角度为45度，如果需要
                ),
                yaxis=dict(
                    title='桨距角',
                    range=[self.axisLowerLimitPitchAngle,
                           self.axisUpperLimitPitchAngle],
                    dtick=self.axisStepPitchAngle
                ),
                coloraxis=dict(
                    colorbar=dict(
                        title="时间",
                        ticks="outside",
                        len=1,  # 设置颜色条的长度，使其占据整个图的高度
                        thickness=20,  # 调整颜色条的宽度
                        orientation='v',  # 设置颜色条为垂直方向
                        tickmode='array',  # 确保刻度按顺序排列
                        tickvals=dataFrameMerge[Field_YearMonth].unique(
                        ).tolist(),  # 确保刻度为唯一的年月
                        ticktext=dataFrameMerge[Field_YearMonth].unique(
                        ).tolist()  # 以%Y-%m格式显示标签
                    )
                )
            )

            # 保存图像
            filePathOfImage = os.path.join(outputAnalysisDir, f"{name[0]}.png")
            fig.write_image(filePathOfImage, width=800, height=600, scale=3)
            filePathOfHtml = os.path.join(outputAnalysisDir, f"{name[0]}.html")
            fig.write_html(filePathOfHtml)

            result_rows1.append({
                Field_Return_TypeAnalyst: self.typeAnalyst(),
                Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
                Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
                Field_CodeOfTurbine: name[1],
                Field_Return_FilePath: filePathOfImage,
                Field_Return_IsSaveDatabase: False
            })

            result_rows1.append({
                Field_Return_TypeAnalyst: self.typeAnalyst(),
                Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
                Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
                Field_CodeOfTurbine: name[1],
                Field_Return_FilePath: filePathOfHtml,
                Field_Return_IsSaveDatabase: True
            })

        result_df1 = pd.DataFrame(result_rows1)

        return result_df1

    def drawScatterGraph(self, dataFrame: pd.DataFrame,  outputAnalysisDir: str, conf: Contract):

        dataFrame = dataFrame[(dataFrame[Field_ActiverPower] > 0)].sort_values(
            by=Field_YearMonth)

        grouped = dataFrame.groupby([Field_NameOfTurbine, Field_CodeOfTurbine])
        '''
        # 遍历每个设备的数据
        result_rows2 = []
        for name, group in grouped:
            if len(group[Field_YearMonth].unique()) > 1:
                fig = px.scatter_3d(dataFrame,
                                    x=Field_PitchAngel1,
                                    y=Field_YearMonth,
                                    z=Field_ActiverPower,
                                    color=Field_YearMonth,
                                    labels={Field_PitchAngel1: '桨距角',
                                            Field_YearMonth: '时间', Field_ActiverPower: '功率'},
                                    )

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

                # 更新图形的布局
                fig.update_layout(
                    title={
                        "text": f'月度桨距角功率3D散点图: {name[0]}',
                        "x": 0.5
                    },
                    scene=dict(
                        xaxis=dict(
                            title='桨距角',
                            dtick=self.axisStepPitchAngle,
                            range=[self.axisLowerLimitPitchAngle,
                                   self.axisUpperLimitPitchAngle],
                        ),
                        yaxis=dict(
                            title='时间',
                            tickformat='%Y-%m',  # 日期格式,
                            dtick='M1',  # 每月一个刻度
                            showgrid=True,  # 显示网格线
                        ),
                        zaxis=dict(
                            title='功率',
                            dtick=self.axisStepActivePower,
                            range=[self.axisLowerLimitActivePower,
                                   self.axisUpperLimitActivePower],
                            showgrid=True,  # 显示网格线
                        )
                    ),
                    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='Time',
                    legend=dict(
                        orientation="h",
                        itemsizing="constant",  # Use constant size for legend items
                        itemwidth=80  # Set the width of legend items to 50 pixels
                    )
                )
        '''
        # 假设 colorsList 已经在代码的其他部分定义
        colorsList = [  
            "#3E409C",
            "#3586BF",
            "#52A3AE",
            "#85D0AE",
            "#A8DCA2",
            "#FBFFBE",
            "#FDF1A9",
            "#FFE286",
            "#FCB06C",
            "#F96F4A",
            "#E4574C",
            "#AF254F"
        ]
        
        # 遍历每个设备的数据
        result_rows2 = []
        for name, group in grouped:
            if len(group[Field_YearMonth].unique()) > 1:
                fig = px.scatter_3d(
                    group,
                    x=Field_PitchAngel1,
                    y=Field_YearMonth,
                    z=Field_ActiverPower,
                    color=Field_YearMonth,
                    color_discrete_sequence=colorsList,  # 使用 colorsList 作为颜色映射
                    labels={
                        Field_PitchAngel1: '桨距角',
                        Field_YearMonth: '时间',
                        Field_ActiverPower: '功率'
                    },
                )
        
                # 设置固定散点大小
                fig.update_traces(marker=dict(size=1.5))
        
                # 更新图形的布局
                fig.update_layout(
                    title={
                        "text": f'月度桨距角功率3D散点图: {name[0]}',
                        "x": 0.5
                    },
                    scene=dict(
                        xaxis=dict(
                            title='桨距角',
                            dtick=self.axisStepPitchAngle,
                            range=[self.axisLowerLimitPitchAngle, self.axisUpperLimitPitchAngle],
                        ),
                        yaxis=dict(
                            title='时间',
                            tickformat='%Y-%m',  # 日期格式,
                            dtick='M1',  # 每月一个刻度
                            showgrid=True,  # 显示网格线
                        ),
                        zaxis=dict(
                            title='功率',
                            dtick=self.axisStepActivePower,
                            range=[self.axisLowerLimitActivePower, self.axisUpperLimitActivePower],
                            showgrid=True,  # 显示网格线
                        )
                    ),
                    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=dict(
                        orientation="h",
                        itemsizing="constant",  # Use constant size for legend items
                        itemwidth=80  # Set the width of legend items to 50 pixels
                    )
                )

                # 保存图像
                filePathOfHtml = os.path.join(
                    outputAnalysisDir, f"{name[0]}_3D.html")
                fig.write_html(filePathOfHtml)

                result_rows2.append({
                    Field_Return_TypeAnalyst: self.typeAnalyst(),
                    Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
                    Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
                    Field_CodeOfTurbine: name[1],
                    Field_Return_FilePath: filePathOfHtml,
                    Field_Return_IsSaveDatabase: True
                })

        result_df2 = pd.DataFrame(result_rows2)

        return result_df2

        # self.drawScatterGraphOfTurbine(
        #     group, outputAnalysisDir, conf, name)

    # def drawScatterGraphOfTurbine(self, dataFrame: pd.DataFrame,  outputAnalysisDir: str, conf: Contract, turbineName: str):
    #     # 创建3D散点图
    #     fig = px.scatter_3d(dataFrame,
    #                         x=Field_PitchAngel1,
    #                         y=Field_YearMonth,
    #                         z=Field_ActiverPower,
    #                         color=Field_YearMonth,
    #                         labels={Field_PitchAngel1: 'Pitch Angle',
    #                                 Field_YearMonth: 'Time', Field_ActiverPower: 'Power'},
    #                         )

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

    #     # 更新图形的布局
    #     fig.update_layout(
    #         title={
    #             "text": f'Monthly Pitch-Power 3D Scatter Plot: {turbineName}',
    #             "x": 0.5
    #         },
    #         scene=dict(
    #             xaxis=dict(
    #                 title='Pitch Angle',
    #                 range=[conf.dataContract.graphSets["pitchAngle"]["min"] if not self.common.isNone(conf.dataContract.graphSets["pitchAngle"]["min"]) else -2,
    #                        conf.dataContract.graphSets["pitchAngle"]["max"] if not self.common.isNone(conf.dataContract.graphSets["pitchAngle"]["max"]) else 28],
    #                 dtick=conf.dataContract.graphSets["pitchAngle"]["step"] if not self.common.isNone(conf.dataContract.graphSets["pitchAngle"]["step"]) else 2,
    #             ),
    #             yaxis=dict(
    #                 title='Time',
    #                 tickformat='%Y-%m',  # 日期格式,
    #                 dtick='M1',  # 每月一个刻度
    #                 showgrid=True,  # 显示网格线
    #             ),
    #             zaxis=dict(
    #                 title='Power',
    #                 dtick=conf.dataContract.graphSets["activePower"]["step"] if not self.common.isNone(
    #                     conf.dataContract.graphSets["activePower"]) and not self.common.isNone(
    #                     conf.dataContract.graphSets["activePower"]["step"]) else 250,
    #                 range=[conf.dataContract.graphSets["activePower"]["min"] if not self.common.isNone(
    #                     conf.dataContract.graphSets["activePower"]["min"]) else 0, conf.dataContract.graphSets["activePower"]["max"] if not self.common.isNone(conf.dataContract.graphSets["activePower"]["max"]) else conf.rated_power*1.2],
    #                 showgrid=True,  # 显示网格线
    #             )
    #         ),
    #         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='Time'
    #     )

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

    #     fig.write_html(outputFileHtml)

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

        ## 参数:  
        ## dataFrameMerge (pd.DataFrame): 包含数据的DataFrame，需要包含设备名、风速和功率列。
        ## outputAnalysisDir (str): 分析输出目录。  
        ## conf (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[Field_PitchAngel1],
                y=group[Field_YearMonth],
                z=group[Field_ActiverPower],
                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'三维散点图{name}',
                    "x": 0.5
                },
                scene=dict(
                    xaxis=dict(
                        title='桨距角',
                        dtick=conf.dataContract.graphSets["pitchAngle"]["step"] if not self.common.isNone(
                            conf.dataContract.graphSets["pitchAngle"]["step"]) else 2,  # 设置y轴刻度间隔为0.1
                        range=[conf.dataContract.graphSets["pitchAngle"]["min"] if not self.common.isNone(
                            conf.dataContract.graphSets["pitchAngle"]["min"]) else -2, conf.dataContract.graphSets["pitchAngle"]["max"] if not self.common.isNone(conf.dataContract.graphSets["pitchAngle"]["max"]) else 28],  # 设置y轴的范围从0到1
                        showgrid=True,  # 显示网格线
                    ),
                    yaxis=dict(
                        title='时间',
                        tickmode='array',
                        tickvals=unique_months,
                        ticktext=unique_months,
                        showgrid=True,  # 显示网格线
                        categoryorder='category ascending'
                    ),
                    zaxis=dict(
                        title='功率',
                        dtick=conf.dataContract.graphSets["activePower"]["step"] if not self.common.isNone(
                            conf.dataContract.graphSets["activePower"]) and not self.common.isNone(
                            conf.dataContract.graphSets["activePower"]["step"]) else 250,
                        range=[conf.dataContract.graphSets["activePower"]["min"] if not self.common.isNone(
                            conf.dataContract.graphSets["activePower"]["min"]) else 0, conf.dataContract.graphSets["activePower"]["max"] if not self.common.isNone(conf.dataContract.graphSets["activePower"]["max"]) else conf.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)
        """