import os from datetime import datetime import numpy as np import pandas as pd import plotly.express as px import plotly.graph_objects as go from algorithmContract.confBusiness import * from algorithmContract.contract import Contract from behavior.analystWithGoodPoint import AnalystWithGoodPoint class GeneratorSpeedTorqueAnalyst(AnalystWithGoodPoint): """ 风电机组发电机转速-转矩分析 """ def typeAnalyst(self): return "speed_torque" def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes): dictionary = self.processTurbineData(turbineCodes, conf, [ Field_DeviceCode, Field_Time,Field_RotorSpeed,Field_GeneratorSpeed,Field_GeneratorTorque, Field_WindSpeed, Field_ActiverPower]) dataFrameOfTurbines = self.userDataFrame( dictionary, conf.dataContract.configAnalysis, self) # 检查所需列是否存在 required_columns = {Field_CodeOfTurbine,Field_RotorSpeed,Field_GeneratorSpeed,Field_GeneratorTorque} if not required_columns.issubset(dataFrameOfTurbines.columns): raise ValueError(f"DataFrame缺少必要的列。需要的列有: {required_columns}") turbrineInfos = self.common.getTurbineInfos( conf.dataContract.dataFilter.powerFarmID, turbineCodes, self.turbineInfo) groupedOfTurbineModel = turbrineInfos.groupby(Field_MillTypeCode) returnDatas = [] for turbineModelCode, group in groupedOfTurbineModel: currTurbineCodes = group[Field_CodeOfTurbine].unique().tolist() currTurbineModeInfo = self.common.getTurbineModelByCode( turbineModelCode, self.turbineModelInfo) currDataFrameOfTurbines = dataFrameOfTurbines[dataFrameOfTurbines[Field_CodeOfTurbine].isin( currTurbineCodes)] # 将 currTurbineInfos 转换为字典 currTurbineInfos_dict = turbrineInfos.set_index(Field_CodeOfTurbine)[Field_NameOfTurbine].to_dict() # 使用 map 函数来填充 Field_NameOfTurbine 列 currDataFrameOfTurbines[Field_NameOfTurbine] = currDataFrameOfTurbines[Field_CodeOfTurbine].map(currTurbineInfos_dict).fillna("") result2D = self.drawScatter2DMonthly( currDataFrameOfTurbines, outputAnalysisDir, conf) returnDatas.extend(result2D) result3D = self.drawScatterGraph( currDataFrameOfTurbines, outputAnalysisDir, conf) returnDatas.extend(result3D) resultTotal = self.drawScatterGraphForTurbines( currDataFrameOfTurbines, outputAnalysisDir, conf, currTurbineModeInfo) returnDatas.extend(resultTotal) returnDataFrame = pd.DataFrame(returnDatas) return returnDataFrame def drawScatter2DMonthlyOfTurbine(self, dataFrame: pd.DataFrame, outputAnalysisDir: str, conf: Contract, turbineName: str): # 设置颜色条参数 dataFrame = dataFrame.sort_values(by=Field_YearMonth) # 绘制 Plotly 散点图 fig = go.Figure(data=go.Scatter( x=dataFrame[Field_GeneratorSpeed], y=dataFrame[Field_GeneratorTorque], # color=Field_YearMonth, # color_continuous_scale='Rainbow', # 颜色条样式 mode='markers', marker=dict( color=dataFrame['monthIntTime'], colorscale='Rainbow', size=3, opacity=0.7, colorbar=dict( tickvals=np.linspace( dataFrame['monthIntTime'].min(), dataFrame['monthIntTime'].max(), 6), ticktext=[datetime.fromtimestamp(ts).strftime('%Y-%m') for ts in np.linspace( dataFrame['monthIntTime'].min(), dataFrame['monthIntTime'].max(), 6)], thickness=18, len=1, # 设置颜色条的长度,使其占据整个图的高度 outlinecolor='rgba(255,255,255,0)' ), showscale=True ), # labels={Field_GeneratorSpeed: 'Generator Speed', # Field_YearMonth: 'Time', Field_GeneratorTorque: 'Torque'}, showlegend=False )) # # 设置固定散点大小 # fig.update_traces(marker=dict(size=3)) # 如果需要颜色轴的刻度和标签 # 以下是以比例方式进行色彩的可视化处理 fig.update_layout( title={ "text": f'月度发电机转速扭矩散点图: {turbineName}', # "x": 0.5 }, xaxis=dict( title='发电机转速', dtick=self.axisStepGeneratorSpeed, range=[self.axisLowerLimitGeneratorSpeed, self.axisUpperLimitGeneratorSpeed], tickangle=-45 ), yaxis=dict( title='扭矩', dtick=self.axisStepGeneratorTorque, range=[self.axisLowerLimitGeneratorTorque, self.axisUpperLimitGeneratorTorque], ) # coloraxis=dict( # colorbar=dict( # title="Time", # ticks="outside", # len=1, # 设置颜色条的长度,使其占据整个图的高度 # thickness=20, # 调整颜色条的宽度 # orientation='v', # 设置颜色条为垂直方向 # tickmode='array', # 确保刻度按顺序排列 # tickvals=dataFrame[Field_YearMonth].unique( # ).tolist(), # 确保刻度为唯一的年月 # ticktext=dataFrame[Field_YearMonth].unique( # ).tolist() # 以%Y-%m格式显示标签 # ) # ) ) # 保存图片 outputFilePathPNG = os.path.join( outputAnalysisDir, f"{turbineName}.png") fig.write_image(outputFilePathPNG, width=800, height=600, scale=3) # 保存html outputFileHtml = os.path.join(outputAnalysisDir, f"{turbineName}.html") fig.write_html(outputFileHtml) result = [] result.append({ Field_Return_TypeAnalyst: self.typeAnalyst(), Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID, Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum, Field_CodeOfTurbine: dataFrame[Field_CodeOfTurbine].iloc[0], Field_Return_FilePath: outputFilePathPNG, Field_Return_IsSaveDatabase: False }) result.append({ Field_Return_TypeAnalyst: self.typeAnalyst(), Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID, Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum, Field_CodeOfTurbine: dataFrame[Field_CodeOfTurbine].iloc[0], Field_Return_FilePath: outputFileHtml, Field_Return_IsSaveDatabase: True }) return result def drawScatterGraphOfTurbine(self, dataFrame: pd.DataFrame, outputAnalysisDir: str, conf: Contract, turbineName: str): # 创建3D散点图 fig = px.scatter_3d(dataFrame, x=Field_GeneratorSpeed, y=Field_YearMonth, z=Field_GeneratorTorque, color=Field_YearMonth, labels={Field_GeneratorSpeed: '发电机转速', Field_YearMonth: '时间', Field_GeneratorTorque: '扭矩'} ) # 设置固定散点大小 fig.update_traces(marker=dict(size=1.5)) # 更新图形的布局 fig.update_layout( title={ "text": f'月度发电机转速扭矩3D散点图: {turbineName}', "x": 0.5 }, scene=dict( xaxis=dict( title='发电机转速', dtick=self.axisStepGeneratorSpeed, # 设置y轴刻度间隔 range=[self.axisLowerLimitGeneratorSpeed, self.axisUpperLimitGeneratorSpeed], # 设置y轴的范围 showgrid=True, # 显示网格线 ), yaxis=dict( title='时间', tickformat='%Y-%m', # 日期格式, dtick='M1', # 每月一个刻度 showgrid=True, # 显示网格线 ), zaxis=dict( title='扭矩', dtick=self.axisStepGeneratorTorque, range=[self.axisLowerLimitGeneratorTorque, self.axisUpperLimitGeneratorTorque], showgrid=True, # 显示网格线 ) ), 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='Time', legend=dict( orientation="h", itemsizing="constant", # Use constant size for legend items itemwidth=80 # Set the width of legend items to 50 pixels ) ) # 保存图像 outputFileHtml = os.path.join( outputAnalysisDir, "{}_3D.html".format(turbineName)) fig.write_html(outputFileHtml) result = [] result.append({ Field_Return_TypeAnalyst: self.typeAnalyst(), Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID, Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum, Field_CodeOfTurbine: dataFrame[Field_CodeOfTurbine].iloc[0], Field_Return_FilePath: outputFileHtml, Field_Return_IsSaveDatabase: True }) return result def drawScatter2DMonthly(self, dataFrameMerge: pd.DataFrame, outputAnalysisDir, conf: Contract): results = [] grouped = dataFrameMerge.groupby(Field_NameOfTurbine) for name, group in grouped: result = self.drawScatter2DMonthlyOfTurbine( group, outputAnalysisDir, conf, name) results.extend(result) return results def drawScatterGraph(self, dataFrame: pd.DataFrame, outputAnalysisDir, conf: Contract): """ 绘制风速-功率分布图并保存为文件。 参数: dataFrameMerge (pd.DataFrame): 包含数据的DataFrame,需要包含设备名、风速和功率列。 outputAnalysisDir (str): 分析输出目录。 confData (Contract): 配置 """ results = [] dataFrame = dataFrame[(dataFrame[Field_GeneratorTorque] > 0)].sort_values( by=Field_YearMonth) grouped = dataFrame.groupby(Field_NameOfTurbine) # 遍历每个设备的数据 for name, group in grouped: if len(group[Field_YearMonth].unique()) > 1: result = self.drawScatterGraphOfTurbine( group, outputAnalysisDir, conf, name) results.extend(result) return results def drawScatterGraphForTurbines(self, dataFrame: pd.DataFrame, outputAnalysisDir, conf: Contract, turbineModelInfo: pd.Series): """ 绘制风速-功率分布图并保存为文件。 参数: dataFrameMerge (pd.DataFrame): 包含数据的DataFrame,需要包含设备名、风速和功率列。 outputAnalysisDir (str): 分析输出目录。 confData (Contract): 配置 """ dataFrame = dataFrame[(dataFrame[Field_GeneratorTorque] > 0)].sort_values( by=Field_NameOfTurbine) # 创建3D散点图 fig = px.scatter_3d(dataFrame, x=Field_GeneratorSpeed, y=Field_NameOfTurbine, z=Field_GeneratorTorque, color=Field_NameOfTurbine, labels={Field_GeneratorSpeed: '发电机转速', Field_NameOfTurbine: '机组', Field_GeneratorTorque: '实际扭矩'} ) # 设置固定散点大小 fig.update_traces(marker=dict(size=1.5)) # 更新图形的布局 fig.update_layout( title={ "text": f'发电机转速扭矩3D散点图-{turbineModelInfo[Field_MachineTypeCode]}', "x": 0.5 }, scene=dict( xaxis=dict( title='发电机转速', dtick=self.axisStepGeneratorSpeed, # 设置y轴刻度间隔 range=[self.axisLowerLimitGeneratorSpeed, self.axisUpperLimitGeneratorSpeed], # 设置y轴的范围 showgrid=True, # 显示网格线 ), yaxis=dict( title='机组', showgrid=True, # 显示网格线 ), zaxis=dict( title='实际扭矩', dtick=self.axisStepGeneratorTorque, range=[self.axisLowerLimitGeneratorTorque, self.axisUpperLimitGeneratorTorque], ) ), 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)间距 legend=dict( orientation="h", itemsizing="constant", # Use constant size for legend items itemwidth=80 # Set the width of legend items to 50 pixels ) ) # 保存图像 outputFileHtml = os.path.join( outputAnalysisDir, "{}-{}.html".format(self.typeAnalyst(),turbineModelInfo[Field_MillTypeCode])) fig.write_html(outputFileHtml) result = [] result.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: outputFileHtml, Field_Return_IsSaveDatabase: True }) return result