02201530

evaluation_processing/analysis_cdq.py

@@ -157,6 +157,10 @@ def put_analysis_report_to_html(args, df_predict, df_accuracy):
             agg_dict['deviationAssessment'] = [np.nanmean, np.nansum]
             rename_cols.append('考核分数平均值')
             rename_cols.append('考核总分数')
+        if 'accuracyAssessment' in df_accuracy.columns:
+            agg_dict['deviationAssessment'] = [np.nanmean, np.nansum]
+            rename_cols.append('考核分数平均值')
+            rename_cols.append('考核总分数')
         # 进行分组聚合，如果有需要聚合的列
         summary_df = df_accuracy.groupby('model').agg(agg_dict).reset_index()
         summary_df.columns = rename_cols

models_processing/losses/loss_cdq.py

@@ -31,8 +31,8 @@ class SouthLoss(tf.keras.losses.Loss):
         :return: 损失值
         """
         # 计算实际和预测的差值
-        y_true = y_true * self.opt.std['C_REAL_VALUE'] + self.opt.mean['C_REAL_VALUE']
-        y_predict = y_predict * self.opt.std['C_REAL_VALUE'] + self.opt.mean['C_REAL_VALUE']
+        # y_true = y_true * self.opt.std['C_REAL_VALUE'] + self.opt.mean['C_REAL_VALUE']
+        # y_predict = y_predict * self.opt.std['C_REAL_VALUE'] + self.opt.mean['C_REAL_VALUE']
         y_true = y_true[:, 15]
         y_predict = y_predict[:, 15]
         diff = y_true - y_predict
@@ -80,8 +80,8 @@ class NorthEastLoss(tf.keras.losses.Loss):
 
     def call(self, y_true, y_predict):
         # 这里我们添加了一个小的 epsilon 值来避免除以 0
-        y_true = y_true * self.opt.std['C_REAL_VALUE'] + self.opt.mean['C_REAL_VALUE']
-        y_predict = y_predict * self.opt.std['C_REAL_VALUE'] + self.opt.mean['C_REAL_VALUE']
+        # y_true = y_true * self.opt.std['C_REAL_VALUE'] + self.opt.mean['C_REAL_VALUE']
+        # y_predict = y_predict * self.opt.std['C_REAL_VALUE'] + self.opt.mean['C_REAL_VALUE']
 
         mask_logical = tf.logical_and(tf.greater(y_true, self.cap), tf.greater(y_predict, self.cap))
         # mask = tf.cast(~mask_logical, tf.float32)

models_processing/model_koi/tf_lstm.py

@@ -5,35 +5,22 @@
 # @Author    :David
 # @Company: shenyang JY
 
-import os.path
-from keras.layers import Input, Dense, LSTM, concatenate, Conv1D, Conv2D, MaxPooling1D, BatchNormalization, Flatten, Dropout
-from keras.models import Model, load_model
-from keras.callbacks import ModelCheckpoint, EarlyStopping, TensorBoard
-from keras import optimizers, regularizers
-import keras.backend as K
-from common.database_dml import *
+from tensorflow.keras.layers import Input, Dense, LSTM, concatenate, Conv1D, Conv2D, MaxPooling1D, Reshape, Flatten
+from tensorflow.keras.models import Model, load_model
+from tensorflow.keras.callbacks import ModelCheckpoint, EarlyStopping, TensorBoard, ReduceLROnPlateau
+from tensorflow.keras import optimizers, regularizers
+from models_processing.losses.loss_cdq import rmse
 import numpy as np
-from sqlalchemy.ext.instrumentation import find_native_user_instrumentation_hook
-np.random.seed(42)
-from models_processing.losses.loss_cdq import SouthLoss, NorthEastLoss
-import tensorflow as tf
-tf.compat.v1.set_random_seed(1234)
+from common.database_dml import *
 from threading import Lock
+import argparse
 model_lock = Lock()
 
-def rmse(y_true, y_pred):
-    return K.sqrt(K.mean(K.square(y_pred - y_true)))
-
-
-var_dir = os.path.dirname(os.path.dirname(__file__))
-
 
 class TSHandler(object):
-    model = None
-    train = False
-
-    def __init__(self, logger):
+    def __init__(self, logger, args):
         self.logger = logger
+        self.opt = argparse.Namespace(**args)
         self.model = None
 
     def get_model(self, args):
@@ -49,41 +36,42 @@ class TSHandler(object):
     @staticmethod
     def get_keras_model(opt):
         # db_loss = NorthEastLoss(opt)
-        south_loss = SouthLoss(opt)
+        # south_loss = SouthLoss(opt)
         l1_reg = regularizers.l1(opt.Model['lambda_value_1'])
         l2_reg = regularizers.l2(opt.Model['lambda_value_2'])
-        nwp_input = Input(shape=(opt.Model['time_step'], opt.Model['input_size_nwp']), name='nwp')
+        nwp_input = Input(shape=(opt.Model['time_step'], opt.Model['input_size']), name='nwp')
 
         con1 = Conv1D(filters=64, kernel_size=5, strides=1, padding='valid', activation='relu', kernel_regularizer=l2_reg)(nwp_input)
-        nwp = MaxPooling1D(pool_size=5, strides=1, padding='valid', data_format='channels_last')(con1)
-        nwp_lstm = LSTM(units=opt.Model['hidden_size'], return_sequences=False, kernel_regularizer=l2_reg)(nwp)
+        con1_p = MaxPooling1D(pool_size=5, strides=1, padding='valid', data_format='channels_last')(con1)
+        nwp_lstm = LSTM(units=opt.Model['hidden_size'], return_sequences=False, kernel_regularizer=l2_reg)(nwp_input)
 
         output = Dense(opt.Model['output_size'], name='cdq_output')(nwp_lstm)
 
         model = Model(nwp_input, output)
         adam = optimizers.Adam(learning_rate=0.001, beta_1=0.9, beta_2=0.999, epsilon=1e-7, amsgrad=True)
-        model.compile(loss=south_loss, optimizer=adam)
+        model.compile(loss=rmse, optimizer=adam)
         return model
 
-    def train_init(self, opt):
+    def train_init(self):
         try:
-            if opt.Model['add_train']:
+            if self.opt.Model['add_train']:
                 # 进行加强训练，支持修模
-                base_train_model = get_h5_model_from_mongo(vars(opt), {'rmse': rmse})
+                base_train_model = get_h5_model_from_mongo(vars(self.opt), {'rmse': rmse})
                 base_train_model.summary()
                 self.logger.info("已加载加强训练基础模型")
             else:
-                base_train_model = self.get_keras_model(opt)
+                base_train_model = self.get_keras_model(self.opt)
             return base_train_model
         except Exception as e:
             self.logger.info("加强训练加载模型权重失败:{}".format(e.args))
 
-    def training(self, opt, train_and_valid_data):
-        model = self.train_init(opt)
+    def training(self, train_and_valid_data):
+        model = self.train_init()
         model.summary()
         train_x, train_y, valid_x, valid_y = train_and_valid_data
-        early_stop = EarlyStopping(monitor='val_loss', patience=opt.Model['patience'], mode='auto')
-        history = model.fit(train_x, train_y, batch_size=opt.Model['batch_size'], epochs=opt.Model['epoch'], verbose=2, validation_data=(valid_x, valid_y), callbacks=[early_stop])
+        early_stop = EarlyStopping(monitor='val_loss', patience=self.opt.Model['patience'], mode='auto')
+        history = model.fit(train_x, train_y, batch_size=self.opt.Model['batch_size'], epochs=self.opt.Model['epoch'],
+                            verbose=2, validation_data=(valid_x, valid_y), callbacks=[early_stop], shuffle=False)
         loss = np.round(history.history['loss'], decimals=5)
         val_loss = np.round(history.history['val_loss'], decimals=5)
         self.logger.info("-----模型训练经过{}轮迭代-----".format(len(loss)))
@@ -91,8 +79,8 @@ class TSHandler(object):
         self.logger.info("验证集损失函数为：{}".format(val_loss))
         return model
 
-    def predict(self, test_X, batch_size=1):
-        result = TSHandler.model.predict(test_X, batch_size=batch_size)
+    def predict(self, test_x, batch_size=1):
+        result = self.model.predict(test_x, batch_size=batch_size)
         self.logger.info("执行预测方法")
         return result
 

models_processing/model_koi/tf_lstm_pre.py

@@ -25,44 +25,50 @@ app = Flask('tf_lstm_pre——service')
 
 with app.app_context():
     with open('../model_koi/bp.yaml', 'r', encoding='utf-8') as f:
-        arguments = yaml.safe_load(f)
+        args = yaml.safe_load(f)
 
-    dh = DataHandler(logger, arguments)
-    ts = TSHandler(logger)
+    dh = DataHandler(logger, args)
+    ts = TSHandler(logger, args)
+    global opt
 
+@app.before_request
+def update_config():
+    # ------------ 整理参数，整合请求参数 ------------
+    args_dict = request.values.to_dict()
+    args_dict['features'] = args_dict['features'].split(',')
+    args.update(args_dict)
+    opt = argparse.Namespace(**args)
+    dh.opt = opt
+    ts.opt = opt
+    logger.info(args)
 
-@app.route('/nn_bp_predict', methods=['POST'])
+@app.route('/nn_lstm_predict', methods=['POST'])
 def model_prediction_bp():
     # 获取程序开始时间
     start_time = time.time()
     result = {}
     success = 0
     print("Program starts execution!")
-    params_dict = request.values.to_dict()
-    args = arguments.deepcopy()
-    args.update(params_dict)
     try:
-        print('args', args)
-        logger.info(args)
         pre_data = get_data_from_mongo(args)
         feature_scaler, target_scaler = get_scaler_model_from_mongo(args)
         scaled_pre_x = dh.pre_data_handler(pre_data, feature_scaler, args)
         ts.get_model(args)
         # result = bp.predict(scaled_pre_x, args)
-        result = list(chain.from_iterable(target_scaler.inverse_transform([ts.predict(scaled_pre_x).flatten()])))
-        pre_data['power_forecast'] = result[:len(pre_data)]
+        res = list(chain.from_iterable(target_scaler.inverse_transform([ts.predict(scaled_pre_x).flatten()])))
+        pre_data['power_forecast'] = res[:len(pre_data)]
         pre_data['farm_id'] = 'J00083'
         pre_data['cdq'] = 1
         pre_data['dq'] = 1
         pre_data['zq'] = 1
-        pre_data.rename(columns={arguments['col_time']: 'date_time'}, inplace=True)
+        pre_data.rename(columns={args['col_time']: 'date_time'}, inplace=True)
         pre_data = pre_data[['date_time', 'power_forecast', 'farm_id', 'cdq', 'dq', 'zq']]
 
         pre_data['power_forecast'] = pre_data['power_forecast'].round(2)
         pre_data.loc[pre_data['power_forecast'] > opt.cap, 'power_forecast'] = opt.cap
         pre_data.loc[pre_data['power_forecast'] < 0, 'power_forecast'] = 0
 
-        insert_data_into_mongo(pre_data, arguments)
+        insert_data_into_mongo(pre_data, args)
         success = 1
     except Exception as e:
         my_exception = traceback.format_exc()

models_processing/model_koi/tf_lstm_train.py

@@ -12,6 +12,8 @@ import logging, argparse
 from data_processing.data_operation.data_handler import DataHandler
 import time, yaml
 from models_processing.model_koi.tf_lstm import TSHandler
+from models_processing.model_koi.tf_cnn import CNNHandler
+
 from common.database_dml import *
 import matplotlib.pyplot as plt
 from common.logs import Log
@@ -23,30 +25,44 @@ app = Flask('tf_lstm_train——service')
 
 with app.app_context():
     with open('../model_koi/lstm.yaml', 'r', encoding='utf-8') as f:
-        arguments = yaml.safe_load(f)
+        args = yaml.safe_load(f)
+
+    dh = DataHandler(logger, args)
+    ts = TSHandler(logger, args)
+    # ts = CNNHandler(logger, args)
+    global opt
 
-    dh = DataHandler(logger, arguments)
-    ts = TSHandler(logger)
+@app.before_request
+def update_config():
+    # ------------ 整理参数，整合请求参数 ------------
+    args_dict = request.values.to_dict()
+    args_dict['features'] = args_dict['features'].split(',')
+    args.update(args_dict)
+    opt = argparse.Namespace(**args)
+    dh.opt = opt
+    ts.opt = opt
+    logger.info(args)
 
-@app.route('/nn_bp_training', methods=['POST'])
+@app.route('/nn_lstm_training', methods=['POST'])
 def model_training_bp():
     # 获取程序开始时间
     start_time = time.time()
     result = {}
     success = 0
     print("Program starts execution!")
-    args_dict = request.values.to_dict()
-    args = arguments.deepcopy()
-    args.update(args_dict)
     try:
-        opt = argparse.Namespace(**args)
-        logger.info(args_dict)
-        train_data = get_data_from_mongo(args_dict)
-        train_x, valid_x, train_y, valid_y, scaled_train_bytes, scaled_target_bytes = dh.train_data_handler(train_data, opt)
-        bp_model = ts.training(opt, [train_x, valid_x, train_y, valid_y])
-        args_dict['params'] = json.dumps(args)
-        args_dict['gen_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
-        insert_trained_model_into_mongo(bp_model, args_dict)
+        # ------------ 获取数据，预处理训练数据 ------------
+        train_data = get_data_from_mongo(args)
+        train_x, valid_x, train_y, valid_y, scaled_train_bytes, scaled_target_bytes = dh.train_data_handler(train_data)
+        # ------------ 训练模型，保存模型 ------------
+        ts.opt.Model['input_size'] = train_x.shape[2]
+        ts_model = ts.training([train_x, valid_x, train_y, valid_y])
+
+        args['params'] = json.dumps(args)
+        args['descr'] = '测试'
+        args['gen_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
+
+        insert_trained_model_into_mongo(ts_model, args)
         insert_scaler_model_into_mongo(scaled_train_bytes, scaled_target_bytes, args)
         success = 1
     except Exception as e:
@@ -69,23 +85,23 @@ if __name__ == "__main__":
     logger = logging.getLogger("model_training_bp log")
     from waitress import serve
 
-    # serve(app, host="0.0.0.0", port=10103, threads=4)
+    serve(app, host="0.0.0.0", port=10103, threads=4)
     print("server start!")
-    args_dict = {"mongodb_database": 'david_test', 'scaler_table': 'j00083_scaler', 'model_name': 'bp1.0.test',
-    'model_table': 'j00083_model', 'mongodb_read_table': 'j00083', 'col_time': 'dateTime',
-    'features': 'speed10,direction10,speed30,direction30,speed50,direction50,speed70,direction70,speed90,direction90,speed110,direction110,speed150,direction150,speed170,direction170'}
-    args_dict['features'] = args_dict['features'].split(',')
-    arguments.update(args_dict)
-    dh = DataHandler(logger, arguments)
-    ts = TSHandler(logger)
-    opt = argparse.Namespace(**arguments)
-    opt.Model['input_size'] = len(opt.features)
-    train_data = get_data_from_mongo(args_dict)
-    train_x, valid_x, train_y, valid_y, scaled_train_bytes, scaled_target_bytes = dh.train_data_handler(train_data, opt)
-    ts_model = ts.training(opt, [train_x, train_y, valid_x, valid_y])
-
-    args_dict['gen_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
-    args_dict['params'] = arguments
-    args_dict['descr'] = '测试'
-    insert_trained_model_into_mongo(ts_model, args_dict)
-    insert_scaler_model_into_mongo(scaled_train_bytes, scaled_target_bytes, args_dict)
+    # args_dict = {"mongodb_database": 'david_test', 'scaler_table': 'j00083_scaler', 'model_name': 'bp1.0.test',
+    # 'model_table': 'j00083_model', 'mongodb_read_table': 'j00083', 'col_time': 'dateTime',
+    # 'features': 'speed10,direction10,speed30,direction30,speed50,direction50,speed70,direction70,speed90,direction90,speed110,direction110,speed150,direction150,speed170,direction170'}
+    # args_dict['features'] = args_dict['features'].split(',')
+    # arguments.update(args_dict)
+    # dh = DataHandler(logger, arguments)
+    # ts = TSHandler(logger)
+    # opt = argparse.Namespace(**arguments)
+    # opt.Model['input_size'] = len(opt.features)
+    # train_data = get_data_from_mongo(args_dict)
+    # train_x, valid_x, train_y, valid_y, scaled_train_bytes, scaled_target_bytes = dh.train_data_handler(train_data, opt)
+    # ts_model = ts.training(opt, [train_x, train_y, valid_x, valid_y])
+    #
+    # args_dict['gen_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
+    # args_dict['params'] = arguments
+    # args_dict['descr'] = '测试'
+    # insert_trained_model_into_mongo(ts_model, args_dict)
+    # insert_scaler_model_into_mongo(scaled_train_bytes, scaled_target_bytes, args_dict)