import os import pandas as pd import numpy as np import plotly.graph_objects as go from plotly.subplots import make_subplots from behavior.analystNotFilter import AnalystNotFilter from utils.directoryUtil import DirectoryUtil as dir import matplotlib.pyplot as plt from algorithmContract.confBusiness import * from algorithmContract.contract import Contract from utils.jsonUtil import JsonUtil class WindDirectionFrequencyAnalyst(AnalystNotFilter): def typeAnalyst(self): return "wind_direction_frequency" ''' def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes): dictionary=self.processTurbineData(turbineCodes,conf,[Field_DeviceCode,Field_Time,Field_WindDirection,Field_WindSpeed,Field_ActiverPower]) # 获取机型信息 turbrineInfos = self.common.getTurbineInfos( conf.dataContract.dataFilter.powerFarmID, turbineCodes, self.turbineInfo) dataFrameMerge=self.userDataFrame(dictionary,conf.dataContract.configAnalysis,self) frequency_data=self.windRoseAnalysis(dataFrameMerge, outputAnalysisDir, conf) returnDatas=[] returnDatas.append(frequency_data) returnJsonData= self.outputJsonData(conf,outputAnalysisDir,turbrineInfos,dataFrameMerge) returnDatas.append(returnJsonData) return returnDatas ''' def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes): dictionary = self.processTurbineData(turbineCodes, conf, [Field_DeviceCode, Field_Time, Field_WindDirection, Field_WindSpeed, Field_ActiverPower]) turbineInfos = self.common.getTurbineInfos(conf.dataContract.dataFilter.powerFarmID, turbineCodes, self.turbineInfo) dataFrameMerge = self.userDataFrame(dictionary, conf.dataContract.configAnalysis, self) results=self.windRoseAnalysis(dataFrameMerge, outputAnalysisDir, conf) returnJsonData = self.outputJsonData(conf, outputAnalysisDir, turbineInfos, dataFrameMerge) returnDatas=pd.concat([results,returnJsonData], axis=0, ignore_index=True) return returnDatas ''' def windRoseAnalysis(self, dataFrameMerge: pd.DataFrame, outputAnalysisDir, conf: Contract): # 检查所需列是否存在 required_columns = {Field_WindDirection, Field_WindSpeed} if not required_columns.issubset(dataFrameMerge.columns) or dataFrameMerge[Field_WindDirection].isna().all() or dataFrameMerge[Field_WindSpeed].isna().all(): msg=f"DataFrame缺少必要的列及值。涉及的列有: {required_columns}" # raise ValueError(msg) self.logger.warning(msg) return pd.DataFrame() # 风速区间 bins = [0, 3, 6, 9, np.inf] speed_labels = ['[0,3)', '[3,6)', '[6,9)', '>=9'] wind_directions = np.arange(0, 360, 22.5) colorscale = { '[0,3)': 'rgba(247.0, 251.0, 255.0, 1.0)', '[3,6)': 'rgba(171.33333333333334, 207.66666666666666, 229.66666666666669, 1.0)', '[6,9)': 'rgba(55.0, 135.0, 192.33333333333334, 1.0)', '>=9': 'rgba(8.0, 48.0, 107.0, 1.0)' } # 按设备名分组数据 grouped = dataFrameMerge.groupby([Field_NameOfTurbine, Field_CodeOfTurbine]) result_rows = [] for name, group in grouped: speed_bins = pd.cut( group[Field_WindSpeed], bins=bins, labels=speed_labels) # 调整风向数据以使东方为0度 # adjusted_wind_dir = (group[Field_WindDirection] - 90) % 360 # group['风向分组'] = pd.cut(adjusted_wind_dir, bins=wind_directions, labels=wind_directions[:-1]) group['风向分组'] = pd.cut( group[Field_WindDirection], bins=wind_directions, labels=wind_directions[:-1]) # 初始化子图，设定为极坐标 fig = make_subplots(rows=1, cols=1, specs=[[{'type': 'polar'}]]) for label in speed_labels: subset = group[speed_bins == label] counts = subset['风向分组'].value_counts().reindex( wind_directions[:-1], fill_value=0) # 转换为百分比 percentage = (counts / counts.sum()) * 100 # 创建 Barpolar 跟踪，并应用单色渐变 trace = go.Barpolar( r=percentage.values, theta=counts.index, # 这里的角度已经适配上北下南左西右东的布局 name=label, marker_color=colorscale[label], # 应用颜色尺度 marker_showscale=False, # 不显示颜色条 marker_line_color='white', # 设置线条颜色，增加扇区之间的分隔 marker_line_width=1 # 设置线条宽度 ) fig.add_trace(trace) # 设置图表的一些基本属性 fig.update_layout( title={ "text": f"机组: {name[0]}", #"x": 0.5 }, polar=dict( radialaxis=dict(visible=True), angularaxis=dict( tickmode="array", tickvals=wind_directions, # 明确标注北、东、南、西等方向，以适应以北为0度的布局 ticktext=['N', 'NNE', 'NE', 'ENE', 'E', 'ESE', 'SE', 'SSE', 'S', 'SSW', 'SW', 'WSW', 'W', 'WNW', 'NW', 'NNW'] # 更新角度标签，以适应以东为0度的布局 # ticktext=['E', 'ESE', 'SE', 'SSE', 'S', 'SSW', 'SW', 'WSW', 'W', 'WNW', 'NW', 'NNW', 'N', 'NNE', 'NE', 'ENE'] ) ), legend_title="风速", margin=dict(t=50, b=10) # t为顶部（top）间距，b为底部（bottom）间距 ) # 保存图像 filePathOfImage = os.path.join(outputAnalysisDir, f"{name[0]}.png") fig.write_image(filePathOfImage, scale=3) filePathOfHtml = os.path.join(outputAnalysisDir, f"{name[0]}.html") fig.write_html(filePathOfHtml) 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: 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: name[1], Field_Return_FilePath: filePathOfHtml, Field_Return_IsSaveDatabase: True }) result_df = pd.DataFrame(result_rows) return result_df ''' def windRoseAnalysis(self, dataFrameMerge: pd.DataFrame, outputAnalysisDir, conf: Contract): # 检查所需列是否存在 required_columns = {Field_WindDirection, Field_WindSpeed} if not required_columns.issubset(dataFrameMerge.columns) or dataFrameMerge[Field_WindDirection].isna().all() or dataFrameMerge[Field_WindSpeed].isna().all(): msg = f"DataFrame缺少必要的列及值。涉及的列有: {required_columns}" self.logger.warning(msg) return pd.DataFrame() # 定义类属性 self.frequency_data self.frequency_data = {} # 风速区间 bins = [0, 3, 6, 9, np.inf] speed_labels = ['[0,3)', '[3,6)', '[6,9)', '>=9'] wind_directions = np.arange(0, 360, 22.5) # 设置颜色 colorscale = { '[0,3)': 'rgba(247.0, 251.0, 255.0, 1.0)', '[3,6)': 'rgba(171.33333333333334, 207.66666666666666, 229.66666666666669, 1.0)', '[6,9)': 'rgba(55.0, 135.0, 192.33333333333334, 1.0)', '>=9': 'rgba(8.0, 48.0, 107.0, 1.0)' } # 按设备名分组数据 grouped = dataFrameMerge.groupby([Field_NameOfTurbine, Field_CodeOfTurbine]) result_rows = [] for name, group in grouped: turbine_code = name[1] self.frequency_data[turbine_code] = {} self.frequency_data[turbine_code]['title'] = f"机组: {name[0]}" # 调整风向数据以使北方为0度 # adjusted_wind_dir = (group[Field_WindDirection] - 90) % 360 group['风向分组'] = pd.cut( group[Field_WindDirection], bins=wind_directions, labels=wind_directions[:-1]) # 初始化子图，设定为极坐标 fig = make_subplots(rows=1, cols=1, specs=[[{'type': 'polar'}]]) # 计算每个风向分组在各个风速区间内的频率 for label in speed_labels: subset = group[pd.cut(group[Field_WindSpeed], bins=bins, labels=speed_labels) == label] counts = subset['风向分组'].value_counts().reindex(wind_directions[:-1], fill_value=0) percentage = (counts / counts.sum()) * 100 self.frequency_data[turbine_code][label] = percentage.to_dict() # 创建 Barpolar 跟踪，并应用单色渐变 trace = go.Barpolar( r=percentage.values, theta=counts.index, # 这里的角度已经适配上北下南左西右东的布局 name=label, marker_color=colorscale[label], # 应用颜色尺度 marker_showscale=False, # 不显示颜色条 marker_line_color='white', # 设置线条颜色，增加扇区之间的分隔 marker_line_width=1 # 设置线条宽度 ) fig.add_trace(trace) # 设置图表的一些基本属性 fig.update_layout( title={ "text": f"机组: {name[0]}", #"x": 0.5 }, polar=dict( radialaxis=dict(visible=True), angularaxis=dict( tickmode="array", tickvals=wind_directions, # 明确标注北、东、南、西等方向，以适应以北为0度的布局 ticktext=['N', 'NNE', 'NE', 'ENE', 'E', 'ESE', 'SE', 'SSE', 'S', 'SSW', 'SW', 'WSW', 'W', 'WNW', 'NW', 'NNW'] # 更新角度标签，以适应以东为0度的布局 # ticktext=['E', 'ESE', 'SE', 'SSE', 'S', 'SSW', 'SW', 'WSW', 'W', 'WNW', 'NW', 'NNW', 'N', 'NNE', 'NE', 'ENE'] ) ), legend_title="风速", margin=dict(t=50, b=10) # t为顶部（top）间距，b为底部（bottom）间距 ) # 保存图像 filePathOfImage = os.path.join(outputAnalysisDir, f"{name[0]}.png") fig.write_image(filePathOfImage, scale=3) # filePathOfHtml = os.path.join(outputAnalysisDir, f"{name[0]}.html") # fig.write_html(filePathOfHtml) 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: 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: name[1], # Field_Return_FilePath: filePathOfHtml, # Field_Return_IsSaveDatabase: True # }) result_df = pd.DataFrame(result_rows) # 返回结果数据框 return result_df def outputJsonData(self, conf: Contract, outputAnalysisDir: str, turbineModelInfo: pd.Series, dataFrameMerge: pd.DataFrame) -> pd.DataFrame: turbineCodes = dataFrameMerge[Field_CodeOfTurbine].unique() result_rows = [] for turbineCode in turbineCodes: jsonDictionary = self.convert2Json(turbineModelInfo, turbineCodes=turbineCode, dataFrameOfTurbines=dataFrameMerge) jsonFileName = f"风向玫瑰图{turbineCode}.json" jsonFilePath = os.path.join(outputAnalysisDir, jsonFileName) JsonUtil.write_json(jsonDictionary, file_path=jsonFilePath) 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: jsonFilePath, Field_Return_IsSaveDatabase: True }) returnDatas = pd.DataFrame(result_rows) return returnDatas def outputJsonData(self, conf: Contract, outputAnalysisDir: str, turbineModelInfo: pd.Series, dataFrameMerge: pd.DataFrame) -> pd.DataFrame: turbineCodes = dataFrameMerge[Field_CodeOfTurbine].unique() result_rows = [] for turbineCode in turbineCodes: if turbineCode in self.frequency_data: # 确保 enginName 是具体的值 enginName = dataFrameMerge[dataFrameMerge[Field_CodeOfTurbine] == turbineCode][Field_NameOfTurbine].iloc[0] if isinstance(enginName, pd.Series): enginName = enginName.iloc[0] jsonDictionary = self.convert2Json(turbineModelInfo, turbineCode, dataFrameMerge) jsonFileName = f"wind_direction_frequency{enginName}.json" jsonFilePath = os.path.join(outputAnalysisDir, jsonFileName) JsonUtil.write_json(jsonDictionary, file_path=jsonFilePath) 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: jsonFilePath, Field_Return_IsSaveDatabase: True }) returnDatas = pd.DataFrame(result_rows) return returnDatas def convert2Json(self, turbineModelInfo: pd.Series, turbineCode, dataFrameOfTurbines: pd.DataFrame): # 确保从 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] if turbineCode in self.frequency_data: # 确保 enginName 是具体的值 enginName = dataFrameOfTurbines[dataFrameOfTurbines[Field_CodeOfTurbine] == turbineCode][Field_NameOfTurbine].iloc[0] if isinstance(enginName, pd.Series): enginName = enginName.iloc[0] turbine_info = { "enginName": enginName, "enginCode": turbineCode, "title": self.frequency_data[turbineCode]['title'], "windRoseData": [] } for speed_label, direction_data in self.frequency_data[turbineCode].items(): if speed_label == 'title': continue for direction, freq in direction_data.items(): turbine_info["windRoseData"].append({ "windDirection": float(direction), "windSpeedRange": speed_label, "frequency": float(freq) }) result = { "analysisTypeCode": "风向玫瑰分析", "engineTypeCode": engineTypeCode, "engineTypeName": engineTypeName, "axes": { "radial": "频率百分比(%)", "angular": "风向", "levelname": "风速(m/s)" }, "data": [turbine_info] } return result return {}