fix bugs and revise for ICRN

Author: BenGalewsky

downscaled_climate_data/assets/county_measures.py

@@ -3,7 +3,7 @@
 import dask_geopandas as dgpd
 from shapely.geometry import Point
 from dask.distributed import Client
-
+import time
 import s3fs
 
 from htcdaskgateway import HTCGateway
@@ -13,7 +13,7 @@
 
 load_dotenv()  # take environment variables from .env.
 
-os.environ['CONDOR_BIN_DIR'] = "/opt/conda/envs/pangeo/bin"
+os.environ['CONDOR_BIN_DIR'] = "/u/bengal1/.conda/envs/downscaled_climate_data/bin"
 gateway = HTCGateway(address="https://dask.software-dev.ncsa.illinois.edu",
                      proxy_address=8786,
                      auth = BasicAuth(
@@ -22,7 +22,7 @@
                     )
 cluster = gateway.new_cluster(image="bengal1/pangeo-ncsa:dev", 
                               container_image="/u/bengal1/condor/pangeo.sif")
-cluster.scale(200)
+cluster.scale(80)
 client = cluster.get_client()
 print(cluster.dashboard_link)
 
@@ -40,15 +40,20 @@
     })
 
 try:
+    start_time = time.time()
+
+    era5_processing_start = time.time()
     era5 = era5_processing({'2m_temperature',
                             'total_precipitation',
                             "sfcWind",
-                            "relative_humidity",
                             "vapor_pressure",
                             "surface_pressure"},
                            2024, 2025, 'analysis_ready')
-    df = era5.to_dask_dataframe()
-    
+    print(f"era processing {time.time() - era5_processing_start:.2f} seconds")
+
+    to_tabular_start = time.time()
+    df = era5.rename("era5").to_dask_dataframe()
+
     era5_gdf = dgpd.from_dask_dataframe(
         df, 
         geometry=dgpd.points_from_xy(df, 'lon', 'lat')) \
@@ -60,6 +65,14 @@
                         filesystem=fs,
                         write_metadata_file=True,
                         schema="infer")
+    print(f"To Tabular {time.time() - to_tabular_start:.2f} seconds")
+    print(f"TOTAL TIME {time.time() - start_time:.2f} seconds")
+
+    info = client.scheduler_info()
+    num_workers = len(info['workers'])
+    print(f"Number of workers: {num_workers}")
+
+
 
 finally:
     cluster.close()

downscaled_climate_data/calculations/calculations.py

@@ -29,7 +29,7 @@ def stats(dataset):
 
 
 
-def heat_index(RH, t2m):
+def heat_index(dataset):
     """
     https://www.wpc.ncep.noaa.gov/html/heatindex_equation.shtml
 
@@ -42,6 +42,8 @@ def heat_index(RH, t2m):
         hi_alone (DataArray) - Heat index array (in K)
         
     """
+    t2m = dataset['2m_temperature']
+    RH = dataset['relative_humidity']
     # Convert to Fahrenheit
     T_F = ((t2m - 273.15) * 1.8) + 32
 
@@ -113,7 +115,7 @@ def wind_tot(uwind, vwind):
 
     return wind_mag, wind_dir
 
-def wind_mag(uwind, vwind):
+def wind_mag(dataset):
     """
     Calculates wind magnitude and angle
     
@@ -124,6 +126,8 @@ def wind_mag(uwind, vwind):
         wind_mag (DataArray) - Wind magnitude (m/s)
         
     """
+    uwind = dataset['10m_u_component_of_wind']
+    vwind = dataset['10m_v_component_of_wind']
     return np.sqrt(vwind**2 + uwind**2)
 
 

downscaled_climate_data/processors/era5_processor.py

@@ -72,7 +72,7 @@ def era5_processing(variables:set[str], year_start:int, year_end:int, dataset:st
     for variable in variables:
         if variable in calc_dict:
             analyis_variables.update(calc_dict[variable]['analysis_variables'])
-            calculations.update(calc_dict[variable]['calculator'])
+            calculations.add(calc_dict[variable]['calculator'])
         else:
             analyis_variables.add(variable)