WTOAAM/dataAnalysisBusiness/algorithm/powerScatter2DAnalyst.py

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

class PowerScatter2DAnalyst(Analyst):
    """
    风电机组功率曲线散点分析。
    秒级scada数据运算太慢，建议使用分钟级scada数据
    """

    def typeAnalyst(self):
        return "power_scatter_2D"

    def turbinesAnalysis(self, dataFrameMerge, outputAnalysisDir, confData: ConfBusiness):
        if len(dataFrameMerge)<=0:
            print("After screening for blade pitch angle less than the configured value, plot power curve scatter points without data")
            return
        
        dataFrameGuaranteePowerCurve=self.common.contractGuaranteePowerCurveData(confData.turbineGuaranteedPowerCurveFilePathCSV,confData)
        self.drawOfPowerCurveScatter(dataFrameMerge,outputAnalysisDir,confData,dataFrameGuaranteePowerCurve)
        
    def drawOfPowerCurveScatter(self, dataFrame: pd.DataFrame,  outputAnalysisDir, confData: ConfBusiness,dataFrameGuaranteePowerCurve:pd.DataFrame):
        """  
        绘制风速-功率分布图并保存为文件。  

        参数:  
        dataFrameMerge (pd.DataFrame): 包含数据的DataFrame，需要包含设备名、风速和功率列。
        csvPowerCurveFilePath (str): 功率曲线文件路径。   
        outputAnalysisDir (str): 分析输出目录。  
        confData (ConfBusiness): 配置   
        """
        x_name = 'wind_speed'
        y_name = 'power'

        # 按设备名分组数据
        grouped = dataFrame.groupby(Field_NameOfTurbine)

        # 遍历每个设备的数据
        for name, group in grouped:
            # 创建图形和坐标轴  
            fig, ax = plt.subplots(figsize=(12, 8), dpi=96)  
            cmap = cm.get_cmap('rainbow')  
            
            # 绘制散点图  
            scatter = ax.scatter(x=group[confData.field_wind_speed],  
                                y=group[confData.field_power], c=group['monthIntTime'], cmap=cmap, s=5)  
            
            # 绘制合同功率曲线  
            ax.plot(dataFrameGuaranteePowerCurve['风速'], dataFrameGuaranteePowerCurve['有功功率'], marker='o',  
                    c='gray', label='Contract Guarantee Power Curve')  
            
            # 设置图形标题和坐标轴标签  
            ax.set_title(f'turbine_name={name}')  
            
            # 设置坐标轴的主刻度定位器  
            ax.xaxis.set_major_locator(MultipleLocator(1))   
            ax.set_xlim(0, 26)  

            # 创建每100个单位一个刻度的定位器
            yloc = 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.yaxis.set_major_locator(yloc)  # 将定位器应用到y轴上
            ax.set_ylim(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.set_xlabel(x_name)  
            ax.set_ylabel(y_name)  
                        
            # 显示图例，并调整位置  
            ax.legend(loc='lower right')  
            
            # 设置颜色条  
            unique_months = len(group['年月'].unique())
            ticks = np.linspace(group['monthIntTime'].min(), group['monthIntTime'].max(), min(unique_months, 6))  # 减少刻度数量 
            ticklabels = [datetime.fromtimestamp(tick).strftime('%Y-%m') for tick in ticks]  
            norm = Normalize(group['monthIntTime'].min(), group['monthIntTime'].max())  
            sm = cm.ScalarMappable(norm=norm, cmap=cmap)  
            
            # 添加颜色条  
            cbar = fig.colorbar(sm, ax=ax)
            cbar.set_ticks(ticks)  
            cbar.set_ticklabels(ticklabels)  
            
            # 旋转x轴刻度标签  
            plt.xticks(rotation=45)  
            
            # 保存图形为文件  
            output_file = os.path.join(outputAnalysisDir, f"{name}-scatter.png")  
            plt.savefig(output_file, bbox_inches='tight')  
            
            # 关闭图形  
            plt.close()