From 28d10ffec2759966782eb4aa58af741da80092d2 Mon Sep 17 00:00:00 2001 From: Zhengtao Cui Date: Tue, 25 Aug 2026 21:10:50 +0000 Subject: [PATCH 1/5] Added USGS streamflow data assimilation to the AnA and Extended AnA cycles. --- .../ush/nwm-realtime/nwm_analysis_assim.py | 24 +++++++++++++++++++ nwm.v4.0.0/ush/nwm-realtime/nwm_forecast.py | 19 +++++++++++++++ nwm.v4.0.0/ush/nwm-realtime/region.json | 8 +++---- 3 files changed, 47 insertions(+), 4 deletions(-) diff --git a/nwm.v4.0.0/ush/nwm-realtime/nwm_analysis_assim.py b/nwm.v4.0.0/ush/nwm-realtime/nwm_analysis_assim.py index 317bbf7..fbea070 100644 --- a/nwm.v4.0.0/ush/nwm-realtime/nwm_analysis_assim.py +++ b/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( diff --git a/nwm.v4.0.0/ush/nwm-realtime/nwm_forecast.py b/nwm.v4.0.0/ush/nwm-realtime/nwm_forecast.py index 0f1b752..75b028e 100644 --- a/nwm.v4.0.0/ush/nwm-realtime/nwm_forecast.py +++ b/nwm.v4.0.0/ush/nwm-realtime/nwm_forecast.py @@ -385,6 +385,25 @@ def configureRTE(self) -> None: else: print("WARNING: Previous RFC Reservoir timeseries does not exist or is empty!") + # + # Copy USGS output timeslices + # + 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!") + # ------------------------------------------------------------------ # # ecFlow / container helpers (shared by all configurations) # ------------------------------------------------------------------ # diff --git a/nwm.v4.0.0/ush/nwm-realtime/region.json b/nwm.v4.0.0/ush/nwm-realtime/region.json index c08b2a1..3f9c06c 100644 --- a/nwm.v4.0.0/ush/nwm-realtime/region.json +++ b/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" } ] } From d78c6a4ff9a833d1118ad11bbc9fe258764002e3 Mon Sep 17 00:00:00 2001 From: Zhengtao Cui Date: Wed, 26 Aug 2026 15:32:40 +0000 Subject: [PATCH 2/5] Moved the codes copying USGS timeslices to the workin directory from the ABS nwm_forecast to the subclass nwm_analyais_assim. The USGS timeslices are only used by AnA and Extended AnA cycles. --- .../ush/nwm-realtime/nwm_analysis_assim.py | 20 +++++++++++++++++++ nwm.v4.0.0/ush/nwm-realtime/nwm_forecast.py | 19 ------------------ 2 files changed, 20 insertions(+), 19 deletions(-) diff --git a/nwm.v4.0.0/ush/nwm-realtime/nwm_analysis_assim.py b/nwm.v4.0.0/ush/nwm-realtime/nwm_analysis_assim.py index fbea070..c9946ca 100644 --- a/nwm.v4.0.0/ush/nwm-realtime/nwm_analysis_assim.py +++ b/nwm.v4.0.0/ush/nwm-realtime/nwm_analysis_assim.py @@ -487,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). diff --git a/nwm.v4.0.0/ush/nwm-realtime/nwm_forecast.py b/nwm.v4.0.0/ush/nwm-realtime/nwm_forecast.py index 75b028e..0f1b752 100644 --- a/nwm.v4.0.0/ush/nwm-realtime/nwm_forecast.py +++ b/nwm.v4.0.0/ush/nwm-realtime/nwm_forecast.py @@ -385,25 +385,6 @@ def configureRTE(self) -> None: else: print("WARNING: Previous RFC Reservoir timeseries does not exist or is empty!") - # - # Copy USGS output timeslices - # - 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!") - # ------------------------------------------------------------------ # # ecFlow / container helpers (shared by all configurations) # ------------------------------------------------------------------ # From dc05fc1b8c0ce5536139a4407c19903c2df87722 Mon Sep 17 00:00:00 2001 From: Zhengtao Cui Date: Thu, 27 Aug 2026 21:02:28 +0000 Subject: [PATCH 3/5] create the add_usga_streamflow_da branch for the development-pw branch. Note that for the development-pw branch, because the Python version in the RTE container is 3.11 not 3.12, the paths to the regionalization cat-group and catchment assignment files in the region file (nwm.v4.0.0/ush/nwm-realtime/region.json) are not the same. --- nwm.v4.0.0/ush/nwm-realtime/region.json | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/nwm.v4.0.0/ush/nwm-realtime/region.json b/nwm.v4.0.0/ush/nwm-realtime/region.json index 3f9c06c..c08b2a1 100644 --- a/nwm.v4.0.0/ush/nwm-realtime/region.json +++ b/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.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" + "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" }, { "type": "vpu", "id": "03N", "nprocs": 6, "hydrofabric": "", - "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" + "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" } ] } From 20c7c68a906874f8b783c04f71f0c9e70e429bd0 Mon Sep 17 00:00:00 2001 From: Zhengtao Cui Date: Mon, 31 Aug 2026 19:35:12 +0000 Subject: [PATCH 4/5] changed preprocess_usgs_n_envca to preprocess_usgs. --- nwm.v4.0.0/ecf/nwm.def | 24 +++++---- .../preprocess/jnwm_preprocess_usgs.ecf | 51 +++++++++++++++++++ 2 files changed, 64 insertions(+), 11 deletions(-) create mode 100644 nwm.v4.0.0/ecf/nwm/da_preprocessing/preprocess/jnwm_preprocess_usgs.ecf diff --git a/nwm.v4.0.0/ecf/nwm.def b/nwm.v4.0.0/ecf/nwm.def index c8505db..cbf9165 100644 --- a/nwm.v4.0.0/ecf/nwm.def +++ b/nwm.v4.0.0/ecf/nwm.def @@ -1,7 +1,7 @@ # Definition of the suite 'nwm' suite nwm - edit ECF_HOME "/contrib/lfs/h1/owp/nwm/noscrub/SOMEBODY/test/packages/nwm-automation-scripts/nwm.v4.0.0/ecf" # path to the package ecf directory - edit ECF_INCLUDE "/contrib/lfs/h1/owp/nwm/noscrub/SOMEBODY/test/packages/nwm-automation-scripts/nwm.v4.0.0/ecf" + edit ECF_HOME "/lfs/h1/owp/nwm/noscrub/NOBODY/test/packages/nwm-automation-scripts/nwm.v4.0.0/ecf" # path to the package ecf directory + edit ECF_INCLUDE "/lfs/h1/owp/nwm/noscrub/NOBODY/test/packages/nwm-automation-scripts/nwm.v4.0.0/ecf" edit ECF_OUT "/home/Zhengtao.Cui/test/tmp" # #enable this for SLURM @@ -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 diff --git a/nwm.v4.0.0/ecf/nwm/da_preprocessing/preprocess/jnwm_preprocess_usgs.ecf b/nwm.v4.0.0/ecf/nwm/da_preprocessing/preprocess/jnwm_preprocess_usgs.ecf new file mode 100644 index 0000000..7c27985 --- /dev/null +++ b/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 +%include + +# +#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 From b5ff87909dc250618c0235edb1374909bb53c69f Mon Sep 17 00:00:00 2001 From: Zhengtao Cui Date: Mon, 31 Aug 2026 19:41:01 +0000 Subject: [PATCH 5/5] update performance script. --- nwm.v4.0.0/performance_test/run_perf_scale_test.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nwm.v4.0.0/performance_test/run_perf_scale_test.sh b/nwm.v4.0.0/performance_test/run_perf_scale_test.sh index 2f56a0b..9e15a23 100755 --- a/nwm.v4.0.0/performance_test/run_perf_scale_test.sh +++ b/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:*]