create single row selection function

Author: chantreux

scripts/derived/reanalysis-cerra-land_accumulation.py

@@ -10,7 +10,7 @@
 import sys
 from datetime import datetime
 sys.path.append('../utilities')
-from utils import load_path_from_df, load_output_path_from_row
+from utils import load_output_path_from_row, require_single_row
 
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 
@@ -136,11 +136,13 @@ def accumulation(ds,var):
 
     for var in derived_variables_list:
         logging.info(f"Calculating {var}")
-        input_row = df_parameters[(df_parameters['filename_variable'] == var) & (df_parameters['product_type'] == 'raw')]
+        mask_input = (df_parameters['filename_variable'] == var) & (df_parameters['product_type'] == 'raw')
+        input_row = require_single_row(df_parameters, mask_input, f"{var}/raw")
 
-        var_row = df_parameters[(df_parameters['filename_variable'] == var) & (df_parameters['product_type'] == 'derived')]
+        mask_var = (df_parameters['filename_variable'] == var) & (df_parameters['product_type'] == 'derived')
+        var_row = require_single_row(df_parameters, mask_var, f"{var}/derived")
         # Use utility function to load input path
-        var_download_path = load_output_path_from_row(input_row.iloc[0], dataset)
+        var_download_path = load_output_path_from_row(input_row], dataset)
         var_files = np.sort(glob.glob(f"{var_download_path}/*.nc"))
         print(f"{var_download_path}/*.nc")
         logging.info(f"List of file variables: {var_files}")
@@ -151,11 +153,11 @@ def accumulation(ds,var):
             date_str = basename.split('_')[-1].replace(".nc","")  
             date_obj = datetime.strptime(date_str, "%Y%m")
             year = date_obj.year
-            logging.info(f"Processing year: {year} and end year: {var_row.cds_years_end.iloc[0]}")
-            if year> var_row.cds_years_end.iloc[0]:
+            logging.info(f"Processing year: {year} and end year: {var_row.cds_years_end}")
+            if year > var_row.cds_years_end:
                 logging.info("Skipping file as it is after the end year")
                 continue
-            dest_dir = load_output_path_from_row(var_row.iloc[0], dataset)
+            dest_dir = load_output_path_from_row(var_row, dataset)
             var_file = os.path.basename(file).replace(".nc", "_daily_accumulated.nc")
             output_file=Path(f"{dest_dir}/{var_file}")
             logging.info(f"Saving calculated {var} to {dest_dir}")

scripts/derived/reanalysis-era5-single-levels.py

@@ -7,7 +7,7 @@
 from pathlib import Path
 import sys
 sys.path.append('../utilities')
-from utils import load_path_from_df, load_output_path_from_row
+from utils import  load_output_path_from_row, require_single_row
 
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 
@@ -19,27 +19,26 @@ def main():
     derived_variables_list = derived_variables.tolist()
     for var in derived_variables_list:
         logging.info(f"Calculating {var}")
-        input_row = df_parameters[(df_parameters['filename_variable'] == var) & (df_parameters['product_type'] == 'raw')]
-        var_row = df_parameters[(df_parameters['filename_variable'] == var) & (df_parameters['product_type'] == 'derived')]
+        mask_input = (df_parameters['filename_variable'] == var) & (df_parameters['product_type'] == 'raw')
+        input_row = require_single_row(df_parameters, mask_input, f"{var}/raw")
+
+        mask_var = (df_parameters['filename_variable'] == var) & (df_parameters['product_type'] == 'derived')
+        var_row = require_single_row(df_parameters, mask_var, f"{var}/derived")
 
         # Create a list of years from start to end
         year_list = list(range(var_row["cds_years_start"].squeeze() , var_row["cds_years_end"].squeeze()  + 1))
         for year in year_list:
 
-            if var == "sfcwind":
-                
-
-
-                input_row_u10 = df_parameters[(df_parameters['filename_variable'] == "u10") & (df_parameters['product_type'] == 'raw')]
-                input_row_v10 = df_parameters[(df_parameters['filename_variable'] == "v10") & (df_parameters['product_type'] == 'raw')]
+            if var == "sfcwind":               
+                input_row_u10 = require_single_row(df_parameters, (df_parameters['filename_variable'] == "u10") & (df_parameters['product_type'] == 'raw'), "u10/raw")
+                input_row_v10 = require_single_row(df_parameters, (df_parameters['filename_variable'] == "v10") & (df_parameters['product_type'] == 'raw'), "v10/raw")
                 # Use utility function to load input paths
