algorithm_platform/models_processing/model_tf/tf_lstm3_pre.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# @FileName  :tf_lstm_pre.py
# @Time      :2025/2/13 10:52
# @Author    :David
# @Company: shenyang JY
import json, copy
import numpy as np
from flask import Flask, request, g
import logging, argparse, traceback
from common.database_dml_koi import *
from common.processing_data_common import missing_features, str_to_list
from data_processing.data_operation.data_handler import DataHandler
from threading import Lock
import time, yaml
from copy import deepcopy
model_lock = Lock()
from itertools import chain
from common.logs import Log
from common.data_utils import deep_update
logger = Log('tf_ts3').logger
np.random.seed(42)  # NumPy随机种子
# tf.set_random_seed(42)  # TensorFlow随机种子
app = Flask('tf_lstm3_pre——service')

current_dir = os.path.dirname(os.path.abspath(__file__))
with open(os.path.join(current_dir, 'lstm.yaml'), 'r', encoding='utf-8') as f:
    global_config = yaml.safe_load(f)  # 只读的全局配置

@app.before_request
def update_config():
    # ------------ 整理参数，整合请求参数 ------------
    # 深拷贝全局配置 + 合并请求参数
    current_config = deepcopy(global_config)
    request_args = request.values.to_dict()
    # features参数规则：1.有传入，解析，覆盖 2. 无传入，不覆盖，原始值
    request_args['features'] = request_args['features'].split(',') if 'features' in request_args else current_config['features']
    request_args['time_series'] = request_args.get('time_series', 3)
    request_args['lstm_type'] = request_args.get('lstm_type', 1)
    current_config = deep_update(current_config, request_args)
    # 存储到请求上下文
    g.opt = argparse.Namespace(**current_config)
    g.dh = DataHandler(logger, current_config)  # 每个请求独立实例
    if int(request_args['lstm_type']) == 1:
        from models_processing.model_tf.tf_bilstm import TSHandler
        g.ts = TSHandler(logger, current_config)
    else:
        from models_processing.model_tf.tf_bilstm_2 import TSHandler2
        g.ts = TSHandler2(logger, current_config)

@app.route('/tf_lstm3_predict', methods=['POST'])
def model_prediction_lstm3():
    # 获取程序开始时间
    start_time = time.time()
    result = {}
    success = 0
    dh = g.dh
    ts = g.ts
    args = deepcopy(g.opt.__dict__)
    logger.info("Program starts execution!")
    try:
        pre_data = get_data_from_mongo(args)
        if args.get('algorithm_test', 0):
            field_mapping = {'clearsky_ghi': 'clearskyGhi', 'dni_calcd': 'dniCalcd','surface_pressure': 'surfacePressure'}
            pre_data = pre_data.rename(columns=field_mapping)
        feature_scaler, target_scaler = get_scaler_model_from_mongo(args)
        ts.opt.cap = round(target_scaler.transform(np.array([[float(args['cap'])]]))[0, 0], 2)
        ts.get_model(args)
        dh.opt.features = json.loads(ts.model_params)['Model']['features'].split(',')
        scaled_pre_x, pre_data = dh.pre_data_handler(pre_data, feature_scaler, time_series=args['time_series'], lstm_type=3)
        res = list(chain.from_iterable(target_scaler.inverse_transform(ts.predict(scaled_pre_x))))
        pre_data['farm_id'] = args.get('farm_id', 'null')
        if int(args.get('algorithm_test', 0)):
            pre_data = pre_data.iloc[dh.opt.Model["time_step"]:]
            pre_data[args['model_name']] = res[:len(pre_data)]
            pre_data.rename(columns={args['col_time']: 'dateTime'}, inplace=True)
            pre_data = pre_data[['dateTime', 'farm_id', args['target'], args['model_name'], 'dq']]
            pre_data = pre_data.melt(id_vars=['dateTime', 'farm_id', args['target']], var_name='model', value_name='power_forecast')
            res_cols = ['dateTime', 'power_forecast', 'farm_id', args['target'], 'model']
            if 'howLongAgo' in args:
                pre_data['howLongAgo'] = int(args['howLongAgo'])
                res_cols += ['howLongAgo']
        else:
            pre_data = pre_data.iloc[dh.opt.Model["time_step"]:]
            pre_data['power_forecast'] = res[:len(pre_data)]
            pre_data.rename(columns={args['col_time']: 'date_time'}, inplace=True)
            res_cols = ['date_time', 'power_forecast', 'farm_id']
        pre_data = pre_data[res_cols]

        pre_data.loc[:, 'power_forecast'] = pre_data['power_forecast'].round(2)
        pre_data.loc[pre_data['power_forecast'] > float(args['cap']), 'power_forecast'] = float(args['cap'])
        pre_data.loc[pre_data['power_forecast'] < 0, 'power_forecast'] = 0

        insert_data_into_mongo(pre_data, args)
        success = 1
    except Exception as e:
        my_exception = traceback.format_exc()
        my_exception.replace("\n", "\t")
        result['msg'] = my_exception
    end_time = time.time()

    result['success'] = success
    result['args'] = args
    result['start_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(start_time))
    result['end_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(end_time))
    print("Program execution ends!")
    return result


if __name__ == "__main__":
    print("Program starts execution!")
    from waitress import serve
    serve(app, host="0.0.0.0", port=10122,
          threads=8,  # 指定线程数（默认4，根据硬件调整）
          channel_timeout=600  # 连接超时时间（秒）
          )
    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)