""" 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}' 不是有效的数值") # 读取各参数值，如果找不到则使用默认值 get_param_value = lambda param_name: df_params.loc[df_params['参数名称'] == param_name, '参数值'].iloc[0] if param_name in df_params['参数名称'].values else None max_storage_capacity = get_param_value('最大储能容量') # 处理空值或字符串"空" if pd.isna(max_storage_capacity) or max_storage_capacity == '空': max_storage_capacity = None try: return SystemParameters( max_curtailment_wind=get_param_value('最大弃风率') or 0.1, max_curtailment_solar=get_param_value('最大弃光率') or 0.1, max_grid_ratio=get_param_value('最大上网电量比例') or 0.2, storage_efficiency=get_param_value('储能效率') or 0.9, discharge_rate=get_param_value('放电倍率') or 1.0, charge_rate=get_param_value('充电倍率') or 1.0, max_storage_capacity=max_storage_capacity, rated_thermal_capacity=get_param_value('额定火电装机容量') or 100.0, rated_solar_capacity=get_param_value('额定光伏装机容量') or 100.0, rated_wind_capacity=get_param_value('额定风电装机容量') or 100.0, available_thermal_energy=get_param_value('火电可用发电量') or 2400.0, available_solar_energy=get_param_value('光伏可用发电量') or 600.0, available_wind_energy=get_param_value('风电可用发电量') or 1200.0 ) except (KeyError, IndexError, Exception) as e: print(f"读取参数失败：{str(e)}，使用默认参数") return 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, rated_thermal_capacity=100.0, rated_solar_capacity=100.0, rated_wind_capacity=100.0, available_thermal_energy=2400.0, available_solar_energy=600.0, available_wind_energy=1200.0 ) 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, # 充电倍率 '', # 最大储能容量（空表示无限制） 100.0, # 额定火电装机容量 100.0, # 额定光伏装机容量 100.0, # 额定风电装机容量 2400.0, # 火电可用发电量 600.0, # 光伏可用发电量 1200.0 # 风电可用发电量 ], '参数说明': [ '允许的最大弃风率（0.0-1.0）', '允许的最大弃光率（0.0-1.0）', '允许的最大上网电量比例（0.0-∞，只限制上网电量）', '储能充放电效率（0.0-1.0）', '储能放电倍率（C-rate，>0）', '储能充电倍率（C-rate，>0）', '储能容量上限（MWh，空表示无限制）', '额定火电装机容量（MW）', '额定光伏装机容量（MW）', '额定风电装机容量（MW）', '火电可用发电量（MWh）', '光伏可用发电量（MWh）', '风电可用发电量（MWh）' ], '取值范围': [ '0.0-1.0', '0.0-1.0', '≥0.0', '0.0-1.0', '>0', '>0', '>0或空', '>0', '>0', '>0', '≥0', '≥0', '≥0' ], '默认值': [ '0.1', '0.1', '0.2', '0.9', '1.0', '1.0', '无限制', '100.0', '100.0', '100.0', '2400.0', '600.0', '1200.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()