pv-current/deviation/devi_spark.py

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField
from pyspark.sql.functions import (pandas_udf, concat,substring,
        PandasUDFType, col, hour)
from pyspark.sql.types import IntegerType, StringType, DoubleType, TimestampType
import logging
import sys
import pandas as pd
import numpy as np
import subprocess
import csv
import json
from typing import Tuple
"""
 * @author Leo
 * 2019.2.21
 *
 * @Description: 光伏组串离散率算子的 Spark 实现
 *
 * 实现标准输入格式到算子输入的转换
 * 保证不论输入的是一个计算单元，还是多个计算单元，不影响算子实现

常量定义：

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

START_TIME = 7
END_TIME = 17
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"
# inputFilePath = "/user/hdfs/pvtest/deviation1122.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)

unitSchema = StructType([
     StructField(DEVIATION, DoubleType(), True)
    ,StructField(STR_STATUS, StringType(), True)
    ])
@pandas_udf(unitSchema, PandasUDFType.GROUPED_MAP)
def unit_deviation(unit: pd.DataFrame) -> pd.DataFrame:
    """计算计算单元的离散度和组串状态
    """
    logging.debug(f'Process unit:\n{unit.head(1)}')
    logging.debug(f'Process unit:\n{unit.info()}')
    unit2 = unit[[TIME_HEADER, 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):
    spark = SparkSession.builder.master("local[*]").appName('pv-deviation').getOrCreate()
    dfSchema = StructType([
       StructField(STATION_ID, StringType(), True)
      ,StructField(DEVICE_ID, StringType(), True)
      ,StructField(TIME_HEADER, TimestampType(), True)
      ,StructField(V_HEADER, DoubleType(), True)
      ,StructField(CURRENT_HEADER, DoubleType(), True)
      ])
    df = spark.read.format("csv").option("header", "true").schema(dfSchema).load(inputFilePath)
    df2 = df.filter((hour(col(TIME_HEADER)) >= START_TIME) & (hour(col(TIME_HEADER)) <= END_TIME))
    df3 = df2.withColumn(GRP_ID, substring(col(DEVICE_ID), 0, 6)).withColumn(
                             GRP_ID, concat(col(STATION_ID), col(GRP_ID)))
    df4 = df3.withColumn(STR_HEADER, substring(col(DEVICE_ID), 7, 2))
    unit_grps = df4.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.toPandas().reset_index(level=GRP_ID)
        r2 = res.toPandas().reset_index()
        r2.to_csv(outf, encoding='utf-8', index=True, quoting=csv.QUOTE_NONE)

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