energy-data-trans/tmp_file/pv_youxiaoxing.py

import multiprocessing
from os import *

import matplotlib

matplotlib.use('Agg')
matplotlib.rcParams['font.family'] = 'SimHei'  # 或者 'Microsoft YaHei'
matplotlib.rcParams['font.sans-serif'] = ['SimHei']  # 或者 ['Microsoft YaHei']

import chardet
import warnings

warnings.filterwarnings("ignore")

import datetime

import pandas as pd


def get_time_space(df, time_str):
    """
    :return: 查询时间间隔
    """
    begin = datetime.datetime.now()
    df1 = pd.DataFrame(df[time_str])
    df1[time_str] = pd.to_datetime(df1[time_str], errors='coerce')
    df1.sort_values(by=time_str, inplace=True)
    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
    print(datetime.datetime.now() - begin)
    return abs(result)


def get_time_space_count(start_time: datetime.datetime, end_time: datetime.datetime, time_space=1):
    """
    获取俩个时间之间的个数
    :return: 查询时间间隔
    """
    delta = end_time - start_time
    total_seconds = delta.days * 24 * 60 * 60 + delta.seconds

    return abs(int(total_seconds / time_space)) + 1


# 获取文件编码
def detect_file_encoding(filename):
    # 读取文件的前1000个字节（足够用于大多数编码检测）
    with open(filename, 'rb') as f:
        rawdata = f.read(1000)
    result = chardet.detect(rawdata)
    encoding = result['encoding']

    if encoding is None:
        encoding = 'gb18030'

    if encoding and encoding.lower() == 'gb2312' or encoding.lower().startswith("windows"):
        encoding = 'gb18030'
    return encoding


def del_blank(df=pd.DataFrame(), cols=list()):
    for col in cols:
        if df[col].dtype == object:
            df[col] = df[col].str.strip()
    return df


# 切割数组到多个数组
def split_array(array, num):
    return [array[i:i + num] for i in range(0, len(array), num)]


# 读取数据到df
def read_file_to_df(file_path, read_cols=list(), header=0):
    try:
        df = pd.DataFrame()
        if str(file_path).lower().endswith("csv") or str(file_path).lower().endswith("gz"):
            encoding = detect_file_encoding(file_path)
            end_with_gz = str(file_path).lower().endswith("gz")
            if read_cols:
                if end_with_gz:
                    df = pd.read_csv(file_path, encoding=encoding, usecols=read_cols, compression='gzip', header=header)
                else:
                    df = pd.read_csv(file_path, encoding=encoding, usecols=read_cols, header=header,
                                     on_bad_lines='warn')
            else:

                if end_with_gz:
                    df = pd.read_csv(file_path, encoding=encoding, compression='gzip', header=header)
                else:
                    df = pd.read_csv(file_path, encoding=encoding, header=header, on_bad_lines='warn')

        else:
            xls = pd.ExcelFile(file_path)
            # 获取所有的sheet名称
            sheet_names = xls.sheet_names
            for sheet in sheet_names:
                if read_cols:
                    now_df = pd.read_excel(xls, sheet_name=sheet, header=header, usecols=read_cols)
                else:
                    now_df = pd.read_excel(xls, sheet_name=sheet, header=header)

                df = pd.concat([df, now_df])

        print('文件读取成功', file_path, '文件数量', df.shape)
    except Exception as e:
        print('读取文件出错', file_path, str(e))
        message = '文件:' + path.basename(file_path) + ',' + str(e)
        raise ValueError(message)

    return df


def __build_directory_dict(directory_dict, path, filter_types=None):
    # 遍历目录下的所有项
    for item in listdir(path):
        item_path = path.join(path, item)
        if path.isdir(item_path):
            __build_directory_dict(directory_dict, item_path, filter_types=filter_types)
        elif path.isfile(item_path):
            if path not in directory_dict:
                directory_dict[path] = []

            if filter_types is None or len(filter_types) == 0:
                directory_dict[path].append(item_path)
            elif str(item_path).split(".")[-1] in filter_types:
                if str(item_path).count("~$") == 0:
                    directory_dict[path].append(item_path)

    # 读取所有文件


# 读取路径下所有的excel文件
def read_excel_files(read_path):
    directory_dict = {}
    __build_directory_dict(directory_dict, read_path, filter_types=['xls', 'xlsx', 'csv', 'gz'])

    return [path for paths in directory_dict.values() for path in paths if path]


# 创建路径
def create_file_path(path, is_file_path=False):
    if is_file_path:
        path = path.dirname(path)

    if not path.exists(path):
        makedirs(path, exist_ok=True)