-                u10_download_path = load_output_path_from_row(input_row_u10.iloc[0], dataset)
+                u10_download_path = load_output_path_from_row(input_row_u10, dataset)
                 u_10_file = glob.glob(f"{u10_download_path}/*{year}*.nc")[0]
-                v10_download_path = load_output_path_from_row(input_row_v10.iloc[0], dataset)
+                v10_download_path = load_output_path_from_row(input_row_v10, dataset)
                 v_10_file = glob.glob(f"{v10_download_path}/*{year}*.nc")[0]
-
                 # Use utility function to build output path
-                dest_dir = load_output_path_from_row(var_row.iloc[0], dataset)
+                dest_dir = load_output_path_from_row(var_row, dataset)
                 os.makedirs(dest_dir, exist_ok=True)
                 sfcwind_file = os.path.basename(u_10_file).replace("u10", "sfcwind")
                 output_file=Path(f"{dest_dir}/{sfcwind_file}")
@@ -54,7 +53,6 @@ def main():
                 sfcwind = operations.sfcwind_from_u_v(ds_merge)
                 sfcwind_daily = operations.resample_to_daily(sfcwind,"valid_time")
 
-
                 logging.info(f"Saving calculated sfcwind to {dest_dir}")
                 sfcwind_daily.to_netcdf(output_file)
 

scripts/interpolation/reanalysis-cerra-single-levels.py

@@ -7,7 +7,7 @@
 from pathlib import Path
 import sys
 sys.path.append('../utilities')
-from utils import load_input_path_from_row, load_output_path_from_row
+from utils import  load_output_path_from_row,require_single_row
 
 def write_to_netcdf(dataset: xr.Dataset, path: str, var: str):
     """
@@ -36,8 +36,9 @@ def main():
     variables_file_path = f"../../requests/{dataset}.csv"
     df_parameters = pd.read_csv(variables_file_path)
 
-    # Load the reference grid file from the first interpolated row
-    interpolated_row = df_parameters[(df_parameters['interpolation'] != 'native') & (df_parameters['product_type'] == 'derived')].iloc[0]
+    mask_ref = (df_parameters['interpolation'] != 'native') & (df_parameters['product_type'] == 'derived')
+    interpolated_row = require_single_row(df_parameters, mask_ref, "interpolated/derived reference row")
+    
     interpolation_file = interpolated_row.get('interpolation_file', 'land_sea_mask_0.0625degree.nc4')
     ds_ref=xr.open_dataset(f"/lustre/gmeteo/WORK/chantreuxa/cica/data/resources/reference-grids/{interpolation_file}")
 
@@ -50,7 +51,8 @@ def main():
         ds_variable=row["filename_variable"]
 
         # Use utility function to load input path (from raw data)
-        raw_row = df_parameters[(df_parameters['filename_variable'] == ds_variable) & (df_parameters['product_type'] == 'raw')].iloc[0]
+        mask_raw = (df_parameters['filename_variable'] == ds_variable) & (df_parameters['product_type'] == 'raw')
+        raw_row = require_single_row(df_parameters, mask_raw, f"{ds_variable}/raw")
         orig_dir = load_output_path_from_row(raw_row, dataset)
 
         # Use utility function to load output path

scripts/utilities/utils.py

@@ -279,4 +279,12 @@ def download_files(dataset, variables_file_path, create_request_func, get_output
                 try:
                     future.result()
                 except Exception as e:
-                    logging.error(f"Failed to download file: {e}")
+                    logging.error(f"Failed to download file: {e}")
+
+def require_single_row(df, mask, desc=None):
+    matches = df[mask]
+    if matches.shape[0] == 0:
+        raise KeyError(f"No row found{': ' + desc if desc else ''}")
+    if matches.shape[0] > 1:
+        raise ValueError(f"{matches.shape[0]} rows found{': ' + desc if desc else ''} — expected exactly 1")
+    return matches.iloc[0]