Weather morpher

cea/datamanagement/weather_helper/epwmorpher.py

@@ -0,0 +1,170 @@
+"""
+Functions to enable climate morphing for the EPW files based on pyepwmorph
+
+Title: pyepwmorph
+Description: This module provides functions to morph EPW files using the pyepwmorph library, a python package authored by Justin McCarty for accessing global climate models and using them to morph time series weather data in EPW files. An independent application can be found at www.morphepw.app. For a peer-reviewed article that uses pyepwmorph, please see https://doi.org/10.1088/1742-6596/2600/8/082005. 
+"""
+
+# general import
+import os
+import shutil
+
+# CEA import
+import cea.inputlocator
+import cea.config
+
+# pyepwmorph import
+# pip install -e pyepwmorph for local dev
+from pyepwmorph.tools import workflow as pyepwmorph_workflow
+from pyepwmorph.tools import utilities as pyepwmorph_utilities
+from pyepwmorph.tools import io as pyepwmorph_io
+from pyepwmorph.tools import configuration as pyepwmorph_config
+
+__author__ = "Justin McCarty"
+__copyright__ = ["Copyright 2025, Architecture and Building Systems - ETH Zurich"]
+__credits__ = ["Justin McCarty"]
+__license__ = "GPLv3+"
+__version__ = "0.1"
+__maintainer__ = ["Justin McCarty"]
+__email__ = ["[email protected]", "[email protected]"]
+__status__ = "Production"
+
+def convert_scenario_names(name):
+    """
+    Convert the scenario names from CEA to the names used in pyepwmorph
+
+    :param string name: name of the scenario in CEA
+    :return: name of the scenario in pyepwmorph
+    :rtype: string
+    """
+    if name == "Best Case":
+        return "Best Case Scenario"
+    elif name == "Moderate":
+        return "Middle of the Road"
+    elif name == "Upper Middle": 
+        return "Upper Middle Scenario"
+    elif name == "Worst Case":
+        return "Worst Case Scenario"
+    else:
+        raise ValueError(f"Could not interpret the climate pathway: {name}. "
+                         f"Please choose one of the following: best_case, moderate_case, semi_moderate_case, worst_case")
+
+def morphing_workflow(locator, config):
+    """Run the morphing workflow for the specified scenario.
+
+    Args:
+        locator (cea.inputlocator.InputLocator): The input locator for the scenario.
+        config (cea.config.Configuration): The configuration for the scenario.
+
+    """
+
+    # 1. Read the inputs and create a epw_morph_configuration 
+    # 1.1. project_name and output_directory
+    project_name = f"{config.general.project.split(os.sep)[-1]}_{config.general.scenario}"
+    # today = datetime.datetime.now().strftime("%Y%m%d")
+
+    output_directory = os.path.dirname(locator.get_weather_file())
+    shutil.copy(locator.get_weather_file(), os.path.join(output_directory, "before_morph_weather.epw"))
+
+    if not os.path.exists(output_directory):
+        os.makedirs(output_directory)
+
+
+    # 1.2. user_epw_file is specified in the config but defaults to the scenario file
+    user_epw_file = locator.get_weather_file()
+    if not os.path.exists(user_epw_file):
+        raise FileNotFoundError(f"Could not find the specified EPW file: {user_epw_file}")
+
+    user_epw_object = pyepwmorph_io.Epw(user_epw_file)
+
+    # 1.3. variable choice is specified in the config but defaults to temperature
+    # variables.choices = temperature, radiation, relative_humidity, wind_speed, pressure, dew_point
+    user_variables = config.weather_helper.variables
+
+    # 1.4 the baseline range against which the calculations are made
+    # this the range of years used to calculate the baseline, should be taken from the EPW files
+    try:
+        baseline_range = user_epw_object.detect_baseline_range()
+    except Exception:
+        print("Could not detect the baseline range from the EPW file, using default of 1985-2014")
+        baseline_range = (1985, 2014)  # default if the EPW file does not have the years in it
+
+    # 1.5 year can be any future year but defaults to 2050
+    user_future_year = config.weather_helper.year
+    user_future_range = pyepwmorph_utilities.calc_period(user_future_year, baseline_range)
+
+    # 1.6 climate pathway can be specified in config but defaults to moderate_case
+    user_climate_pathway = config.weather_helper.climate_pathway
+    print(f"User pathway is: {user_climate_pathway}")
+
+    # 1.7. percentile can be specified in config but defaults to 50 (single choice)
+    # percentile.choices = 1, 5, 10, 25, 50, 75, 90, 95, 99
+    user_percentile = int(config.weather_helper.percentile)
+
+    # 1.8 model_sources is not something the user can change in CEA
+    # Sources as of 2025.09.23 using r1i1p1f1
+    # 'KACE-1-0-G', 'MRI-ESM2-0', 'GFDL-ESM4', 'INM-CM4-8', 'IPSL-CM6A-LR', 'INM-CM5-0', 'ACCESS-CM2', 'MIROC6', 'EC-Earth3-Veg-LR', 'BCC-CSM2-MR'
+    model_sources = [
+        'KACE-1-0-G', 'MRI-ESM2-0', 'GFDL-ESM4', 'INM-CM4-8', 'IPSL-CM6A-LR',
+        'INM-CM5-0', 'ACCESS-CM2', 'MIROC6', 'EC-Earth3-Veg-LR', 'BCC-CSM2-MR'
+    ]
+
+    print(f"User variables are: {user_variables}")
+    # 1. create the config object for the morpher
+    morph_config = pyepwmorph_config.MorphConfig(project_name, 
+                                            user_epw_file, 
+                                            user_variables, 
+                                            [convert_scenario_names(user_climate_pathway)], 
+                                            [user_percentile],
+                                            user_future_range, 
+                                            output_directory, 
+                                            model_sources=model_sources, 
+                                            baseline_range=baseline_range)
+
+    # # 2. Morph the EPW file
+    print(f"Config pathways are: {morph_config.model_pathways}")
+    print(f"Config variables are: {morph_config.model_variables}")
+
+    # 2.1 get climate model data
+    year_model_dict = pyepwmorph_workflow.iterate_compile_model_data(morph_config.model_pathways,
+                                                                    morph_config.model_variables,
+                                                                    morph_config.model_sources,
+                                                                    morph_config.epw.location['longitude'],
+                                                                    morph_config.epw.location['latitude'],
+                                                                    morph_config.percentiles)
+
+
+    # write the new epw file to the output directory
+    print("Morphing")
+    morphed_data = pyepwmorph_workflow.morph_epw(morph_config.epw,
+                                                morph_config.user_variables,
+                                                morph_config.baseline_range,
+                                                user_future_range,
+                                                year_model_dict,
+                                                [p for p in morph_config.model_pathways if p!="historical"][0],
+                                                user_percentile)
+    morphed_data.dataframe['year'] = int(user_future_year)
+
+    # morphed_data.write_to_file(os.path.join(morph_config.output_directory,
+    #                                         f"{str(user_future_year)}_{user_climate_pathway}_{percentile_key}.epw"))
+
+    morphed_data.write_to_file(os.path.join(morph_config.output_directory,"weather.epw"))
+
+
+
+def main(config):
+    """
+    This script uses the pyepwmorph library to morph an EPW file based on user-defined parameters in the config file.
+
+    Args:
+        config (cea.config.Configuration): The configuration for the scenario.
+    """
+    assert os.path.exists(
+        config.scenario), 'Scenario not found: %s' % config.scenario
+    locator = cea.inputlocator.InputLocator(config.scenario)
+    print(f"{'=' * 10} Starting the climate morphing workflow for scenario {config.general.scenario} {'=' * 10}")
+    morphing_workflow(locator, config)
+
+
+if __name__ == '__main__':
+    main(cea.config.Configuration())

