wgz_forecast/run.py

from flask import request, Flask, jsonify
from pv.pv_inference import load_model, pv_forecast
from carbon.carbon_inference import load_carbon_model,carbon_forecast
from water.water_inference import load_water_model,water_forecast
from load.load_inference import load_load_model,load_forecast
from logzero import logger
import numpy as np
app = Flask(__name__)
pv_model_gbm = load_model('./pv/models/pv_pred.joblib')
carbon_model_gbm = load_carbon_model('./carbon/models/carbon_pred.joblib')
water_model_gbm = load_water_model('./water/models/water_pred.joblib')
load_model_gbm = load_load_model('./load/models/load_pred.joblib')


@app.route('/pv', methods=['POST'])
def run_pv_forecast():
    try:
        data = request.json
        if not data or 'inputs' not in data:
            return jsonify({"error": "Invalid data"}), 400
        inputs = data.get('inputs')
        # 转为1维数组
        inputs = np.array(inputs)
        if inputs.ndim != 1:
            inputs = inputs.flatten()
        logger.info(f"pv history inputs: {inputs}")
        # 调用预测函数
        out = pv_forecast(inputs, pv_model_gbm)
        # 将 NumPy 数组转换为列表
        results = {"result": out.tolist()}  # 使用 .tolist() 转换为列表
        return jsonify(results), 200
    except Exception as e:
        logger.error(f"Error in pv_forecast: {e}", exc_info=True)
        return jsonify({"error": str(e)}), 500
    
@app.route('/carbon', methods=['POST'])
def run_carbon_forecast():

    try:
        data = request.json
        if not data or 'inputs' not in data:
            return jsonify({"error": "Invalid data"}), 400
        inputs = data.get('inputs')
        # 转为1维数组
        inputs = np.array(inputs)
        if inputs.ndim != 1:
            inputs = inputs.flatten()
        logger.info(f"carbon history inputs: {inputs}")
        # 调用预测函数
        out = carbon_forecast(inputs, carbon_model_gbm)
        # 将 NumPy 数组转换为列表
        results = {"result": out.tolist()}  # 使用 .tolist() 转换为列表
        return jsonify(results), 200
    except Exception as e:
        logger.error(f"Error in carbon_forecast: {e}", exc_info=True)
        return jsonify({"error": str(e)}), 500

@app.route('/water', methods=['POST'])
def run_water_forecast():

    try:
        data = request.json
        if not data or 'inputs' not in data:
            return jsonify({"error": "Invalid data"}), 400
        inputs = data.get('inputs')
        # 转为1维数组
        inputs = np.array(inputs)
        if inputs.ndim != 1:
            inputs = inputs.flatten()
        logger.info(f"water history inputs: {inputs}")
        # 调用预测函数
        out = water_forecast(inputs, water_model_gbm)
        # 将 NumPy 数组转换为列表
        results = {"result": out.tolist()}  # 使用 .tolist() 转换为列表
        return jsonify(results), 200
    except Exception as e:
        logger.error(f"Error in water_forecast: {e}", exc_info=True)
        return jsonify({"error": str(e)}), 500
@app.route('/load', methods=['POST'])
def run_load_forecast():

    try:
        data = request.json
        if not data or 'inputs' not in data:
            return jsonify({"error": "Invalid data"}), 400
        inputs = data.get('inputs')
        # 转为1维数组
        inputs = np.array(inputs)
        if inputs.ndim != 1:
            inputs = inputs.flatten()
        logger.info(f"load history inputs: {inputs}")
        # 调用预测函数
        out = load_forecast(inputs, load_model_gbm)
        # 将 NumPy 数组转换为列表
        results = {"result": out.tolist()}  # 使用 .tolist() 转换为列表
        return jsonify(results), 200
    except Exception as e:
        logger.error(f"Error in load_forecast: {e}", exc_info=True)
        return jsonify({"error": str(e)}), 500

if __name__=='__main__':
    app.run(host='0.0.0.0', port='2467')