Merge pull request #244 from CIROH-UA/test_nrds_fp

add tests for full conus nrds forcingprocessing

Author: JordanLaserGit

.github/workflows/forcingprocessor_nrds.yaml

@@ -0,0 +1,56 @@
+# This workflow will install Python dependencies, run tests and lint with a single version of Python
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
+
+name: Test Forcing Processor for NRDS
+on:
+  workflow_dispatch:
+  push:
+    branches:
+      - main
+    paths:
+      - 'forcingprocessor/**'
+      - '!forcingprocessor/README.md'
+  pull_request:
+    branches:
+      - main    
+    paths:      
+      - 'forcingprocessor/**'
+      - '!forcingprocessor/README.md'
+
+permissions:
+  contents: read
+jobs:
+  test-forcingprocessor-nrds:
+    runs-on: ubuntu-latest
+    
+    steps:
+    - uses: actions/checkout@v3
+
+    - name: Extract Python version from setup.cfg
+      id: pyver
+      run: |
+        # Read python_requires from setup.cfg
+        VERSION=$(awk -F'>=' '/^python_requires/ {print $2}' forcingprocessor/setup.cfg | tr -d ' ')
+        echo "version=$VERSION" >> $GITHUB_OUTPUT
+
+    - name: Set up Python
+      uses: actions/setup-python@v5
+      with:
+        python-version: ${{ steps.pyver.outputs.version }}
+
+    - name: Configure AWS
+      run: |
+        aws configure set aws_access_key_id ${{ secrets.AWS_ACCESS_KEY_ID }}
+        aws configure set aws_secret_access_key ${{ secrets.AWS_SECRET_ACCESS_KEY }}
+        aws configure set region us-east-1           
+        
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -e ./forcingprocessor
+        pip install pytest
+
+    - name: Test with pytest
+      run: |
+        cd forcingprocessor
+        python -m pytest -vv tests/test_nrds_fp.py

forcingprocessor/src/forcingprocessor/utils.py

@@ -27,6 +27,8 @@
         "DSWRF_surface",
     ] 
 
+vpus = ["01","02","03W","03S","03N","04","05","06","07","08","09","10L","10U","11","12","13","14","15","16","17","18"]
+
 def get_window(weights_df):
     """
     Providing window on weights for which number of catchments is over 50,000

forcingprocessor/tests/conftest.py

@@ -0,0 +1,6 @@
+import pytest, os
+
+@pytest.fixture(scope="session")
+def clean_s3_test():
+    yield
+    os.system("aws s3 rm s3://ciroh-community-ngen-datastream/test/nrds_fp_test/ --recursive")

forcingprocessor/tests/test_nrds_fp.py

@@ -0,0 +1,113 @@
+# test_nrds_fp.py
+#
+# Author: Jordan Laser <jlaser@lynker.com>
+# 
+# 
+# Test the NRDS forcing processing by inputing all 21 VPU's weight files, 
+# processing a single nwm forcing file, and the writing to a test location in the producting bucket.
+
+import shutil, os
+from pathlib import Path
+from datetime import datetime, timedelta, timezone
+from forcingprocessor.processor import prep_ngen_data
+from forcingprocessor.nwm_filenames_generator import generate_nwmfiles
+import pytest
+from forcingprocessor.utils import vpus
+import boto3
+from botocore.exceptions import ClientError
+import re
+
+HF_VERSION="v2.2"
+TODAY = datetime.now(timezone.utc)
+TODAY_YYMMDD = TODAY.strftime('%Y%m%d')
+hourminute  = '0000'
+TODAY_YYMMDDHHMM = TODAY_YYMMDD + hourminute
+YESTERDAY = TODAY - timedelta(hours=24)
+YESTERDAY_YYMMDD = YESTERDAY.strftime('%Y%m%d')
+YESTERDAY_YYMMDDHHMM = YESTERDAY_YYMMDD + hourminute
+test_dir = Path(__file__).parent
+data_dir = (test_dir/'data').resolve()
+forcings_dir = (data_dir/'forcings').resolve()
+pwd      = Path.cwd()
+data_dir = data_dir
+if os.path.exists(data_dir):
+    os.system(f"rm -rf {data_dir}")
+os.system(f"mkdir {data_dir}")
+pwd      = Path.cwd()
+filenamelist = str((pwd/"filenamelist.txt").resolve())
+
+weight_files = [f"https://ciroh-community-ngen-datastream.s3.amazonaws.com/v2.2_resources/weights/nextgen_VPU_{x}_weights.json" for x in vpus]
+local_weight_files = [str((data_dir/f"nextgen_VPU_{x}_weights.json").resolve()) for x in vpus]
+
+# download weight files
+for j, wf in enumerate(weight_files):
+    local_file = local_weight_files[j]
+    if not os.path.exists(local_file):
+        os.system(f"wget {wf} -P {data_dir}")
+
+conf = {
+    "forcing"  : {
+        "nwm_file"   : filenamelist,
+        "gpkg_file"  : local_weight_files
+    },
+
+    "storage":{
+        "output_path"       : "s3://ciroh-community-ngen-datastream/test/nrds_fp_test",
+        "output_file_type"  : ["netcdf"]
+    },    
+
+    "run" : {
+        "verbose"       : False,
+        "collect_stats" : False,
+        "nprocs"         : 1
+    }
+    }
+
+nwmurl_conf = {
+        "forcing_type" : "operational_archive",
+        "start_date"   : YESTERDAY_YYMMDDHHMM,
+        "end_date"     : YESTERDAY_YYMMDDHHMM,
+        "runinput"     : 1,
+        "varinput"     : 5,
+        "geoinput"     : 1,
+        "meminput"     : 0,
+        "urlbaseinput" : 7,
+        "fcst_cycle"   : [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18],
+        "lead_time"    : [1]
+    }
+
+s3 = boto3.client("s3")
+
+@pytest.fixture(autouse=True)
+def clean_dir():
+    if os.path.exists(forcings_dir):
+        shutil.rmtree(forcings_dir)
+
+def s3_object_exists(url: str) -> bool:
+    m = re.match(r"s3://([^/]+)/(.+)", url)
+    if not m:
+        raise ValueError(f"Invalid S3 URL: {url}")
+    bucket, key = m.groups()
+
+    try:
+        s3.head_object(Bucket=bucket, Key=key)
+        return True
+    except ClientError as e:
+        if e.response["Error"]["Code"] == "404":
+            return False
+        else:
+            raise
+
+def test_nrds_fp(clean_s3_test):
+    generate_nwmfiles(nwmurl_conf)  
+    conf['run']['collect_stats'] = False 
+    prep_ngen_data(conf)
+
+    for vpu in vpus:
+        url = f"s3://ciroh-community-ngen-datastream/test/nrds_fp_test/ngen.t01z.short_range.forcing.f001_f001.VPU_{vpu}.nc"
+        print(f"Checking for {url}")
+        assert s3_object_exists(url)
+
+if __name__ == "__main__":
+    test_nrds_fp()
+