WTOAAM/dataAnalysisBusiness/algorithm/dataIntegrityOfSecondAnalyst.py

import os
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from geopy.distance import geodesic
from behavior.analystNotFilter import AnalystNotFilter
from utils.directoryUtil import DirectoryUtil as dir
from algorithmContract.confBusiness import *
import calendar
import random
from datetime import datetime
from algorithmContract.contract import Contract


class DataIntegrityOfSecondAnalyst(AnalystNotFilter):
    """
    风电机组秒级数据完整度分析
    """

    def typeAnalyst(self):
        return "data_integrity_second"

    def turbinesAnalysis(self,  outputAnalysisDir, conf: Contract, turbineCodes):
        select = [Field_DeviceCode, Field_Time, Field_ActiverPower, Field_WindSpeed, Field_NacPos, Field_WindDirection, Field_RotorSpeed, Field_GeneratorSpeed, Field_GeneratorTorque, Field_AngleIncluded, Field_EnvTemp, Field_NacTemp, Field_PitchAngel1, Field_PitchAngel2, Field_PitchAngel3]

        dictionary = self.processTurbineData(turbineCodes, conf, select)
        dataFrameOfTurbines = self.userDataFrame(
            dictionary, conf.dataContract.configAnalysis, self)

        turbrineInfos = self.common.getTurbineInfos(
            conf.dataContract.dataFilter.powerFarmID, turbineCodes, self.turbineInfo)

        currDataFrameOfTurbines = dataFrameOfTurbines[dataFrameOfTurbines[Field_CodeOfTurbine].isin(
            turbineCodes)]
        # 将 currTurbineInfos 转换为字典
        currTurbineInfosDict = turbrineInfos.set_index(
            Field_CodeOfTurbine)[Field_NameOfTurbine].to_dict()
        # 使用 map 函数来填充 Field_NameOfTurbine 列
        currDataFrameOfTurbines[Field_NameOfTurbine] = currDataFrameOfTurbines[Field_CodeOfTurbine].map(
            currTurbineInfosDict).fillna("")

        groupedDataFrame = self.dataIntegrityByMonth(
            dataFrameOfTurbines, conf, Field_NameOfTurbine)
        print("groupedDataFrame : \n {}".format(groupedDataFrame.head()))

        return self.plotByAllMonth(groupedDataFrame, outputAnalysisDir, self.powerFarmInfo[Field_PowerFarmName].iloc[0], Field_NameOfTurbine, conf)

    def fullMonthIndex(self, start_time, end_time, turbine_name, new_frame):
        months = (end_time.year - start_time.year) * \
            12 + end_time.month - start_time.month
        month_range = ['%04d-%02d' % (int(start_time.year + mon//12), int(mon % 12+1))
                       for mon in range(start_time.month-1, start_time.month+months)]
        month_index = pd.DataFrame(month_range, columns=[Field_YearMonth])

        plot_res = pd.DataFrame()
        grouped = new_frame.groupby(turbine_name)
        for name, group in grouped:
            group = pd.merge(group, month_index,
                             on=Field_YearMonth, how='outer')
            group['数据完整度%'] = group['数据完整度%'].fillna(0)
            group[turbine_name] = name
            group['year'] = group[Field_YearMonth].apply(
                lambda x: str(x).split('-')[0])
            group['month'] = group[Field_YearMonth].apply(
                lambda x: str(x).split('-')[1])
            plot_res = pd.concat([plot_res, group], axis=0, sort=False)

        return plot_res
    
    def get_time_space(self,df, time_str):
        """
        :return: 查询时间间隔（单位：秒）
        """
        df1 = pd.DataFrame(df[time_str])
        df1['chazhi'] = df1[time_str].shift(-1) - df1[time_str]
        result = df1.sample(int(df1.shape[0] / 100))['chazhi'].value_counts().idxmax().seconds
        del df1
        return result

    def dataIntegrityByMonth(self, dataFrameMerge: pd.DataFrame, conf: Contract, Field_NameOfTurbine):
        grouped = dataFrameMerge.groupby([dataFrameMerge.loc[:, Field_Time].dt.year.rename('year'),
                                          dataFrameMerge.loc[:, Field_Time].dt.month.rename(
                                              'month'),
                                          dataFrameMerge.loc[:, Field_NameOfTurbine]]).agg({'count'})[Field_Time].rename({'count': '长度'}, axis=1)

        new_frame = grouped.reset_index('month')
        # timeGranularity = self.dataTransfer[self.dataTransfer[Field_TransferType] == Const_TimeGranularity_Second][Field_TimeGranularity].iloc[0] if self.typeAnalyst(
        # ) == "data_integrity_second" else self.dataTransfer[self.dataTransfer[Field_TransferType] == Const_TimeGranularity_Minute][Field_TimeGranularity].iloc[0]
        timeGranularity=self.get_time_space(dataFrameMerge,Field_Time)
        self.logger.info(f"{self.typeAnalyst()}   timeGranularity-->{timeGranularity}")

        new_frame = new_frame.reset_index()

        new_frame['数据完整度'] = (100 * new_frame['长度'] / (new_frame.apply(lambda row: calendar.monthrange(
            row['year'], row['month'])[1] * 24 * 3600 / timeGranularity, axis=1))).round(decimals=0)

        new_frame = new_frame.rename(columns={'数据完整度': '数据完整度%'})
        new_frame['month'] = new_frame['month'].astype(
            str).apply(lambda x: x.zfill(2))
        new_frame[Field_YearMonth] = new_frame['year'].astype(
            str) + '-' + new_frame['month'].astype(str)

        beginTime = None
        if not self.common.isNone(conf.dataContract.dataFilter.beginTime):
            beginTime = conf.dataContract.dataFilter.beginTime
        else:
            beginTime = dataFrameMerge[Field_Time].min().strftime(
                '%Y-%m-%d %H:%M:%S')
        endTime = None
        if not self.common.isNone(conf.dataContract.dataFilter.endTime):
            endTime = conf.dataContract.dataFilter.endTime
        else:
            endTime = dataFrameMerge[Field_Time] .max().strftime(
                '%Y-%m-%d %H:%M:%S')

        beginTime = datetime.strptime(beginTime, '%Y-%m-%d %H:%M:%S')
        endTime = datetime.strptime(endTime, '%Y-%m-%d %H:%M:%S')

        new_frame = self.fullMonthIndex(
            beginTime, endTime, Field_NameOfTurbine, new_frame)

        return new_frame

    def plotByAllMonth(self, groupedDataFrame, outputAnalysisDir, farmName, fieldTurbineName, conf: Contract):
        title = '数据完整度检测(%)'
        # 根据场景决定索引和列的方向
        if len(set(groupedDataFrame[Field_YearMonth])) > len(set(groupedDataFrame[fieldTurbineName])):
            result = groupedDataFrame.pivot(
                values="数据完整度%", index=fieldTurbineName, columns=Field_YearMonth)
            x_labels = result.columns.tolist()  # 月份
            y_labels = result.index.tolist()    # 风机名
            x_axis_title = "日期"
            y_axis_title = "机组"
            # 构建最终的JSON对象
            json_output = {
                "analysisTypeCode": "数据完整度检测(%)",
                "engineCode": "",
                "engineTypeName": "",
                "xaixs": "日期",
                "yaixs": "机组",
                "data": [{
                    "engineName": "",
                    "engineCode": "",
                    "title": f' 数据完整度%',
                    "xData": x_labels,
                    "yData": y_labels,
                    "ZData": result.values.tolist(),
                }]
            }
        else:
            result = groupedDataFrame.pivot(
                values="数据完整度%", index=Field_YearMonth, columns=fieldTurbineName)
            x_labels = result.columns.tolist()  # 风机名
            y_labels = result.index.tolist()    # 月份
            x_axis_title = "机组"
            y_axis_title = "日期"
            # 构建最终的JSON对象
            json_output = {
                "analysisTypeCode": "数据完整度检测(%)",
                "engineCode": "",
                "engineTypeName": "",
                "xaixs": "机组",
                "yaixs": "日期",
                "data": [{
                    "engineName": "",
                    "engineCode": "",
                    "title": f' 数据完整度%',
                    "xData": x_labels,
                    "yData": y_labels,
                    "ZData": result.values.tolist(),
                }]
            }

        # # 创建热图
        # fig = go.Figure(data=go.Heatmap(
        #     z=result.values,
        #     x=x_labels,
        #     y=y_labels,
        #     colorscale='Viridis',
        #     # colorbar=dict(title='数据完整度%'),
        #     showscale=False,  # 显示颜色条
        #     text=result.values,
        #     texttemplate="%{text}",  # Format the text display inside cells
        #     # hoverinfo='text'
        # ))
        # 创建热图
        fig = go.Figure(data=go.Heatmap(
            z=result.values,
            x=x_labels,
            y=y_labels,
            colorscale=[
                [0.0, 'rgb(255, 102, 102)'],  # 柔和的红色
                [0.5, 'rgb(255, 102, 102)'],
                [0.5, 'rgb(255, 255, 153)'],  # 柔和的黄色
                [0.85, 'rgb(255, 255, 153)'],
                [0.85, 'rgb(153, 255, 153)'],  # 柔和的绿色
                [1.0, 'rgb(153, 255, 153)']
            ],
            zmin=0,  # 设置颜色范围的最小值
            zmax=100,  # 设置颜色范围的最大值
            showscale=True,  # 显示颜色条
            text=result.values,
            texttemplate="%{text}",  # Format the text display inside cells
        ))

        # 更新图形布局
        fig.update_layout(
            title={'text': title, 'x': 0.5},
            # xaxis_nticks=len(x_labels),
            xaxis=dict(tickmode='array', tickvals=x_labels,
                       ticktext=x_labels, tickangle=-45, title=x_axis_title),
            yaxis=dict(tickmode='array', tickvals=y_labels,
                       ticktext=y_labels, title=y_axis_title),
            # xaxis=dict(tickmode='array', tickvals=list(range(len(x_labels))), ticktext=x_labels, tickangle=-45, title=x_axis_title),
            # yaxis=dict(tickmode='array', tickvals=list(range(len(y_labels))), ticktext=y_labels, title=y_axis_title),
            autosize=True,
            # width=len(x_labels) * 80,  # Adjust width and height as needed
            # height=len(y_labels) * 80,
            margin=dict(l=50, r=50, b=100, t=100),  # 调整边距以确保标签完整显示
            # Transparent background to show cell borders
            plot_bgcolor='rgba(0,0,0,0)'
        )

        fig.update_traces(
            xgap=1,
            ygap=1
        )

        result_rows = []

        # 将JSON对象保存到文件
        output_json_path = os.path.join(outputAnalysisDir, f"Data_Integrity_Of_Second_Analyst.json")
        with open(output_json_path, 'w', encoding='utf-8') as f:
            import json
            json.dump(json_output, f, ensure_ascii=False, indent=4)

        # 保存图像
        pngFileName = f'{farmName}数据完整度分析.png'
        pngFilePath = os.path.join(outputAnalysisDir, pngFileName)
        fig.write_image(pngFilePath, scale=3)

        # 保存HTML
        # htmlFileName = f'{farmName}数据完整度分析.html'
        # htmlFilePath = os.path.join(outputAnalysisDir, htmlFileName)
        # fig.write_html(htmlFilePath)

        result_rows.append({
            Field_Return_TypeAnalyst: self.typeAnalyst(),
            Field_PowerFarmCode: conf.dataContract.dataFilter.powerFarmID,
            Field_Return_BatchCode: conf.dataContract.dataFilter.dataBatchNum,
            Field_CodeOfTurbine: Const_Output_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: Const_Output_Total,
            Field_MillTypeCode: 'total',
            Field_Return_FilePath: output_json_path,
            Field_Return_IsSaveDatabase: True
        })

        result_df = pd.DataFrame(result_rows)

        return result_df