def time_biaozhun(df):
    time_space = get_time_space(df, '时间')
    query_df = df[['时间']]
    query_df['时间'] = pd.to_datetime(df['时间'], errors="coerce")
    query_df = query_df.dropna(subset=['时间'])
    total = get_time_space_count(query_df['时间'].min(), query_df['时间'].max(), time_space)
    return total, save_percent(1 - query_df.shape[0] / total), save_percent(1 - df.shape[0] / total)


def save_percent(value, save_decimal=7):
    return round(value, save_decimal) * 100


def calc(df, file_name):
    error_dict = {}
    lose_dict = {}
    error_dict['箱变'] = "".join(file_name.split(".")[:-1])
    lose_dict['箱变'] = "".join(file_name.split(".")[:-1])

    total, lose_time, error_time = time_biaozhun(df)
    error_dict['时间'] = error_time
    lose_dict['时间'] = lose_time

    error_df = pd.DataFrame()
    lose_df = pd.DataFrame()

    try:
        df.columns = ["".join(["逆变器" + "".join(col.split("逆变器")[1:])]) if col.find("逆变器") > -1 else col for col in
                      df.columns]

        for col in df.columns:
            if col == '时间':
                continue
            query_df = df[[col]]
            query_df[col] = pd.to_numeric(query_df[col], errors="coerce")
            query_df = query_df.dropna(subset=[col])
            lose_dict[col] = save_percent(1 - query_df.shape[0] / total)

            if col.find('电压') > -1:
                error_dict[col] = save_percent(query_df[query_df[col] < 0].shape[0] / total)

            if col.find('电流') > -1:
                error_dict[col] = save_percent(query_df[query_df[col] < -0.1].shape[0] / total)

            if col.find('逆变器效率') > -1:
                error_dict[col] = save_percent(query_df[(query_df[col] <= 0) | (query_df[col] >= 100)].shape[0] / total)

            if col.find('温度') > -1:
                error_dict[col] = save_percent(query_df[(query_df[col] < 0) | (query_df[col] > 100)].shape[0] / total)

            if col.find('功率因数') > -1:
                error_dict[col] = save_percent(query_df[(query_df[col] < 0) | (query_df[col] > 1)].shape[0] / total)

        total, count = 0, 0
        for k, v in error_dict.items():
            if k != '箱变':
                total = total + error_dict[k]
                count = count + 1

        error_dict['平均异常率'] = save_percent(total / count / 100)

        total, count = 0, 0
        for k, v in lose_dict.items():
            if k != '箱变':
                total = total + lose_dict[k]
                count = count + 1

        lose_dict['平均缺失率'] = save_percent(total / count / 100)

        error_df = pd.concat([error_df, pd.DataFrame(error_dict, index=[0])])
        lose_df = pd.concat([lose_df, pd.DataFrame(lose_dict, index=[0])])

        error_df_cols = ['箱变', '平均异常率']
        for col in error_df.columns:
            if col not in error_df_cols:
                error_df_cols.append(col)

        lose_df_cols = ['箱变', '平均缺失率']
        for col in lose_df.columns:
            if col not in lose_df_cols:
                lose_df_cols.append(col)

        error_df = error_df[error_df_cols]
        lose_df = lose_df[lose_df_cols]
    except Exception as e:
        print("异常文件", path.basename(file_name))
        raise e

    return error_df, lose_df


def run(file_path):
    df = read_file_to_df(file_path)
    return calc(df, path.basename(file_path))


if __name__ == '__main__':
    # read_path = r'/data/download/大唐玉湖性能分析离线分析/05整理数据/逆变器数据'
    # save_path = r'/data/download/大唐玉湖性能分析离线分析/06整理数据/逆变器数据'

    read_path = r'D:\trans_data\大唐玉湖性能分析离线分析\test\yuanshi'
    save_path = r'D:\trans_data\大唐玉湖性能分析离线分析\test\zhengli'
    all_files = read_excel_files(read_path)

    with multiprocessing.Pool(2) as pool:
        df_arrys = pool.starmap(run, [(file,) for file in all_files])

    error_df = pd.concat([df[0] for df in df_arrys])
    lose_df = pd.concat([df[1] for df in df_arrys])
    with pd.ExcelWriter(path.join(save_path, "玉湖光伏数据统计.xlsx")) as writer:
        error_df.to_excel(writer, sheet_name='error_percent', index=False)
        lose_df.to_excel(writer, sheet_name='lose_percent', index=False)