#!/usr/bin/env python # -*- coding:utf-8 -*- # @FileName :tf_lstm_pre.py # @Time :2025/2/13 10:52 # @Author :David # @Company: shenyang JY import os.path import numpy as np import logging, argparse, traceback from app.common.data_handler import DataHandler, write_number_to_file from threading import Lock import time, json model_lock = Lock() from itertools import chain from app.common.logs import logger, params from app.common.tf_cnn import CNNHandler from app.common.dbmg import MongoUtils np.random.seed(42) # NumPy随机种子 dh = DataHandler(logger, params) cnn = CNNHandler(logger, params) mgUtils = MongoUtils(logger) def model_prediction(pre_data, input_file, cap): # 获取程序开始时间 start_time = time.time() success = 0 print("Program starts execution!") farm_id = input_file.split('/')[-2] output_file = input_file.replace('IN', 'OUT') file = 'DQYC_OUT_PREDICT_POWER.txt' status_file = 'STATUS.TXT' try: params['model_table'] += farm_id params['scaler_table'] += farm_id feature_scaler, target_scaler = mgUtils.get_scaler_model_from_mongo(params) cnn.opt.cap = round(target_scaler.transform(np.array([[cap]]))[0, 0], 2) cnn.get_model(params) dh.opt.features = json.loads(cnn.model_params).get('Model').get('features', ','.join(cnn.opt.features)).split(',') scaled_pre_x, pre_data = dh.pre_data_handler(pre_data, feature_scaler) success = 1 # 更新算法状态:1. 启动成功 write_number_to_file(os.path.join(output_file, status_file), 1, 1, 'rewrite') logger.info("算法启动成功") res = list(chain.from_iterable(target_scaler.inverse_transform([cnn.predict(scaled_pre_x).flatten()]))) pre_data['Power'] = res[:len(pre_data)] pre_data['PlantID'] = farm_id pre_data = pre_data[['PlantID', params['col_time'], 'Power']] pre_data.loc[:, 'Power'] = pre_data['Power'].round(2) pre_data.loc[pre_data['Power'] > cap, 'Power'] = cap pre_data.loc[pre_data['Power'] < 0, 'Power'] = 0 pre_data.to_csv(os.path.join(output_file, file), sep=' ', index=False) # 更新算法状态:正常结束 write_number_to_file(os.path.join(output_file, status_file), 2, 2) logger.info("算法正常结束") except Exception as e: # 如果算法状态没启动,不更新 if success: write_number_to_file(os.path.join(output_file, status_file), 2, 3) my_exception = traceback.format_exc() my_exception.replace("\n", "\t") result['msg'] = my_exception logger.info("算法状态异常:{}".format(my_exception)) end_time = time.time() logger.info("cnn预测任务:用了 %s 秒 " % (end_time - start_time)) if __name__ == "__main__": print("Program starts execution!") logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger("model_training_bp log") # 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', '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) dh = DataHandler(logger, arguments) ts = TSHandler(logger) opt = argparse.Namespace(**arguments) opt.Model['input_size'] = len(opt.features) pre_data = get_data_from_mongo(args_dict) feature_scaler, target_scaler = get_scaler_model_from_mongo(arguments) pre_x = dh.pre_data_handler(pre_data, feature_scaler, opt) ts.get_model(arguments) result = ts.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 insert_data_into_mongo(pre_data, arguments)