02131024

data_processing/data_operation/data_handler.py

@@ -59,7 +59,12 @@ class DataHandler(object):
         feature_data = norm_data.reset_index(drop=True)
         time_step_loc = time_step - 1
         iters = int(len(feature_data)) // self.opt.Model['time_step']
+        end = int(len(feature_data)) % self.opt.Model['time_step']
         features_x = np.array([feature_data.loc[i*time_step:i*time_step + time_step_loc, features].reset_index(drop=True) for i in range(iters)])
+        if end > 0:
+            df = feature_data.tail(end)
+            df_repeated = pd.concat([df] + [pd.DataFrame([df.iloc[0]]* (time_step-end))]).reset_index(drop=True)
+            features_x = np.concatenate((features_x, np.expand_dims(df_repeated, 0)), axis=0)
         return features_x
 
     def get_timestep_features(self, norm_data, col_time, features, target, is_train):

models_processing/model_koi/tf_bp_pre.py

@@ -45,13 +45,25 @@ def model_prediction_bp():
     try:
         print('args', args)
         logger.info(args)
-        predict_data = get_data_from_mongo(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(predict_data, feature_scaler, args)
+        scaled_pre_x = dh.pre_data_handler(pre_data, feature_scaler, args)
         bp.get_model(args)
         # result = bp.predict(scaled_pre_x, args)
         result = list(chain.from_iterable(target_scaler.inverse_transform([bp.predict(scaled_pre_x).flatten()])))
-        insert_data_into_mongo(result, args)
+        pre_data['power_forecast'] = result[: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 = 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)
         success = 1
     except Exception as e:
         my_exception = traceback.format_exc()
@@ -75,8 +87,10 @@ if __name__ == "__main__":
 
     # serve(app, host="0.0.0.0", port=1010x, 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_test', 'col_time': 'date_time',
+                 'model_table': 'j00083_model', 'mongodb_read_table': 'j00083_test', 'col_time': 'date_time', 'mongodb_write_table': 'j00083_rs',
                  '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)
@@ -90,8 +104,17 @@ if __name__ == "__main__":
     pre_x = dh.pre_data_handler(pre_data, feature_scaler, opt)
     bp.get_model(arguments)
     result = bp.predict(pre_x)
+    result1 = list(chain.from_iterable(target_scaler.inverse_transform([result.flatten()])))
+    pre_data['power_forecast'] = result1[: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 = 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
 
-    args_dict['gen_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
-    args_dict['params'] = arguments
-    args_dict['descr'] = '测试'
-    insert_data_into_mongo(result, arguments)
+    insert_data_into_mongo(pre_data, arguments)