WTOAAM/dataAnalysisService/service/serviceOfDataAnalysis.py

import sys
import pandas as pd
from logging import Logger
from utils.jsonUtil import JsonUtil
from utils.minioUtil.threadSafeMinioClient import ThreadSafeMinioClient
from utils.rdbmsUtil.databaseUtil import DatabaseUtil
from algorithmContract.confBusiness import *
from algorithmContract.contract import Contract,LoadAnalysisInput, Analysis
from algorithmContract.const import *
from behavior.baseAnalyst import BaseAnalyst
from behavior.dalAnalyst import DALAnalyst
from pathlib import Path

# 获取项目根目录（WTOAAM文件夹）
BASE_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)  # 向上四级到项目根目录

if BASE_DIR not in sys.path:
    sys.path.append(BASE_DIR)
from dataAnalysisBusiness.algorithm.dataProcessor import DataProcessor
from common.commonBusiness import CommonBusiness
from common.appConfig import AppConfig
from sqlalchemy.orm import Session
from sqlalchemy.sql import text

def buildDynamicInstance(modulePath, className, conf: Contract,logger:Logger,dbUtil:dict[str,DatabaseUtil],minioUtil:ThreadSafeMinioClient,
                         powerFarmInfo:pd.DataFrame,turbineInfo:pd.DataFrame,
                         turbineModelInfo:pd.DataFrame,dataTransfer:pd.DataFrame,
                         weatherStationInfo:pd.DataFrame,dataFrameContractOfTurbine:pd.DataFrame,
                         ):
    # 动态导入模块
    module = importlib.import_module(modulePath)
    # 获取类
    cls = getattr(module, className)
    # 创建类实例
    instance:BaseAnalyst = cls(logger,dbUtil,minioUtil,conf,powerFarmInfo,turbineInfo,turbineModelInfo,dataTransfer,weatherStationInfo,dataFrameContractOfTurbine)

    instance.powerFarmInfo=powerFarmInfo
    instance.turbineInfo=turbineInfo
    instance.turbineModelInfo=turbineModelInfo
    # instance.dataTransfer=dataTransfer
    instance.weatherStationInfo=weatherStationInfo
    instance.dataFrameContractOfTurbine=dataFrameContractOfTurbine
    
    return instance


def dynamic_instance_and_call(module_path, class_name, method_name, *args, **kwargs):
    # 动态导入模块
    module = importlib.import_module(module_path)
    # 获取类
    cls = getattr(module, class_name)
    # 创建类实例
    instance = cls()
    # 获取方法
    method = getattr(instance, method_name)
    # 调用方法
    method(*args, **kwargs)


class ServiceOfDataAnalysis:
    def __init__(self, logger:Logger,dbUtil:dict,minioUtil:ThreadSafeMinioClient,conf:Contract):
        self.logger = logger        
        self.dbUtil=dbUtil
        self.minioUtil=minioUtil
        self.conf=conf

        self.common = CommonBusiness()

        dal = DALAnalyst(logger, dbUtil)
        # 加载所有新能源场站信息
        self.powerFarmInfo:pd.DataFrame = dal.loadPowerFarmInfos(
            conf.dataContract.dataFilter.powerFarmID)
        
        if self.powerFarmInfo.empty:
            raise CustomError(104)

        # 加载所有风电机组信息
        self.turbineInfo:pd.DataFrame= dal.loadTurbineInfos(
            conf.dataContract.dataFilter.powerFarmID)
        
        if self.turbineInfo.empty:
            raise CustomError(105)

        # 加载数据转换信息
        # self.dataTransfer:pd.DataFrame=dal.loadDataTransfer(conf.dataContract.dataFilter.powerFarmID,conf.dataContract.dataFilter.dataBatchNum)

        # if self.dataTransfer.empty:
        #     raise CustomError(106)
        self.dataTransfer:pd.DataFrame=None

        # 加载机型信息
        self.turbineModelInfo:pd.DataFrame= dal.loadTurbineModelInfos(
            self.turbineInfo[Field_MillTypeCode].unique().tolist())
        
        if self.turbineModelInfo.empty:
            raise CustomError(107)
        
        # 加载所有新能源场站，及所属风电机组机型的合同功率曲线
        self.dataFrameContractOfTurbine:pd.DataFrame = dal.processContractData(
            self.common, conf.dataContract.dataFilter.powerFarmID, self.powerFarmInfo[Field_AirDensity].iloc[0], self.turbineModelInfo)
        
        if self.dataFrameContractOfTurbine.empty:
            raise CustomError(108)

        # 加载所有测风塔信息
        self.weatherStationInfo:pd.DataFrame = dal.loadWeatherStationInfos(
            conf.dataContract.dataFilter.powerFarmID)

    def loadJson(self, filePath):
        # 打开并读取JSON文件
        with open(filePath, 'r') as f:
            data = JsonUtil.read_json(filePath)

        return data

    def executeAnalysis(self):
        analysts = []

        for dynamicAnalyst in self.conf.dataContract.configAnalysis:
            package = dynamicAnalyst.package
            className = dynamicAnalyst.className
            methodName = dynamicAnalyst.methodName
            analyst = buildDynamicInstance(package, className, self.conf,self.logger,self.dbUtil,self.minioUtil,
                                           self.powerFarmInfo,self.turbineInfo,
                                           self.turbineModelInfo,self.dataTransfer,
                                           self.weatherStationInfo,self.dataFrameContractOfTurbine)

            analysts.append(analyst)

        process = DataProcessor(logger=self.logger,dbUtil=self.dbUtil,minioUtil=self.minioUtil,conf=self.conf)
        # 加载所有新能源场站信息
        process.powerFarmInfo = self.powerFarmInfo
        # 加载所有风电机组信息
        process.turbineInfo= self.turbineInfo
        # # 加载机型信息
        process.turbineModelInfo= self.turbineModelInfo
        # 加载所有测风塔信息
        process.weatherStationInfo = self.weatherStationInfo
        # 加载所有新能源场站，及所属风电机组机型的合同功率曲线
        process.dataFrameContractOfTurbine = self.dataFrameContractOfTurbine
        # process.dataTransfer=self.dataTransfer

        for analyst in analysts:
            process.attach(analyst)
        process.executeAnalysis(self.conf)

        for analyst in analysts:
            process.detach(analyst)

    def analysis(self):
        self.logger.info("analysis executing....")
        self.executeAnalysis()

        # 使用多线程时，matplotlib绘图报错
        # 在使用Python语言的matplotlib包时，如果尝试在多线程环境中创建图形界面，你可能会遇到“QWidget-: Must construct a QApplication before a QWidget”的错误。
        # 这个错误通常是因为matplotlib的图形后端（backend）依赖于Qt框架，而Qt框架要求在一个QApplication实例被创建之后再创建任何QWidget对象。
        # 在多线程环境中，每个线程都应该有一个它自己的事件循环。然而，QApplication实例通常是全局的，并且应该只在主线程中创建一次。
        # 如果你尝试在一个非主线程中创建QApplication或者QWidget，就会遇到这个问题。

        # with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        #     futures=[executor.submit(executeAnalysis, config) for config in configs]

        #     for future in concurrent.futures.as_completed(futures):
        #         try:
        #             result = future.result()
        #         except Exception as exc:
        #             print(f'生成异常: {exc}')
        #         else:
        #             print(f'任务返回结果: {result}')