WTOAAM/dataAnalysisBusiness/algorithm/dataProcessor.py

import concurrent.futures
import os
import traceback
from logging import Logger

import pandas as pd
from algorithmContract.confBusiness import *
from algorithmContract.configAnalysis import ConfigAnalysis
from algorithmContract.const import *
from algorithmContract.contract import Contract
from behavior.analyst import Analyst
from behavior.dalAnalyst import DALAnalyst
from behavior.outputProcessor import OutputProcessor
from common.appConfig import GetBusinessFoundationDbUtil
from common.commonBusiness import CommonBusiness
from utils.minioUtil.threadSafeMinioClient import ThreadSafeMinioClient
from utils.rdbmsUtil.databaseUtil import DatabaseUtil


class DataProcessor:
    def __init__(self, logger: Logger, dbUtil: dict[str, DatabaseUtil], minioUtil: ThreadSafeMinioClient, conf: Contract):
        self.logger = logger
        self.dbUtil = dbUtil
        self.minioUtil = minioUtil
        self.conf = conf
        self.common = CommonBusiness()
        self._baseAnalysts = []
        self._analysts = []
        dal = DALAnalyst(logger, dbUtil)

        # 加载所有新能源场站信息
        self.powerFarmInfo: pd.DataFrame

        # 加载所有风电机组信息
        self.turbineInfo: pd.DataFrame

        # 加载数据转换信息
        self.dataTransfer: pd.DataFrame

        # 加载机型信息
        self.turbineModelInfo: pd.DataFrame

        # 加载所有测风塔信息
        self.weatherStationInfo: pd.DataFrame

        # 加载所有新能源场站，及所属风电机组机型的合同功率曲线
        self.dataFrameContractOfTurbine: pd.DataFrame

    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 userDataFrame(self, dictionary: dict, configs: list[ConfigAnalysis], analyst: Analyst) -> pd.DataFrame:
        timeGranularity = next(
            (config.scada for config in configs if config.className.lower() == type(analyst).__name__.lower()), None)
        return pd.DataFrame() if dictionary[timeGranularity].empty else dictionary[timeGranularity]

    def singleTurbineDataNotify(self,analyst,conf:Contract,turbineCode:str):
            outputAnalysisDir = analyst.getOutputAnalysisDir()
            outputFilePath = r"{}/{}{}".format(outputAnalysisDir,
                                               turbineCode, CSVSuffix)
            return analyst.analysisOfTurbine(
                 outputAnalysisDir, outputFilePath, conf, turbineCode)

    def notifyOfTurbine(self,  conf: Contract, turbineCode: str) -> pd.DataFrame:
        results =[]
        maxWorkers = 4
        with concurrent.futures.ThreadPoolExecutor(max_workers=maxWorkers) as executor:
            futures = [
                        executor.submit(self.singleTurbineDataNotify,analyst, conf,turbineCode)
                        for analyst in self._analysts
                    ]

            for future in concurrent.futures.as_completed(futures):
                try:
                    result = future.result()
                    results.append(result)
                except Exception as exc:
                    raise exc

        returnResult = pd.concat(results, ignore_index=True)
        return returnResult

    def multiTubineDataNotify(self,analyst,conf:Contract,turbineCodes):
        try:
            outputAnalysisDir = analyst.getOutputAnalysisDir()
            dataFram = analyst.analysisOfTurbines(outputAnalysisDir, conf,turbineCodes)
            self.logger.info(f"analysis type execute result : dataFrame = {dataFram}")
            return dataFram
        except Exception as exception:
            self.logger.info(f"analysis type execute error : {traceback.format_exc()}")
            raise exception

    def notifyOfTurbines(self, conf: Contract,turbineCodes) -> pd.DataFrame:
        results =[]
        maxWorkers = 4
        with concurrent.futures.ThreadPoolExecutor(max_workers=maxWorkers) as executor:
            futures = [
                        executor.submit(self.multiTubineDataNotify,analyst, conf,turbineCodes)
                        for analyst in self._analysts
                    ]
            self.logger.info("Waiting for all futures to complete...")
            for future in concurrent.futures.as_completed(futures):
                self.logger.info(f"future is : {future}")
                try:
                    result = future.result()
                    results.append(result)
                except Exception as exc:
                    self.logger.info(f"future deal error. future : {future} , error is {traceback.format_exc()}")
                    raise exc

        returnResult = pd.concat(results, ignore_index=True)
        return returnResult

    def loadDataCSV(self, dataBatchNum: str, timeGranularity: str, turbineCode: str, condition: str, csvFileDir: str = f"data/mima_dt/second") -> pd.DataFrame:
        csvFilePath = os.path.join(csvFileDir, f"{turbineCode}.csv")
        print("current csv file path:", csvFilePath)

        dataFrame = pd.read_csv(csvFilePath, header=0)

        dataFrame = dataFrame.astype({
            'wind_turbine_number': 'string',
            'wind_turbine_name': 'string',
            'time_stamp': 'datetime64[ns]',
            'active_power': 'float32',
            'rotor_speed': 'float32',
            'generator_speed': 'float32',
            'wind_velocity': 'float32',
            'pitch_angle_blade_1': 'float32',
            'pitch_angle_blade_2': 'float32',
            'pitch_angle_blade_3': 'float32',
            'cabin_position': 'float32',
            'true_wind_direction': 'float32',
            'yaw_error1': 'float32',
            'set_value_of_active_power': 'float32',
            'gearbox_oil_temperature': 'float32',
            'generatordrive_end_bearing_temperature': 'float32',
            'generatornon_drive_end_bearing_temperature': 'float32',
            'cabin_temperature': 'float32',
            'twisted_cable_angle': 'float32',
            'front_back_vibration_of_the_cabin': 'float32',
            'side_to_side_vibration_of_the_cabin': 'float32',
            'actual_torque': 'float32',
            'given_torque': 'float32',
            'clockwise_yaw_count': 'float32',
            'counterclockwise_yaw_count': 'float32',
            'unusable': 'float32',
            'power_curve_available': 'float32',
            'required_gearbox_speed': 'float32',
            'inverter_speed_master_control': 'float32',
            'outside_cabin_temperature': 'float32',
            'main_bearing_temperature': 'float32',
            'gearbox_high_speed_shaft_bearing_temperature': 'float32',
            'gearboxmedium_speed_shaftbearing_temperature': 'float32',
            'gearbox_low_speed_shaft_bearing_temperature': 'float32',
            'generator_winding1_temperature': 'float32',
            'generator_winding2_temperature': 'float32',
            'generator_winding3_temperature': 'float32',
            'wind_turbine_status': 'float32',
            'wind_turbine_status2': 'float32',
            'turbulence_intensity': 'float32',
            'year': 'int32',
            'month': 'int32',
            'day': 'int32'
        })

        return dataFrame

    def getTurbines(self,conf:Contract,turbineInfo:pd.DataFrame):
        return conf.dataContract.dataFilter.turbines if not self.common.isNone(
                conf.dataContract.dataFilter.turbines) and len(conf.dataContract.dataFilter.turbines) > 0 else turbineInfo[Field_CodeOfTurbine].unique()

    def executeAnalysis(self, conf: Contract):
        # 根据输入参数 conf ，加载台账（电场、机型、机组、测风塔信息）、发电机组scada的分钟级与秒级、测风塔运行、事件等数据
        # 若同一电场，具有多种机型的机组，注意按机型分组分析
        outputProcessor = OutputProcessor(
            conf, self.logger, self.dbUtil, self.minioUtil)

        # Initialize an empty DataFrame to store merged results
        DataFrameOutput = pd.DataFrame()

        try:
            foundationDB: DatabaseUtil = self.dbUtil[DATABASE_BusinessFoundationDb]
            with foundationDB.session_scope() as foundationSession:
                outputProcessor.analysisState(
                    foundationSession, conf.dataContract.dataFilter.dataBatchNum, AnalysisState.Analyzing.value, analysisProgress=10)

            # 机组筛选
            turbines = self.getTurbines(conf,self.turbineInfo)
            # turbineSqlInStr = ", ".join(f"'{turbine}'" for turbine in turbines)  # 使用%s作为占位符，稍后可以替换为实际值
            # dictionary= self.processTurbineData(turbines,conf)

            foundationDB: DatabaseUtil = self.dbUtil[DATABASE_BusinessFoundationDb]
            with foundationDB.session_scope() as foundationSession:
                outputProcessor.analysisState(
                    foundationSession, conf.dataContract.dataFilter.dataBatchNum, AnalysisState.Analyzing.value, analysisProgress=25)
            self.logger.info(
                f"批次：{conf.dataContract.dataFilter.dataBatchNum} 完成机组运行数据加载及再处理，执行请求发电性能分析...")

            for turbine in turbines:
                self.logger.info(
                    f"Power Farm: {conf.dataContract.dataFilter.powerFarmID} Batch: {conf.dataContract.dataFilter.dataBatchNum}  Turbine: {turbine} .")
                dataFrameOfReturn = self.notifyOfTurbine(
                    conf, turbine)
                DataFrameOutput = pd.concat(
                    [DataFrameOutput, dataFrameOfReturn], ignore_index=True)

            # Notify processing for all turbines
            Outputs = self.notifyOfTurbines( conf,turbines)
            DataFrameOutput = pd.concat(
                [Outputs, DataFrameOutput], ignore_index=True)

            self.logger.info(
                f"批次：{conf.dataContract.dataFilter.dataBatchNum} 执行完成请求发电性能分析，准备输出及更新状态")
            if DataFrameOutput.empty:
                raise CustomError(103)
            else:
                outputProcessor.process(conf.dataContract.dataFilter.powerFarmID,
                                        conf.dataContract.dataFilter.dataBatchNum, DataFrameOutput)
            self.logger.info(
                f"批次：{conf.dataContract.dataFilter.dataBatchNum} 完成请求发电性能分析、输出及更新状态")
        except (CustomError, Exception) as e:
            self.logger.info(f" executeAnalysis error: {traceback.format_exc()}")
            try:
                dbUtil = GetBusinessFoundationDbUtil()
                ex = e if isinstance(e, CustomError) else CustomError(-1)
                code = ex.code
                message = ex.message
                with dbUtil.session_scope() as session:
                    outputProcessor.analysisState(session, conf.dataContract.dataFilter.dataBatchNum,
                                                  AnalysisState.Analyzed.value, ErrorState.Err.value, code, message)
            except Exception as e1:
                # 使用 traceback 模块获取完整的异常信息
                error_message = ''.join(traceback.format_exception(e1))
                # 记录异常信息
                self.logger.error(f"捕获到异常: {traceback.format_exc()}")

        return DataFrameOutput