import os import pandas as pd import numpy as np import plotly.graph_objects as go from plotly.subplots import make_subplots import plotly.express as px import seaborn as sns import matplotlib.pyplot as plt from matplotlib.ticker import MultipleLocator from behavior.analyst import Analyst from utils.directoryUtil import DirectoryUtil as dir from algorithmContract.confBusiness import * class GeneratorSpeedTorqueAnalyst(Analyst): """ 风电机组发电机转速-转矩分析 """ def typeAnalyst(self): return "speed_torque" def turbinesAnalysis(self, dataFrameMerge, outputAnalysisDir, confData: ConfBusiness): self.create_and_save_plots( dataFrameMerge, outputAnalysisDir, confData) self.drawScatterGraph(dataFrameMerge, outputAnalysisDir, confData) self.drawScatterGraphForTurbines( dataFrameMerge, outputAnalysisDir, confData) def create_and_save_plots(self, dataFrame: pd.DataFrame, outputAnalysisDir, confData: ConfBusiness): # 检查所需列是否存在 required_columns = {confData.field_gen_speed, Field_GeneratorTorque} if not required_columns.issubset(dataFrame.columns): raise ValueError(f"DataFrame缺少必要的列。需要的列有: {required_columns}") x_name = 'generator_speed' y_name = 'generator_torque' maxTurque = dataFrame[Field_GeneratorTorque].max() grouped = dataFrame.groupby(Field_NameOfTurbine) for name, group in grouped: groupNew = group.copy() if not self.common.isNone(confData.value_gen_speed_multiple): groupNew[confData.field_gen_speed] = group[confData.field_gen_speed] * \ confData.value_gen_speed_multiple # sns.lmplot函数参数scatter_kws： 设置为{"s": 5}时，会出现颜色丢失问题；改为={"s": 5, "color": "b"}后，则造成图形风格不统一问题； g = sns.lmplot(x=confData.field_gen_speed, y=Field_GeneratorTorque, data=groupNew, fit_reg=False, scatter_kws={ "s": 5, "color": "b"}, legend=False, height=6, aspect=1.2) # g = sns.lmplot(x=fieldGeneratorSpeed, y=Field_GeneratorTorque, data=group, fit_reg=False, scatter_kws={ # "s": 5}, legend=False, height=6, aspect=1.2) for ax in g.axes.flat: # 创建每100个单位一个刻度的定位器 loc = MultipleLocator(confData.graphSets["generatorSpeed"]["step"] if not self.common.isNone( confData.graphSets["generatorSpeed"]) and not self.common.isNone( confData.graphSets["generatorSpeed"]["step"]) else 200) ax.xaxis.set_major_locator(loc) # 将定位器应用到x轴上 ax.set_xlim(confData.graphSets["generatorSpeed"]["min"] if not self.common.isNone( confData.graphSets["generatorSpeed"]["min"]) else 1000, confData.graphSets["generatorSpeed"]["max"] if not self.common.isNone(confData.graphSets["generatorSpeed"]["max"]) else 2000) yloc = MultipleLocator(confData.graphSets["generatorTorque"]["step"] if not self.common.isNone( confData.graphSets["generatorTorque"]["step"]) else 200) ax.yaxis.set_major_locator(yloc) # 将定位器应用到y轴上 ax.set_ylim(confData.graphSets["generatorTorque"]["min"] if not self.common.isNone( confData.graphSets["generatorTorque"]["min"]) else 0, confData.graphSets["generatorTorque"]["max"] if not self.common.isNone(confData.graphSets["generatorTorque"]["max"]) else 2000) ax.set_xlabel(x_name) ax.set_ylabel(y_name) plt.tight_layout() plt.title(f'{Field_NameOfTurbine}={name}') # 保存图片到指定路径 output_file = os.path.join(outputAnalysisDir, f"{name}.png") plt.savefig(output_file, bbox_inches='tight', dpi=120) plt.close() def drawScatterGraph(self, dataFrame: pd.DataFrame, outputAnalysisDir, confData: ConfBusiness): """ 绘制风速-功率分布图并保存为文件。 参数: dataFrameMerge (pd.DataFrame): 包含数据的DataFrame，需要包含设备名、风速和功率列。 outputAnalysisDir (str): 分析输出目录。 confData (ConfBusiness): 配置 """ dataFrame = dataFrame[(dataFrame[Field_GeneratorTorque] > 0)] # 按设备名分组数据 colorsList = ['#1f77b4', '#ff7f0e', '#2ca02c', '#d62728', '#9467bd', '#8c564b', '#e377c2', '#7f7f7f', '#bcbd22', '#17becf', '#aec7e8', '#ffbb78'] grouped = dataFrame.groupby(Field_NameOfTurbine) # 遍历每个设备的数据 for name, group in grouped: # 创建颜色映射，将每个年月映射到一个唯一的颜色 unique_months = group[Field_YearMonth].unique() colors = [ colorsList[i % len(colorsList)] for i in range(len(unique_months))] color_map = dict(zip(unique_months, colors)) # 使用go.Scatter3d创建3D散点图 trace = go.Scatter3d( x=group[confData.field_gen_speed], y=group[Field_YearMonth], z=group[Field_GeneratorTorque], mode='markers', marker=dict( color=[color_map[month] for month in group[Field_YearMonth]], size=2, line=dict( color='rgba(0, 0, 0, 0)', # 设置边框颜色为透明，以去掉白色边框 width=0 # 设置边框宽度为0，进一步确保没有边框 ), opacity=0.8 # 调整散点的透明度，增加透视效果 ) ) # 创建图形 fig = go.Figure(data=[trace]) # 更新图形的布局 fig.update_layout( title={ "text": f'Monthly generator speed torque 3D scatter plot {name}', "x": 0.5 }, scene=dict( xaxis=dict( title='Generator Speed', dtick=confData.graphSets["generatorSpeed"]["step"] if not self.common.isNone( confData.graphSets["generatorSpeed"]["step"]) else 200, # 设置y轴刻度间隔为0.1 range=[confData.graphSets["generatorSpeed"]["min"] if not self.common.isNone( confData.graphSets["generatorSpeed"]["min"]) else 1000, confData.graphSets["generatorSpeed"]["max"] if not self.common.isNone(confData.graphSets["generatorSpeed"]["max"]) else 2000], # 设置y轴的范围从0到1 showgrid=True, # 显示网格线 ), yaxis=dict( title='Time', tickmode='array', tickvals=unique_months, ticktext=unique_months, # dtick=500, # 设置y轴刻度间隔 # range=[0, # group[Field_GeneratorTorque].max()], # 设置y轴的范围 showgrid=True, # 显示网格线 # categoryorder='category ascending' ), zaxis=dict( title='Generator Torque', dtick=confData.graphSets["generatorTorque"]["step"] if not self.common.isNone( confData.graphSets["generatorTorque"]["step"]) else 200, # 设置y轴刻度间隔为0.1 range=[confData.graphSets["generatorTorque"]["min"] if not self.common.isNone( confData.graphSets["generatorTorque"]["min"]) else 0, confData.graphSets["generatorTorque"]["max"] if not self.common.isNone(confData.graphSets["generatorTorque"]["max"]) else 2000], # 设置y轴的范围从0到1 ) ), 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° ), margin=dict(t=50, b=10) # t为顶部（top）间距，b为底部（bottom）间距 ) # 保存图像 outputFileHtml = os.path.join( outputAnalysisDir, "{}.html".format(name)) fig.write_html(outputFileHtml) def drawScatterGraphForTurbines(self, dataFrame: pd.DataFrame, outputAnalysisDir, confData: ConfBusiness): """ 绘制风速-功率分布图并保存为文件。 参数: dataFrameMerge (pd.DataFrame): 包含数据的DataFrame，需要包含设备名、风速和功率列。 outputAnalysisDir (str): 分析输出目录。 confData (ConfBusiness): 配置 """ dataFrame = dataFrame[(dataFrame[confData.field_power] > 0)].sort_values( by=Field_NameOfTurbine) # 创建3D散点图 fig = px.scatter_3d(dataFrame, x=confData.field_gen_speed, y=Field_NameOfTurbine, z=Field_GeneratorTorque, color=Field_NameOfTurbine, labels={confData.field_gen_speed: 'Generator Speed', Field_NameOfTurbine: 'Turbine', Field_GeneratorTorque: 'Generator Torque'}, ) # 设置固定散点大小 fig.update_traces(marker=dict(size=1.5)) # 更新图形的布局 fig.update_layout( title={ "text": 'Turbine generator speed Turque 3D scatter plot', "x": 0.5 }, scene=dict( xaxis=dict( title='Generator Speed', dtick=confData.graphSets["generatorSpeed"]["step"] if not self.common.isNone( confData.graphSets["generatorSpeed"]["step"]) else 200, # 设置y轴刻度间隔为0.1 range=[confData.graphSets["generatorSpeed"]["min"] if not self.common.isNone( confData.graphSets["generatorSpeed"]["min"]) else 1000, confData.graphSets["generatorSpeed"]["max"] if not self.common.isNone(confData.graphSets["generatorSpeed"]["max"]) else 2000], # 设置y轴的范围从0到1 showgrid=True, # 显示网格线 ), yaxis=dict( title='Turbine', showgrid=True, # 显示网格线 ), zaxis=dict( title='Generator Turque', dtick=confData.graphSets["generatorTorque"]["step"] if not self.common.isNone( confData.graphSets["generatorTorque"]["step"]) else 200, # 设置y轴刻度间隔为0.1 range=[confData.graphSets["generatorTorque"]["min"] if not self.common.isNone( confData.graphSets["generatorTorque"]["min"]) else 0, confData.graphSets["generatorTorque"]["max"] if not self.common.isNone(confData.graphSets["generatorTorque"]["max"]) else 2000], # 设置y轴的范围从0到1 ) ), 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', margin=dict(t=50, b=10) # t为顶部（top）间距，b为底部（bottom）间距 ) # 保存图像 outputFileHtml = os.path.join( outputAnalysisDir, "{}.html".format(self.typeAnalyst())) fig.write_html(outputFileHtml)