WTOAAM/dataAnalysisBusiness/algorithm/generatorSpeedPowerAnalyst.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.analystWithGoodPoint import AnalystWithGoodPoint


class GeneratorSpeedPowerAnalyst(AnalystWithGoodPoint):
    """
    风电机组发电机转速-有功功率分析
    """

    def typeAnalyst(self):
        return "speed_power"

    def turbinesAnalysis(self,  outputAnalysisDir, conf: Contract, turbineCodes):
        dictionary = self.processTurbineData(turbineCodes, conf, [
                                             Field_DeviceCode, Field_Time,Field_RotorSpeed,Field_GeneratorSpeed, Field_WindSpeed, Field_ActiverPower])
        dataFrameOfTurbines = self.userDataFrame(
            dictionary, conf.dataContract.configAnalysis, self)
        # 检查所需列是否存在
        required_columns = {Field_CodeOfTurbine,Field_RotorSpeed,Field_GeneratorSpeed,Field_ActiverPower}
        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)]

            # 将 currTurbineInfos 转换为字典
            currTurbineInfos_dict = turbrineInfos.set_index(Field_CodeOfTurbine)[Field_NameOfTurbine].to_dict()
            # 使用 map 函数来填充 Field_NameOfTurbine 列
            currDataFrameOfTurbines[Field_NameOfTurbine] = currDataFrameOfTurbines[Field_CodeOfTurbine].map(currTurbineInfos_dict).fillna("")

            result2D = self.drawScatter2DMonthly(
                currDataFrameOfTurbines, outputAnalysisDir, conf)
            returnDatas.extend(result2D)

            result3D = self.drawScatterGraph(
                currDataFrameOfTurbines, outputAnalysisDir, conf)
            returnDatas.extend(result3D)

            resultTotal = self.drawScatterGraphForTurbines(
                currDataFrameOfTurbines, outputAnalysisDir, conf, currTurbineModeInfo)
            returnDatas.extend(resultTotal)

        returnDataFrame = pd.DataFrame(returnDatas)

        return returnDataFrame

    def drawScatter2DMonthlyOfTurbine(self, dataFrame: pd.DataFrame, outputAnalysisDir: str, conf: Contract, turbineName: str):
        # 设置颜色条参数
        dataFrame = dataFrame.sort_values(by=Field_YearMonth)

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

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

        # 如果需要颜色轴的刻度和标签
        # 以下是以比例方式进行色彩的可视化处理
        fig.update_layout(
            title={
                "text": f'月度发电机转速功率散点图: {turbineName}',
                # "x": 0.5
            },
            xaxis=dict(
                title='发电机转速',
                dtick=self.axisStepGeneratorSpeed,
                range=[self.axisLowerLimitGeneratorSpeed,
                       self.axisUpperLimitGeneratorSpeed],
                tickangle=-45
            ),
            yaxis=dict(
                title='功率',
                dtick=self.axisStepActivePower,
                range=[self.axisLowerLimitActivePower,
                       self.axisUpperLimitActivePower],
            ),
            coloraxis=dict(
                colorbar=dict(
                    title="时间",
                    ticks="outside",
                    len=1,  # 设置颜色条的长度，使其占据整个图的高度
                    thickness=20,  # 调整颜色条的宽度
                    orientation='v',  # 设置颜色条为垂直方向
                    tickmode='array',  # 确保刻度按顺序排列
                    tickvals=dataFrame[Field_YearMonth].unique(
                    ).tolist(),  # 确保刻度为唯一的年月
                    ticktext=dataFrame[Field_YearMonth].unique(
                    ).tolist()  # 以%Y-%m格式显示标签
                )
            )
        )

        # 保存图片
        outputFilePathPNG = os.path.join(
            outputAnalysisDir, f"{turbineName}.png")
        fig.write_image(outputFilePathPNG, width=800, height=600, scale=3)
        # 保存html
        outputFileHtml = os.path.join(outputAnalysisDir, f"{turbineName}.html")
        fig.write_html(outputFileHtml)

        result = []
        result.append({
            Field_Return_TypeAnalyst: self.typeAnalyst(),
            Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
            Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
            Field_CodeOfTurbine: dataFrame[Field_CodeOfTurbine].iloc[0],
            Field_Return_FilePath: outputFilePathPNG,
            Field_Return_IsSaveDatabase: False
        })
        result.append({
            Field_Return_TypeAnalyst: self.typeAnalyst(),
            Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
            Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
            Field_CodeOfTurbine: dataFrame[Field_CodeOfTurbine].iloc[0],
            Field_Return_FilePath: outputFileHtml,
            Field_Return_IsSaveDatabase: True
        })

        return result

    def drawScatterGraphOfTurbine(self, dataFrame: pd.DataFrame,  outputAnalysisDir: str, conf: Contract, turbineName: str):
        # 创建3D散点图
        fig = px.scatter_3d(dataFrame,
                            x=Field_GeneratorSpeed,
                            y=Field_YearMonth,
                            z=Field_ActiverPower,
                            color=Field_YearMonth,
                            labels={Field_GeneratorSpeed: '发电机转速',
                                    Field_YearMonth: '时间', Field_ActiverPower: '功率'}
                            )

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

        # 更新图形的布局
        fig.update_layout(
            title={
                "text": f'月度发电机转速功率3D散点图: {turbineName}',
                # "x": 0.5
            },
            scene=dict(
                xaxis=dict(
                    title='发电机转速',
                    dtick=self.axisStepGeneratorSpeed,  # 设置y轴刻度间隔
                    range=[self.axisLowerLimitGeneratorSpeed,
                           self.axisUpperLimitGeneratorSpeed],  # 设置y轴的范围
                    showgrid=True,  # 显示网格线
                ),
                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
            )
        )

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

        result = []

        result.append({
            Field_Return_TypeAnalyst: self.typeAnalyst(),
            Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
            Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
            Field_CodeOfTurbine: dataFrame[Field_CodeOfTurbine].iloc[0],
            Field_Return_FilePath: outputFileHtml,
            Field_Return_IsSaveDatabase: True
        })

        return result

    def drawScatter2DMonthly(self, dataFrameMerge: pd.DataFrame, outputAnalysisDir, conf: Contract):
        """
        生成每台风电机组二维有功功率、发电机转速散点图表
        """
        results = []
        grouped = dataFrameMerge.groupby(Field_NameOfTurbine)
        for groupKey, group in grouped:
            result = self.drawScatter2DMonthlyOfTurbine(
                group, outputAnalysisDir, conf, groupKey)
            results.extend(result)

        return results

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

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

        grouped = dataFrame.groupby(Field_NameOfTurbine)

        # 遍历每个设备的数据
        for groupKey, group in grouped:
            if len(group[Field_YearMonth].unique()) > 1:
                result = self.drawScatterGraphOfTurbine(
                    group, outputAnalysisDir, conf, groupKey)
                results.extend(result)

        return results

    def drawScatterGraphForTurbines(self, dataFrame: pd.DataFrame,  outputAnalysisDir, conf: Contract, turbineModelInfo: pd.Series):
        """
        绘制风速-功率分布图并保存为文件。  (须按照机型分组)

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

        # 创建3D散点图
        fig = px.scatter_3d(dataFrame,
                            x=Field_GeneratorSpeed,
                            y=Field_NameOfTurbine,
                            z=Field_ActiverPower,
                            color=Field_NameOfTurbine,
                            labels={Field_GeneratorSpeed: '发电机转速',
                                    Field_NameOfTurbine: '风机', Field_ActiverPower: '功率'},
                            )

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

        # 更新图形的布局
        fig.update_layout(
            title={
                "text": f'风机发电机转速功率3D散点图-{turbineModelInfo[Field_MachineTypeCode]}',
                "x": 0.5
            },
            scene=dict(
                xaxis=dict(
                    title='发电机转速',
                    dtick=self.axisStepGeneratorSpeed,  # 设置y轴刻度间隔
                    range=[self.axisLowerLimitGeneratorSpeed,
                           self.axisUpperLimitGeneratorSpeed],  # 设置y轴的范围
                    showgrid=True,  # 显示网格线
                ),
                yaxis=dict(
                    title='机组',
                    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='Turbine'
            legend=dict(
                orientation="h",
                itemsizing="constant",  # Use constant size for legend items
                itemwidth=80  # Set the width of legend items to 50 pixels
            )
        )

        # 保存图像
        outputFileHtml = os.path.join(
            outputAnalysisDir, "{}-{}.html".format(self.typeAnalyst(),turbineModelInfo[Field_MillTypeCode]))
        fig.write_html(outputFileHtml)

        result = []

        result.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: outputFileHtml,
            Field_Return_IsSaveDatabase: True
        })

        return result