pv-current/coef_matrix.py

# Created by: leo
# Created on: 2018.11.14

import pandas as pd
import numpy as np
import logging
from max_current import (CURRENT_HEADER, TIME_HEADER, CLUSTER_HEADER, OUTLIER,
                         CONST_ZERO, CONST_NONZERO, get_thr_from_coefs)


def coef_median(inp: pd.DataFrame, min_sample_no: int = 2) -> pd.Series:
    """
    基于相关系数矩阵列中值法计算指定数据集所在的时间窗口内所有组串的相关系数向量

    :param inp: 包含时间和电流数值的 Dataframe，需包含3列：时间戳, 组串ID, 电流值
    :param min_sample_no: 最小有效观测数，数据清洗后如果小于此值则返回异常值
    :returns: 包含每个组串与最大电流所在组串比较的相关系数向量，长度为 inp 内包含所有组串数量
    """
    if len(inp) == 0:
        return pd.Series([OUTLIER])

    # 若输入为纵表，转换为宽表（组串ID作列名）
    if CLUSTER_HEADER in inp.columns:
        raw = pd.pivot_table(inp, values=CURRENT_HEADER, index=[TIME_HEADER], columns=[CLUSTER_HEADER])
        data = raw.dropna()
        data.index = pd.to_datetime(data.index)
    else:
        data = inp

    # 校验有效数据长度，若低于最小阈值则返回异常指标序列
    if len(data) < min_sample_no:
        return pd.Series([OUTLIER])

    coef_matrix = data.corr()


    coef_med = coef_matrix.median()
    if coef_med.dropna().empty:
        return pd.Series([OUTLIER])

    # 保证电流高的组串相关系数高
    coefs = coef_med if data[coef_med.idxmax()].mean() >= data[coef_med.idxmin()].mean() else 1 - coef_med

    # mark strings with constant current
    consts = coefs[coefs.isna()]
    for idx in consts.index:
        if sum(abs(data[idx])) < 1e-7:
            coefs[idx] = CONST_ZERO
        else:
            coefs[idx] = CONST_NONZERO
    return coefs


def string_coef_matrix(inp: pd.DataFrame, str_id: int) -> float:
    """
    基于相关系数矩阵方法计算指定数据集所在的窗口时间内指定组串的电流相关性系数

    :param inp: 包含时间和电流数值的 Dataframe，需包含3列：时间戳, 组串ID, 电流值
    :param str_id: 指定组串的ID
    :return: 指定组串与最大电流所在组串比较的相关系数
    """
    coefs = coef_median(inp)
    if len(coefs) == 1:
        return coefs[0]
    if str_id in coefs.index:
        return coefs[str_id]
    else:
        return OUTLIER


def calc_matrix_coefs(inp: pd.DataFrame) -> pd.DataFrame:
    """
    基于相关系数矩阵方法计算所有组串在输入 DataFrame 上的所有相关系数，作为后续聚类的输入

    :param inp: 包含多个月份一个汇流箱内所有组串电流值的数据集，需包含3列：时间戳, 组串ID, 电流值
    :return: DataFrame, index 为时间，列为输入 DataFrame 中包含的所有组串ID
    """
    raw = pd.pivot_table(inp, values=CURRENT_HEADER, index=[TIME_HEADER], columns=[CLUSTER_HEADER])
    data = raw.dropna()
    data.index = pd.to_datetime(data.index)
    daily = data.groupby(pd.Grouper(freq='D'))
    coef_tbl = {str_id: daily.apply(string_coef_matrix, str_id) for str_id in data.columns}
    return pd.DataFrame(coef_tbl)


def train_median_thresholds(inp: pd.DataFrame) -> np.ndarray:
    """
    光伏故障分类算子训练函数，计算分隔阈值

    :param inp: 包含多个月份一个汇流箱内所有组串电流值的数据集，需包含3列：时间戳, 组串ID, 电流值
    :param thr: 月度最大电流的百分比，低于此阈值的数据被分入低电流组
    :return: 二元组，分别为正常、关注、告警组的分隔阈值
    """
    coefs = calc_matrix_coefs(inp)
    return get_thr_from_coefs(coefs)


def classify(inp: pd.DataFrame) -> pd.DataFrame:
    """
    根据输入参数对输出数据做状态分类

    :param inp: 输入电流数据，需包含3列：时间戳, 组串ID, 电流值
    :return: 输入数据集中每个组串的电流中值，为了配合PySpark 的 PandasUDFType.GROUPED_MAP,
             类型改为 DataFrame
    """
    # 需要将电流值由字符串类型转为实数类型
    fix_type = inp.astype({'current': 'float'})
    coefs = coef_median(fix_type)

    return coefs.to_frame('coef')