WTOAAM/dataAnalysisBusiness/algorithm/powerCurveAnalyst.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 PowerCurveAnalyst(Analyst):
    """
    风电机组功率曲线散点分析。
    秒级scada数据运算太慢，建议使用分钟级scada数据
    """

    def typeAnalyst(self):
        return "power_curve"
    
    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
        
        self.drawOfPowerCurve(dataFrameMerge, outputAnalysisDir, confData,self.dataFrameContractOfTurbine)
        # 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()

    def drawOfPowerCurve(self, dataFrameMerge: pd.DataFrame,  outputAnalysisDir, confData: ConfBusiness,dataFrameGuaranteePowerCurve:pd.DataFrame):
        """  
        生成功率曲线并保存为文件。  

        参数:  
        frames (pd.DataFrame): 包含数据的DataFrame，需要包含设备名、风速和功率列。  
        outputAnalysisDir (str): 分析输出目录。  
        confData (ConfBusiness): 配置 
        """
        
        # 定义风速区间        
        bins = np.arange(0, 26, 0.5)
        
        # 初始化结果DataFrame
        all_res = pd.DataFrame()

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

        # 计算每个设备的功率曲线
        for name, group in grouped:
            act_line = self.power_curve_helper(
                group, confData.field_wind_speed, confData.field_power, bins)
            act_line[Field_NameOfTurbine] = name
            all_res = pd.concat([all_res, act_line], axis=0, sort=False)
        
        # 绘制全场功率曲线图
        ress = all_res.reset_index(drop=True)        
        
        self.plot_power_curve(ress,  outputAnalysisDir, dataFrameGuaranteePowerCurve,Field_NameOfTurbine,
                              '全场-{}功率曲线.png'.format(confData.farm_name),confData)

        # 绘制每个设备的功率曲线图
        grouped=ress.groupby(Field_NameOfTurbine)
        for name, group in grouped:
            self.plot_single_power_curve(ress, group,dataFrameGuaranteePowerCurve, name, outputAnalysisDir,confData)

    def power_curve_helper(self, group, wind_speed_col, power_col, bins):
        """  
        计算设备的功率曲线。  
        """
        powerCut = group.groupby(pd.cut(group[wind_speed_col], bins, labels=np.arange(0, 25.5, 0.5))).agg({
            power_col: 'mean',
            wind_speed_col: ['mean', 'count']
        })
        wind_count = powerCut[wind_speed_col]['count'].tolist()
        line = powerCut[power_col]['mean'].round(decimals=2).tolist()
        act_line = pd.DataFrame([powerCut.index, wind_count, line]).T
        act_line.columns = ['风速区间', '有效数量', '实际功率曲线']
        return act_line

    def plot_power_curve(self, ress, output_path,dataFrameGuaranteePowerCurve:pd.DataFrame, Field_NameOfTurbine, filename,confData:ConfBusiness):
        """  
        绘制全场功率曲线图。  
        """
        sns.set_palette('deep')
        fig, ax = plt.subplots(figsize=(16, 8)) 
        ax = sns.lineplot(x='风速区间', y='实际功率曲线', data=ress, hue=Field_NameOfTurbine)
        # # 绘制合同功率曲线  
        # ax.plot(dataFrameGuaranteePowerCurve['风速'], dataFrameGuaranteePowerCurve['有功功率'], marker='o',  
        #         c='red', label='Contract Guarantee Power Curve') 
        ax.set_xlabel('wind speed')
        ax.set_ylabel('power')
        ax.set_title('power curve')
        # 设置坐标轴的主刻度定位器  
        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)
        
        plt.legend(title='turbine',bbox_to_anchor=(1.02, 0.5),ncol=2, loc='center left', borderaxespad=0.) 
        plt.xticks(rotation=45)  # 旋转45度
        plt.savefig(os.path.join(output_path, filename),
                    bbox_inches='tight', dpi=120)
        plt.close()

    def plot_single_power_curve(self, ress, group,dataFrameGuaranteePowerCurve:pd.DataFrame , turbineName, outputAnalysisDir,confData:ConfBusiness):
        
        color = ["lightgrey"]*len(ress[Field_NameOfTurbine].unique())
        fig, ax = plt.subplots(figsize=(8, 8))
        ax = sns.lineplot(x='风速区间', y='实际功率曲线', data=ress, hue=Field_NameOfTurbine,
                          palette=sns.set_palette(color), legend=False)
        ax = sns.lineplot(x='风速区间', y='实际功率曲线', data=group,
                          color='darkblue', legend=False)
        
        # 绘制合同功率曲线  
        ax.plot(dataFrameGuaranteePowerCurve['风速'], dataFrameGuaranteePowerCurve['有功功率'], marker='o',  
                c='red', label='Contract Guarantee Power Curve')  
        
        ax.set_xlabel('wind speed')
        ax.set_ylabel('power')
        ax.set_title('turbine_name={}'.format(turbineName))
        # 设置坐标轴的主刻度定位器  
        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.legend(loc='lower right')  
        plt.xticks(rotation=45)  # 旋转45度
        plt.savefig(outputAnalysisDir + r"/{}-curve.png".format(turbineName),
                    bbox_inches='tight', dpi=120)
        plt.close()