""" 测试程序 - 验证Excel数据输入和储能容量优化 该程序使用data_template_24.xlsx和data_template_8760.xlsx作为输入， 测试储能容量优化系统的功能和错误处理。 作者: iFlow CLI 创建日期: 2025-12-25 """ import sys import os sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'src')) import numpy as np import pandas as pd import matplotlib.pyplot as plt from typing import Dict, Any from excel_reader import read_excel_data, analyze_excel_data, read_system_parameters, validate_system_parameters from storage_optimization import optimize_storage_capacity, SystemParameters from advanced_visualization import create_comprehensive_plot, create_time_series_plot def test_excel_file(file_path: str, test_name: str) -> Dict[str, Any]: """ 测试单个Excel文件 Args: file_path: Excel文件路径 test_name: 测试名称 Returns: 测试结果字典 """ print(f"\n{'='*60}") print(f"测试：{test_name}") print(f"文件：{file_path}") print(f"{'='*60}") result = { 'test_name': test_name, 'file_path': file_path, 'success': False, 'error': None, 'data_stats': {}, 'optimization_result': {} } try: # 1. 检查文件是否存在 if not os.path.exists(file_path): raise FileNotFoundError(f"文件不存在：{file_path}") print("[OK] 文件存在检查通过") # 2. 读取Excel数据 print("正在读取Excel数据...") data = read_excel_data(file_path, include_parameters=True) print(f"[OK] 数据读取成功，类型：{data['data_type']}") print(f" 原始数据长度：{data['original_length']}") print(f" 处理后数据长度：{len(data['solar_output'])}") # 2.1 测试参数读取 if 'system_parameters' in data: params = data['system_parameters'] print(f"[OK] 系统参数读取成功") print(f" 最大弃风率: {params.max_curtailment_wind}") print(f" 最大弃光率: {params.max_curtailment_solar}") print(f" 最大上网电量比例: {params.max_grid_ratio}") print(f" 储能效率: {params.storage_efficiency}") print(f" 放电倍率: {params.discharge_rate}") print(f" 充电倍率: {params.charge_rate}") print(f" 最大储能容量: {params.max_storage_capacity}") # 验证参数 validation = validate_system_parameters(params) if validation['valid']: print("[OK] 系统参数验证通过") else: print("[ERROR] 系统参数验证失败:") for error in validation['errors']: print(f" - {error}") raise ValueError(f"系统参数验证失败: {validation['errors']}") if validation['warnings']: print("[WARNING] 系统参数警告:") for warning in validation['warnings']: print(f" - {warning}") else: print("[WARNING] 未找到系统参数，使用默认参数") params = SystemParameters() # 3. 分析数据统计信息 print("正在分析数据统计信息...") stats = analyze_excel_data(file_path) result['data_stats'] = stats print("[OK] 数据统计分析完成") # 4. 验证数据完整性 print("正在验证数据完整性...") solar_output = data['solar_output'] wind_output = data['wind_output'] thermal_output = data['thermal_output'] load_demand = data['load_demand'] # 检查数据长度一致性 if not (len(solar_output) == len(wind_output) == len(thermal_output) == len(load_demand)): raise ValueError("数据长度不一致") print("[OK] 数据长度一致性检查通过") # 检查非负值 for name, values in [ ("光伏出力", solar_output), ("风电出力", wind_output), ("火电出力", thermal_output), ("负荷需求", load_demand) ]: if any(v < 0 for v in values): raise ValueError(f"{name}包含负值") print("[OK] 非负值检查通过") # 5. 储能容量优化测试 print("正在进行储能容量优化计算...") # 使用Excel中的参数和不同的测试配置 excel_params = params # 从Excel中读取的参数 test_configs = [ { 'name': 'Excel配置', 'params': excel_params }, { 'name': '基础配置', 'params': SystemParameters( max_curtailment_wind=0.1, max_curtailment_solar=0.1, max_grid_ratio=0.2, storage_efficiency=0.9, discharge_rate=1.0, charge_rate=1.0 ) }, { 'name': '高弃风弃光配置', 'params': SystemParameters( max_curtailment_wind=0.2, max_curtailment_solar=0.2, max_grid_ratio=0.3, storage_efficiency=0.85, discharge_rate=1.5, charge_rate=1.5 ) }, { 'name': '低储能效率配置', 'params': SystemParameters( max_curtailment_wind=0.05, max_curtailment_solar=0.05, max_grid_ratio=0.1, storage_efficiency=0.75, discharge_rate=0.8, charge_rate=0.8 ) } ] optimization_results = [] for config in test_configs: print(f" 测试配置：{config['name']}") opt_result = optimize_storage_capacity( solar_output, wind_output, thermal_output, load_demand, config['params'] ) print(f" 所需储能容量：{opt_result['required_storage_capacity']:.2f} MWh") print(f" 弃风率：{opt_result['total_curtailment_wind_ratio']:.3f}") print(f" 弃光率：{opt_result['total_curtailment_solar_ratio']:.3f}") print(f" 上网电量比例：{opt_result['total_grid_feed_in_ratio']:.3f}") print(f" 能量平衡校验：{'通过' if opt_result['energy_balance_check'] else '未通过'}") optimization_results.append({ 'config_name': config['name'], 'result': opt_result }) result['optimization_result'] = optimization_results print("[OK] 储能容量优化计算完成") # 6. 可视化测试（仅对24小时数据进行，避免8760小时数据生成过大图像） if data['data_type'] == '24': print("正在生成可视化图表...") try: # 使用基础配置的结果生成图表 base_result = optimization_results[0]['result'] base_params = test_configs[0]['params'] # 生成基础图表（截取前24小时数据） solar_24 = solar_output[:24] if len(solar_output) > 24 else solar_output wind_24 = wind_output[:24] if len(wind_output) > 24 else wind_output thermal_24 = thermal_output[:24] if len(thermal_output) > 24 else thermal_output load_24 = load_demand[:24] if len(load_demand) > 24 else load_demand # 截取结果的24小时数据 result_24 = { 'required_storage_capacity': base_result['required_storage_capacity'], 'charge_profile': base_result['charge_profile'][:24] if len(base_result['charge_profile']) > 24 else base_result['charge_profile'], 'discharge_profile': base_result['discharge_profile'][:24] if len(base_result['discharge_profile']) > 24 else base_result['discharge_profile'], 'curtailed_wind': base_result['curtailed_wind'][:24] if len(base_result['curtailed_wind']) > 24 else base_result['curtailed_wind'], 'curtailed_solar': base_result['curtailed_solar'][:24] if len(base_result['curtailed_solar']) > 24 else base_result['curtailed_solar'], 'grid_feed_in': base_result['grid_feed_in'][:24] if len(base_result['grid_feed_in']) > 24 else base_result['grid_feed_in'], 'storage_profile': base_result['storage_profile'][:24] if len(base_result['storage_profile']) > 24 else base_result['storage_profile'], 'total_curtailment_wind_ratio': base_result['total_curtailment_wind_ratio'], 'total_curtailment_solar_ratio': base_result['total_curtailment_solar_ratio'], 'total_grid_feed_in_ratio': base_result['total_grid_feed_in_ratio'], 'energy_balance_check': base_result['energy_balance_check'] } create_comprehensive_plot(solar_24, wind_24, thermal_24, load_24, result_24, base_params) create_time_series_plot(solar_24, wind_24, thermal_24, load_24, result_24) print("[OK] 可视化图表生成完成") except Exception as e: print(f"⚠ 可视化图表生成失败：{str(e)}") print(" 这不是严重错误，可能是字体或显示问题") # 标记测试成功 result['success'] = True print(f"\n[OK] 测试 '{test_name}' 成功完成！") except Exception as e: result['error'] = str(e) result['traceback'] = traceback.format_exc() print(f"\n[ERROR] 测试 '{test_name}' 失败：{str(e)}") print("详细错误信息：") traceback.print_exc() return result def run_comprehensive_tests(): """运行综合测试""" print("开始运行Excel数据输入综合测试...") print(f"当前工作目录：{os.getcwd()}") # 测试文件列表 test_files = [ { 'path': 'data_template_24.xlsx', 'name': '24小时数据模板测试' }, { 'path': 'data_template_8760.xlsx', 'name': '8760小时数据模板测试' } ] # 运行所有测试 results = [] for test_file in test_files: result = test_excel_file(test_file['path'], test_file['name']) results.append(result) # 生成测试报告 print(f"\n{'='*80}") print("测试报告总结") print(f"{'='*80}") total_tests = len(results) successful_tests = sum(1 for r in results if r['success']) failed_tests = total_tests - successful_tests print(f"总测试数：{total_tests}") print(f"成功测试：{successful_tests}") print(f"失败测试：{failed_tests}") print(f"成功率：{successful_tests/total_tests*100:.1f}%") print("\n详细结果：") for i, result in enumerate(results, 1): status = "[OK] 成功" if result['success'] else "[ERROR] 失败" print(f"{i}. {result['test_name']}: {status}") if not result['success']: print(f" 错误：{result['error']}") # 如果有失败的测试，返回非零退出码 if failed_tests > 0: print(f"\n有{failed_tests}个测试失败，请检查上述错误信息") return False else: print("\n所有测试通过！系统运行正常。") return True def main(): """主函数""" try: success = run_comprehensive_tests() sys.exit(0 if success else 1) except KeyboardInterrupt: print("\n测试被用户中断") sys.exit(1) except Exception as e: print(f"\n测试过程中发生未预期的错误：{str(e)}") traceback.print_exc() sys.exit(1) if __name__ == "__main__": main()