WTOAAM/dataAnalysisBusiness/algorithm/minPitchAnalyst.py

import os
import pandas as pd
import plotly.express as px
from algorithmContract.confBusiness import *
from algorithmContract.contract import Contract
from behavior.analystWithGoodBadPoint import AnalystWithGoodBadPoint


class MinPitchAnalyst(AnalystWithGoodBadPoint):
    """
    风电机组最小桨距角分析
    """

    def typeAnalyst(self):
        return "min_pitch"

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

    def processDateTime(self, dataFrame: pd.DataFrame, fieldTime:str = None):
        super().processDateTime(dataFrame,Field_YearMonthDay)

    def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes):
        dictionary = self.processTurbineData(turbineCodes,conf,self.selectColums())
        turbineInfos = self.common.getTurbineInfos(conf.dataContract.dataFilter.powerFarmID, turbineCodes, self.turbineInfo)
        dataFrameMerge = self.userDataFrame(dictionary,conf.dataContract.configAnalysis,self)
        return self.drawTrendGraph(dataFrameMerge,turbineInfos, outputAnalysisDir, conf)

    def drawTrendGraph(self, dataFrameMerge: pd.DataFrame,  turbineModelInfo: pd.Series,outputAnalysisDir, conf: Contract):
        """
        Generates pitch angle distribution scatter plots for turbines in a wind farm using plotly.

        Parameters:
        - dataFrameMerge: pd.DataFrame, DataFrame containing turbine data.
        - outputAnalysisDir: str, path to save the output plots.
        - conf: Contract, configuration object containing field names.
        """
        # 检查所需列是否存在
        required_columns = {Field_YearMonthDay, Field_PitchAngel1}
        if not required_columns.issubset(dataFrameMerge.columns):
            raise ValueError(f"DataFrame缺少必要的列。需要的列有: {required_columns}")

        pitchAngleRate = 'pitch_angle_rate'
        fieldPitchAngleBin = 'pitch_angle_bin'
        # Custom color scale: Blue (high pitch_angle_rate) to Light Grey (low pitch_angle_rate)
        # custom_color_scale = [
        #     (0.0, "rgb(255, 255, 255)"),  # white for the lowest values
        #     (0.05, "rgb(240, 240, 240)"),  # Light grey for the lowest values
        #     (0.5, "rgba(55.0, 135.0, 192.33333333333334, 1.0)"),  # Medium blue-grey
        #     # (0.75, "rgba(55.0, 135.0, 192.33333333333334, 1.0)"),  # Medium blue-grey
        #     # Dark blue for the highest values
        #     (1.0, "rgba(55.0, 135.0, 192.33333333333334, 1.0)")
        # ]
        custom_color_scale = [
            (0.0, "rgb(255, 255, 255)"),   # White for the lowest values (0%)
            # (0.05, "rgb(135, 206, 235)"),# Sky blue for low but non-zero values
            (0.05, "rgb(30, 144, 255)"),   # Dodger blue for for low but non-zero values
            (0.5, "rgb(0, 0, 255)"),       # Blue for higher values
            (1.0, "rgb(0, 0, 139)")        # Dark blue (Dark slate blue)
        ]

        # Group data by turbine identifier
        # dataFrameMerge = dataFrameMerge[dataFrameMerge[Field_PitchAngel1] <= 5]
        grouped = dataFrameMerge.groupby(
            [Field_NameOfTurbine, Field_CodeOfTurbine])
                # 创建一个列表来存储各个风电机组的数据
        turbine_data_list = []
        result_rows = []
        for name, group in grouped:
            # Convert the date column to datetime type
            group[Field_YearMonthDay] = pd.to_datetime(
                group[Field_YearMonthDay])

            # Creating bins of 0.2 intervals for pitch angles
            bins = pd.interval_range(start=group[Field_PitchAngel1].min(),
                                     end=group[Field_PitchAngel1].max(
            ) + 0.1, freq=0.2, closed='right')
            group[fieldPitchAngleBin] = pd.cut(
                group[Field_PitchAngel1], bins=bins)
            group[fieldPitchAngleBin] = group[fieldPitchAngleBin].apply(
                lambda x: x.left)  # 提取每个区间的左端点作为值
            # Calculate the pitch angle rate within each day
            # df = group.groupby([Field_YearMonthDay, Field_PitchAngel1]
            #                    ).size().reset_index(name='count')
            df = group.groupby([Field_YearMonthDay, fieldPitchAngleBin]
                               ).size().reset_index(name='count')
            total_counts = group.groupby(Field_YearMonthDay
                                         ).size().reset_index(name='total_count')
            df = df.merge(total_counts, on=Field_YearMonthDay)
            # df[pitchAngleRate] = df['count'] / df['total_count'] * 100
            # df[pitchAngleRate] = (df['count'] / df['total_count']).apply(lambda x: x ** 0.5)*100
            df[pitchAngleRate] = (df['count'] / df['total_count']) * 100
            
            # 提取数据
            turbine_data = {
                "engineName": name[0],
                "engineCode": name[1],
                "title":f' {name[0]}的最小桨距角分布',
                "xData": df[Field_YearMonthDay].dt.strftime('%Y-%m-%d').tolist(),
                "yData": df[fieldPitchAngleBin].tolist(),
                "colorbar": df[pitchAngleRate].tolist(),
                "colorbartitle": "百分比(%)"
                
            }

            turbine_data_list.append(turbine_data)
            # Plotting using plotly
            fig = px.scatter(df,
                             x=Field_YearMonthDay,
                             #  y=Field_PitchAngel1,  # 桨距角不分仓方式
                             y=fieldPitchAngleBin,  # 桨距角分仓方式
                             size='count',
                             color=pitchAngleRate,
                             #  color_continuous_scale='Blues',
                             color_continuous_scale=custom_color_scale
                             )

            # Set date format on x-axis
            fig.update_xaxes(
                title='时间', tickformat='%Y-%m-%d', tickangle=-45)

            fig.update_yaxes(title='桨距角',
                             dtick=self.axisStepPitchAngle,
                             range=[self.axisLowerLimitPitchAngle, 10],
                             )

            # Customizing legend
            fig.update_layout(
                title={
                    "text": f' {name[0]}的最小桨距角分布',
                    "x": 0.5
                },
                coloraxis_colorbar=dict(title='百分比'),
                margin=dict(t=50, b=10),  # t为顶部（top）间距，b为底部（bottom）间距
                # plot_bgcolor='rgb(240, 240, 240)'
            )


            # Set marker size if fixed size is needed
            # Fixed size for all points
            fig.update_traces(marker=dict(size=3, opacity=0.9))

            # Save plot
            filePathOfImage = os.path.join(outputAnalysisDir, f"{name[0]}.png")
            fig.write_image(filePathOfImage, scale=3)
            filePathOfHtml = os.path.join(outputAnalysisDir, f"{name[0]}.html")
            fig.write_html(filePathOfHtml)

            result_rows.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_rows.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
            })
        # 确保从 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": turbine_data_list
        }

        # 将JSON对象保存到文件
        output_json_path = os.path.join(outputAnalysisDir, "min_pitch_analysis.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: Const_Output_Total,
            Field_Return_FilePath: output_json_path,
            Field_Return_IsSaveDatabase: True
        })
        result_df = pd.DataFrame(result_rows)

        return result_df