cea/datamanagement/weather_helper/weather_helper.py

@@ -10,6 +10,7 @@
 
 import geopandas as gpd
 import requests
+import cea.datamanagement.weather_helper.epwmorpher as epwmorpher
 
 import cea.config
 import cea.inputlocator
@@ -86,17 +87,22 @@ def main(config):
     locator = cea.inputlocator.InputLocator(config.scenario)
     weather = config.weather_helper.weather
 
-    if not weather:
-        raise ValueError("No weather file provided. "
-                         "Please specify a weather file or select an option to fetch data automatically. "
-                         "e.g --weather climate.onebuilding.org")
-
-    locator.ensure_parent_folder_exists(locator.get_weather_file())
-    if config.weather_helper.weather == 'climate.onebuilding.org':
-        print("No weather provided, fetching from online sources.")
-        fetch_weather_data(locator.get_weather_file(), locator.get_zone_geometry())
+    if config.weather_helper.morph:
+        epwmorpher.main(config)
+        print("Weather morphing complete. The original weather file has been backed up as 'before_morph_weather.epw' in the inputs folder. The morphed weather file according to your percentile of warming choice is now set as 'weather.epw'.")
+
     else:
-        copy_weather_file(weather, locator)
+        if not weather:
+            raise ValueError("No weather file provided. "
+                            "Please specify a weather file or select an option to fetch data automatically. "
+                            "e.g --weather climate.onebuilding.org")
+
+        locator.ensure_parent_folder_exists(locator.get_weather_file())
+        if config.weather_helper.weather == 'climate.onebuilding.org':
+            print("No weather provided, fetching from online sources.")
+            fetch_weather_data(locator.get_weather_file(), locator.get_zone_geometry())
+        else:
+            copy_weather_file(weather, locator)
 
 
 if __name__ == '__main__':

