energy-data-trans/tmp_file/power_derating_for_chunlin.py

import os

import matplotlib
import numpy as np
from matplotlib import pyplot as plt

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

import pandas as pd
import chardet
import warnings

warnings.filterwarnings("ignore")


# 获取文件编码
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 = '文件:' + os.path.basename(file_path) + ',' + str(e)
        raise ValueError(message)

    return df


def __build_directory_dict(directory_dict, path, filter_types=None):
    # 遍历目录下的所有项
    for item in os.listdir(path):
        item_path = os.path.join(path, item)
        if os.path.isdir(item_path):
            __build_directory_dict(directory_dict, item_path, filter_types=filter_types)
        elif os.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]


class ContractPowerCurve(object):

    def __init__(self, df: pd.DataFrame, wind_velocity='风速', active_power='功率'):
        self.df = df
        self.wind_velocity = wind_velocity
        self.active_power = active_power


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

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


def scatter(title, x_label, y_label, x_values, y_values, color='blue', size=10, save_file_path=''):
    if save_file_path:
        create_file_path(save_file_path, True)
    else:
        save_file_path = title + '.png'

    plt.figure(figsize=(8, 6))
    plt.title(title, fontsize=16)
    plt.xlabel(x_label, fontsize=14)
    plt.ylabel(y_label, fontsize=14)
    plt.scatter(x_values, y_values, s=size, c=color)
    plt.savefig(save_file_path)
    plt.close()


def marker_active_power(contract_power_curve_class: ContractPowerCurve, df: pd.DataFrame, active_power='有功功率 kW均值',
                        wind_velocity='风速 m/s均值'):
    """
    标记有功功率为正的记录
    :param contract_power_curve_class: 合同功率曲线
    :param df: 原始数据
    :return: 标记有功功率为正的原始数据
    """
    contract_power_curve_df = contract_power_curve_class.df
    curve_wv = contract_power_curve_df[contract_power_curve_class.wind_velocity].values
    curve_ap = contract_power_curve_df[contract_power_curve_class.active_power].values

    df.dropna(subset=[active_power, wind_velocity], inplace=True)
    ap_gt_0_df = df[df[active_power] > 0]
    ap_le_0_df = df[df[active_power] <= 0]
    ap_le_0_df["marker"] = -1

    active_power_values = ap_gt_0_df[active_power].values
    wind_speed_values = ap_gt_0_df[wind_velocity].values
    ap_gt_0_in = [0] * ap_gt_0_df.shape[0]

    for i in range(len(ap_gt_0_in)):
        wind_speed = wind_speed_values[i]
        active_power = active_power_values[i]

        # if active_power >= 2200 - 200:
        #     ap_gt_0_in[i] = 1
        # else:
        diffs = np.abs(curve_wv - wind_speed)
        # 找到差值最小的索引和对应的差值
        minDiff, idx = np.min(diffs), np.argmin(diffs)

        # 使用找到的索引获取对应的值
        closestValue = curve_ap[idx]
        if active_power - closestValue >= -100:
            ap_gt_0_in[i] = 1

    ap_gt_0_df['marker'] = ap_gt_0_in
    return pd.concat([ap_gt_0_df, ap_le_0_df])


if __name__ == '__main__':
    wind_power_df = read_file_to_df(r"D:\中能智能\matlib计算相关\标记derating\PV_Curve.csv")

    all_files = read_excel_files(r"Z:\collection_data\1进行中\诺木洪风电场-甘肃-华电\清理数据\min-666")
    save_path = r"D:\trans_data\诺木洪\清理数据\min-666-derating"

    wind_power_df_class = ContractPowerCurve(wind_power_df)

    for file in all_files:
        name = os.path.basename(file).split("@")[0]
        try:
            df = read_file_to_df(file)
            df = marker_active_power(wind_power_df_class, df)
            df = df[df['marker'] == 1]
            # 保存筛选后数据
            name = name.replace('HD', 'HD2')
            df.to_csv(os.path.join(save_path, name + '.csv'), index=False, encoding='utf-8')

            # 使用scatter函数绘制散点图
            if not df.empty:
                scatter(name, x_label='风速均值', y_label='有功功率均值', x_values=df['风速 m/s均值'].values,
                        y_values=df['有功功率 kW均值'].values, color='green',
                        save_file_path=os.path.join(save_path, name + '均值.png'))

        except Exception as e:
            print(os.path.basename(file), "出错", str(e))
            raise e