#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 负荷预测脚本 功能：根据模型名称加载模型并执行预测 """ import requests import argparse import json import os import subprocess from datetime import datetime # 正确的路径计算 SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) SKILL_DIR = os.path.dirname(SCRIPT_DIR) # API基础地址 API_BASE_URL = "http://47.96.16.84:5005" def get_model_list(): """ 获取模型列表及其性能指标 """ url = f"{API_BASE_URL}/api/load-forecast/model/available-with-metrics" try: response = requests.get(url) response.raise_for_status() data = response.json() if data.get('code') == 200: return data['data'] else: print(f"获取模型列表失败: {data.get('error')}") return [] except Exception as e: print(f"获取模型列表异常: {e}") return [] def find_model_by_name(model_name): """ 根据模型名称找到模型信息（模糊匹配） """ # 先获取包含完整信息的模型列表 url = f"{API_BASE_URL}/api/load-forecast/model/list" models_full = [] try: response = requests.get(url) response.raise_for_status() data = response.json() if data.get('code') == 200: models_full = data['data'] except Exception as e: print(f"获取完整模型信息异常: {e}") # 再获取包含性能指标的模型列表 models_with_metrics = get_model_list() # 优先匹配包含模型名称的模型 for model_full in models_full: if model_name.lower() in model_full['name'].lower(): # 查找对应的性能指标 for model_metrics in models_with_metrics: if model_metrics['name'] == model_full['name']: # 合并性能指标到完整模型信息中 model_full.update(model_metrics) return model_full return model_full # 如果没有找到，返回第一个模型 if models_full: first_model = models_full[0] # 查找对应的性能指标 for model_metrics in models_with_metrics: if model_metrics['name'] == first_model['name']: first_model.update(model_metrics) return first_model return first_model # 如果没有完整模型信息，尝试从性能指标列表中返回 if models_with_metrics: for model_metrics in models_with_metrics: if model_name.lower() in model_metrics['name'].lower(): # 添加默认的模型类型 model_metrics['model_type'] = 'cnn_rbilstm' # 添加默认的路径 model_metrics['path'] = '/home/nvidia/workspace/Load_forecasting/dynamic_train/results' return model_metrics # 返回第一个模型 first_model = models_with_metrics[0] first_model['model_type'] = 'cnn_rbilstm' first_model['path'] = '/home/nvidia/workspace/Load_forecasting/dynamic_train/results' return first_model return None def load_model(model_dir, model_type): """ 加载模型 """ url = f"{API_BASE_URL}/api/load-forecast/model/load" data = { "model_dir": model_dir } try: response = requests.post(url, json=data) response.raise_for_status() result = response.json() if result.get('code') == 200: print(f"模型加载成功: {model_type}") return True else: print(f"模型加载失败: {result.get('error')}") return False except Exception as e: print(f"模型加载异常: {e}") return False def predict_7days(backtest=False): """ 7天预测 """ url = f"{API_BASE_URL}/api/load-forecast/predict/7days" data = { "backtest": backtest, "model_type": "lightgbm" } try: response = requests.post(url, json=data) response.raise_for_status() result = response.json() if result.get('code') == 200: return result['data'] else: print(f"7天预测失败: {result.get('error')}") return None except Exception as e: print(f"7天预测异常: {e}") return None def predict_multi(start_time, prediction_hours=24): """ 多步预测 """ url = f"{API_BASE_URL}/api/load-forecast/predict/multi" data = { "start_time": start_time, "prediction_hours": prediction_hours, "model_type": "cnn_rbilstm" } try: response = requests.post(url, json=data) response.raise_for_status() result = response.json() if result.get('code') == 200: return result['data'] else: print(f"多步预测失败: {result.get('error')}") return None except Exception as e: print(f"多步预测异常: {e}") return None def generate_report(predictions, model_name, start_time, prediction_hours): """ 生成预测报告 """ # 准备报告数据 report_data = [] for pred in predictions: report_data.append({ "time": pred['time'], "predicted": pred['power'] }) # 调用step1_select.py选择模板 step1_script = os.path.join(os.path.dirname(SKILL_DIR), "report-generator", "scripts", "step1_select.py") if not os.path.exists(step1_script): raise Exception(f"Step1 script not found at {step1_script}") user_query = f"使用{model_name}模型从{start_time}开始的{prediction_hours}小时负荷预测" step1_cmd = ["python", step1_script, "--query", user_query, "--data", json.dumps(report_data, ensure_ascii=False)] step1_result = subprocess.run(step1_cmd, capture_output=True, text=True, timeout=60) if step1_result.returncode != 0: raise Exception(f"Step 1 failed: {step1_result.stderr}") lines = step1_result.stdout.strip().split('\n') step1_output = None for line in reversed(lines): try: step1_output = json.loads(line) break except: continue if step1_output is None or step1_output.get("status") != "success": raise Exception(f"Template selection failed") template_name = step1_output.get("template_name", "lineChart") # 调用step2_render.py生成报告 step2_script = os.path.join(os.path.dirname(SKILL_DIR), "report-generator", "scripts", "step2_render.py") if not os.path.exists(step2_script): raise Exception(f"Step2 script not found at {step2_script}") step2_cmd = ["python", step2_script, "--query", user_query, "--template", template_name, "--data", json.dumps(report_data, ensure_ascii=False)] step2_result = subprocess.run(step2_cmd, capture_output=True, text=True, timeout=60) if step2_result.returncode != 0: raise Exception(f"Step 2 failed: {step2_result.stderr}") lines = step2_result.stdout.strip().split('\n') step2_output = None for line in reversed(lines): try: step2_output = json.loads(line) break except: continue if step2_output is None or step2_output.get("status") != "success": raise Exception(f"Report generation failed") report_address = step2_output.get('address', '') summary = step2_output.get('summary', '') print("\nGENERATING REPORT") print("=" * 60) print("PROGRESSION: ⏳ 正在分析数据，选择合适的可视化模板...") print(f"Selected template: {template_name}") print("PROGRESSION: ⏳ 正在将数据渲染为图表报告...") print("") print("✓ Report generated successfully!") print(f"Summary: {summary}") print("") print(f"@@@@@{report_address}@@@@@") print("") print("#####%查看详细数据%￥点击上方链接查看完整的数据分析报告￥ %继续深入分析%￥基于当前数据进行更深度的挖掘￥ %导出报告%￥将当前报告导出为PDF或图片格式￥#####") return report_address def main(): parser = argparse.ArgumentParser(description='负荷预测脚本') parser.add_argument('--model_name', type=str, required=True, help='模型名称') parser.add_argument('--start_time', type=str, default="2025-03-27 22:17:00", help='预测起始时间') parser.add_argument('--prediction_hours', type=int, default=24, help='预测小时数') args = parser.parse_args() print(f"Step 1: 查找模型 {args.model_name}...") model = find_model_by_name(args.model_name) if not model: print(f"未找到模型: {args.model_name}") return print(f"找到模型: {model['name']}") # 显示性能指标 if 'MAE' in model: print(f"MAE: {model['MAE']}") if 'RMSE' in model: print(f"RMSE: {model['RMSE']}") if 'R2' in model: print(f"R2: {model['R2']}") print(f"Step 2: 加载模型...") model_type = model['model_type'] if not load_model(model['path'], model_type): return print(f"Step 3: 执行预测...") if model_type == 'lightgbm': print("使用LightGBM模型执行7天预测") result = predict_7days(backtest=False) else: print(f"使用CNN-RBiLSTM模型执行{args.prediction_hours}小时预测") result = predict_multi(args.start_time, args.prediction_hours) if result: print("预测成功!") if 'predictions' in result: print(f"预测数据点数: {len(result['predictions'])}") if 'summary' in result: print("预测摘要:") print(f" 峰值功率: {result['summary'].get('peak_power', 'N/A')} kW") print(f" 谷值功率: {result['summary'].get('valley_power', 'N/A')} kW") print(f" 平均功率: {result['summary'].get('avg_power', 'N/A')} kW") # 生成报告 if 'predictions' in result: try: generate_report(result['predictions'], args.model_name, args.start_time, args.prediction_hours) except Exception as e: print(f"报告生成失败: {e}") else: print("预测失败") if __name__ == '__main__': main()