cea/default.config

@@ -125,6 +125,32 @@ weather =
 weather.type = WeatherPathParameter
 weather.help = Path to a weather file (.epw) or the label of one CEA Built-in weather file. Leave blank to fetch from third-party sources.
 
+morph = False
+morph.type = BooleanParameter
+morph.help = Morph the existing weather file to match the desired climate pathway and year.
+
+year = 2060
+year.type = IntegerParameter
+year.help = Year to which the weather file is morphed.
+year.category = Morph Settings
+
+climate-pathway = moderate_case
+climate-pathway.type = ChoiceParameter
+climate-pathway.choices = Best Case, Moderate, Upper Middle, Worst Case
+climate-pathway.help = Climate pathway to morph the weather file to. The "Best Case" pathway corresponds to a 1.8°C increase in global temperature (SSP126), the "Moderate" pathway corresponds to a 2.7°C increase (SSP245), the "Upper Middle" pathway corresponds to a 3.6˚C increase (SSP370), and the "Worst Case" pathway corresponds to a 4.4°C increase (SSP585). More information about these pathways can be read here https://www.carbonbrief.org/explainer-how-shared-socioeconomic-pathways-explore-future-climate-change/.
+climate-pathway.category = Morph Settings
+
+percentile = 50
+percentile.type = ChoiceParameter
+percentile.choices = 1, 5, 10, 25, 50, 75, 90, 95, 99
+percentile.help = Percentile of the weather file to morph to. The percentile is applied to the temperature and radiation data of the weather file.
+
+variables = Temperature
+variables.type = MultiChoiceParameter
+variables.choices = Temperature, Clouds and Radiation, Humidity, Wind, Pressure, Dew Point
+variables.help = List of weather variables to morph.
+variables.category = Morph Settings
+
 [radiation-crax]
 buildings =
 buildings.type = BuildingsParameter

cea/utilities/epwreader.py

@@ -72,6 +72,23 @@ def epw_reader(weather_path):
 
     return epw_data
 
+def epw_location(weather_path):
+    """
+    Returns the location of the EPW file as a tuple (latitude, longitude, utc, elevation).
+
+    :param weather_path: Path to the EPW file
+    :return: dict containing latitude, longitude, utc, and elevation
+    """
+    with open(weather_path, 'r') as f:
+        epw_data = f.readlines()
+
+    lat = float(epw_data[0].split(',')[6])
+    lon = float(epw_data[0].split(',')[7])
+    utc = float(epw_data[0].split(',')[8])
+    elevation = float(epw_data[0].split(',')[9].strip("\n"))
+
+    return {"latitude": lat, "longitude": lon, "utc": utc, "elevation": elevation}
+
 
 def calc_horirsky(Tdrybulb, Tdewpoint, N):
     """