diff --git a/cea/analysis/lca/primary_energy.py b/cea/analysis/lca/primary_energy.py new file mode 100644 index 0000000000..365013ee8e --- /dev/null +++ b/cea/analysis/lca/primary_energy.py @@ -0,0 +1,225 @@ +""" +Primary Energy Calculation Tool + +Calculates primary energy consumption for buildings in a scenario using +Primary Energy Factors (PEF) with optional PV offsetting. + +Part of Life Cycle Analysis family. +""" + +import os +import pandas as pd +from cea.config import Configuration +from cea.inputlocator import InputLocator +from cea.utilities.standardize_coordinates import get_geographic_coordinate_system +from cea.analysis.lca.primary_energy_calculator import ( + calculate_primary_energy, + calculate_normalized_metrics, + calculate_hourly_primary_energy +) + + +__author__ = "Zhongming Shi" +__copyright__ = "Copyright 2025, Architecture and Building Systems - ETH Zurich" +__credits__ = ["Zhongming Shi"] +__license__ = "MIT" +__version__ = "0.1" +__maintainer__ = "Reynold Mok" +__email__ = "cea@arch.ethz.ch" +__status__ = "Production" + +def main(config: Configuration) -> None: + """ + Calculate primary energy for all buildings in scenario. + + Parameters + ---------- + config : Configuration + CEA configuration with primary-energy section + + Outputs + ------- + Creates CSV files: + - {scenario}/outputs/data/primary-energy/Total_annual_primary_energy.csv + """ + locator = InputLocator(config.scenario) + + # Get list of buildings + building_names = config.primary_energy.buildings + if not building_names: + # Use all buildings in zone + zone_df = get_building_names_from_zone(locator) + # Handle both 'Name' and 'name' column naming conventions + if 'Name' in zone_df.columns: + name_col = 'Name' + elif 'name' in zone_df.columns: + name_col = 'name' + else: + raise KeyError(f"Zone geometry must have either 'Name' or 'name' column. Available columns: {zone_df.columns.tolist()}") + building_names = zone_df[name_col].tolist() + + # Check PV requirements BEFORE processing any buildings + include_pv = config.primary_energy.include_pv + if include_pv: + pv_codes_param = config.primary_energy.pv_codes + # Handle both string (CLI) and list (GUI) input + if isinstance(pv_codes_param, list): + pv_codes = pv_codes_param if pv_codes_param else None + elif isinstance(pv_codes_param, str): + pv_codes = [code.strip() for code in pv_codes_param.split(',')] if pv_codes_param else None + else: + pv_codes = None + + if pv_codes and building_names: + # Check if PV files exist for the first building + first_building = building_names[0] + missing_panels = [] + for pv_code in pv_codes: + pv_path = locator.PV_results(first_building, pv_code) + if not os.path.exists(pv_path): + missing_panels.append(pv_code) + + if missing_panels: + missing_list = ', '.join(missing_panels) + error_msg = ( + f"PV electricity results missing for panel type(s): {missing_list}. " + f"Please run the 'photovoltaic (PV) panels' script first to generate PV potential results for these panel types." + ) + print(f"ERROR: {error_msg}") + raise FileNotFoundError(error_msg) + + # Ensure output folders exist BEFORE processing buildings + output_folder = locator.get_primary_energy_folder() + if not os.path.exists(output_folder): + os.makedirs(output_folder) + + hourly_folder = locator.get_primary_energy_hourly_folder() + if not os.path.exists(hourly_folder): + os.makedirs(hourly_folder) + + print(f"Calculating primary energy for {len(building_names)} buildings...") + + # Calculate primary energy for each building + annual_results = [] + district_hourly_results = [] + + for i, building in enumerate(building_names, 1): + print(f" [{i}/{len(building_names)}] {building}") + try: + # Annual totals + building_result = calculate_primary_energy(locator, building, config) + building_result_normalized = calculate_normalized_metrics(building_result) + annual_results.append(building_result_normalized) + + # Hourly timeseries (GRID + PV only) + hourly_result = calculate_hourly_primary_energy(locator, building, config) + + # Round numeric columns to 2 decimals + numeric_columns = hourly_result.select_dtypes(include=['float64', 'int64']).columns + hourly_result[numeric_columns] = hourly_result[numeric_columns].round(2) + + # Save per-building hourly file + building_hourly_path = locator.get_primary_energy_hourly_building(building) + hourly_result.to_csv(building_hourly_path, index=False) + print(f" Saved: {building_hourly_path}") + + # Add building name for district aggregation + hourly_result_with_name = hourly_result.copy() + hourly_result_with_name.insert(0, 'Name', building) + district_hourly_results.append(hourly_result_with_name) + + except Exception as e: + print(f" WARNING: Failed to calculate primary energy for {building}: {e}") + continue + + if not annual_results: + print("ERROR: No buildings successfully calculated") + return + + # === Save Annual Results === + annual_df = pd.DataFrame(annual_results) + + # Round all numeric columns to 2 decimals + numeric_columns = annual_df.select_dtypes(include=['float64', 'int64']).columns + annual_df[numeric_columns] = annual_df[numeric_columns].round(2) + + # Write annual results + annual_output_path = locator.get_primary_energy_annual() + annual_df.to_csv(annual_output_path, index=False) + + print(f"\nAnnual results saved to: {annual_output_path}") + print(f"Total buildings: {len(annual_df)}") + + # === Save District-Level Hourly Results === + if district_hourly_results: + # Sum across all buildings for each hour + # Remove 'Name' column from each building's data before summing + hourly_data_without_name = [df.drop(columns=['Name']) for df in district_hourly_results] + + # Concatenate all building DataFrames (handles missing columns automatically with NaN) + concatenated_df = pd.concat(hourly_data_without_name, ignore_index=True) + + # Identify numeric columns from the concatenated frame + numeric_cols = concatenated_df.select_dtypes(include=['float64', 'int64']).columns.tolist() + + # Group by 'date' column and sum numeric columns (NaN values are ignored in sum) + district_hourly_df = concatenated_df.groupby('date', as_index=False)[numeric_cols].sum() + + # Round numeric columns to 2 decimals + district_hourly_df[numeric_cols] = district_hourly_df[numeric_cols].round(2) + + # Write district hourly results + district_hourly_path = locator.get_primary_energy_hourly_district() + district_hourly_df.to_csv(district_hourly_path, index=False) + + print("\nHourly results:") + print(f" Per-building files: {len(district_hourly_results)} buildings") + print(f" District aggregation (8760 hours): {district_hourly_path}") + + # Print summary + print("\n=== Primary Energy Summary ===") + + # Sum base PE carriers (exclude NetGRID variations) + base_pe_cols = [col for col in annual_df.columns + if col.startswith('PE_') and col.endswith('_MJyr') + and 'NetGRID' not in col] + if base_pe_cols: + print(f"Total PE (all carriers): {annual_df[base_pe_cols].sum().sum() / 1000:.0f} GJ/yr") + + if config.primary_energy.include_pv: + # Find all NetGRID columns + netgrid_cols = [col for col in annual_df.columns if col.startswith('PE_NetGRID_') and col.endswith('_MJyr')] + for col in netgrid_cols: + pv_code = col.replace('PE_NetGRID_', '').replace('_MJyr', '') + pv_gen_col = f'PV_{pv_code}_generation_MJyr' + print(f"Total PE NetGRID ({pv_code}): {annual_df[col].sum() / 1000:.0f} GJ/yr") + if pv_gen_col in annual_df.columns: + print(f"Total PV generation ({pv_code}): {annual_df[pv_gen_col].sum() / 1000:.0f} GJ/yr") + + +def get_building_names_from_zone(locator): + """ + Get building names from zone geometry. + + Parameters + ---------- + locator : InputLocator + File path resolver + + Returns + ------- + pd.DataFrame + Zone geometry with 'Name' or 'name' column (caller should check both) + """ + import geopandas as gpd + + zone_path = locator.get_zone_geometry() + crs = get_geographic_coordinate_system() + zone_df = gpd.read_file(zone_path).to_crs(crs) + + return zone_df + + +if __name__ == '__main__': + config = Configuration() + main(config) diff --git a/cea/analysis/lca/primary_energy_calculator.py b/cea/analysis/lca/primary_energy_calculator.py new file mode 100644 index 0000000000..096356d9e1 --- /dev/null +++ b/cea/analysis/lca/primary_energy_calculator.py @@ -0,0 +1,292 @@ +""" +Primary Energy Calculator + +Calculates primary energy consumption using Primary Energy Factors (PEF). + +Primary Energy = Final Energy × PEF + +Supports PV offsetting using net metering approach. +""" + +import pandas as pd +from cea.analysis.lca.pv_offsetting import calculate_net_energy + +__author__ = "Zhongming Shi" +__copyright__ = "Copyright 2025, Architecture and Building Systems - ETH Zurich" +__credits__ = ["Zhongming Shi"] +__license__ = "MIT" +__version__ = "0.1" +__maintainer__ = "Reynold Mok" +__email__ = "cea@arch.ethz.ch" +__status__ = "Production" + +def calculate_primary_energy(locator, building, config): + """ + Calculate annual primary energy for a building. + + Parameters + ---------- + locator : InputLocator + File path resolver for scenario + building : str + Building name (e.g., "B001") + config : Configuration + CEA configuration with primary-energy section containing: + - include_pv : bool + - pv_codes : str (comma-separated, e.g., "PV1,PV2") + - pef_grid : float (default: 2.5) + - pef_naturalgas : float (default: 1.1) + - pef_coal : float (default: 1.1) + - pef_oil : float (default: 1.1) + - pef_wood : float (default: 1.0) + + Note: DH and DC are excluded - their primary energy is calculated + separately in thermal network and district optimization features. + + Returns + ------- + dict + Dictionary with primary energy results: + { + 'building': str, + 'FE_GRID_MJyr': float, + 'FE_NATURALGAS_MJyr': float, + 'FE_COAL_MJyr': float, + 'FE_OIL_MJyr': float, + 'FE_WOOD_MJyr': float, + 'PE_GRID_MJyr': float, + 'PE_NATURALGAS_MJyr': float, + 'PE_COAL_MJyr': float, + 'PE_OIL_MJyr': float, + 'PE_WOOD_MJyr': float, + 'NetGRID_{pv_code}_MJyr': float, # Net grid after PV offset (per panel) + 'PE_NetGRID_{pv_code}_MJyr': float, # Primary energy of net grid (per panel) + 'PV_{pv_code}_generation_MJyr': float, # PV generation (per panel) + 'GFA_m2': float # For normalisation + } + """ + # Parse PV codes from config + include_pv = config.primary_energy.include_pv + pv_codes_param = config.primary_energy.pv_codes + + # Handle both string (CLI) and list (GUI) input + if isinstance(pv_codes_param, list): + pv_codes = pv_codes_param if pv_codes_param else None + elif isinstance(pv_codes_param, str): + pv_codes = [code.strip() for code in pv_codes_param.split(',')] if pv_codes_param else None + else: + pv_codes = None + + # Get PEF values from config + # Note: DH and DC excluded - calculated in district optimization + pef = { + 'GRID': config.primary_energy.pef_grid, + 'NATURALGAS': config.primary_energy.pef_naturalgas, + 'COAL': config.primary_energy.pef_coal, + 'OIL': config.primary_energy.pef_oil, + 'WOOD': config.primary_energy.pef_wood, + } + + # Calculate net energy using shared utility + net_energy = calculate_net_energy(locator, building, include_pv, pv_codes) + + # Convert kWh to MJ (1 kWh = 3.6 MJ) + kWh_to_MJ = 3.6 + + # Extract final energy (FE) in MJ + # Note: DH and DC excluded - their primary energy calculated in district optimization + fe = { + 'GRID': net_energy['FE_GRID_kWh'] * kWh_to_MJ, + 'NATURALGAS': net_energy['FE_NATURALGAS_kWh'] * kWh_to_MJ, + 'COAL': net_energy['FE_COAL_kWh'] * kWh_to_MJ, + 'OIL': net_energy['FE_OIL_kWh'] * kWh_to_MJ, + 'WOOD': net_energy['FE_WOOD_kWh'] * kWh_to_MJ, + } + + # Calculate primary energy (PE) in MJ + pe = { + carrier: fe[carrier] * pef[carrier] + for carrier in fe.keys() + } + + # Get GFA for normalisation + demand_path = locator.get_demand_results_file(building) + demand_df = pd.read_csv(demand_path) + gfa_m2 = demand_df['GFA_m2'].iloc[0] if 'GFA_m2' in demand_df.columns else 0.0 + + # Build base results + results = { + 'building': building, + 'FE_GRID_MJyr': fe['GRID'], + 'FE_NATURALGAS_MJyr': fe['NATURALGAS'], + 'FE_COAL_MJyr': fe['COAL'], + 'FE_OIL_MJyr': fe['OIL'], + 'FE_WOOD_MJyr': fe['WOOD'], + 'PE_GRID_MJyr': pe['GRID'], + 'PE_NATURALGAS_MJyr': pe['NATURALGAS'], + 'PE_COAL_MJyr': pe['COAL'], + 'PE_OIL_MJyr': pe['OIL'], + 'PE_WOOD_MJyr': pe['WOOD'], + 'GFA_m2': gfa_m2 + } + + # Add per-panel NetGRID columns if PV is included + pv_by_type = net_energy.get('PV_by_type', {}) + for pv_code, pv_generation_kWh in pv_by_type.items(): + pv_generation_MJ = pv_generation_kWh * kWh_to_MJ + net_grid_MJ = fe['GRID'] - pv_generation_MJ + pe_net_grid_MJ = net_grid_MJ * pef['GRID'] + + results[f'NetGRID_{pv_code}_MJyr'] = net_grid_MJ + results[f'PE_NetGRID_{pv_code}_MJyr'] = pe_net_grid_MJ + results[f'PV_{pv_code}_generation_MJyr'] = pv_generation_MJ + + return results + + +def calculate_normalized_metrics(building_results): + """ + Add normalised metrics (per m²) to building results. + + Parameters + ---------- + building_results : dict + Results from calculate_primary_energy() + + Returns + ------- + dict + Same dict with added normalised columns: + - FE_GRID_MJm2yr + - PE_GRID_MJm2yr + - NetGRID_{pv_code}_MJm2yr (per panel) + - etc. + """ + gfa_m2 = building_results['GFA_m2'] + + if gfa_m2 == 0: + # Avoid division by zero + return building_results + + # Add normalised metrics + normalized = building_results.copy() + + # Normalize base carriers + for carrier in ['GRID', 'NATURALGAS', 'COAL', 'OIL', 'WOOD']: + normalized[f'FE_{carrier}_MJm2yr'] = building_results[f'FE_{carrier}_MJyr'] / gfa_m2 + normalized[f'PE_{carrier}_MJm2yr'] = building_results[f'PE_{carrier}_MJyr'] / gfa_m2 + + # Normalize per-panel NetGRID columns + for key in building_results.keys(): + if key.startswith('NetGRID_') and key.endswith('_MJyr'): + pv_code = key.replace('NetGRID_', '').replace('_MJyr', '') + normalized[f'NetGRID_{pv_code}_MJm2yr'] = building_results[key] / gfa_m2 + normalized[f'PE_NetGRID_{pv_code}_MJm2yr'] = building_results[f'PE_NetGRID_{pv_code}_MJyr'] / gfa_m2 + normalized[f'PV_{pv_code}_generation_MJm2yr'] = building_results[f'PV_{pv_code}_generation_MJyr'] / gfa_m2 + + return normalized + + +def calculate_hourly_primary_energy(locator, building, config): + """ + Calculate hourly primary energy for GRID and PV only. + + Other carriers (NATURALGAS, COAL, OIL, WOOD) have constant PEF, + so hourly breakdown provides no additional value. + + Parameters + ---------- + locator : InputLocator + File path resolver for scenario + building : str + Building name (e.g., "B001") + config : Configuration + CEA configuration with primary-energy section + + Returns + ------- + pd.DataFrame + Hourly timeseries with columns: + - date (datetime) + - GRID_MJ: Hourly grid electricity demand + - PE_GRID_MJ: Hourly primary energy from grid + - PV_{code}_generation_MJ: Hourly PV generation per panel + - NetGRID_{code}_MJ: Hourly net grid per panel + - PE_NetGRID_{code}_MJ: Hourly primary energy net grid per panel + """ + # Get PEF for grid + pef_grid = config.primary_energy.pef_grid + + # Parse PV codes from config + include_pv = config.primary_energy.include_pv + pv_codes_param = config.primary_energy.pv_codes + + # Handle both string (CLI) and list (GUI) input + if isinstance(pv_codes_param, list): + pv_codes = pv_codes_param if pv_codes_param else None + elif isinstance(pv_codes_param, str): + pv_codes = [code.strip() for code in pv_codes_param.split(',')] if pv_codes_param else None + else: + pv_codes = None + + # Read building demand (hourly) + demand_path = locator.get_demand_results_file(building) + demand_df = pd.read_csv(demand_path) + + # Convert Wh to MJ (1 Wh = 0.0036 MJ) + Wh_to_MJ = 0.0036 + + # Extract hourly GRID demand + if 'GRID_kWh' in demand_df.columns: + hourly_grid_MJ = demand_df['GRID_kWh'].values * 3.6 # kWh to MJ + else: + hourly_grid_MJ = demand_df['GRID_Wh'].values * Wh_to_MJ if 'GRID_Wh' in demand_df.columns else 0.0 + + # Calculate primary energy for grid + hourly_pe_grid_MJ = hourly_grid_MJ * pef_grid + + # Get date column (case-insensitive search) + date_col = next((col for col in demand_df.columns if col.lower() == 'date'), None) + if date_col: + date_values = demand_df[date_col].values + else: + raise ValueError(f"Date column not found in demand file for building {building}") + + # Build base DataFrame + hourly_df = pd.DataFrame({ + 'date': date_values, + 'GRID_MJ': hourly_grid_MJ, + 'PE_GRID_MJ': hourly_pe_grid_MJ + }) + + # Add per-panel PV columns if enabled + if include_pv and pv_codes: + # Get available PV panels + import os + available_panels = [] + for pv_code in pv_codes: + pv_path = locator.PV_results(building, pv_code) + if os.path.exists(pv_path): + available_panels.append(pv_code) + + # Process each panel + for pv_code in available_panels: + pv_path = locator.PV_results(building, pv_code) + pv_df = pd.read_csv(pv_path) + + # Get hourly PV generation + if 'E_PV_gen_kWh' in pv_df.columns: + hourly_pv_MJ = pv_df['E_PV_gen_kWh'].values * 3.6 # kWh to MJ + else: + hourly_pv_MJ = pv_df['E_PV_gen_Wh'].values * Wh_to_MJ if 'E_PV_gen_Wh' in pv_df.columns else 0.0 + + # Calculate net grid for this panel + hourly_netgrid_MJ = hourly_grid_MJ - hourly_pv_MJ + hourly_pe_netgrid_MJ = hourly_netgrid_MJ * pef_grid + + # Add columns + hourly_df[f'PV_{pv_code}_generation_MJ'] = hourly_pv_MJ + hourly_df[f'NetGRID_{pv_code}_MJ'] = hourly_netgrid_MJ + hourly_df[f'PE_NetGRID_{pv_code}_MJ'] = hourly_pe_netgrid_MJ + + return hourly_df diff --git a/cea/analysis/lca/pv_offsetting.py b/cea/analysis/lca/pv_offsetting.py new file mode 100644 index 0000000000..86346525b9 --- /dev/null +++ b/cea/analysis/lca/pv_offsetting.py @@ -0,0 +1,164 @@ +""" +Shared utility for calculating net energy with PV offsetting. + +Used by both: +- primary_energy module (Life Cycle Analysis) +- operational_emission module (Life Cycle Analysis) + +Net metering approach: NetGRID = GRID_demand - PV_total +""" +from __future__ import annotations +import pandas as pd +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from cea.inputlocator import InputLocator + + +__author__ = "Yiqiao Wang, Zhongming Shi" +__copyright__ = "Copyright 2025, Architecture and Building Systems - ETH Zurich" +__credits__ = ["Yiqiao Wang", "Zhongming Shi"] +__license__ = "MIT" +__version__ = "0.1" +__maintainer__ = "Reynold Mok" +__email__ = "cea@arch.ethz.ch" +__status__ = "Production" + +def calculate_net_energy( + locator: InputLocator, + building: str, + include_pv: bool = False, + pv_codes: list[str] | None = None, +): + """ + Calculate final energy consumption with optional PV offsetting. + + This function implements net metering for grid electricity: + - Reads building demand (final energy by carrier) + - Optionally reads PV generation from selected panels + - Calculates net grid electricity: NetGRID = GRID_demand - PV_total + + Parameters + ---------- + locator : InputLocator + File path resolver for scenario + building : str + Building name (e.g., "B001") + include_pv : bool + Whether to include PV offsetting (default: False) + pv_codes : list[str] or None + List of PV panel codes to include (e.g., ["PV1", "PV2"]) + If None and include_pv=True, includes all available panels + + Returns + ------- + dict + Dictionary with energy data: + { + 'FE_GRID_kWh': float, # Original grid demand (annual) + 'FE_NATURALGAS_kWh': float, # Natural gas (annual) + 'FE_COAL_kWh': float, # Coal (annual) + 'FE_OIL_kWh': float, # Oil (annual) + 'FE_WOOD_kWh': float, # Wood (annual) + 'PV_by_type': dict, # {pv_code: generation_kWh} by panel type + 'hourly_data': pd.DataFrame # Hourly timeseries (optional future use) + } + + Notes + ----- + - DH (District Heating) and DC (District Cooling) are excluded from primary energy + calculations as their distribution losses and generation primary energy are + calculated separately in thermal network and district optimization features. + + Notes + ----- + - All non-GRID carriers pass through unchanged + - Uses demand_energycarrier columns from building demand file + - PV data returned per-panel for flexibility in downstream calculations + """ + # Read building demand + demand_path = locator.get_demand_results_file(building) + demand_df = pd.read_csv(demand_path, index_col=None) + demand_df.index.set_names(['hour'], inplace=True) + + # Extract final energy by carrier (kWh = Wh / 1000) + # Use demand_energycarrier columns from demand output + # Note: DH and DC excluded - their primary energy calculated in district optimization + carriers = { + 'GRID': demand_df['GRID_kWh'].sum() if 'GRID_kWh' in demand_df.columns else 0.0, + 'NATURALGAS': demand_df['NG_hs_kWh'].sum() if 'NG_hs_kWh' in demand_df.columns else 0.0, + 'COAL': demand_df['COAL_hs_kWh'].sum() if 'COAL_hs_kWh' in demand_df.columns else 0.0, + 'OIL': demand_df['OIL_hs_kWh'].sum() if 'OIL_hs_kWh' in demand_df.columns else 0.0, + 'WOOD': demand_df['WOOD_hs_kWh'].sum() if 'WOOD_hs_kWh' in demand_df.columns else 0.0, + } + + # Initialize PV offsetting + pv_by_type: dict[str, float] = {} # Store PV generation by panel type + + if include_pv: + # Get available PV panels + available_panels = _get_available_pv_panels(locator, building) + + # Determine which panels to include + if pv_codes is None: + # Include all available panels + panels_to_include = available_panels + else: + # Include only requested panels that exist + panels_to_include = [code for code in pv_codes if code in available_panels] + + # Get PV generation for each panel separately + for pv_code in panels_to_include: + pv_path = locator.PV_results(building, pv_code) + try: + pv_df = pd.read_csv(pv_path) + # PV generation column: E_PV_gen_kWh + if 'E_PV_gen_kWh' in pv_df.columns: + pv_generation = float(pv_df['E_PV_gen_kWh'].sum()) + pv_by_type[pv_code] = pv_generation + except FileNotFoundError: + # Panel file doesn't exist, skip + continue + + # Return results with per-panel data + return { + 'FE_GRID_kWh': carriers['GRID'], + 'FE_NATURALGAS_kWh': carriers['NATURALGAS'], + 'FE_COAL_kWh': carriers['COAL'], + 'FE_OIL_kWh': carriers['OIL'], + 'FE_WOOD_kWh': carriers['WOOD'], + 'PV_by_type': pv_by_type, # Dict: {pv_code: generation_kWh} + 'hourly_data': demand_df # For potential future hourly analysis + } + + +def _get_available_pv_panels(locator: InputLocator, building: str) -> list[str]: + """ + Get list of available PV panel codes for a building. + + Parameters + ---------- + locator : InputLocator + File path resolver + building : str + Building name + + Returns + ------- + list[str] + List of PV codes (e.g., ["PV1", "PV2"]) + """ + import os + + pv_folder = locator.solar_potential_folder_PV() + + if not os.path.exists(pv_folder): + return [] + + pv_codes_path = locator.get_db4_components_conversion_conversion_technology_csv(conversion_technology="PHOTOVOLTAIC_PANELS") + if os.path.exists(pv_codes_path): + pv_codes_df = pd.read_csv(pv_codes_path) + pv_codes: list[str] = pv_codes_df['code'].tolist() + else: + raise FileNotFoundError(f"PV components conversion file not found at {pv_codes_path}") + return sorted(pv_codes) diff --git a/cea/config.pyi b/cea/config.pyi index 08feecb2d5..66d474d37e 100644 --- a/cea/config.pyi +++ b/cea/config.pyi @@ -34,6 +34,7 @@ class Configuration: demand: DemandSection costs: CostsSection emissions: EmissionsSection + primary_energy: PrimaryEnergySection extract_reference_case: ExtractReferenceCaseSection solar: SolarSection dbf_tools: DbfToolsSection @@ -130,6 +131,8 @@ class Configuration: @overload def __getattr__(self, item: Literal["emissions"]) -> EmissionsSection: ... @overload + def __getattr__(self, item: Literal["primary_energy"]) -> PrimaryEnergySection: ... + @overload def __getattr__(self, item: Literal["extract_reference_case"]) -> ExtractReferenceCaseSection: ... @overload def __getattr__(self, item: Literal["solar"]) -> SolarSection: ... @@ -579,6 +582,35 @@ class EmissionsSection(Section): def __getattr__(self, item: Literal["grid_decarbonise_target_emission_factor"]) -> float | None: ... def __getattr__(self, item: str) -> Any: ... +class PrimaryEnergySection(Section): + """Typed section for primary-energy configuration""" + buildings: list[str] + include_pv: bool + pv_codes: list[str] + pef_grid: float + pef_naturalgas: float + pef_coal: float + pef_oil: float + pef_wood: float + + @overload + def __getattr__(self, item: Literal["buildings"]) -> list[str]: ... + @overload + def __getattr__(self, item: Literal["include_pv"]) -> bool: ... + @overload + def __getattr__(self, item: Literal["pv_codes"]) -> list[str]: ... + @overload + def __getattr__(self, item: Literal["pef_grid"]) -> float: ... + @overload + def __getattr__(self, item: Literal["pef_naturalgas"]) -> float: ... + @overload + def __getattr__(self, item: Literal["pef_coal"]) -> float: ... + @overload + def __getattr__(self, item: Literal["pef_oil"]) -> float: ... + @overload + def __getattr__(self, item: Literal["pef_wood"]) -> float: ... + def __getattr__(self, item: str) -> Any: ... + class ExtractReferenceCaseSection(Section): """Typed section for extract-reference-case configuration""" destination: str diff --git a/cea/default.config b/cea/default.config index 83a1463b77..12ff64f229 100644 --- a/cea/default.config +++ b/cea/default.config @@ -442,6 +442,61 @@ grid-decarbonise-target-emission-factor.help = Target emission factor for grid d grid-decarbonise-target-emission-factor.nullable = True grid-decarbonise-target-emission-factor.category = Grid Decarbonisation +[primary-energy] +include-pv = false +include-pv.type = BooleanParameter +include-pv.help = True to use PV generation offsets grid electricity (NetGRID = GRID - PV_total). False to ignore PV offsetting. Ensure Photovoltaic Panels Feature has been executed. + +pv-codes = +pv-codes.type = ColumnMultiChoiceParameter +pv-codes.help = PV panel types to include in primary energy calculations. Leave empty to include all available PV panels. Ignored relevant when include-pv is False. +pv-codes.locator = get_db4_components_conversion_conversion_technology_csv +pv-codes.kwargs = conversion_technology=PHOTOVOLTAIC_PANELS +pv-codes.column = code + +pef-grid = 2.5 +pef-grid.type = RealParameter +pef-grid.help = Primary Energy Factor for grid electricity. + Accounts for generation, transmission, and distribution losses. + + Typical values: + - EU: 2.3-2.5 (EN 15603) + - US: 3.0-3.4 (site-to-source ratio) + - Switzerland: 2.5 (SIA 380/1) +pef-grid.category = Primary Energy Factors + +pef-naturalgas = 1.1 +pef-naturalgas.type = RealParameter +pef-naturalgas.help = Primary Energy Factor for natural gas. + Typical values: 1.1 (includes extraction and distribution losses) +pef-naturalgas.category = Primary Energy Factors + +pef-coal = 1.1 +pef-coal.type = RealParameter +pef-coal.help = Primary Energy Factor for coal. + Typical values: 1.1 (includes mining and transportation losses) +pef-coal.category = Primary Energy Factors + +pef-oil = 1.1 +pef-oil.type = RealParameter +pef-oil.help = Primary Energy Factor for oil. + Typical values: 1.1 (includes extraction, refining, and distribution losses) +pef-oil.category = Primary Energy Factors + +pef-wood = 1.0 +pef-wood.type = RealParameter +pef-wood.help = Primary Energy Factor for wood/biomass. + Typical values: 1.0-1.2 (includes harvesting and transportation losses) + + Note: This does NOT account for carbon neutrality or sustainability. + Use 1.0 if considering wood as renewable primary energy. +pef-wood.category = Primary Energy Factors + +buildings = +buildings.type = BuildingsParameter +buildings.help = Buildings to include in primary energy calculation. Leave empty to include all buildings in the scenario. +buildings.category = Customisation + [extract-reference-case] destination = {general:scenario}/../.. destination.type = PathParameter diff --git a/cea/inputlocator.py b/cea/inputlocator.py index c181ec7e34..6e57cf2738 100644 --- a/cea/inputlocator.py +++ b/cea/inputlocator.py @@ -1460,6 +1460,31 @@ def get_lca_operational_hourly_building(self, building: str): """scenario/outputs/data/emissions/timeline/{building}_operational_hourly.csv""" return os.path.join(self.get_lca_timeline_folder(), f"{building}_operational_hourly.csv") + # PRIMARY ENERGY + def get_primary_energy_folder(self): + """scenario/outputs/data/primary-energy""" + return os.path.join(self.scenario, 'outputs', 'data', 'primary-energy') + + def get_primary_energy_hourly_folder(self): + """scenario/outputs/data/primary-energy/hourly""" + return os.path.join(self.get_primary_energy_folder(), 'hourly') + + def get_primary_energy_annual(self): + """scenario/outputs/data/primary-energy/Total_primary_energy_buildings.csv""" + return os.path.join(self.get_primary_energy_folder(), 'Total_primary_energy_buildings.csv') + + def get_primary_energy_building(self, building_name): + """scenario/outputs/data/primary-energy/{building_name}_primary_energy.csv""" + return os.path.join(self.get_primary_energy_folder(), f'{building_name}_primary_energy.csv') + + def get_primary_energy_hourly_building(self, building_name): + """scenario/outputs/data/primary-energy/hourly/{building_name}_primary_energy_hourly.csv""" + return os.path.join(self.get_primary_energy_hourly_folder(), f'{building_name}_primary_energy_hourly.csv') + + def get_primary_energy_hourly_district(self): + """scenario/outputs/data/primary-energy/Total_primary_energy_hourly.csv""" + return os.path.join(self.get_primary_energy_folder(), 'Total_primary_energy_hourly.csv') + # COSTS def get_costs_folder(self): """scenario/outputs/data/costs""" diff --git a/cea/scripts.yml b/cea/scripts.yml index 291bbb4bf4..ecc7d3fad5 100644 --- a/cea/scripts.yml +++ b/cea/scripts.yml @@ -292,6 +292,37 @@ Life Cycle Analysis: - [get_radiation_metadata, building_name] - [get_radiation_building, building_name] + - name: primary-energy + label: Primary Energy + short_description: Calculate primary energy consumption from final energy using PEF + description: | + This Feature calculates primary energy consumption for each building using Primary Energy Factors (PEF). + + **Primary Energy** = Final Energy × PEF + + Primary Energy Factors account for energy losses during generation, transmission, and distribution. + For example, grid electricity typically has PEF = 2.5, meaning 2.5 MJ of primary energy is needed + to deliver 1 MJ of electricity to the building. + + Primary energy represents the total energy resources consumed at source (e.g., fuel burned at power plant), + while final energy is what is delivered to the building boundary. + + **PV Offsetting**: When enabled, PV generation offsets grid electricity using net metering: + NetGRID = GRID_demand - PV_total + + **Key outputs**: + - Final Energy by carrier (GRID, NATURALGAS, WOOD, etc.) + - Primary Energy by carrier (applying PEF to each carrier) + - Net grid electricity after PV offsetting (can be negative for net exporters) + - Normalised metrics per floor area (MJ/m²/year) + - Hourly timeseries for GRID and PV (other carriers use constant PEF) + interfaces: [cli] + module: cea.analysis.lca.primary_energy + parameters: ['general:scenario', primary-energy] + input-files: + - [get_zone_geometry] + - [get_demand_results_file, building_name] + - name: system-costs label: Energy Supply System Costs short_description: Calculate costs for energy supply systems