#!/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/list" 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): """ 根据模型名称找到模型信息（模糊匹配） """ models = get_model_list() # 优先匹配包含模型名称的模型 for model in models: if model_name.lower() in model['name'].lower(): return model # 如果没有找到，返回第一个模型 if models: return models[0] 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 generate_report(predictions, target_date, model_name): """ 生成对比报告 """ # 准备报告数据 report_data = [] for pred in predictions: time_str = pred['time'].split('T')[1][:5] # 提取时间部分 actual = pred.get('actual', 0) predicted = pred['power'] diff = predicted - actual report_data.append({ "time": time_str, "actual": actual, "predicted": predicted, "difference": diff }) # 调用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"对比{target_date}的历史数据和预测数据，使用{model_name}模型" 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", "lineChartCompare") # 调用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 compare_prediction(target_date, model_name="CNN-RBiLSTM"): """ 对比指定日期的历史数据和预测数据 """ # Step 1: 查找并加载模型 print(f"Step 1: 查找模型 {model_name}...") model = find_model_by_name(model_name) if not model: print(f"未找到模型: {model_name}") return False print(f"找到模型: {model['name']}") print(f"模型路径: {model['path']}") print(f"模型类型: {model['model_type']}") print(f"Step 2: 加载模型...") if not load_model(model['path'], model['model_type']): return False # Step 3: 构建目标时间并执行预测 target_time = f"{target_date} 00:00:00" print(f"Step 3: 执行预测，目标时间: {target_time}") # 调用多步预测接口 url = f"{API_BASE_URL}/api/load-forecast/predict/multi" data = { "start_time": target_time, "prediction_hours": 24 } try: response = requests.post(url, json=data) response.raise_for_status() result = response.json() if result.get('code') == 200: predictions = result['data']['predictions'] print("\n对比结果:") print("") print("时间 实际值(kW) 预测值(kW) 差值(kW)") print("------------------------------------------------------------") for pred in predictions: time_str = pred['time'].split('T')[1][:5] # 提取时间部分 actual = pred.get('actual', 0) predicted = pred['power'] diff = predicted - actual print(f"{time_str} {actual:5.0f} {predicted:5.0f} {diff:5.0f}") # 生成报告 try: generate_report(predictions, target_date, model_name) except Exception as e: print(f"报告生成失败: {e}") return True else: print(f"预测失败: {result.get('error')}") return False except Exception as e: print(f"对比异常: {e}") return False def main(): parser = argparse.ArgumentParser(description='对比历史数据和预测数据') parser.add_argument('--target_date', type=str, required=True, help='目标日期，格式: "2024-07-01"') parser.add_argument('--model_name', type=str, default="CNN-RBiLSTM", help='模型名称') args = parser.parse_args() # 验证日期格式 try: datetime.strptime(args.target_date, '%Y-%m-%d') except ValueError: print('错误：日期格式不正确，应为 "YYYY-MM-DD"') return compare_prediction(args.target_date, args.model_name) if __name__ == '__main__': main()