Add usga streamflow da

nwm.v4.0.0/ecf/nwm.def

@@ -46,12 +46,12 @@ suite nwm
        task jnwm_fetch_usgs
          edit CASETYPE FETCH_USGS 
          time +00:01
-       task jnwm_fetch_env_ca
-         edit CASETYPE FETCH_ENVCA 
-         cron 00:02 23:52 01:00
-       task jnwm_fetch_usace
-         edit CASETYPE FETCH_USACE 
-         cron 00:05 23:55 01:00
+#       task jnwm_fetch_env_ca
+#         edit CASETYPE FETCH_ENVCA 
+#         cron 00:02 23:52 01:00
+#       task jnwm_fetch_usace
+#         edit CASETYPE FETCH_USACE 
+#         cron 00:05 23:55 01:00
        task jnwm_fetch_rfc
          edit CASETYPE FETCH_RFC 
          cron 00:10 23:55 01:00
@@ -60,10 +60,12 @@ suite nwm
        cron 00:00 23:55 00:15
        # USGS and Env. Canada are preprocessed together so that their timeslices
        # can be merged in the same pass
-       task jnwm_preprocess_usgs_n_envca
+       task jnwm_preprocess_usgs
          edit CASETYPE PREPROCESS_USGS_N_ENVCA
-       task jnwm_preprocess_usace
-         edit CASETYPE PREPROCESS_USACE
+#       task jnwm_preprocess_usgs_n_envca
+#         edit CASETYPE PREPROCESS_USGS_N_ENVCA
+#       task jnwm_preprocess_usace
+#         edit CASETYPE PREPROCESS_USACE
        task jnwm_preprocess_rfc
          edit CASETYPE PREPROCESS_RFC
      endfamily

nwm.v4.0.0/ecf/nwm/da_preprocessing/preprocess/jnwm_preprocess_usgs.ecf

@@ -0,0 +1,51 @@
+#!/usr/bin/env bash
+
+%manual
+   Definition: This task preprocesses the USGS and Env. Canada realtime
+               streamflow observations.
+   Operations: Create the USGS and Env. Canada timeslices, and merge each
+               matching pair into a combined timeslice
+%end
+
+# SLURM settings
+##SBATCH -J PREPROCESS_USGS_N_ENVCA
+##SBATCH -o %ECF_JOBOUT%
+##SBATCH -e %ECF_JOBOUT%.err
+##SBATCH -t 00:60:00
+##SBATCH --nodes 1
+##SBATCH --nodelist=zhengtaocui-opertestbed-00015-1-0001
+###SBATCH --exclusive
+##SBATCH --ntasks-per-node=1
+
+export model=nwm
+
+%include <head.h>
+%include <model_envir.h>
+
+#
+#Enable module loading when LMOD Environment Module 
+# is installed on the system
+#
+#module load PrgEnv-intel/${PrgEnv_intel_ver}
+#module load craype/${craype_ver}
+#module load intel/${intel_ver}
+#module load cray-pals/${pals_ver}
+#module load cfp/${cfp_ver}
+#module load python/${python_ver}
+#module load netcdf/${netcdf_ver}
+#module load wgrib2/${grib2_ver}
+#module list
+
+# EXPORT list here
+
+export ECF_HOST=%ECF_HOST%
+
+echo "I, %ECF_NAME%, am part of a suite that lives in %ECF_HOME%.  My date is '%DATE%' "
+
+export cyc=%CYC%
+export PDY=`date -u +%%Y%%m%%d`
+export CASETYPE=%CASETYPE%
+
+${HOMEnwm}/jobs/JNWM_PREPROCESS_DA_OBSERVATIONS
+
+%include <tail.h>

nwm.v4.0.0/performance_test/run_perf_scale_test.sh

@@ -268,7 +268,7 @@ EOD
       set output 'VPU_mpi_03S_runtime.png'
 
       # Set title and labels
-      set title "MPI parallel number of cores vs runtime (03S)"
+      set title "MPI parallel number of cores vs runtime (03S and 03N)"
       set xlabel 'Number of cores'
       set ylabel "Wall clock time (mins) " offset 0,0,0
       set yrange [0:*]

nwm.v4.0.0/ush/nwm-realtime/nwm_analysis_assim.py

