Merge branch 'development' into 'release-candidate'

Merge development into release-candidate for release 3.1.1.3.0-rc1

See merge request NGWPC/nwm-ngen/ngen-fcst!39

Author: cmaynard-ngwpc

README.md

@@ -44,7 +44,7 @@ Follow the following steps to test the program:
 where [NGEN-FCST_ROOT] is where ngen-fcst is installed
 
 The program takes three command line arguments:
-1) Path to the NetCDF forcing file
+1) Path to the NetCDF forcing file or folder containing csv forcing files for all catchments in the basin
 2) Path to the config yaml file for a validation run (from ngen-cal)
 3) Path to the folder to be created for storing inputs/outputs from running ngen, relative to the Output directory of the calibration run as indicated in the config yaml file. For example, if "fcst_run1" is the 3rd argument, and "yaml_file" in the "general" section of the config file is '/home/yuqiong.liu/work/Gitlab/run/kge_DDS/noah_cfes/01123000/Output/Validation_Run/01123000_config_valid_best.yaml', then the new output directory to be created for the ngen-fcst run would be:
 
@@ -79,7 +79,7 @@ This will print a usage statement for the container:
 ```
 Usage: run-ngen-fcst.sh <forcing_file> <config_file> <output_path> [log_file] [venv_path]
 
-FORCING_FILE: Path to the NetCDF forcing file.
+FORCING_FILE: Path to the NetCDF forcing file or a folder containing csv forcing files for all catchments in the basin.
 CONFIG_FILE: Path to the config yaml file for a validation run (from ngen-cal).
 OUTPUT_PATH: Path to the folder to be created for storing inputs/outputs from running ngen.
 LOG_FILE (optional): Path to the output file where the script's output will be saved.  Used when running in LOCAL or DOCKER environment

docker/run-ngen-fcst.sh

@@ -14,13 +14,13 @@ umask 000
 
 # Function to display help message
 show_help() {
-  echo "Usage: $(basename "$0") <command> <forcing_file> <config_file> <output_dir> [stdout_file] [venv_path]"
+  echo "Usage: $(basename "$0") <command> <forcing_dir> <config_file> <output_dir> [stdout_file] [venv_path]"
   echo ""
   echo ""
   echo "COMMAND:"
   echo "  forecast          Run forecast script."
   echo ""
-  echo "FORCING_FILE: Path to the NetCDF forcing file."
+  echo "FORCING_DIR: Path to the directory container csv forcing files."
   echo "CONFIG_FILE: Path to the config yaml file for a validation run (from ngen-cal)."
   echo "FORECAST_DIR: Name of the folder to to store the forecast output."
   echo "STDOUT_FILE (optional): Path to the stdout file where the script's console output will be saved.  Used when running in LOCAL or DOCKER environment"
@@ -71,18 +71,18 @@ if [ $# -lt $REQUIRED_ARGS ]; then
   show_help
 fi
 
-FORCING_FILE=$1
+FORCING_DIR=$1
 CONFIG_FILE=$2
 FORECAST_DIR=$3
 shift $REQUIRED_ARGS
 
-echo "FORCING_FILE: ${FORCING_FILE}"
+echo "FORCING_DIR: ${FORCING_DIR}"
 echo "CONFIG_FILE: ${CONFIG_FILE}"
 echo "FORECAST_DIR: ${FORECAST_DIR}"
 
 # Check if the forcing data exists
-if [ ! -f "${FORCING_FILE}" ]; then
-  echo "Forcing data not found at ${FORCING_FILE}"
+if [[ ! -f "${FORCING_DIR}" && ! -d "${FORCING_DIR}" ]]; then
+  echo "Forcing data not found at ${FORCING_DIR}"
 fi
 
 # Check if the configuration file exists
@@ -124,9 +124,9 @@ fi
 # Run the Python script, redirecting its output if an output file is provided
 echo "   Running $(basename "$SCRIPT_PATH") with input file: $CONFIG_FILE"
 if [ -z "$STDOUT_FILE" ]; then
-  python "${SCRIPT_PATH}" "${FORCING_FILE}" "${CONFIG_FILE}" "${FORECAST_DIR}"
+  python "${SCRIPT_PATH}" "${FORCING_DIR}" "${CONFIG_FILE}" "${FORECAST_DIR}"
 else
-  python "${SCRIPT_PATH}" "${FORCING_FILE}" "${CONFIG_FILE}" "${FORECAST_DIR}" &> "${STDOUT_FILE}" 2>&1
+  python "${SCRIPT_PATH}" "${FORCING_DIR}" "${CONFIG_FILE}" "${FORECAST_DIR}" &> "${STDOUT_FILE}" 2>&1
 fi
 
 python_exit_code=$?

python/log_level.py

@@ -0,0 +1,64 @@
+import logging
+import os
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+def create_timestamp() -> str:
+    now = datetime.now(timezone.utc)
+    return now.strftime("%Y-%m-%d")
+
+
+def log_level_set():
+    '''
+    Set logging level and specify logger configuration.
+    
+    Arguments
+    ---------
+    input_parameters (dict): User input logging parameters
+    
+    Returns
+    -------
+    None
+    
+    Notes
+    -----
+    In the absense of user-specified logging level, level defaults to DEBUG
+    See also https://docs.python.org/3/library/logging.html
+    
+    '''
+
+    log_level = 'INFO'
+    if True:
+        BASE_DIR = Path(__file__).resolve().parent.parent
+
+        if Path("/ngencerf/data").exists():
+            log_file_dir = Path(f'/ngencerf/data/run-logs/ngen_fcst_{create_timestamp()}/')
+        else:
+            log_file_dir = Path(BASE_DIR) / f'run-logs/ngen_fcst_{create_timestamp()}/'
+
+        log_file_name = "ngen_fcst.log"
+        os.makedirs(log_file_dir, exist_ok=True)
+        logFilePath = os.path.join(log_file_dir, log_file_name)
+        try:
+            logFile = open(logFilePath, "w")
+            print(f"Logging into: {logFilePath}")
+        except IOError:
+            print(f"Can't Open local directory Log File: {logFilePath}", file=sys.stderr)
+
+        logging.Formatter.converter = time.gmtime
+        logging.basicConfig(
+            force=True,
+            level=log_level,
+            format='%(asctime)s.%(msecs)03d NGEN_FCST %(levelname)s    %(message)s',
+            datefmt='%Y-%m-%dT%H:%M:%S',
+            handlers=[
+                logging.FileHandler(logFilePath, mode='a'),  # Log to a file
+                #logging.StreamHandler(sys.stdout)
+            ])
+    else:
+        logging.basicConfig(
+            level=log_level,
+            format='%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)s - %(funcName)s]: %(message)s',
+            stream=sys.stderr,
+        )

python/process_forcing.py

@@ -0,0 +1,79 @@
+import logging
+from pathlib import Path
+import netCDF4
+import pandas as pd
+import geopandas as gpd
+
+from log_level import log_level_set
+
+# setup the logger
+log_level_set()
+logger = logging.getLogger(__name__)
+
+def update_forcing_in_realization(
+        forc_file: Path, 
+        real_config: dict, 
+        gpkg_file:Path,
+) -> dict:
+    """
+    Read forcing file(s) to retrieve start time and end time of the forcing data, 
+    and adjust the realization configuration accordingly:
+        1) update forcing information
+        2) update start and end times
+    
+    Arguments
+    ---------
+    forc_file: file path to forcing data (a single .nc file or a folder containing a csv file for each catchment)
+    real_config: dictionary containing the realization configuration
+    gpkg_file: file path the GeoPacakge file
+    
+    Returns
+    -------
+    dictionary containing the adjusted realization config  
+    
+    """
+
+    # make sure forcing is provided via a single netcdf file or a folder containing a csv file for each catchment
+    if forc_file.is_dir():
+        # if it is a dir, the folder must contain a csv file for each catchment in the gpkg file
+        catids = gpd.read_file(gpkg_file, layer='divides')['divide_id'].tolist()
+        cats = [c1 for c1 in catids if not Path(forc_file,c1 + '.csv').exists()]
+        if len(cats) > 0:
+            raise FileNotFoundError(f'csv files not found in {forc_file} for these catchments: {cats}')
+        else:
+            # get start and end times from one of the csv files
+            df1 = pd.read_csv(Path(forc_file, catids[0] + '.csv'))
+            start_time = pd.to_datetime(df1['Time'].iloc[0],format="%Y-%m-%d %H:%M:%S")
+            end_time = pd.to_datetime(df1['Time'].iloc[-1],format="%Y-%m-%d %H:%M:%S")
+
+            # update realization file for forcing
+            real_config['global']['forcing'] = dict([
+                                                ('file_pattern', '.*{{id}}.*.csv'),
+                                                ('path', str(forc_file)), 
+                                                ('provider', 'CsvPerFeature')])
+
+    elif forc_file.is_file():
+        # read start and end times from netcdf file
+        try:
+            with netCDF4.Dataset(forc_file, 'r') as ncvar:
+                t0 = pd.to_datetime(ncvar.model_initialization_time, format="%Y-%m-%d_%H:%M:%S")
+                times = [t1 for t1 in ncvar['Time']]
+                start_time = t0 + pd.Timedelta(seconds=3600)
+                end_time = t0 + pd.Timedelta(seconds=(times[-1] - times[0] + 60) * 60)
+
+                # update forcing in realization file
+                real_config['global']['forcing'] = dict([('path', str(forc_file)), ('provider', 'NetCDF')])
+
+        except Exception:
+            logger.error(f'{forc_file} is not a valid NetCDF file')
+    else:
+        raise Exception(f'{forc_file} must be a valid NetCDF file or a folder containing a csv file for each catchment in {gpkg_file}')
+
+    logger.info(f'Start time: {start_time}')
+    logger.info(f'End time: {end_time}')
+
+    # update time period in realization file
+    real_config['time']['start_time'] = str(start_time)
+    real_config['time']['end_time'] = str(end_time)
+
+    return real_config

python/read_output.py

@@ -0,0 +1,63 @@
+import json
+from pathlib import Path
+
+import geopandas as gpd
+import pandas as pd
+import netCDF4
+
+def read_troute_output(
+        gage0: str,
+        cwt_file:Path,
+        gpkg_file: Path,
+        out_file:Path,
+) -> pd.DataFrame:
+
+    """
+    Arguments:
+    ---------
+    gage0: gage ID to retrieve streamflow simulations for
+    cwt_file: path to crosswalk file mapping gage to catchments
+    gpkg_file: path to geopackage file
+    out_file: path to t-route output file (in NetCDF format)
+
+    Returns:
+    ---------
+    dataframe containing time and streamflow simulations
+    
+    """
+    # Handle crosswalk file (in order to get the correct feature_id when reading t-route data)
+    x_walk = pd.Series(dtype=object)
+    try:
+        with open(cwt_file) as fp:
+            data = json.load(fp)
+            for id, values in data.items():
+                gage = values.get('Gage_no')
+                if gage:
+                    if not isinstance(gage, str):
+                        gage = gage[0]
+                    if gage==gage0:
+                        x_walk[id] = gage
+                        break
+    except FileNotFoundError:
+        raise FileNotFoundError(f"Crosswalk file '{cwt_file}' not found.")
+    except json.JSONDecodeError:
+        raise ValueError(f"Failed to parse JSON from crosswalk file '{cwt_file}'.")
+
+    if x_walk.empty:
+        raise Exception(f'{gage0} is not found in crosswalk file {cwt_file}')
+
+    # get catchment at basin outlet for reading from t-route output
+    catchment_hydro_fabric = gpd.read_file(gpkg_file, layer='divides')
+    catchment_hydro_fabric.set_index('id', inplace=True)
+    nexus_id = catchment_hydro_fabric.loc[x_walk.index[0].replace('cat', 'wb')]['toid']
+    wb_lst = [x.split('-')[1] for x in list(catchment_hydro_fabric.query('toid==@nexus_id').index)]
+
+    # read troute output
+    ncvar = netCDF4.Dataset(out_file, "r")
+    fid_index = [list(ncvar['feature_id'][0:]).index(int(fid)) for fid in wb_lst]
+    output = pd.DataFrame(data={'sim_flow': pd.DataFrame(ncvar['flow'][fid_index], index=fid_index).T.sum(axis=1)})
+    t0 = pd.to_datetime(ncvar.file_reference_time, format="%Y-%m-%d_%H:%M:%S")
+    output.index = [t0 + pd.Timedelta(seconds=int(t1)) for t1 in ncvar['time']]
+    output.index.name = 'Time'
+
+    return output