WTOAAM/dataAnalysisBusiness/algorithm/dataProcessor.py

import os
from datetime import datetime
import concurrent.futures
import numpy as np
import pandas as pd
from utils.directoryUtil import DirectoryUtil as dir
from algorithmContract.confBusiness import *
from behavior.baseAnalyst import BaseAnalyst
from behavior.analyst import Analyst
from common.commonBusiness import CommonBusiness
from algorithm.dataMarker import DataMarker


class DataProcessor:
    def __init__(self):
        self.common = CommonBusiness()
        self._baseAnalysts = []
        self._analysts = []

    def attachBaseAnalyst(self, analyst: BaseAnalyst):
        if analyst not in self._analysts:
            self._baseAnalysts.append(analyst)

    def detachBaseAnalyst(self, analyst: BaseAnalyst):
        try:
            self._baseAnalysts.remove(analyst)
        except ValueError:
            pass

    def attach(self, analyst: Analyst):
        if analyst not in self._analysts:
            self._analysts.append(analyst)

    def detach(self, analyst: Analyst):
        try:
            self._analysts.remove(analyst)
        except ValueError:
            pass

    def turbineNotify(self,
                      dataFrameOfTurbine: pd.DataFrame,
                      confData: ConfBusiness,
                      turbineName):
        for analyst in self._analysts:
            outputAnalysisDir = analyst.getOutputAnalysisDir()

            outputFilePath = r"{}/{}{}".format(
                outputAnalysisDir, turbineName, CSVSuffix)

            analyst.analysisOfTurbine(
                dataFrameOfTurbine, outputAnalysisDir, outputFilePath, confData, turbineName)

    def turbinesNotify(self, dataFrameOfTurbines: pd.DataFrame,  confData: ConfBusiness):
        for analyst in self._analysts:
            outputAnalysisDir = analyst.getOutputAnalysisDir()
            analyst.analysisOfTurbines(
                dataFrameOfTurbines, outputAnalysisDir, confData)

    def baseAnalystTurbineNotify(self,
                                 dataFrameOfTurbine: pd.DataFrame,
                                 confData: ConfBusiness,
                                 turbineName):
        for analyst in self._baseAnalysts:
            outputAnalysisDir = analyst.getOutputAnalysisDir()

            outputFilePath = r"{}/{}{}".format(
                outputAnalysisDir, turbineName, CSVSuffix)

            analyst.analysisOfTurbine(
                dataFrameOfTurbine, outputAnalysisDir, outputFilePath, confData, turbineName)

    def baseAnalystNotify(self, dataFrameOfTurbines: pd.DataFrame,  confData: ConfBusiness):
        for analyst in self._baseAnalysts:
            outputAnalysisDir = analyst.getOutputAnalysisDir()
            analyst.analysisOfTurbines(
                dataFrameOfTurbines, outputAnalysisDir, confData)

    def calculateAngleIncluded(self, array1, array2):
        """
        计算两个相同长度角度数组中两两对应角度值的偏差。
        结果限制在-90°到+90°之间，并保留两位小数。

        参数:
        array1 (list): 第一个角度数组
        array2 (list): 第二个角度数组

        返回:
        list: 两两对应角度的偏差列表
        """
        deviations = []
        for angle1, angle2 in zip(array1, array2):
            # 计算原始偏差
            deviation = angle1 - angle2

            # 调整偏差，使其位于-180°到+180°范围内
            if deviation == 0.0:
                deviation = 0.0
            else:
                deviation = (deviation + 180) % 360 - 180

            # 将偏差限制在-90°到+90°范围内
            if deviation > 90:
                deviation -= 180
            elif deviation < -90:
                deviation += 180

            # 保留两位小数
            deviations.append(round(deviation, 2))

        return deviations

    def recalculationOfIncludedAngle(self, dataFrame: pd.DataFrame, fieldAngleIncluded, fieldWindDirect, fieldNacellePos):
        """
        依据机舱位置（角度）、风向计算两者夹角
        """
        if not self.common.isNone(fieldAngleIncluded) and fieldAngleIncluded in dataFrame.columns:
            dataFrame[Field_AngleIncluded] = dataFrame[fieldAngleIncluded]

        if self.common.isNone(fieldAngleIncluded) and fieldAngleIncluded not in dataFrame.columns and fieldWindDirect in dataFrame.columns and fieldNacellePos in dataFrame.columns:
            dataFrame[Field_AngleIncluded] = self.calculateAngleIncluded(
                dataFrame[fieldNacellePos], dataFrame[fieldWindDirect])

    def recalculationOfGeneratorSpeed(self, dataFrame: pd.DataFrame, fieldRotorSpeed, fieldGeneratorSpeed, rotationalSpeedRatio):
        """
        风电机组发电机转速再计算，公式：转速比=发电机转速/叶轮或主轴转速
        """
        if fieldGeneratorSpeed in dataFrame.columns:
            dataFrame[Field_GeneratorSpeed] = dataFrame[fieldGeneratorSpeed]

        if fieldGeneratorSpeed not in dataFrame.columns and fieldRotorSpeed in dataFrame.columns:
            dataFrame[fieldGeneratorSpeed] = rotationalSpeedRatio * \
                dataFrame[fieldRotorSpeed]

    def recalculationOfRotorSpeed(self, dataFrame: pd.DataFrame, fieldRotorSpeed, fieldGeneratorSpeed, rotationalSpeedRatio):
        """
        风电机组发电机转速再计算，公式：转速比=发电机转速/叶轮或主轴转速
        """
        if fieldRotorSpeed not in dataFrame.columns and fieldGeneratorSpeed in dataFrame.columns:
            dataFrame[fieldRotorSpeed] = dataFrame[fieldGeneratorSpeed] / \
                rotationalSpeedRatio

        if not self.common.isNone(fieldRotorSpeed) and fieldRotorSpeed in dataFrame.columns:
            dataFrame[Field_RotorSpeed] = dataFrame[fieldRotorSpeed]

    def recalculationOfRotorTorque(self, dataFrame: pd.DataFrame, fieldGeneratorTorque, fieldActivePower, fieldGeneratorSpeed):
        """
        风电机组发电机转矩计算，P的单位换成KW转矩计算公式：
        P*1000= pi/30*T*n  
        30000/pi*P=T*n
        30000/3.1415926*P=T*n
        9549.297*p=T*n  
        其中：n为发电机转速，p为有功功率，T为转矩
        """
        if self.common.isNone(fieldGeneratorTorque) and fieldActivePower in dataFrame.columns and fieldGeneratorSpeed in dataFrame.columns:
            dataFrame[Field_GeneratorTorque] = 9549.297 * \
                dataFrame[fieldActivePower]/dataFrame[fieldGeneratorSpeed]

        if fieldGeneratorTorque in dataFrame.columns:
            dataFrame[Field_GeneratorTorque] = dataFrame[fieldGeneratorTorque]

    def recalculation(self, dataFrame: pd.DataFrame, confData: ConfBusiness):
        """
        再计算数据测点
        参数：
        dataFrame 原始数据
        confData  配置数据
        """
        self.recalculationOfGeneratorSpeed(
            dataFrame, confData.field_rotor_speed, confData.field_gen_speed, confData.rotational_Speed_Ratio)
        self.recalculationOfRotorSpeed(
            dataFrame, confData.field_rotor_speed, confData.field_gen_speed, confData.rotational_Speed_Ratio)
        self.recalculationOfRotorTorque(
            dataFrame, confData.field_torque, confData.field_power, confData.field_gen_speed)
        self.recalculationOfIncludedAngle(
            dataFrame, confData.field_angle_included, confData.field_wind_dir, confData.field_nacelle_pos)

        self.common.calculateTSR(dataFrame, confData)
        self.common.calculateCp(dataFrame, confData)

    def filterWithDateTime(self, dataFrame: pd.DataFrame, confData: ConfBusiness):
        dataFrame = dataFrame[(dataFrame[confData.field_turbine_time] >= confData.start_time) & (
            dataFrame[confData.field_turbine_time] < confData.end_time)]

        return dataFrame

    def processDateTime(self, dataFrame: pd.DataFrame, confData: ConfBusiness):
        dataFrame["年月"] = pd.to_datetime(
            dataFrame[confData.field_turbine_time], format="%Y-%m")
        dataFrame['日期'] = pd.to_datetime(
            dataFrame[confData.field_turbine_time], format="%Y-%m")
        dataFrame['monthIntTime'] = dataFrame['日期'].apply(
            lambda x: x.timestamp())

        dataFrame[Field_YearMonth] = dataFrame[confData.field_turbine_time].dt.strftime(
            '%Y-%m')

        dataFrame[Field_YearMonthDay] = dataFrame[confData.field_turbine_time].dt.strftime(
            '%Y-%m-%d')

        if not self.common.isNone(confData.excludingMonths) and len(confData.excludingMonths) > 0:
            # 给定的日期列表
            date_strings = []
            for month in confData.excludingMonths:
                if not self.common.isNone(month):
                    date_strings.append(month)

            if len(date_strings) > 0:
                mask = ~dataFrame[Field_YearMonth].isin(date_strings)

                # 使用掩码过滤DataFrame，删除指定日期的行
                dataFrame = dataFrame[mask]

        return dataFrame

    def setColumnDataType(self, dataFrame: pd.DataFrame,  confData: ConfBusiness):
        # 选择所有的数值型列
        dataFrame = dataFrame.convert_dtypes()
        numeric_cols = dataFrame.select_dtypes(
            include=['float64', 'float16']).columns
        
        # 将这些列转换为float32
        dataFrame[Field_NameOfTurbine] = dataFrame[Field_NameOfTurbine].astype(str)

        # 将这些列转换为float32
        dataFrame[numeric_cols] = dataFrame[numeric_cols].astype(
            self.common.getFloat32())

        if not self.common.isNone(confData.field_turbine_time) and confData.field_turbine_time in dataFrame.columns:
            # 首先尝试去除字符串前后的空白
            dataFrame[confData.field_turbine_time] = dataFrame[confData.field_turbine_time].str.strip()
            dataFrame[confData.field_turbine_time] = pd.to_datetime(
                dataFrame[confData.field_turbine_time], format='%Y-%m-%d %H:%M:%S', errors="coerce")
            dataFrame[confData.field_turbine_time] = dataFrame[confData.field_turbine_time].dt.strftime(
                '%Y-%m-%d %H:%M:%S')
            dataFrame[confData.field_turbine_time] = pd.to_datetime(
                dataFrame[confData.field_turbine_time])

            # 删除时间字段为空的行记录
            dataFrame.dropna(
                axis=0, subset=[confData.field_turbine_time], inplace=True)

        if confData.field_wind_speed in dataFrame.columns:
            dataFrame[confData.field_wind_speed] = dataFrame[confData.field_wind_speed].astype(
                self.common.getFloat32())

        if confData.field_wind_dir in dataFrame.columns:
            dataFrame[confData.field_wind_dir] = dataFrame[confData.field_wind_dir].astype(
                self.common.getFloat32())

        if confData.field_angle_included in dataFrame.columns:
            dataFrame[confData.field_angle_included] = dataFrame[confData.field_angle_included].astype(
                self.common.getFloat32())

        if confData.field_power in dataFrame.columns:
            dataFrame[confData.field_power] = dataFrame[confData.field_power].astype(
                self.common.getFloat32())

        if confData.field_wind_dir in dataFrame.columns:
            dataFrame[confData.field_wind_dir] = dataFrame[confData.field_wind_dir].astype(
                self.common.getFloat32())

        if confData.field_pitch_angle1 in dataFrame.columns:
            dataFrame[confData.field_pitch_angle1] = dataFrame[confData.field_pitch_angle1].astype(
                self.common.getFloat32())

        if confData.field_pitch_angle2 in dataFrame.columns:
            dataFrame[confData.field_pitch_angle2] = dataFrame[confData.field_pitch_angle2].astype(
                self.common.getFloat32())

        if confData.field_pitch_angle3 in dataFrame.columns:
            dataFrame[confData.field_pitch_angle3] = dataFrame[confData.field_pitch_angle3].astype(
                self.common.getFloat32())

        if confData.field_gen_speed in dataFrame.columns:
            dataFrame[confData.field_gen_speed] = dataFrame[confData.field_gen_speed].astype(
                self.common.getFloat32())

        if confData.field_rotor_speed in dataFrame.columns:
            dataFrame[confData.field_rotor_speed] = dataFrame[confData.field_rotor_speed].astype(
                self.common.getFloat32())
            
        if confData.field_Cabin_Vibrate_X in dataFrame.columns:
            dataFrame[confData.field_Cabin_Vibrate_X] = dataFrame[confData.field_Cabin_Vibrate_X].astype(
                self.common.getFloat32())
        
        if confData.field_Cabin_Vibrate_Y in dataFrame.columns:
            dataFrame[confData.field_Cabin_Vibrate_Y] = dataFrame[confData.field_Cabin_Vibrate_Y].astype(
                self.common.getFloat32())
            
        if confData.field_activePowerSet in dataFrame.columns:
            dataFrame[confData.field_activePowerSet] = dataFrame[confData.field_activePowerSet].astype(
                self.common.getFloat32())
            
        if confData.field_activePowerAvailable in dataFrame.columns:
            dataFrame[confData.field_activePowerAvailable] = dataFrame[confData.field_activePowerAvailable].astype(
                self.common.getFloat32())

        return dataFrame

    def loadData(self, csvFilePath, confData: ConfBusiness, turbineName):
        useColumns = self.common.getUseColumns(confData)
        # Load the CSV, skipping the specified initial rows
        dataFrame = pd.read_csv(csvFilePath, header=0, usecols=useColumns, skiprows=range(
            1, confData.skip_row_number+1))

        if not self.common.isNone(confData.field_turbine_name) and confData.field_turbine_name in dataFrame.columns:
            dataFrame[Field_NameOfTurbine] = dataFrame[confData.field_turbine_name]
        else:
            dataFrame[Field_NameOfTurbine] = turbineName
        # 对除了“时间”列之外的所有列进行自下而上的填充（先反转后填充）
        # 注意：补植须要考虑业务合理性
        # dataFrame = dataFrame.fillna(method='ffill')
        # dataFrame = dataFrame.fillna(method='bfill')

        return dataFrame

    def execute(self, confData: ConfBusiness):
        outputDataAfterFilteringDir = r"{}/{}".format(
            confData.output_path,  "DataAfterFiltering")
        dir.create_directory(outputDataAfterFilteringDir)

        labler = DataMarker()  #类的实例化
        
        dataFrameMerge = pd.DataFrame()
        for rootDir, subDirs, files in dir.list_directory(confData.input_path):
            files = sorted(files)
            for file in files:
                if not file.endswith(CSVSuffix):
                    continue

                csvFilePath = os.path.join(rootDir, file)
                print(f"current csv file path: {csvFilePath}")
                turbineName = confData.add_W_if_starts_with_digit(file.split(confData.csvFileNameSplitStringForTurbine)[
                    confData.index_turbine])

                dataFrame = self.loadData(
                    csvFilePath, confData, turbineName)
                turbineName = dataFrame[Field_NameOfTurbine].loc[0]

                if len(dataFrame) <= 0:
                    print("dataFrameFilter not data.")
                    continue

                dataFrame = self.setColumnDataType(
                    dataFrame, confData)

                dataFrame = self.processDateTime(dataFrame, confData)

                dataFrame = self.filterWithDateTime(dataFrame, confData)

                # self.baseAnalystTurbineNotify(dataFrame,
                #                               confData,
                #                               turbineName)

                self.recalculation(dataFrame, confData)

                # dataFrame = labler.main(confData,dataFrame)

                dataFrameMerge = pd.concat(
                    [dataFrameMerge, dataFrame], axis=0, sort=False)

                # dataFrame.to_csv(os.path.join(
                #     outputDataAfterFilteringDir, "{}{}".format(turbineName,CSVSuffix)), index=False)
                dataFrame = self.turbineNotify(dataFrame,
                                               confData,
                                               turbineName)

        # self.baseAnalystNotify(dataFrameMergeFilter,  confData)
        self.turbinesNotify(dataFrameMerge,  confData)