489 lines
17 KiB
Python
489 lines
17 KiB
Python
"""
|
||
Excel数据读取模块
|
||
|
||
该模块提供从Excel文件中读取8760小时负荷和发电曲线数据的功能。
|
||
|
||
作者: iFlow CLI
|
||
创建日期: 2025-12-25
|
||
"""
|
||
|
||
import pandas as pd
|
||
import numpy as np
|
||
from typing import Dict, List, Optional, Tuple, Any
|
||
import os
|
||
from storage_optimization import SystemParameters
|
||
|
||
|
||
def validate_excel_data(df: pd.DataFrame, data_type: str = "8760") -> bool:
|
||
"""
|
||
验证Excel数据格式是否正确
|
||
|
||
Args:
|
||
df: pandas DataFrame对象
|
||
data_type: 数据类型，"24"或"8760"
|
||
|
||
Returns:
|
||
bool: 验证是否通过
|
||
"""
|
||
expected_length = 8760 if data_type == "8760" else 24
|
||
|
||
# 检查行数
|
||
if len(df) != expected_length:
|
||
print(f"错误：数据行数应为{expected_length}，实际为{len(df)}")
|
||
return False
|
||
|
||
# 检查必需的列
|
||
required_columns = ['光伏出力(MW)', '风电出力(MW)', '火电出力(MW)', '负荷需求(MW)']
|
||
missing_columns = [col for col in required_columns if col not in df.columns]
|
||
|
||
if missing_columns:
|
||
print(f"错误：缺少必需的列：{missing_columns}")
|
||
return False
|
||
|
||
# 检查数据类型和非负值
|
||
for col in required_columns:
|
||
if not pd.api.types.is_numeric_dtype(df[col]):
|
||
print(f"错误：列'{col}'必须为数值类型")
|
||
return False
|
||
|
||
if (df[col] < 0).any():
|
||
print(f"错误：列'{col}'包含负值")
|
||
return False
|
||
|
||
return True
|
||
|
||
|
||
def read_system_parameters(file_path: str) -> SystemParameters:
|
||
"""
|
||
从Excel文件读取系统参数
|
||
|
||
Args:
|
||
file_path: Excel文件路径
|
||
|
||
Returns:
|
||
SystemParameters对象
|
||
|
||
Raises:
|
||
FileNotFoundError: 文件不存在
|
||
ValueError: 参数格式错误
|
||
"""
|
||
# 检查文件是否存在
|
||
if not os.path.exists(file_path):
|
||
raise FileNotFoundError(f"文件不存在：{file_path}")
|
||
|
||
try:
|
||
# 读取参数工作表
|
||
df_params = pd.read_excel(file_path, sheet_name='参数')
|
||
|
||
# 验证参数工作表格式
|
||
required_columns = ['参数名称', '参数值', '参数说明']
|
||
missing_columns = [col for col in required_columns if col not in df_params.columns]
|
||
|
||
if missing_columns:
|
||
raise ValueError(f"参数工作表缺少必需的列：{missing_columns}")
|
||
|
||
# 提取参数值
|
||
params_dict = {}
|
||
for _, row in df_params.iterrows():
|
||
param_name = row['参数名称']
|
||
param_value = row['参数值']
|
||
|
||
# 跳过空行
|
||
if pd.isna(param_name) or pd.isna(param_value):
|
||
continue
|
||
|
||
# 转换参数值
|
||
try:
|
||
if isinstance(param_value, str):
|
||
# 尝试转换为浮点数
|
||
param_value = float(param_value)
|
||
params_dict[param_name] = param_value
|
||
except (ValueError, TypeError):
|
||
raise ValueError(f"参数 '{param_name}' 的值 '{param_value}' 不是有效的数值")
|
||
|
||
# 创建SystemParameters对象
|
||
return SystemParameters(
|
||
max_curtailment_wind=params_dict.get('最大弃风率', 0.1),
|
||
max_curtailment_solar=params_dict.get('最大弃光率', 0.1),
|
||
max_grid_ratio=params_dict.get('最大上网电量比例', 0.2),
|
||
storage_efficiency=params_dict.get('储能效率', 0.9),
|
||
discharge_rate=params_dict.get('放电倍率', 1.0),
|
||
charge_rate=params_dict.get('充电倍率', 1.0),
|
||
max_storage_capacity=params_dict.get('最大储能容量', None)
|
||
)
|
||
|
||
except Exception as e:
|
||
print(f"读取参数工作表失败，使用默认参数：{str(e)}")
|
||
# 如果参数工作表不存在或读取失败，返回默认参数
|
||
return SystemParameters()
|
||
|
||
|
||
def read_excel_data(file_path: str, sheet_name: str = 0, include_parameters: bool = True) -> Dict[str, List[float]]:
|
||
"""
|
||
从Excel文件读取8760小时数据
|
||
|
||
Args:
|
||
file_path: Excel文件路径
|
||
sheet_name: 工作表名称或索引，默认为第一个工作表
|
||
include_parameters: 是否同时读取系统参数
|
||
|
||
Returns:
|
||
包含所有数据的字典
|
||
|
||
Raises:
|
||
FileNotFoundError: 文件不存在
|
||
ValueError: 数据格式错误
|
||
"""
|
||
# 检查文件是否存在
|
||
if not os.path.exists(file_path):
|
||
raise FileNotFoundError(f"文件不存在：{file_path}")
|
||
|
||
try:
|
||
# 读取Excel文件
|
||
df = pd.read_excel(file_path, sheet_name=sheet_name)
|
||
|
||
# 自动检测数据类型
|
||
data_type = "8760" if len(df) >= 8760 else "24"
|
||
|
||
# 验证数据格式
|
||
if not validate_excel_data(df, data_type):
|
||
raise ValueError("Excel数据格式验证失败")
|
||
|
||
# 提取数据并转换为列表
|
||
solar_output = df['光伏出力(MW)'].tolist()
|
||
wind_output = df['风电出力(MW)'].tolist()
|
||
thermal_output = df['火电出力(MW)'].tolist()
|
||
load_demand = df['负荷需求(MW)'].tolist()
|
||
|
||
# 如果是24小时数据，扩展到8760小时（重复365天）
|
||
if data_type == "24" and len(df) == 24:
|
||
print("检测到24小时数据，自动扩展到8760小时（重复365天）")
|
||
solar_output = solar_output * 365
|
||
wind_output = wind_output * 365
|
||
thermal_output = thermal_output * 365
|
||
load_demand = load_demand * 365
|
||
|
||
# 构建返回结果
|
||
result = {
|
||
'solar_output': solar_output,
|
||
'wind_output': wind_output,
|
||
'thermal_output': thermal_output,
|
||
'load_demand': load_demand,
|
||
'data_type': data_type,
|
||
'original_length': len(df)
|
||
}
|
||
|
||
# 如果需要读取参数
|
||
if include_parameters:
|
||
try:
|
||
result['system_parameters'] = read_system_parameters(file_path)
|
||
print("成功读取系统参数")
|
||
except Exception as e:
|
||
print(f"读取系统参数失败，使用默认参数：{str(e)}")
|
||
result['system_parameters'] = SystemParameters()
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
raise ValueError(f"读取Excel文件失败：{str(e)}")
|
||
|
||
|
||
def create_excel_template(file_path: str, data_type: str = "8760"):
|
||
"""
|
||
创建Excel数据模板文件
|
||
|
||
Args:
|
||
file_path: 保存路径
|
||
data_type: 数据类型，"24"或"8760"
|
||
"""
|
||
# 生成示例数据
|
||
if data_type == "24":
|
||
hours = 24
|
||
# 24小时典型日数据
|
||
solar = [0.0] * 6 + [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] * 6
|
||
wind = [2.0, 3.0, 4.0, 3.0, 2.0, 1.0] * 4
|
||
thermal = [5.0] * 24
|
||
load = [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]
|
||
description = "24小时典型日数据模板"
|
||
else:
|
||
hours = 8760
|
||
# 生成8760小时的模拟数据（基于日模式加季节变化）
|
||
daily_solar = [0.0] * 6 + [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] * 6
|
||
daily_wind = [2.0, 3.0, 4.0, 3.0, 2.0, 1.0] * 4
|
||
daily_thermal = [5.0] * 24
|
||
daily_load = [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]
|
||
|
||
solar = []
|
||
wind = []
|
||
thermal = []
|
||
load = []
|
||
|
||
np.random.seed(42) # 确保可重复性
|
||
|
||
for day in range(365):
|
||
# 季节性因子
|
||
season_factor = 1.0 + 0.3 * np.sin(2 * np.pi * day / 365)
|
||
|
||
for hour in range(24):
|
||
# 添加随机变化
|
||
solar_variation = 1.0 + 0.2 * (np.random.random() - 0.5)
|
||
wind_variation = 1.0 + 0.3 * (np.random.random() - 0.5)
|
||
load_variation = 1.0 + 0.1 * (np.random.random() - 0.5)
|
||
|
||
solar.append(daily_solar[hour] * season_factor * solar_variation)
|
||
wind.append(daily_wind[hour] * wind_variation)
|
||
thermal.append(daily_thermal[hour])
|
||
load.append(daily_load[hour] * (2.0 - season_factor) * load_variation)
|
||
|
||
description = "8760小时全年数据模板"
|
||
|
||
# 创建DataFrame
|
||
df = pd.DataFrame({
|
||
'小时': range(1, hours + 1),
|
||
'光伏出力(MW)': solar,
|
||
'风电出力(MW)': wind,
|
||
'火电出力(MW)': thermal,
|
||
'负荷需求(MW)': load
|
||
})
|
||
|
||
# 保存到Excel
|
||
with pd.ExcelWriter(file_path, engine='openpyxl') as writer:
|
||
df.to_excel(writer, sheet_name='数据', index=False)
|
||
|
||
# 添加参数工作表
|
||
parameters_df = pd.DataFrame({
|
||
'参数名称': [
|
||
'最大弃风率',
|
||
'最大弃光率',
|
||
'最大上网电量比例',
|
||
'储能效率',
|
||
'放电倍率',
|
||
'充电倍率',
|
||
'最大储能容量'
|
||
],
|
||
'参数值': [
|
||
0.1, # 最大弃风率
|
||
0.1, # 最大弃光率
|
||
0.2, # 最大上网电量比例
|
||
0.9, # 储能效率
|
||
1.0, # 放电倍率
|
||
1.0, # 充电倍率
|
||
'' # 最大储能容量（空表示无限制）
|
||
],
|
||
'参数说明': [
|
||
'允许的最大弃风率（0.0-1.0）',
|
||
'允许的最大弃光率（0.0-1.0）',
|
||
'允许的最大上网电量比例（0.0-∞，只限制上网电量）',
|
||
'储能充放电效率（0.0-1.0）',
|
||
'储能放电倍率（C-rate，>0）',
|
||
'储能充电倍率（C-rate，>0）',
|
||
'储能容量上限（MWh，空表示无限制）'
|
||
],
|
||
'取值范围': [
|
||
'0.0-1.0',
|
||
'0.0-1.0',
|
||
'≥0.0',
|
||
'0.0-1.0',
|
||
'>0',
|
||
'>0',
|
||
'>0或空'
|
||
],
|
||
'默认值': [
|
||
'0.1',
|
||
'0.1',
|
||
'0.2',
|
||
'0.9',
|
||
'1.0',
|
||
'1.0',
|
||
'无限制'
|
||
]
|
||
})
|
||
parameters_df.to_excel(writer, sheet_name='参数', index=False)
|
||
|
||
# 添加说明工作表
|
||
description_df = pd.DataFrame({
|
||
'项目': ['数据说明', '数据类型', '时间范围', '单位', '注意事项', '参数说明'],
|
||
'内容': [
|
||
description,
|
||
f'{data_type}小时电力数据',
|
||
f'1-{hours}小时',
|
||
'MW (兆瓦)',
|
||
'所有数值必须为非负数',
|
||
'系统参数请在"参数"工作表中修改'
|
||
]
|
||
})
|
||
description_df.to_excel(writer, sheet_name='说明', index=False)
|
||
|
||
print(f"Excel模板已创建：{file_path}")
|
||
|
||
|
||
def analyze_excel_data(file_path: str) -> Dict[str, float]:
|
||
"""
|
||
分析Excel数据的基本统计信息
|
||
|
||
Args:
|
||
file_path: Excel文件路径
|
||
|
||
Returns:
|
||
包含统计信息的字典
|
||
"""
|
||
try:
|
||
data = read_excel_data(file_path)
|
||
|
||
solar = data['solar_output']
|
||
wind = data['wind_output']
|
||
thermal = data['thermal_output']
|
||
load = data['load_demand']
|
||
|
||
return {
|
||
'data_length': len(solar),
|
||
'total_solar': sum(solar),
|
||
'total_wind': sum(wind),
|
||
'total_thermal': sum(thermal),
|
||
'total_generation': sum(solar) + sum(wind) + sum(thermal),
|
||
'total_load': sum(load),
|
||
'max_solar': max(solar),
|
||
'max_wind': max(wind),
|
||
'max_thermal': max(thermal),
|
||
'max_load': max(load),
|
||
'avg_solar': np.mean(solar),
|
||
'avg_wind': np.mean(wind),
|
||
'avg_thermal': np.mean(thermal),
|
||
'avg_load': np.mean(load)
|
||
}
|
||
except Exception as e:
|
||
print(f"分析数据失败：{str(e)}")
|
||
return {}
|
||
|
||
|
||
def validate_system_parameters(params: SystemParameters) -> Dict[str, Any]:
|
||
"""
|
||
验证系统参数的有效性
|
||
|
||
Args:
|
||
params: SystemParameters对象
|
||
|
||
Returns:
|
||
验证结果字典
|
||
"""
|
||
validation_result = {
|
||
'valid': True,
|
||
'errors': [],
|
||
'warnings': []
|
||
}
|
||
|
||
# 检查弃风率
|
||
if not (0.0 <= params.max_curtailment_wind <= 1.0):
|
||
validation_result['valid'] = False
|
||
validation_result['errors'].append(f"弃风率必须在0.0-1.0之间，当前值：{params.max_curtailment_wind}")
|
||
|
||
# 检查弃光率
|
||
if not (0.0 <= params.max_curtailment_solar <= 1.0):
|
||
validation_result['valid'] = False
|
||
validation_result['errors'].append(f"弃光率必须在0.0-1.0之间，当前值：{params.max_curtailment_solar}")
|
||
|
||
# 检查上网电量比例
|
||
if not (0.0 <= params.max_grid_ratio):
|
||
validation_result['valid'] = False
|
||
validation_result['errors'].append(f"上网电量比例必须为非负值，当前值：{params.max_grid_ratio}")
|
||
|
||
# 检查储能效率
|
||
if not (0.0 < params.storage_efficiency <= 1.0):
|
||
validation_result['valid'] = False
|
||
validation_result['errors'].append(f"储能效率必须在0.0-1.0之间，当前值：{params.storage_efficiency}")
|
||
|
||
# 检查放电倍率
|
||
if params.discharge_rate <= 0:
|
||
validation_result['valid'] = False
|
||
validation_result['errors'].append(f"放电倍率必须大于0，当前值：{params.discharge_rate}")
|
||
|
||
# 检查充电倍率
|
||
if params.charge_rate <= 0:
|
||
validation_result['valid'] = False
|
||
validation_result['errors'].append(f"充电倍率必须大于0，当前值：{params.charge_rate}")
|
||
|
||
# 检查储能容量上限
|
||
if params.max_storage_capacity is not None and params.max_storage_capacity <= 0:
|
||
validation_result['valid'] = False
|
||
validation_result['errors'].append(f"储能容量上限必须大于0，当前值：{params.max_storage_capacity}")
|
||
|
||
# 添加警告信息
|
||
if params.storage_efficiency < 0.8:
|
||
validation_result['warnings'].append("储能效率较低，可能影响系统性能")
|
||
|
||
if params.max_curtailment_wind > 0.3 or params.max_curtailment_solar > 0.3:
|
||
validation_result['warnings'].append("弃风弃光率较高，可能造成能源浪费")
|
||
|
||
if params.max_grid_ratio > 0.5:
|
||
validation_result['warnings'].append("上网电量比例较高，可能影响电网稳定性")
|
||
|
||
return validation_result
|
||
|
||
|
||
def main():
|
||
"""主函数，演示Excel数据读取功能"""
|
||
print("=== Excel数据读取模块演示 ===")
|
||
|
||
# 创建模板文件
|
||
template_8760 = "data_template_8760.xlsx"
|
||
template_24 = "data_template_24.xlsx"
|
||
|
||
print("\n1. 创建Excel模板文件...")
|
||
create_excel_template(template_8760, "8760")
|
||
create_excel_template(template_24, "24")
|
||
|
||
# 分析模板数据
|
||
print(f"\n2. 分析{template_8760}数据...")
|
||
stats = analyze_excel_data(template_8760)
|
||
if stats:
|
||
print("数据统计信息：")
|
||
for key, value in stats.items():
|
||
print(f" {key}: {value:.2f}")
|
||
|
||
print(f"\n3. 演示读取{template_24}数据...")
|
||
try:
|
||
data = read_excel_data(template_24)
|
||
print(f"成功读取数据，类型：{data['data_type']}")
|
||
print(f"光伏出力前10小时：{data['solar_output'][:10]}")
|
||
print(f"风电出力前10小时：{data['wind_output'][:10]}")
|
||
print(f"负荷需求前10小时：{data['load_demand'][:10]}")
|
||
|
||
# 演示参数读取
|
||
if 'system_parameters' in data:
|
||
params = data['system_parameters']
|
||
print(f"\n系统参数：")
|
||
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}")
|
||
|
||
if validation['warnings']:
|
||
print("[WARNING] 参数警告:")
|
||
for warning in validation['warnings']:
|
||
print(f" - {warning}")
|
||
|
||
except Exception as e:
|
||
print(f"读取失败：{str(e)}")
|
||
|
||
print("\n=== 演示完成 ===")
|
||
print("模板文件已创建，您可以根据实际数据修改Excel文件。")
|
||
print("系统参数可以在Excel的'参数'工作表中直接修改。")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|