WTOAAM/dataAnalysisBusiness/algorithm/dataIntegrityOfSecondAnalyst.py

import os
import pandas as pd
import numpy as np
import pandas as pd
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.analyst import Analyst
from utils.directoryUtil import DirectoryUtil as dir
from algorithmContract.confBusiness import *
import calendar
import random 


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

    def typeAnalyst(self):
        return "data_integrity_second"
    
    def filterCommon(self,dataFrame:pd.DataFrame, confData:ConfBusiness):        
        return dataFrame

    def turbinesAnalysis(self, dataFrameMerge, outputAnalysisDir, confData: ConfBusiness):
        groupedDataFrame = self.dataIntegrityByMonth(
            dataFrameMerge,confData, Field_NameOfTurbine )
        print("groupedDataFrame : \n {}".format(groupedDataFrame.head()))
        self.plotByAllMonth(groupedDataFrame, outputAnalysisDir,
                  confData.farm_name, Field_NameOfTurbine)
        
    def generate_weighted_random(self):  
        # 首先，尝试生成一个在91至100之间的随机数，这样大部分值都会落在这个区间  
        if random.random() < 0.8:  # 假设80%的随机数应该在91至100之间  
            return random.randint(91, 100)  
        else:  # 剩下的20%则均匀分布在79至100之间（包括79但不包括100）  
            return random.randint(79, 100)  
        
    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 dataIntegrityByMonth(self, dataFrameMerge:pd.DataFrame, confData:ConfBusiness,fieldTurbineName):
        grouped = dataFrameMerge.groupby([dataFrameMerge.loc[:, confData.field_turbine_time].dt.year.rename('year'),
                                          dataFrameMerge.loc[:, confData.field_turbine_time].dt.month.rename(
                                              'month'),
                                          dataFrameMerge.loc[:, fieldTurbineName]]).agg({'count'})[confData.field_turbine_time].rename({'count': '长度'}, axis=1)
        
        new_frame = grouped.reset_index('month')
        new_frame = new_frame.assign(数据完整度=(100 * new_frame['长度'] / (
            new_frame['month'].map(lambda x: calendar.mdays[x] * 24 * 3600 / confData.time_period))).round(decimals=0))
        
        # new_frame['数据完整度'] = [self.generate_weighted_random() for _ in range(len(new_frame))] 
        new_frame = new_frame.rename(columns={'数据完整度': '数据完整度%'})

        new_frame = new_frame.reset_index()
        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)
        
        new_frame = self.fullMonthIndex(confData.start_time,confData.end_time,fieldTurbineName,new_frame)

        return new_frame

    def plotByAllMonth(self, groupedDataFrame, outputAnalysisDir, farmName, fieldTurbineName):
        title = 'time integrity check(%)'
        fig, ax = plt.subplots(figsize=(18, 15), dpi=300)
        # 风机数量小于月份
        if len(set(groupedDataFrame.loc[:, Field_YearMonth])) > len(set(groupedDataFrame.loc[:, fieldTurbineName])):
            result = pd.pivot(groupedDataFrame, index=fieldTurbineName,
                              columns=Field_YearMonth, values="数据完整度%")
            ax = sns.heatmap(data=result, square=True, annot=True,
                             linewidths=0.3, cbar=False, fmt='g',)
            bottom, top = ax.get_ylim()
            ax.set_ylim(bottom + 0.5, top - 0.5)
            ax.set_title(title)
            plt.setp(ax.get_yticklabels(), rotation=0)
            plt.setp(ax.get_xticklabels(), rotation=90)
            plt.savefig(outputAnalysisDir +
                        r'/{}数据完整度分析.png'.format(farmName), bbox_inches='tight')
            plt.close()
        else:
            result = pd.pivot(groupedDataFrame, index=Field_YearMonth,
                              columns=fieldTurbineName, values="数据完整度%")
            ax = sns.heatmap(data=result, square=True, annot=True,
                             linewidths=0.3, cbar=False, fmt='g',)
            bottom, top = ax.get_ylim()
            ax.set_ylim(bottom + 0.5, top - 0.5)
            ax.set_title(title)
            plt.setp(ax.get_yticklabels(), rotation=0)
            plt.setp(ax.get_xticklabels(), rotation=90)
            # 设置x轴标签斜向展示
            plt.xticks(rotation=45)  # 旋转45度
            plt.savefig(outputAnalysisDir +
                        r'/{}数据完整度分析.png'.format(farmName), bbox_inches='tight')
            plt.close()

    def draw(self, groupedDataFrame, outputAnalysisDir, farmName, fieldTurbineName):
        fig = make_subplots(rows=1, cols=1)
        if len(set(groupedDataFrame[Field_YearMonth])) > len(set(groupedDataFrame[fieldTurbineName])):
            result = groupedDataFrame.pivot(
                index=fieldTurbineName, columns=Field_YearMonth, values="数据完整度%")
            fig.add_trace(
                go.Heatmap(
                    z=result.values,
                    x=result.columns,
                    y=result.index,
                    colorscale='Viridis',
                    colorbar=dict(title='数据完整度%'),
                    text=[[f"{value}" for value in row]
                        for row in result.values],  # 显示的文本（百分比）
                    texttemplate="%{text}",  # 使用文本模板
                    yaxis=dict(  
                        tickformat="%Y-%m",  # 设置y轴刻度的格式  
                        tickmode="array",    # 如果需要，可以指定自定义的刻度标签  
                        tickvals=result.columns.strftime("%Y-%m").tolist()  # 如果需要自定义刻度位置  
                    )  
                )
            )
        else:
            result = groupedDataFrame.pivot(
                index=Field_YearMonth, columns=fieldTurbineName, values="数据完整度%")
            fig.add_trace(
                go.Heatmap(
                    z=result.values,
                    x=result.columns,
                    y=result.index,
                    colorscale='Viridis',
                    colorbar=dict(title='数据完整度%'),
                    text=[[f"{value}" for value in row]
                        for row in result.values],  # 显示的文本（百分比）
                    texttemplate="%{text}",  # 使用文本模板
                    xaxis=dict(  
                        tickformat="%Y-%m",  # 设置y轴刻度的格式  
                        tickmode="array",    # 如果需要，可以指定自定义的刻度标签  
                        tickvals=result.index.strftime("%Y-%m").tolist()  # 如果需要自定义刻度位置  
                    )
                )
            )

        fig.update_layout(
            title_text='{}-time integrity check(%)'.format(farmName),
            xaxis_nticks=36
        )

        fig.write_image(outputAnalysisDir + '/' +
                        '{}数据完整度分析.png'.format(farmName))