bmi config generation tests

python_tools/src/python_tools/ngen_configs_gen.py

@@ -72,7 +72,12 @@ def generate_troute_conf(out_dir,start,max_loop_size,geo_file_path):
     with open(Path(out_dir,"ngen.yaml"),'w') as fp:
         fp.writelines(troute_conf_str)  
 
-def gen_petAORcfe(hf_file,out,models,include):
+def gen_petAORcfe(hf_file,out,include):
+    models = []
+    if 'PET' in include:
+        models.append(Pet)
+    if 'CFE' in include:
+        models.append(Cfe)        
     for j, jmodel in enumerate(include):
         hf: gpd.GeoDataFrame = gpd.read_file(hf_file, layer="divides")
         hf_lnk_data: pd.DataFrame = gpd.read_file(hf_file,layer="model-attributes")
@@ -187,14 +192,14 @@ def gen_petAORcfe(hf_file,out,models,include):
             print(f'ignoring CFE')
         else:
             print(f'Generating CFE configs from pydantic models',flush = True)
-            gen_petAORcfe(args.hf_file,args.outdir,[Cfe],["CFE"])
+            gen_petAORcfe(args.hf_file,args.outdir,["CFE"])
 
     if "PET" in model_names: 
         if "PET" in ignore:
             print(f'ignoring PET')
         else:
             print(f'Generating PET configs from pydantic models',flush = True)
-            gen_petAORcfe(args.hf_file,args.outdir,[Pet],["PET"])
+            gen_petAORcfe(args.hf_file,args.outdir,["PET"])
 
     globals = [x[0] for x in serialized_realization]
     if serialized_realization.routing is not None:

python_tools/tests/test_bmi_config_generation.py

@@ -0,0 +1,54 @@
+import pytest, os
+from python_tools.ngen_configs_gen import gen_noah_owp_confs_from_pkl, gen_petAORcfe, generate_troute_conf
+from python_tools.noahowp_pkl import gen_noah_owp_pkl
+import datetime as dt
+
+TEST_DIR = os.path.dirname(os.path.realpath(__file__))
+DATA_DIR = os.path.join(TEST_DIR,'data')
+CONF_DIR = os.path.join(DATA_DIR,'cat_config')
+NOAH_DIR = os.path.join(CONF_DIR,'NOAH-OWP-M')
+CFE_DIR  = os.path.join(CONF_DIR,'CFE')
+PET_DIR  = os.path.join(CONF_DIR,'PET')
+GEOPACKAGE_NAME = "palisade.gpkg"
+GEOPACKAGE_PATH = os.path.join(DATA_DIR,GEOPACKAGE_NAME)
+os.system(f"curl -o {GEOPACKAGE_PATH} -L -O https://ngen-datastream.s3.us-east-2.amazonaws.com/{GEOPACKAGE_NAME}")
+PKL_FILE = os.path.join(DATA_DIR,"noah-owp-modular-init.namelist.input.pkl")
+START    = dt.datetime.strptime("202006200100",'%Y%m%d%H%M')
+END      = dt.datetime.strptime("202006200100",'%Y%m%d%H%M')
+
+@pytest.fixture(autouse=True)
+def clean_dir():
+    if os.path.exists(CONF_DIR):
+        os.system(f'rm -rf {str(CONF_DIR)}')
+    os.system(f'mkdir {str(CONF_DIR)}')
+
+def test_pkl():
+    gen_noah_owp_pkl(GEOPACKAGE_PATH,DATA_DIR)
+    assert os.path.exists(PKL_FILE)
+
+def test_noah_owp_m():
+    os.system(f'mkdir -p {NOAH_DIR}')
+    gen_noah_owp_confs_from_pkl(PKL_FILE, NOAH_DIR, START, END)
+    noah_config_example = os.path.join(NOAH_DIR,"noah-owp-modular-init-cat-2586011.namelist.input")
+    assert os.path.exists(noah_config_example)
+
+def test_cfe():
+    os.system(f'mkdir -p {CFE_DIR}')
+    gen_petAORcfe(GEOPACKAGE_PATH,DATA_DIR,["CFE"])
+    cfe_example = os.path.join(CFE_DIR,"CFE_cat-2586011.ini")
+    assert os.path.exists(cfe_example)
+
+def test_pet():
+    os.system(f'mkdir -p {PET_DIR}')
+    gen_petAORcfe(GEOPACKAGE_PATH,DATA_DIR,["PET"])
+    pet_example = os.path.join(PET_DIR,"PET_cat-2586011.ini")
+    assert os.path.exists(pet_example)    
+
+def test_routing():
+    max_loop_size = (END - START + dt.timedelta(hours=1)).total_seconds() / (3600)
+    generate_troute_conf(DATA_DIR,START,max_loop_size,GEOPACKAGE_PATH) 
+    yml_example = os.path.join(DATA_DIR,'ngen.yaml')
+    assert os.path.exists(yml_example)
+
+
+

python_tools/tests/test_configurer.py

@@ -1,6 +1,6 @@
 import os, pytest, json
 from datetime import datetime
-from python_tools.configure_datastream import config_class2dict, create_confs
+from python_tools.configure_datastream import create_confs
 
 SCRIPT_DIR        = os.path.dirname(os.path.realpath(__file__))
 DATA_DIR          = os.path.join(SCRIPT_DIR,'data')
@@ -71,12 +71,12 @@ def __init__(self,
 )
 
 @pytest.fixture
-def clean_dir():
+def clean_dir(autouse=True):
     if os.path.exists(DATA_DIR):
         os.system(f'rm -rf {str(DATA_DIR)}')
     os.system(f'mkdir {str(DATA_DIR)}')
 
-def test_conf_basic(clean_dir):
+def test_conf_basic():
     create_confs(inputs)
     assert os.path.exists(CONF_NWM)
     assert os.path.exists(CONF_FP)
@@ -85,7 +85,7 @@ def test_conf_basic(clean_dir):
     assert os.path.exists(REALIZATION_META_DS)   
     assert os.path.exists(REALIZATION_RUN) 
 
-def test_conf_daily(clean_dir):
+def test_conf_daily():
     inputs.start_date = "DAILY"
     inputs.end_date   = ""
     create_confs(inputs)