@@ -158,6 +158,12 @@ def _run_two_pass_ana(self, case_type: str, fconfig: str, rname: str,
         else:
             print("WARNING: RFC Reservoir timeseries not found! Skipping reservoir data assimilation!")
 
+        working_usgs_path = os.path.join(paths.host_run_dir, "usgs_timeslices")
+        if os.path.isdir( working_usgs_path ) and any(os.scandir(working_usgs_path)):
+            docker_args += f' -usgs {os.path.join(paths.container_run_dir, "usgs_timeslices")}'
+        else:
+            print("WARNING: UGS timeslices not found! Skipping streamflow data assimilation!")
+
         rc = self._docker_run(docker_args)
         if rc is None or rc != 0:
             print(
@@ -223,6 +229,12 @@ def _run_two_pass_ana(self, case_type: str, fconfig: str, rname: str,
         else:
             print("WARNING: RFC Reservoir timeseries not found! Skipping reservoir data assimilation!")
 
+        working_usgs_path = os.path.join(paths.host_run_dir, "usgs_timeslices")
+        if os.path.isdir( working_usgs_path ) and any(os.scandir(working_usgs_path)):
+            docker_args += f' -usgs {os.path.join(paths.container_run_dir, "usgs_timeslices")}'
+        else:
+            print("WARNING: UGS timeslices not found! Skipping streamflow data assimilation!")
+
         rc = self._docker_run(docker_args)
         if rc is None or rc != 0:
             print(
@@ -315,6 +327,12 @@ def ext_ana_restart(self, working_dir: str, pass_num: int,
                else:
                    print("WARNING: RFC Reservoir timeseries not found! Skipping reservoir data assimilation!")
 
+               working_usgs_path = os.path.join(paths.host_run_dir, "usgs_timeslices")
+               if os.path.isdir( working_usgs_path ) and any(os.scandir(working_usgs_path)):
+                   docker_args += f' -usgs {os.path.join(paths.container_run_dir, "usgs_timeslices")}'
+               else:
+                   print("WARNING: UGS timeslices not found! Skipping streamflow data assimilation!")
+
                rc = self._docker_run(docker_args)
             else:
                rc = self._docker_restart(src_dir, dst_dir, checkpoint_dir)
@@ -369,6 +387,12 @@ def ext_ana_restart(self, working_dir: str, pass_num: int,
         else:
             print("WARNING: RFC Reservoir timeseries not found! Skipping reservoir data assimilation!")
 
+        working_usgs_path = os.path.join(paths.host_run_dir, "usgs_timeslices")
+        if os.path.isdir( working_usgs_path ) and any(os.scandir(working_usgs_path)):
+            docker_args += f' -usgs {os.path.join(paths.container_run_dir, "usgs_timeslices")}'
+        else:
+            print("WARNING: UGS timeslices not found! Skipping streamflow data assimilation!")
+
         rc = self._docker_run(docker_args)
         if rc is None or rc != 0:
             print(
@@ -463,6 +487,26 @@ def move_outputs_to_storage(self) -> int:
     # ------------------------------------------------------------------ #
 
     def runRTE(self) -> int:
+        #
+        # Copy USGS output timeslices
+        #
+        paths = self._paths()
+        usgs_path = os.path.join(self.comout, "usgs_timeslices")
+        working_usgs_path = os.path.join(paths.host_run_dir, "usgs_timeslices")
+        if os.path.isdir( usgs_path ) and any(os.scandir(usgs_path)):
+            shutil.copytree( usgs_path, working_usgs_path, dirs_exist_ok=True )
+        else:
+            print("WARNING: USGS timeslices does not exist or is empty! Skipping USGS data assimilation!")
+
+        # Copy the previous day USGS output timeslices
+        if self.start_time.hour <= 6 or self.start_time.date() < self.t0.date():
+           usgs_path = os.path.join(self.previous_day_comout, "usgs_timeslices")
+           working_usgs_path = os.path.join(paths.host_run_dir, "usgs_timeslices")
+           if os.path.isdir( usgs_path ) and any(os.scandir(usgs_path)):
+                shutil.copytree( usgs_path, working_usgs_path, dirs_exist_ok=True )
+           else:
+                print("WARNING: Previous day USGS timeslices does not exist or is empty!")
+
         case_type = self.case_type
         if case_type == "CONUS_ANALYSIS_ASSIM":
             # AnA 3h window split into 1h (run 1) + 2h (run 2).

nwm.v4.0.0/ush/nwm-realtime/region.json

@@ -5,16 +5,16 @@
       "id": "03S",
       "nprocs": 2,
       "hydrofabric": "",
-      "formulation_assignment": "/ngen-app/ngen-python/lib/python3.11/site-packages/mswm/example_inputs/regionalization/vpu_03S/formulation_assignment.csv",
-      "catchment_group": "/ngen-app/ngen-python/lib/python3.11/site-packages/mswm/example_inputs/regionalization/vpu_03S/catchment_groups.csv"
+      "formulation_assignment": "/ngen-app/ngen-python/lib/python3.12/site-packages/mswm/example_inputs/regionalization/vpu_03S/formulation_assignment.csv",
+      "catchment_group": "/ngen-app/ngen-python/lib/python3.12/site-packages/mswm/example_inputs/regionalization/vpu_03S/catchment_groups.csv"
     },
     {
       "type": "vpu",
       "id": "03N",
       "nprocs": 6,
       "hydrofabric": "",
-      "formulation_assignment": "/ngen-app/ngen-python/lib/python3.11/site-packages/mswm/example_inputs/regionalization/vpu_03N/formulation_assignment.csv",
-      "catchment_group": "/ngen-app/ngen-python/lib/python3.11/site-packages/mswm/example_inputs/regionalization/vpu_03N/catchment_groups.csv"
+      "formulation_assignment": "/ngen-app/ngen-python/lib/python3.12/site-packages/mswm/example_inputs/regionalization/vpu_03N/formulation_assignment.csv",
+      "catchment_group": "/ngen-app/ngen-python/lib/python3.12/site-packages/mswm/example_inputs/regionalization/vpu_03N/catchment_groups.csv"
     }
   ]
 }