import argparse import requests import json import math import sys import subprocess import os from datetime import datetime, timedelta API_BASE_URL = "http://47.96.16.84:5006" # 正确的路径计算 SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) SKILL_DIR = os.path.dirname(SCRIPT_DIR) # 计算report-generator的路径 REPORT_GENERATOR_DIR = os.path.join(os.path.dirname(SKILL_DIR), "report-generator") def get_model_list(): url = f"{API_BASE_URL}/api/model/list" try: response = requests.get(url, timeout=30) response.raise_for_status() return response.json() except: return None def get_time_range(): url = f"{API_BASE_URL}/api/data/timerange" try: response = requests.get(url, timeout=30) response.raise_for_status() return response.json() except: return None def quick_predict(target_time, steps=96): url = f"{API_BASE_URL}/api/prediction/quick" payload = {"target_time": target_time, "prediction_steps": steps} try: response = requests.post(url, json=payload, timeout=60) response.raise_for_status() return response.json() except: return None def calculate_metrics(actuals, predicteds): valid_pairs = [(a, p) for a, p in zip(actuals, predicteds) if a is not None and p is not None] n = len(valid_pairs) if n == 0: return None mae = sum(abs(p - a) for a, p in valid_pairs) / n rmse = (sum((p - a) ** 2 for a, p in valid_pairs) / n) ** 0.5 mean_actual = sum(a for a, p in valid_pairs) / n ss_tot = sum((a - mean_actual) ** 2 for a, p in valid_pairs) ss_res = sum((a - p) ** 2 for a, p in valid_pairs) r2 = 1 - (ss_res / ss_tot) if ss_tot != 0 else 0 mape_sum = 0 mape_count = 0 for a, p in valid_pairs: if a != 0: mape_sum += abs((a - p) / a) mape_count += 1 mape = (100 / mape_count * mape_sum) if mape_count > 0 else float('inf') return {"MAE": mae, "RMSE": rmse, "R2": r2, "MAPE": mape, "valid_samples": n} def generate_report(results, model_id, target_time, user_query="模型评估报告"): report_data = [] for station, metrics in results.items(): report_data.append({ "station": station, "MAE": round(metrics['MAE'], 2), "RMSE": round(metrics['RMSE'], 2), "R2": round(metrics['R2'], 4), "MAPE": round(metrics['MAPE'], 2) if metrics['MAPE'] != float('inf') else 0, "valid_samples": metrics['valid_samples'] }) step1_script = os.path.join(REPORT_GENERATOR_DIR, "scripts", "step1_select.py") if not os.path.exists(step1_script): raise Exception(f"Step1 script not found at {step1_script}") 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", "dataTable") step2_script = os.path.join(REPORT_GENERATOR_DIR, "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") return step2_output def main(): parser = argparse.ArgumentParser() parser.add_argument("--action", type=str, required=True, choices=["list_models", "evaluate_model"]) parser.add_argument("--model_id", type=str) parser.add_argument("--station_id", type=str) parser.add_argument("--mae_threshold", type=float, default=50.0) parser.add_argument("--rmse_threshold", type=float, default=100.0) parser.add_argument("--r2_threshold", type=float, default=0.85) args = parser.parse_args() if args.action == "list_models": models_data = get_model_list() if models_data and models_data.get("success"): print(json.dumps(models_data.get("models", []), indent=2, ensure_ascii=False)) else: sys.exit(1) elif args.action == "evaluate_model": if not args.model_id: sys.exit(1) time_range_data = get_time_range() if not time_range_data or not time_range_data.get("success"): sys.exit(1) end_time_str = time_range_data.get("end_time") end_dt = datetime.strptime(end_time_str, "%Y-%m-%d %H:%M:%S") target_time_str = end_dt.strftime("%Y-%m-%d 00:00:00") predict_data = quick_predict(target_time_str, 96) if not predict_data or not predict_data.get("success"): sys.exit(1) stations_data = predict_data.get("stations", {}) stations_to_evaluate = [args.station_id] if args.station_id else list(stations_data.keys()) results = {} for station in stations_to_evaluate: if station not in stations_data: continue station_pred = stations_data[station] predicteds = station_pred.get("predicted", []) actuals = station_pred.get("actual", []) if not predicteds: continue metrics = calculate_metrics(actuals, predicteds) if metrics: results[station] = metrics try: report_output = generate_report(results, args.model_id, target_time_str, f"模型{args.model_id}在{target_time_str}的评估报告") report_address = report_output.get('address', '') print(f"@@@@@{report_address}@@@@@") except Exception as e: print(f"报告生成失败: {e}") # 即使报告生成失败，也不退出程序 pass if __name__ == "__main__": main()