WTOAAM/dataAnalysisBusiness/algorithm/ratedWindSpeedAnalyst.py

import os

import pandas as pd
import plotly.graph_objects as go
from algorithmContract.confBusiness import *
from algorithmContract.contract import Contract
from behavior.analystWithGoodBadPoint import AnalystWithGoodBadPoint


class RatedWindSpeedAnalyst(AnalystWithGoodBadPoint):
    """
    风电机组额定风速分析。
    秒级scada数据运算太慢，建议使用分钟级scada数据
    """

    def typeAnalyst(self):
        return "rated_windspeed"

    def turbinesAnalysis(self, outputAnalysisDir, conf: Contract, turbineCodes):
        # dictionary=self.processTurbineData(turbineCodes,conf,[Field_DeviceCode,Field_Time,Field_WindSpeed,Field_ActiverPower])
        # dataFrameMerge=self.userDataFrame(dictionary,conf.dataContract.configAnalysis,self)
        dictionary = self.processTurbineData(turbineCodes, conf,
                                             [Field_DeviceCode, Field_Time, Field_WindSpeed, Field_ActiverPower])
        turbineInfos = self.common.getTurbineInfos(conf.dataContract.dataFilter.powerFarmID, turbineCodes,
                                                   self.turbineInfo)
        dataFrameMerge = self.userDataFrame(dictionary, conf.dataContract.configAnalysis, self)
        return self.draw(dataFrameMerge, turbineInfos, outputAnalysisDir, conf)


    def draw(self, dataFrameMerge: pd.DataFrame, turbineModelInfo: pd.Series, outputAnalysisDir, conf: Contract):
        """
        绘制并保存满发风速区间数据计数图。

        参数:
        dataFrameMerge (pd.DataFrame): 包含数据的DataFrame，需要包含设备名、风速和功率列。
        outputAnalysisDir (str): 分析输出目录。
        confData (ConfBusiness): 配置
        """

        # 初始化结果列表
        res = []

        # 按设备名分组并计算统计数据
        grouped = dataFrameMerge.groupby(Field_NameOfTurbine)
        for name, group in grouped:
            group = group[group[Field_WindSpeed] >= 11]
            res.append([name, group[Field_ActiverPower].min(), group[Field_ActiverPower].max(
            ), group[Field_ActiverPower].median(), group.shape[0]])

        # 创建结果DataFrame
        data = pd.DataFrame(res, columns=[
                            Field_NameOfTurbine, 'power-min', 'power-max', 'power-median', 'count'])

        fig = go.Figure(data=[go.Bar(
                        x=data[Field_NameOfTurbine],
                        y=data['count'],
                        marker_color='dodgerblue'
                        )
        ]
        )

        fig.update_layout(
            title={
                "text": f'额定风速间隔数据计数',
                        'x': 0.5
            },
            xaxis=dict(
                title='机组',
                tickangle=-45
            ),
            yaxis=dict(
                title='总数'
            )
        )

        # 确保从 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": "总数",
            "data": [{
                "engineName": "",
                "engineCode": "",
                "title": f' 额定风速间隔数据计数',
                "xData": data[Field_NameOfTurbine].tolist(),
                "yData": data['count'].tolist(),
            }]
        }

        result_rows = []
        # 保存图像
        pngFileName = '风速区间数据计数.png'
        pngFilePath = os.path.join(outputAnalysisDir, pngFileName)
        fig.write_image(pngFilePath, scale=3)

        # 保存HTML
        # htmlFileName = '风速区间数据计数.html'
        # htmlFilePath = os.path.join(outputAnalysisDir, htmlFileName)
        # fig.write_html(htmlFilePath)
        # 将JSON对象保存到文件
        output_json_path = os.path.join(outputAnalysisDir, f"rated_WindSpeed.json")
        with open(output_json_path, 'w', encoding='utf-8') as f:
            import json
            json.dump(json_output, f, ensure_ascii=False, indent=4)

        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: pngFilePath,
            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: 'total',
            Field_MillTypeCode: 'total',
            Field_Return_FilePath: output_json_path,
            Field_Return_IsSaveDatabase: True
        })

        result_df = pd.DataFrame(result_rows)

        return result_df