import os import pandas as pd import numpy as np import plotly.graph_objects as go from plotly.subplots import make_subplots import seaborn as sns import matplotlib.pyplot as plt from matplotlib.ticker import MultipleLocator from behavior.analystExcludeRatedPower import AnalystExcludeRatedPower from utils.directoryUtil import DirectoryUtil as dir from algorithmContract.confBusiness import * class CpAnalyst(AnalystExcludeRatedPower): """ 风电机组风能利用系数分析 """ def typeAnalyst(self): return "cp" def turbinesAnalysis(self, dataFrameMerge:pd.DataFrame, outputAnalysisDir, confData: ConfBusiness): # 检查所需列是否存在 required_columns = {confData.field_wind_speed, Field_Cp, Field_PowerFloor} if not required_columns.issubset(dataFrameMerge.columns): raise ValueError(f"DataFrame缺少必要的列。需要的列有: {required_columns}") self.drawLineGraphForTurbine(dataFrameMerge,outputAnalysisDir,confData) def drawLineGraphForTurbine(self,dataFrameMerge:pd.DataFrame, outputAnalysisDir, confData: ConfBusiness): sns.set_palette('deep') upLimitOfPower = confData.rated_power*0.9 value_step = 0.5 grouped = dataFrameMerge.groupby([Field_NameOfTurbine,Field_PowerFloor]).agg( cp=('cp', 'median'), cp_max=('cp', 'max'), cp_min=('cp', 'min'), ).reset_index() # Rename columns post aggregation for clarity grouped.columns = [Field_NameOfTurbine,Field_PowerFloor, Field_CpMedian, 'cp_max', 'cp_min'] # Sort by power_floor grouped = grouped.sort_values(by=[Field_NameOfTurbine,Field_PowerFloor]) fig, ax = plt.subplots() ax = sns.lineplot(x=Field_PowerFloor, y=Field_CpMedian, data=grouped, hue=Field_NameOfTurbine) # 绘制合同功率曲线 ax.plot(self.dataFrameContractOfTurbine[Field_PowerFloor], self.dataFrameContractOfTurbine[Field_Cp], marker='o', c='red', label='Contract Guarantee Cp') ax.xaxis.set_major_locator(MultipleLocator(confData.graphSets["activePower"]["step"] if not self.common.isNone( confData.graphSets["activePower"]) and not self.common.isNone( confData.graphSets["activePower"]["step"]) else 250)) # 创建一个刻度 ，将定位器应用到y轴上 ax.set_xlim(0, upLimitOfPower) ax.yaxis.set_major_locator(MultipleLocator( confData.graphSets["cp"]["step"] if not self.common.isNone(confData.graphSets["cp"]["step"]) else value_step)) # 创建一个刻度 ，将定位器应用到y轴上 ax.set_ylim(confData.graphSets["cp"]["min"] if not self.common.isNone(confData.graphSets["cp"]["min"]) else 0, confData.graphSets["cp"]["max"] if not self.common.isNone(confData.graphSets["cp"]["max"]) else 2) ax.set_title('Cp-Distribution') # plt.legend(ncol=4) plt.xticks(rotation=45) # 旋转45度 plt.legend(title='Turbine', bbox_to_anchor=(1.02, 0.5), ncol=2, loc='center left', borderaxespad=0.) plt.savefig(os.path.join( outputAnalysisDir, "{}-Cp-Distribution.png".format(confData.farm_name)), bbox_inches='tight', dpi=300) plt.close() groupedX=grouped.groupby(Field_NameOfTurbine) for name,group in groupedX: color = ["lightgrey"] * len(dataFrameMerge[Field_NameOfTurbine].unique()) fig, ax = plt.subplots(figsize=(8, 8)) ax = sns.lineplot(x=Field_PowerFloor, y=Field_CpMedian, data=grouped, hue=Field_NameOfTurbine, palette=sns.color_palette(color), legend=False) ax = sns.lineplot(x=Field_PowerFloor, y=Field_CpMedian, data=group, color='darkblue', legend=False) # 绘制合同功率曲线 ax.plot(self.dataFrameContractOfTurbine[Field_PowerFloor], self.dataFrameContractOfTurbine[Field_Cp], marker='o', c='red', label='Contract Guarantee Cp') ax.xaxis.set_major_locator( MultipleLocator(confData.graphSets["activePower"]["step"] if not self.common.isNone( confData.graphSets["activePower"]) and not self.common.isNone( confData.graphSets["activePower"]["step"]) else 250)) # 创建一个刻度 ，将定位器应用到y轴上 ax.set_xlim(0, upLimitOfPower) ax.yaxis.set_major_locator(MultipleLocator( confData.graphSets["cp"]["step"] if not self.common.isNone(confData.graphSets["cp"]["step"]) else value_step)) # 创建一个刻度 ，将定位器应用到y轴上 ax.set_ylim(confData.graphSets["cp"]["min"] if not self.common.isNone(confData.graphSets["cp"]["min"]) else 0, confData.graphSets["cp"]["max"] if not self.common.isNone(confData.graphSets["cp"]["max"]) else 1) ax.set_title('turbine name={}'.format(name)) plt.xticks(rotation=45) # 旋转45度 plt.legend(title='Turbine', bbox_to_anchor=(1.02, 0.5), ncol=2, loc='center left', borderaxespad=0.) plt.savefig(os.path.join(outputAnalysisDir, "{}.png".format( name)), bbox_inches='tight', dpi=120) plt.close() def generate_cp_distribution(self, csvFileDirOfCp, confData: ConfBusiness, encoding='utf-8'): """ Generates Cp distribution plots for turbines in a wind farm. Parameters: - csvFileDirOfCp: str, path to the directory containing input CSV files. - farm_name: str, name of the wind farm. - encoding: str, encoding of the input CSV files. Defaults to 'utf-8'. """ output_path = csvFileDirOfCp field_Name_Turbine = "turbine_name" x_name = 'power_floor' y_name = 'cp' upLimitOfPower = confData.rated_power*0.9 value_step = 0.5 sns.set_palette('deep') res = pd.DataFrame() for root, dir_names, file_names in dir.list_directory(csvFileDirOfCp): for file_name in file_names: if not file_name.endswith(CSVSuffix): continue file_path = os.path.join(root, file_name) print(file_path) frame = pd.read_csv(file_path, encoding=encoding) frame = frame[(frame[x_name] > 0)] turbine_name = file_name.split(CSVSuffix)[0] frame[field_Name_Turbine] = turbine_name res = pd.concat( [res, frame.loc[:, [field_Name_Turbine, x_name, y_name]]], axis=0) ress = res.reset_index() fig, ax = plt.subplots() ax = sns.lineplot(x=x_name, y=y_name, data=ress, hue=field_Name_Turbine) ax.xaxis.set_major_locator(MultipleLocator(200)) # 创建一个刻度 ，将定位器应用到y轴上 ax.set_xlim(0, upLimitOfPower) ax.yaxis.set_major_locator(MultipleLocator( confData.graphSets["cp"]["step"] if not self.common.isNone(confData.graphSets["cp"]["step"]) else value_step)) # 创建一个刻度 ，将定位器应用到y轴上 ax.set_ylim(confData.graphSets["cp"]["min"] if not self.common.isNone(confData.graphSets["cp"]["min"]) else 0, confData.graphSets["cp"]["max"] if not self.common.isNone(confData.graphSets["cp"]["max"]) else 2) ax.set_title('Cp-Distribution') # plt.legend(ncol=4) plt.xticks(rotation=45) # 旋转45度 plt.legend(title='turbine', bbox_to_anchor=(1.02, 0.5), ncol=2, loc='center left', borderaxespad=0.) plt.savefig(os.path.join( output_path, "{}-Cp-Distribution.png".format(confData.farm_name)), bbox_inches='tight', dpi=300) plt.close() grouped = ress.groupby(field_Name_Turbine) for name, group in grouped: color = ["lightgrey"] * len(ress[field_Name_Turbine].unique()) fig, ax = plt.subplots(figsize=(8, 8)) ax = sns.lineplot(x=x_name, y=y_name, data=ress, hue=field_Name_Turbine, palette=sns.color_palette(color), legend=False) ax = sns.lineplot(x=x_name, y=y_name, data=group, color='darkblue', legend=False) ax.xaxis.set_major_locator( MultipleLocator(200)) # 创建一个刻度 ，将定位器应用到y轴上 ax.set_xlim(0, upLimitOfPower) ax.yaxis.set_major_locator(MultipleLocator( confData.graphSets["cp"]["step"] if not self.common.isNone(confData.graphSets["cp"]["step"]) else value_step)) # 创建一个刻度 ，将定位器应用到y轴上 ax.set_ylim(confData.graphSets["cp"]["min"] if not self.common.isNone(confData.graphSets["cp"]["min"]) else 0, confData.graphSets["cp"]["max"] if not self.common.isNone(confData.graphSets["cp"]["max"]) else 1) ax.set_title('turbine name={}'.format(name)) plt.xticks(rotation=45) # 旋转45度 plt.savefig(os.path.join(output_path, "{}.png".format( name)), bbox_inches='tight', dpi=120) plt.close() def plot_cp_distribution(self, csvFileDir, farm_name): field_Name_Turbine = "设备名" x_name = 'power_floor' y_name = 'cp' # Create the output path based on the farm name output_path = csvFileDir # output_path_template.format(farm_name) # Ensure the output directory exists os.makedirs(output_path, exist_ok=True) print(csvFileDir) # Initialize a DataFrame to store results res = pd.DataFrame() # Walk through the input directory to process each file for root, _, file_names in dir.list_directory(csvFileDir): for file_name in file_names: full_path = os.path.join(root, file_name) frame = pd.read_csv(full_path, encoding='gbk') turbine_name = file_name.split(CSVSuffix)[0] print("turbine_name={}".format(turbine_name)) frame[field_Name_Turbine] = turbine_name res = pd.concat( [res, frame.loc[:, [field_Name_Turbine, x_name, y_name]]], axis=0) # Reset index for plotting ress = res.reset_index(drop=True) # Plot combined Cp distribution for all turbines fig = make_subplots(rows=1, cols=1) for name, group in ress.groupby(field_Name_Turbine): fig.add_trace(go.Scatter( x=group[x_name], y=group[y_name], mode='lines', name=name)) fig.update_layout(title_text='{} Cp分布'.format( farm_name), xaxis_title=x_name, yaxis_title=y_name) fig.write_image(os.path.join( output_path, "{}Cp分布.png".format(farm_name)), scale=3) # Plot individual Cp distributions unique_turbines = ress[field_Name_Turbine].unique() for name in unique_turbines: individual_fig = make_subplots(rows=1, cols=1) # Add all turbines in grey for turbine in unique_turbines: group = ress[ress[field_Name_Turbine] == turbine] individual_fig.add_trace(go.Scatter( x=group[x_name], y=group[y_name], mode='lines', name=turbine, line=dict(color='lightgrey'))) # Highlight the current turbine in dark blue group = ress[ress[field_Name_Turbine] == name] individual_fig.add_trace(go.Scatter( x=group[x_name], y=group[y_name], mode='lines', name=name, line=dict(color='darkblue'))) individual_fig.update_layout(title_text='设备名={}'.format(name)) individual_fig.write_image(os.path.join( output_path, "all-{}.png".format(name)), scale=2)