232 lines
8.7 KiB
Python
232 lines
8.7 KiB
Python
|
#!/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']}")
|