#!/usr/bin/env python3 # -*- coding: utf-8 -*- """测试场景储能配置优化模块""" import sys import os sys.path.append(os.path.join(os.path.dirname(__file__), 'src')) from multi_scenario import MultiScenarioAnalyzer from storage_optimization import SystemParameters import numpy as np def test_scenario_storage_optimization(): """测试聚类场景的储能配置优化""" print("=== 场景储能配置优化测试 ===") # 生成模拟数据 np.random.seed(42) solar_output = [] wind_output = [] load_demand = [] for day in range(30): # 30天数据用于测试 # 光伏：白天有出力，夜间为0 daily_solar = [0.0] * 6 + list(np.random.uniform(2, 8, 12)) + [0.0] * 6 solar_output.extend(daily_solar) # 风电：相对随机 daily_wind = np.random.exponential(3.0, 24).tolist() wind_output.extend(daily_wind) # 负荷：日内变化，夜间低，白天高 daily_load = [4.0, 5.0, 6.0, 7.0, 9.0, 11.0, 13.0, 15.0, 17.0, 19.0, 21.0, 19.0, 17.0, 15.0, 13.0, 11.0, 9.0, 7.0, 6.0, 5.0, 4.0, 3.0, 2.0, 3.0] load_demand.extend(daily_load) try: # 1. 执行聚类分析 print("1. 执行多场景聚类分析...") analyzer = MultiScenarioAnalyzer(n_clusters=4, random_state=42) result = analyzer.fit_predict(solar_output, wind_output, load_demand) print(f" 识别出 {result.n_scenarios} 个场景") # 2. 对每个场景进行储能配置优化 print("\n2. 对每个场景进行储能配置优化...") # 系统参数 system_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 ) # 使用 analyzer 的优化功能 optimization_results = analyzer.optimize_storage_for_scenarios( result, solar_output, wind_output, load_demand, system_params, safety_factor=1.2 ) # 3. 显示优化结果 print("\n3. 储能配置优化结果...") analyzer.print_storage_optimization_summary(optimization_results) # 4. 测试结果验证 assert 'summary' in optimization_results, "缺少汇总结果" assert optimization_results['summary']['n_scenarios'] == result.n_scenarios, "场景数不匹配" assert optimization_results['summary']['recommended_capacity'] > 0, "推荐容量应该大于0" print("\n✅ 场景储能配置优化测试通过") return True, optimization_results except Exception as e: print(f"❌ 场景储能配置优化测试失败: {str(e)}") import traceback traceback.print_exc() return False, None def test_storage_optimization_with_different_scenarios(): """测试不同场景数量的储能优化""" print("\n=== 不同场景数量的储能优化测试 ===") # 生成测试数据 np.random.seed(123) solar_output = [] wind_output = [] load_demand = [] for day in range(60): # 60天数据 # 光伏：模拟夏季和冬季差异 if day < 30: # 夏季 daily_solar = [0.0] * 6 + list(np.random.uniform(3, 10, 12)) + [0.0] * 6 else: # 冬季 daily_solar = [0.0] * 6 + list(np.random.uniform(1, 6, 12)) + [0.0] * 6 solar_output.extend(daily_solar) # 风电：模拟季节性变化 daily_wind = np.random.exponential(2.5, 24).tolist() wind_output.extend(daily_wind) # 负荷：模拟夏冬负荷差异 if day < 30: # 夏季负荷较高（空调） daily_load = [6.0, 7.0, 8.0, 9.0, 11.0, 13.0, 15.0, 17.0, 19.0, 21.0, 23.0, 21.0, 19.0, 17.0, 15.0, 13.0, 11.0, 9.0, 8.0, 7.0, 6.0, 5.0, 4.0, 5.0] else: # 冬季负荷相对较低 daily_load = [4.0, 5.0, 6.0, 7.0, 9.0, 11.0, 13.0, 15.0, 17.0, 19.0, 21.0, 19.0, 17.0, 15.0, 13.0, 11.0, 9.0, 7.0, 6.0, 5.0, 4.0, 3.0, 2.0, 3.0] load_demand.extend(daily_load) analyzer = MultiScenarioAnalyzer(random_state=123) # 测试不同聚类数 for n_clusters in [3, 5, 7]: print(f"\n测试 {n_clusters} 个场景的储能优化...") try: result = analyzer.fit_predict(solar_output, wind_output, load_demand, n_clusters=n_clusters) print(f" 实际识别出 {result.n_scenarios} 个场景") # 执行储能优化 optimization_results = analyzer.optimize_storage_for_scenarios( result, solar_output, wind_output, load_demand, safety_factor=1.3 ) # 显示结果 summary = optimization_results['summary'] print(f" 加权平均储能需求: {summary['weighted_average_storage']:.2f} MWh") print(f" 推荐储能容量: {summary['recommended_capacity']:.2f} MWh") except Exception as e: print(f" ❌ {n_clusters}场景测试失败: {str(e)}") return False print("\n✅ 不同场景数量的储能优化测试通过") return True def test_edge_cases(): """测试边界情况""" print("\n=== 边界情况测试 ===") # 测试短时间数据 print("1. 测试短时间数据...") analyzer = MultiScenarioAnalyzer(n_clusters=2, random_state=42) # 只生成24小时数据 solar_24h = [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 5.0, 4.0, 3.0, 2.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0] wind_24h = [2.0, 3.0, 4.0, 3.0, 2.0, 1.0, 1.0, 2.0, 3.0, 4.0, 3.0, 2.0, 2.0, 3.0, 4.0, 3.0, 2.0, 1.0, 1.0, 2.0, 3.0, 4.0, 3.0, 2.0] load_24h = [3.0, 4.0, 5.0, 6.0, 8.0, 10.0, 12.0, 14.0, 16.0, 18.0, 20.0, 18.0, 16.0, 14.0, 12.0, 10.0, 8.0, 6.0, 5.0, 4.0, 3.0, 2.0, 1.0, 2.0] try: result = analyzer.fit_predict(solar_24h, wind_24h, load_24h) # 对于短时间数据，应该会被扩展到24小时 optimization_results = analyzer.optimize_storage_for_scenarios( result, solar_24h, wind_24h, load_24h ) print(" ✅ 短时间数据测试通过") except Exception as e: print(f" ❌ 短时间数据测试失败: {str(e)}") return False # 测试极大储能需求情况 print("2. 测试极大储能需求情况...") # 创建储能需求很大的场景 solar_extreme = [10.0] * 24 # 始终高出力 wind_extreme = [8.0] * 24 # 始终高出力 load_extreme = [1.0] * 24 # 始终低负荷 try: result_extreme = analyzer.fit_predict(solar_extreme, wind_extreme, load_extreme) optimization_results_extreme = analyzer.optimize_storage_for_scenarios( result_extreme, solar_extreme, wind_extreme, load_extreme ) print(f" 极端场景推荐储能容量: {optimization_results_extreme['summary']['recommended_capacity']:.2f} MWh") print(" ✅ 极大储能需求情况测试通过") except Exception as e: print(f" ❌ 极大储能需求情况测试失败: {str(e)}") return False print("\n✅ 所有边界情况测试通过") return True if __name__ == "__main__": print("=== 场景储能配置优化模块测试 ===\n") success = True # 基础功能测试 success1, optimization_results = test_scenario_storage_optimization() success &= success1 # 不同场景数量测试 success2 = test_storage_optimization_with_different_scenarios() success &= success2 # 边界情况测试 success3 = test_edge_cases() success &= success3 print(f"\n{'='*60}") if success: print("🎉 所有场景储能配置优化测试通过！") print(" 储能优化模块功能正常工作。") else: print("💥 部分测试失败，请检查代码。") print(f"{'='*60}") # 返回测试结果供其他模块使用 if success: print(f"\n测试数据示例:") if optimization_results: summary = optimization_results['summary'] print(f" - 加权平均储能需求: {summary['weighted_average_storage']:.2f} MWh") print(f" - 推荐储能容量: {summary['recommended_capacity']:.2f} MWh") print(f" - 安全系数: {summary['safety_factor']}") print(f" - 分析场景数: {summary['n_scenarios']}")