import os import pandas as pd import numpy as np from plotly.subplots import make_subplots import plotly.graph_objects as go import matplotlib.pyplot as plt from matplotlib.ticker import MultipleLocator from behavior.analystWithGoodPoint import AnalystWithGoodPoint from utils.directoryUtil import DirectoryUtil as dir from algorithmContract.confBusiness import * from algorithmContract.contract import Contract class CpTrendAnalyst(AnalystWithGoodPoint): """ 风电机组风能利用系数时序分析 """ def typeAnalyst(self): return "cp_trend" def turbinesAnalysis(self, outputAnalysisDir, conf: Contract,turbineCodes): dictionary = self.processTurbineData(turbineCodes, conf, [ Field_DeviceCode, Field_Time, Field_WindSpeed, Field_ActiverPower]) dataFrameOfTurbines = self.userDataFrame( dictionary, conf.dataContract.configAnalysis, self) # 检查所需列是否存在 required_columns = {Field_Cp, Field_YearMonthDay} if not required_columns.issubset(dataFrameOfTurbines.columns): raise ValueError(f"DataFrame缺少必要的列。需要的列有: {required_columns}") return self.drawCpTrend(dataFrameOfTurbines, outputAnalysisDir, conf) def drawCpTrend(self, dataFrameOfTurbines: pd.DataFrame, outputAnalysisDir, conf: Contract): # 按设备名分组数据 grouped = dataFrameOfTurbines.groupby(Field_CodeOfTurbine) result_rows = [] for turbineCode, group in grouped: currTurbineInfo=self.common.getTurbineInfo(conf.dataContract.dataFilter.powerFarmID,turbineCode,self.turbineInfo) # # 计算四分位数和IQR Q1 = group[Field_Cp].quantile(0.25) Q3 = group[Field_Cp].quantile(0.75) IQR = Q3 - Q1 # 定义离群值的范围 lower_bound = Q1 - 1.5 * IQR upper_bound = Q3 + 1.5 * IQR # 筛选掉离群值 filtered_group = group[(group[Field_Cp] >= lower_bound) & ( group[Field_Cp] <= upper_bound)] # 创建箱线图 fig = go.Figure() fig.add_trace(go.Box( x=filtered_group[Field_YearMonthDay], # 设置x轴数据为日期 y=filtered_group[Field_Cp], # 设置y轴数据为风能利用系数 # boxpoints='outliers', # 显示异常值（偏离值），不显示数据的所有点（只显示异常值） boxpoints=False, # 不显示偏离值 marker=dict(color='lightgoldenrodyellow', size=1), # 设置偏离值的颜色和大小 line=dict(color='lightgray', width=2), # 设置箱线和须线的颜色为灰色，粗细为2 fillcolor='rgba(200, 200, 200, 0.5)', # 设置箱体的填充颜色和透明度 name='风能利用系数' # 图例名称 )) # 对于每个箱线图的中位数，绘制一个蓝色点 medians = filtered_group.groupby(filtered_group[Field_YearMonthDay])[ Field_Cp].median() fig.add_trace(go.Scatter( x=medians.index, y=medians.values, mode='markers', marker=dict(color='orange', size=3), name='风能利用系数-中位数' # 中位数标记的图例名称 )) # 设置图表的标题和轴标签 fig.update_layout( title={ 'text': f'机组: {currTurbineInfo[Field_NameOfTurbine]}', # 'x': 0.5, }, xaxis_title='时间', yaxis_title='风能利用系数', xaxis=dict( tickmode='auto', # 自动设置x轴刻度，以适应日期数据 tickformat='%Y-%m-%d', # 设置x轴时间格式 showgrid=True, # 显示网格线 gridcolor='lightgray', # setting y-axis gridline color to black tickangle=-45, linecolor='black', # 设置y轴坐标系线颜色为黑色 ticklen=5, # 设置刻度线的长度 ), yaxis=dict( dtick=self.axisStepCp, range=[self.axisLowerLimitCp, self.axisUpperLimitCp], # 设置y轴的范围从0到1 showgrid=True, # 显示网格线 gridcolor='lightgray', # setting y-axis gridline color to black linecolor='black', # 设置y轴坐标系线颜色为黑色 ticklen=5, # 设置刻度线的长度 ), paper_bgcolor='white', # 设置纸张背景颜色为白色 plot_bgcolor='white', # 设置图表背景颜色为白色 margin=dict(t=50, b=10) # t为顶部（top）间距，b为底部（bottom）间距 ) # 确保从 Series 中提取的是具体的值 engineTypeCode = currTurbineInfo.get(Field_MillTypeCode, "") if isinstance(engineTypeCode, pd.Series): engineTypeCode = engineTypeCode.iloc[0] engineTypeName = currTurbineInfo.get(Field_MachineTypeCode, "") if isinstance(engineTypeName, pd.Series): engineTypeName = engineTypeName.iloc[0] # 构建最终的JSON对象 json_output = { "analysisTypeCode": "风能利用系数时序分析", "engineCode": engineTypeCode, "engineTypeName": engineTypeName, "xaixs": "时间", "yaixs": "风能利用系数", "data": [{ "engineName":currTurbineInfo[Field_NameOfTurbine], "engineCode":turbineCode, "title":f'机组-{currTurbineInfo[Field_NameOfTurbine]}', "xData": filtered_group[Field_YearMonthDay].tolist(), "yData": filtered_group[Field_Cp].tolist(), "color":'lightgray', "width":2, "type":"box_plot", "medians": { "x": medians.index.tolist(), # 中位数的 x 轴数据 "y": medians.values.tolist(), # 中位数的 y 轴数据 "mode":'markers', "color":'orange', "size":3 } }] } # 保存图像 filePathOfImage = os.path.join(outputAnalysisDir, f"{currTurbineInfo[Field_NameOfTurbine]}.png") fig.write_image(filePathOfImage, scale=3) # filePathOfHtml = os.path.join(outputAnalysisDir, f"{currTurbineInfo[Field_NameOfTurbine]}.html") # fig.write_html(filePathOfHtml) # 将JSON对象保存到文件 output_json_path = os.path.join(outputAnalysisDir, f"{currTurbineInfo[Field_NameOfTurbine]}.json") with open(output_json_path, 'w', encoding='utf-8') as f: import json json.dump(json_output, f, ensure_ascii=False, indent=4) # 如果需要返回DataFrame，可以包含文件路径 result_rows.append({ Field_Return_TypeAnalyst: self.typeAnalyst(), Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID, Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum, Field_CodeOfTurbine: turbineCode, Field_Return_FilePath: output_json_path, Field_Return_IsSaveDatabase: True }) result_rows.append({ Field_Return_TypeAnalyst: self.typeAnalyst(), Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID, Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum, Field_CodeOfTurbine: turbineCode, Field_Return_FilePath: filePathOfImage, Field_Return_IsSaveDatabase: False }) # result_rows.append({ # Field_Return_TypeAnalyst: self.typeAnalyst(), # Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID, # Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum, # Field_CodeOfTurbine: turbineCode, # Field_Return_FilePath: filePathOfHtml, # Field_Return_IsSaveDatabase: True # }) result_df = pd.DataFrame(result_rows) return result_df