pv-current/deviation/app.py

import sys
import pandas as pd
import subprocess
import logging
import csv
import json
import numpy as np
from typing import Tuple
"""
 * @author Leo
 * 2019.1.25
 *
 * @Description: 光伏组串离散率算子入口函数
 *
 * 实现标准输入格式到算子输入的转换
 * 保证不论输入的是一个计算单元，还是多个计算单元，不影响算子实现

常量定义：

大写字段名用于输入、输出，小写字段名用于计算过程中的存储字段
"""

START_TIME = '07:00'
END_TIME = '17:00'
TF = 30
SAMPLE_INT = 5 # 采用频率，目前规定为5min
I0 = 1
IM = 12
STATION_ID = 'STATION_ID'
DEVICE_ID = 'DEVICE_ID'
TIME_HEADER = 'MONITOR_TIME'
REPORT_TIME = 'ALERT_TIME'
V_HEADER = 'VOLTAGE_VALUE'
CURRENT_HEADER = 'CURRENT_VALUE'
GRP_ID = 'grp_id'
STR_HEADER = 'str_no'
DEVIATION = 'DEVIATION'    # 输出结果中 离散率 列名
STR_STATUS = 'STR_STATUS'  # 输出结果中 组串状态 列名
COM_FAIL = -1
UNIT_INVALID = -2
STR_INVALID = -3

COM_FAIL_DF = pd.DataFrame({DEVIATION: [COM_FAIL], STR_STATUS: ["-"]}, index=pd.Index([END_TIME]))
UNIT_INVALID_DF = pd.DataFrame({DEVIATION: [UNIT_INVALID], STR_STATUS: ["-"]}, index=pd.Index([END_TIME]))

inputFilePath = sys.argv[1]
outputFilePath = sys.argv[2]
isTrain = sys.argv[3]

# inputFilePath = "/home/leo//docs/znbt/photovoltaics/HuBeiShaYang/fullTable/comb1122.csv"
# outputFilePath = "res.csv"
# isTrain = "false"

logging.basicConfig(level=logging.INFO)
logging.debug(f"input file: {inputFilePath}, output file: {outputFilePath}")
logging.debug('-------- Head of input CSV file: ---------')
head_lines = subprocess.run(['head', inputFilePath], check=True, stdout=subprocess.PIPE)
logging.debug(head_lines)

def get_valid_zone(indf: pd.DataFrame, thr: float):
    """获取有效计算区间 [t1, t2]
    """
    above_i0_flags = indf[CURRENT_HEADER] >= thr
    above_i0_vals = indf.loc[above_i0_flags.any(axis=1)]
    assert not above_i0_vals.empty  # 通过 step 2 的验证，本步中至少应该有一个大于 I0 的电流值
    return above_i0_vals.index[0]

def filter_window(indf: pd.DataFrame, start_index) -> pd.DataFrame:
    """按照文档第3步要求对过滤和校正组串电流值
    """
    df2 = indf[start_index:]
    df3 = df2[(df2[CURRENT_HEADER] >= -0.5) & (df2[CURRENT_HEADER] <= 12)]
    # 如果 df3 长度为 0，直接返回空 DataFrame，在后续计算中不产生效果，
    # 所以不需要这里专门标记为 组串无有效值

    pd.options.mode.chained_assignment = None
    df3[CURRENT_HEADER] = df3[CURRENT_HEADER].apply(lambda x: x if (x > 0.1) else 0)
    return df3[[CURRENT_HEADER, V_HEADER]]  # 去掉重复的 组串编号 列

def mark_unit(powers: pd.Series, str_info: dict, d_1st: float=None) -> Tuple[float, dict]:
    """根据一个计算单元中各个组串的平均功率计算 d 值和组串异常状态
    """
    for str_no in str_info:
        powers[str_no] = np.nan
    if powers.dropna().empty:
        return (d_1st, str_info)
    p_u = np.mean(powers)
    d = np.sqrt(sum((powers - p_u).pow(2).dropna())) / p_u
    d_1st = d if d_1st is None else d_1st

    if d <= 0.05:
        return (d_1st, str_info)
    elif (d > 0.05) and (d <= 0.1):
        str_info[powers.idxmin()] = 'E3'
    elif (d > 0.1) and (d <= 0.2):
        str_info[powers.idxmin()] = 'E2'
    elif d > 0.2:
        str_info[powers.idxmin()] = 'E1'

    return mark_unit(powers, str_info, d_1st)

def unit_deviation(unit: pd.DataFrame) -> pd.DataFrame:
    """计算计算单元的离散度和组串状态
    """
    logging.debug(f'Process unit:\n{unit.head(1)}')
    unit2 = unit[[CURRENT_HEADER, V_HEADER, STR_HEADER]]
    # step 1
    pivt = pd.pivot_table(unit2, values=[CURRENT_HEADER,V_HEADER], index=[TIME_HEADER], columns=[STR_HEADER])
    if pivt.rolling(int(TF / SAMPLE_INT)).std().min().sum() == 0:
        return COM_FAIL_DF
    # step 2
    str_grps = unit2.groupby(STR_HEADER)
    if str_grps.apply(lambda x: max(x[CURRENT_HEADER]) < I0).all():
        return UNIT_INVALID_DF
    # step 3
    start = get_valid_zone(pivt, I0)
    logging.debug(f'Unit start time: {start}')
    filtered_data = str_grps.apply(filter_window, (start)) # apply 的函数参数必须是 tuple
    # step 4
    str_avg = filtered_data.groupby(STR_HEADER).mean()
    if str_avg[CURRENT_HEADER].mean() < I0:
        return UNIT_INVALID_DF
    # step 5
    p_s = str_avg.apply(np.prod, axis=1)
    # step 6 ~ 9
    d, str_info = mark_unit(p_s, {})
    # step 10
    return pd.DataFrame({DEVIATION: [d], STR_STATUS: [json.dumps(str_info, separators=('|', ':'))]},
                        index=pd.Index([END_TIME]))

def main(inf, outf, train):
    inp = pd.read_csv(inputFilePath,
                      usecols=[0, 1, 2, 3, 4],
                      dtype={STATION_ID: str, DEVICE_ID: str, CURRENT_HEADER: float, V_HEADER: float},
                      index_col=2,
                      parse_dates = [TIME_HEADER]).dropna()

    inp = inp.between_time(START_TIME, END_TIME)
    inp[GRP_ID] = inp[STATION_ID] + inp[DEVICE_ID].apply(lambda x: x[:6])  # 设备标识是一个8位数字组成的字符串，最后两位是组串编号
    inp[STR_HEADER] = inp[DEVICE_ID].apply(lambda x: x[6:])
    unit_grps = inp.groupby(GRP_ID)
    res = unit_grps.apply(unit_deviation)

    if isTrain == 'true':
        fh = open(outf, 'w', encoding='utf-8')
        fh.write("this is train's result")
        fh.close()
    else:
        r2 = res.reset_index(level=GRP_ID)
        res.to_csv(outf, encoding='utf-8', index=True, quoting=csv.QUOTE_NONE)
        # res.to_csv(outf, encoding='utf-8', index=False,
                   # columns=[STATION_ID, DEVICE_ID, REPORT_TIME, 'ALERT_TYPE',  'ALERT_LEVEL', JSON_MAT,
                            # 'IS_PROCESSED', COEF_COL], quoting=csv.QUOTE_NONE, escapechar='|')

if __name__ == '__main__':
    main(inputFilePath, outputFilePath, isTrain)