WTOAAM/dataAnalysisBusiness/algorithm/yawErrorAnalyst.py

import os
import pandas as pd
import numpy as np
import plotly.graph_objects as go
from algorithmContract.confBusiness import *
from algorithmContract.contract import Contract
from behavior.analystWithGoodPoint import AnalystWithGoodPoint
from plotly.subplots import make_subplots
from scipy.optimize import curve_fit


class YawErrorAnalyst(AnalystWithGoodPoint):
    """
    风电机组静态偏航误差分析
    """
    fieldWindDirFloor = 'wind_dir_floor'
    fieldPower = 'power'
    fieldPowerMean = 'mean_power'
    fieldPowerMax = 'max_power'
    fieldPowerMin = 'min_power'
    fieldPowerMedian = 'median_power'
    fieldYawError = 'yaw_error'
    fieldStep = 'wsstep'
    fieldK = 'k' 
    fieldWindSpeed = 'mean_WindSpeed'
    fieldcount = 'point_num'

    def typeAnalyst(self):
        return "yaw_error"

    def selectColumns(self):
        return [Field_DeviceCode,Field_Time,Field_WindSpeed,Field_ActiverPower,Field_AngleIncluded,Field_PitchAngel1,Field_RotorSpeed, Field_GeneratorSpeed]

    def filterCommon(self, dataFrame: pd.DataFrame,conf:Contract):
        #-------------------1.物理筛选--------
        # 使用 loc 方法获取 Field_RatedPower 列的值
        fullpower  = self.turbineInfo[Field_RatedPower].iloc[0]
        # rated_power  = self.turbineInfo.loc[self.turbineInfo[Field_CodeOfTurbine].isin(dataFrame[Field_CodeOfTurbine]), Field_RatedPower]
        # fullpower=rated_power.iloc[0]
        # rated_power=self.turbineInfo[Field_RatedPower]

        # 删除小于0的有功功率
        dataFrame = dataFrame[~(dataFrame[Field_ActiverPower] < 0)]
        # 删除小于2.5的风速
        dataFrame = dataFrame[~(dataFrame[Field_WindSpeed] < 2.5)]
        # 删除有功小于额定功率-100，变桨角度大于0.2的数据
        dataFrame = dataFrame[~((dataFrame[Field_ActiverPower] < fullpower - 100) & (dataFrame[Field_PitchAngel1] > 3))]
        # 删除有功小于额定功率-40，变桨角度大于1的数据
        dataFrame = dataFrame[~((dataFrame[Field_ActiverPower] < fullpower - 40) & (dataFrame[Field_PitchAngel1] > 5))]
        # 删除对风角度小于-30，大于30的数据
        dataFrame = dataFrame[~((dataFrame[Field_AngleIncluded].abs() > 30))]
        #-------------------2.数学筛选--------
        wsstep = 0.5
        stdup = 6
        stddown = 2.2
        for wscol in np.arange(3, 15.25, wsstep/2):
            dataFrame_lv = dataFrame[(dataFrame[Field_WindSpeed] >= (wscol-wsstep/4)) & (dataFrame[Field_WindSpeed] < (wscol+wsstep/4))]
            meanpowr = np.mean(dataFrame_lv[Field_ActiverPower])
            stddataws = np.std(dataFrame_lv[Field_ActiverPower])
            index1 = dataFrame[(dataFrame[Field_WindSpeed] >= (wscol-wsstep/4)) & (dataFrame[Field_WindSpeed] < (wscol+wsstep/4))
                             & ((dataFrame_lv[Field_ActiverPower] - meanpowr) > stdup * stddataws) | ((meanpowr - dataFrame_lv
            [Field_ActiverPower]) > stddown * stddataws)].index
            dataFrame.drop(index1, inplace=True)

        return dataFrame

    def calculateYawError(self, dataFrame: pd.DataFrame, fieldAngleInclude,fieldActivePower, fieldWindSpeed):
        
        dataFrame = dataFrame.dropna(
            subset=[Field_NameOfTurbine, fieldAngleInclude, fieldActivePower,fieldWindSpeed])
        df_math_yaw = dataFrame.copy()
        # 偏航误差角度计算
        degree_length = 20  # 对风角度最大值
        yaw_data_out = np.empty((0, 4), float)
        wsstep = 0.5
        # 初始化一个字典来存储每 k 次循环的平均功率结果
        power_dict = {}
        for k in np.arange(3, 15.5, wsstep):
            yaw_data_value_co = np.empty((0, 4), float)
            yaw_data = df_math_yaw[(df_math_yaw[fieldWindSpeed] >= (k - 0.25)) & (df_math_yaw[fieldWindSpeed] < (k + 0.25))]
            # print(yaw_data)
            for m in np.arange((0 - degree_length), degree_length + 1, 0.5):
                wd = yaw_data[(yaw_data[fieldAngleInclude] >= (m - 0.5)) & (yaw_data[fieldAngleInclude] < (m + 0.5))]
                wd_row = wd.shape[0]
                if wd_row < 10:
                    continue
                if wd_row >= 10:
                    max_value_row = wd[fieldActivePower].idxmax()
                    min_value_row = wd[fieldActivePower].idxmin()
                    wd = wd.drop(max_value_row)
                    wd = wd.drop(min_value_row)
                wd_row = wd.shape[0]
                mean_wd = wd[fieldAngleInclude].mean()
                mean_power = wd[fieldActivePower].mean()
                mean_ws = wd[fieldWindSpeed].mean()
                yaw_data_value = [mean_power, mean_wd, mean_ws, wd_row]
                yaw_data_value_co = np.vstack((yaw_data_value_co, yaw_data_value))
                # 存储当前 k 和 m 对应的 mean_power
                if k not in power_dict:
                    power_dict[k] = {}
                power_dict[k][m] = mean_power
            # 检查 yaw_data_value_co 是否为空
            if yaw_data_value_co.size == 0:
                print(f"yaw_data_value_co k={k}是空的,无法计算argmax")
                continue
            else:
                max_row = np.argmax(yaw_data_value_co[:, 0])
                yaw_data_out = np.vstack((yaw_data_out, yaw_data_value_co[max_row, :]))

        # 将结果转换为 DataFrame
        result_df = pd.DataFrame(yaw_data_out, columns=[self.fieldPowerMean, self.fieldYawError,self.fieldWindSpeed,self.fieldcount])
        # result_df.to_csv('D:\\dashengkeji\\project\\WTOAAM\\debug_tset_20241008\\result_df_JHS_.csv',index=False)
        # 构建最终画图的 DataFrame
        final_df = pd.DataFrame(columns=[self.fieldWindDirFloor] + [f'{k}-{k+wsstep}' for k in np.arange(4, 9, wsstep)])
        rows = []
        for m in np.arange((0 - degree_length), degree_length + 1, 1):
            row = {self.fieldWindDirFloor: m}
            for k in np.arange(4, 9, wsstep):
                if k in power_dict and power_dict[k]:  # 检查 power_dict[k] 是否存在且不为空
                    row[f'{k}-{k+wsstep}'] = power_dict[k].get(m, np.nan)
                else:
                    row[f'{k}-{k+wsstep}'] = np.nan  # 如果 power_dict[k] 为空，则填充 np.nan
            rows.append(row)
        final_df = pd.concat([final_df, pd.DataFrame(rows)], ignore_index=True)
        return result_df,final_df


    def turbinesAnalysis(self,  outputAnalysisDir, conf: Contract, turbineCodes):
        dictionary = self.processTurbineData(turbineCodes,conf,self.selectColumns())
        dataFrameMerge = self.userDataFrame(dictionary,conf.dataContract.configAnalysis,self)
        turbineInfos = self.common.getTurbineInfos(conf.dataContract.dataFilter.powerFarmID, turbineCodes, self.turbineInfo)
        return self.yawErrorAnalysis(dataFrameMerge,turbineInfos, outputAnalysisDir, conf)

    def yawErrorAnalysis(self, dataFrameMerge: pd.DataFrame,turbineModelInfo: pd.Series, outputAnalysisDir, conf: Contract):
        # 检查所需列是否存在
        required_columns = {Field_ActiverPower, Field_YawError,Field_WindSpeed,Field_PitchAngel1,Field_Cp,Field_TSR}
        if not required_columns.issubset(dataFrameMerge.columns):
            raise ValueError(f"DataFrame缺少必要的列。需要的列有: {required_columns}")
        
        results = []
        result_rows = []
        grouped = dataFrameMerge.groupby(
            [Field_NameOfTurbine, Field_CodeOfTurbine])
 
        for name, group in grouped:
            group = self.filterCommon(group,conf)
            df,final_df = self.calculateYawError(
                group, Field_YawError, Field_ActiverPower,Field_WindSpeed)
            
            df.dropna(inplace=True)
            
            # final_df.dropna(inplace=True)
            """
            自动化选择tsr平稳段的风速段作为风速最后的筛选区间
           
            # 滑动窗口size
            window_size=5
            # 差分阈值
            threshold=2
            diff = group[Field_TSR].diff().abs()
            stable_mask = diff.rolling(window=window_size).mean() < threshold
            # print("diff的索引:", diff.index)
            # print("stable_mask的索引:", stable_mask.index)
            # 找到最长的平稳段
            longest_stable_segment = None
            max_length = 0
            current_segment = None

            for i in range(len(stable_mask)):
                if i in stable_mask.index:
                    if stable_mask[i]:
                        if current_segment is None:
                            current_segment = [i]
                        else:
                            current_segment.append(i)
                    else:
                        if current_segment is not None:
                            if len(current_segment) > max_length:
                                longest_stable_segment = current_segment
                                max_length = len(current_segment)
                            current_segment = None

            if current_segment is not None and len(current_segment) > max_length:
                longest_stable_segment = current_segment

            if longest_stable_segment is not None:
                start_idx = longest_stable_segment[0]
                end_idx = longest_stable_segment[-1]
                # 获取tsr平稳区间对应的风俗区间
                if start_idx is not None and end_idx is not None:
                    # 提取对应的windspeed区间
                    stable_windspeed_segment = group.loc[start_idx:end_idx, Field_WindSpeed]
                    # print(f"平稳段的起始索引: {start_idx}, 结束索引: {end_idx}")
                    # print("对应的windspeed区间:")
                    # print(stable_windspeed_segment)
                    # 计算风速平稳段的最大值和最小值
                    max_windspeed = stable_windspeed_segment.max()
                    min_windspeed = stable_windspeed_segment.min()
                    # print(f"风速平稳段的最大值: {max_windspeed}")
                    # print(f"风速平稳段的最小值: {min_windspeed}")
                else:
                    print("未找到平稳段")
             """      
            # 筛选出 self.fieldWindSpeed 列的数值处于 4.5 到 8 区间内的行
            filtered_df = df[(df[self.fieldWindSpeed] >= 4.5) & (df[self.fieldWindSpeed] <= 8)]
            # 查找 df 中  self.fieldYawError列的平均值，并向下取整作为偏航误差
            max_yaw_error = filtered_df[self.fieldYawError].mean()
            max_yaw_error_floor = np.floor(max_yaw_error)

            # 存储结果
            results.append({
                Field_NameOfTurbine: name[0],
                Field_YawError: max_yaw_error_floor
            })

            #对 final_df 中的数据进行可视化
            fig = go.Figure()
            for col in final_df.columns[1:]:  # 跳过第一列 self.fieldWindDirFloor
                fig.add_trace(go.Scatter(
                    x=final_df[self.fieldWindDirFloor],
                    y=final_df[col],
                    mode='lines+markers',
                    name=col
                ))
            
            fig.update_layout(
                title=f'偏航误差分析 {name[0]} ',
                xaxis_title='偏航角度',
                yaxis_title='平均功率',
                legend_title='风速区间',
                showlegend=True
            )
            # 确保从 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": "偏航误差分析",
                    "engineCode": engineTypeCode,    
                    "engineTypeName": engineTypeName, 
                    "xaixs": "偏航角度",
                    "yaixs": "有功功率(kw)",
                    "data": [
                        {
                            "engineName": name[0],
                            "engineCode": name[1],
                            "title": f'静态偏航误差分析 {name[0]}',
                            "xData": final_df[self.fieldWindDirFloor].tolist(),
                            "yData": final_df[col].tolist(),
                            "legend": col,
                            "type": "lines+markers",
                            
                        }
                        for col in final_df.columns[1:]  # 跳过第一列 self.fieldWindDirFloor
                    ]
                    }
            
            # Save to file
            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)
            # 将JSON对象保存到文件
            output_json_path = os.path.join(outputAnalysisDir, f"{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: 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
            
            
            # })
            # # 初始化一个空的DataFrame，指定列名
            # columns = [Field_NameOfTurbine, Field_YawError]
            # dataFrameResult = pd.DataFrame(results, columns=columns)

            # filePathOfYawErrorResult = os.path.join(
            #     outputAnalysisDir, f"{name[0]}yaw_error_result{CSVSuffix}")
            # df.to_csv(filePathOfYawErrorResult, index=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: filePathOfYawErrorResult,
            #             Field_Return_IsSaveDatabase: True
            # })

        # 初始化一个空的DataFrame，指定列名
        columns = [Field_NameOfTurbine, Field_YawError]
        dataFrameResult = pd.DataFrame(results, columns=columns)

        filePathOfYawError = os.path.join(
            outputAnalysisDir, f"yaw_error_result{CSVSuffix}")
        dataFrameResult.to_csv(filePathOfYawError, index=False)

        result_rows.append({
            Field_Return_TypeAnalyst: self.typeAnalyst(),
            Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
            Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
            Field_CodeOfTurbine: "total",
            Field_Return_FilePath: filePathOfYawError,
                    Field_Return_IsSaveDatabase: True
        })

        result_df = pd.DataFrame(result_rows)

        return result_df