import os import pandas as pd import plotly.graph_objects as go from algorithmContract.confBusiness import * from algorithmContract.contract import Contract from behavior.analystWithGoodBadPoint import AnalystWithGoodBadPoint class PowerScatterAnalyst(AnalystWithGoodBadPoint): """ 风电机组功率曲线散点分析。 秒级scada数据运算太慢，建议使用分钟级scada数据 """ def typeAnalyst(self): return "power_scatter" def selectColumns(self): return [Field_DeviceCode, Field_Time, Field_WindSpeed, Field_ActiverPower] def processDateTime(self, dataFrame: pd.DataFrame, fieldTime:str = None): super().processDateTime(dataFrame,Field_YearMonth) def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes): dictionary = self.processTurbineData(turbineCodes, conf, self.selectColumns()) dataFrame = self.userDataFrame(dictionary, conf.dataContract.configAnalysis, self) if len(dataFrame) <= 0: print("After screening for blade pitch angle less than the configured value, plot power curve scatter points without data") return # dataFrameGuaranteePowerCurve=self.powerFarmInfo[Field_PowerContractURL] # grouped=self.dataFrameContractOfTurbine.groupby([Field_PowerFarmCode, Field_MillTypeCode]) # for groupByKey,contractPowerCurveOfMillType in grouped: # break turbineInfos = self.common.getTurbineInfos(conf.dataContract.dataFilter.powerFarmID, turbineCodes, self.turbineInfo) return self.drawOfPowerCurveScatter(dataFrame,turbineInfos, outputAnalysisDir, conf) """ def contractGuaranteePowerCurveData(self, csvPowerCurveFilePath): dataFrameGuaranteePowerCurve = pd.read_csv( csvPowerCurveFilePath, encoding=charset_unify) return dataFrameGuaranteePowerCurve """ def drawOfPowerCurveScatter(self, dataFrame: pd.DataFrame, turbineModelInfo: pd.Series, outputAnalysisDir, conf: Contract): """ 绘制风速-功率分布图并保存为文件。 参数: dataFrameMerge (pd.DataFrame): 包含数据的DataFrame，需要包含设备名、风速和功率列。 csvPowerCurveFilePath (str): 功率曲线文件路径。 outputAnalysisDir (str): 分析输出目录。 conf (ConfBusiness): 配置 """ colorsList = [ "#3E409C", "#3586BF", "#52A3AE", "#85D0AE", "#A8DCA2", "#FBFFBE", "#FDF1A9", "#FFE286", "#FCB06C", "#F96F4A", "#E4574C", "#AF254F"] cutInWsField = self.turbineModelInfo[Field_CutInWS] cut_in_ws = cutInWsField.min() - 1 if cutInWsField.notna().any() else 2 # if not dataFrame.empty and Field_CutInWS in dataFrame.columns and dataFrame[Field_CutInWS].notna().any(): # cut_in_ws = dataFrame[Field_CutInWS].min() - 1 # else: # cut_in_ws = 2 grouped = dataFrame.groupby([Field_NameOfTurbine, Field_CodeOfTurbine]) result_rows = [] # 遍历每个设备的数据 for name, group in grouped: # 创建颜色映射，将每个年月映射到一个唯一的颜色 unique_months = group[Field_YearMonth].unique() colors = [ colorsList[i % 12] for i in range(len(unique_months))] color_map = dict(zip(unique_months, colors)) # 使用go.Scatter3d创建3D散点图 trace = go.Scatter3d( x=group[Field_WindSpeed], y=group[Field_YearMonth], z=group[Field_ActiverPower], mode='markers', marker=dict( color=[color_map[month] for month in group[Field_YearMonth]], size=1.5, 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'逐月有功功率散点3D分析: {name[0]}', "x": 0.5 }, scene=dict( xaxis=dict( title='风速', range=[cut_in_ws, 25], ), yaxis=dict( title='时间', tickmode='array', tickvals=unique_months, ticktext=unique_months, categoryorder='category ascending' ), zaxis=dict( title='功率', range=[self.axisLowerLimitActivePower, self.axisUpperLimitActivePower], dtick=self.axisStepActivePower, ) ), 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）间距 ) # 确保从 Series 中提取的是具体的值 engineTypeCode = turbineModelInfo.get(Field_MillTypeCode, "") if isinstance(engineTypeCode, pd.Series): engineTypeCode = engineTypeCode.iloc[0] engineTypeName = turbineModelInfo.get(Field_MachineTypeCode, "") if isinstance(engineTypeName, pd.Series): engineTypeName = engineTypeName.iloc[0] # 构建最终的JSON对象 json_output = { "analysisTypeCode": "逐月有功功率散点3D分析", "engineCode": engineTypeCode, "engineTypeName": engineTypeName, "xaixs": "风速（m/s）", "yaixs": "时间", "zaixs": "有功功率（kW)", "data": [{ "engineName": name[0], "engineCode": name[1], "title":f'逐月有功功率散点3D分析-机组: {name[0]}', "xData": group[Field_WindSpeed].tolist(), "xrange":[cut_in_ws, 25], "yData":group[Field_YearMonth].tolist(), "zData":group[Field_ActiverPower].tolist(), "zrange":[self.axisLowerLimitActivePower, self.axisUpperLimitActivePower], "color":group[Field_YearMonth].tolist() }] } # # Save plot # filePathOfHtml = os.path.join(outputAnalysisDir, f"{name[0]}.html") # fig.write_html(filePathOfHtml) # 将JSON对象保存到文件 output_json_path = os.path.join(outputAnalysisDir, f"power_scatter{name[0]}.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: name[1], 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: name[1], # Field_Return_FilePath: filePathOfHtml, # Field_Return_IsSaveDatabase: True # }) result_df = pd.DataFrame(result_rows) return result_df