Merge pull request #396 from nasa/release/3.0.0-er.5.1

DISP-S1 Release v3.0.0-er.5.1

.ci/docker/Dockerfile_disp_s1

@@ -52,12 +52,15 @@ RUN set -ex \
     && chmod +x ${CONDA_ROOT}/bin/*_entrypoint.sh \
     && source /usr/local/bin/_activate_current_env.sh \
     && ${MAMBA_EXE} install --yes --channel conda-forge --file ${PGE_DEST_DIR}/opera/requirements.txt \
-    && ${MAMBA_EXE} install --yes --channel conda-forge eccodes jpeg \
     && chmod 775 ${PGE_DEST_DIR} \
+    # Create a separate conda environment for the eccodes install needed to run grib_to_netcdf
+    && ${MAMBA_EXE} create -n eccodes \
+    && ${MAMBA_EXE} install -n eccodes --yes --channel conda-forge eccodes jpeg \
+    # Install grib_to_netcdf and make it available to the eccodes conda env
     && curl -o /tmp/eccodes-2.28.1-1.x86_64.rpm https://artifactory-fn.jpl.nasa.gov:443/artifactory/general/gov/nasa/jpl/opera/sds/pge/disp_s1/eccodes/eccodes-2.28.1-1.x86_64.rpm \
     && dnf install -y /tmp/eccodes-2.28.1-1.x86_64.rpm \
     && rm -rf /tmp/eccodes-2.28.1-1.x86_64.rpm \
-    && ln -sf /opt/eccodes/bin/grib_to_netcdf /opt/conda/bin/
+    && ln -sf /opt/eccodes/bin/grib_to_netcdf /opt/conda/envs/eccodes/bin/
 
 # Set the Docker entrypoint and clear the default command
 ENTRYPOINT ["sh", "-c", "source /usr/local/bin/_activate_current_env.sh;exec ${CONDA_ROOT}/bin/pge_docker_entrypoint.sh \"${@}\"", "--"]

examples/disp_s1_sample_runconfig-v3.0.0-er.5.0.yaml → examples/disp_s1_sample_runconfig-v3.0.0-er.5.1.yaml

@@ -181,7 +181,10 @@ RunConfig:
         ProgramOptions: []
 
       DebugLevelGroup:
-        # Set to True to enable Debug mode (TODO this is currently a no-op)
+        # Set to True to enable Debug mode
+        # For the DISP-S1 PGE, enabling Debug mode will bypass the input product
+        # validation step, which can be useful when executing the PGE with
+        # an incomplete set of ancillary inputs (missing bursts, etc...)
         DebugSwitch: False
 
         # Set to True to have the PGE invoke the SAS/QA executables via

src/opera/pge/disp_s1/disp_s1_pge.py

@@ -58,7 +58,10 @@ def run_preprocessor(self, **kwargs):
         """
         super().run_preprocessor(**kwargs)
 
-        validate_disp_inputs(self.runconfig, self.logger, self.name)
+        # If debug mode is enabled, skip the input validation, since we might
+        # be working with only a partial set of inputs/ancillaries
+        if not self.runconfig.debug_switch:
+            validate_disp_inputs(self.runconfig, self.logger, self.name)
 
         validate_algorithm_parameters_config(self.name,
                                              self.runconfig.algorithm_parameters_schema_path,
@@ -85,21 +88,27 @@ def convert_troposphere_model_files(self):
             if splitext(tropo_file)[-1] == '.grb':
                 # change the extension to .nc
                 netcdf_file = join(scratch_dir, splitext(basename(tropo_file))[0] + '.nc')
+
                 # This list of will the new paths to the converted files in the in-memory runconfig file.
-                netcdf_file_list.append(netcdf_file)
                 grib_file_name = tropo_file
                 subprocess.run(
                     [
-                        "grib_to_netcdf",
+                        "/opt/conda/envs/eccodes/bin/grib_to_netcdf",
                         "-D",
                         "NC_FLOAT",
                         "-o",
                         netcdf_file,
                         grib_file_name[:-4],
                     ],
+                    env={'ENV_NAME': 'eccodes', 'LD_LIBRARY_PATH': '/opt/conda/envs/eccodes/lib'},
                     shell=False,
                     check=False,
                 )
+            else:
+                # no conversion necessary, carry NetCDF file along as-is
+                netcdf_file = tropo_file
+
+            netcdf_file_list.append(netcdf_file)
 
         # Update the in-memory runconfig instance
         self.runconfig.sas_config['dynamic_ancillary_file_group']['troposphere_files'] = netcdf_file_list
@@ -726,6 +735,8 @@ class DispS1Executor(DispS1PreProcessorMixin, DispS1PostProcessorMixin, PgeExecu
     LEVEL = "L3"
     """Processing Level for DISP-S1 Products"""
 
+    PGE_VERSION = "3.0.0-er.5.1"
+
     SAS_VERSION = "0.1.0"  # Beta release https://github.com/opera-adt/disp-s1/releases/tag/v0.1.0
     """Version of the SAS wrapped by this PGE, should be updated as needed"""
 

src/opera/test/pge/disp_s1/test_disp_s1_pge.py

@@ -41,7 +41,7 @@ def mock_grib_to_netcdf(*popenargs, input=None, capture_output=False, timeout=No
     here we do not want this mocked version to run.
 
     """
-    if popenargs[0][0] == "grib_to_netcdf":
+    if popenargs[0][0].endswith("grib_to_netcdf"):
         output_path = popenargs[0][4]
         open(output_path, 'w').close()
     else:
@@ -863,38 +863,49 @@ def test_disp_s1_pge_validate_product_output(self):
     def test_scratch_sas_runconfig_for_grib_to_netcdf_files(self):
         """
         Test that the grib_to_netcdf files are in 'disp_s1_pge_test/scratch_dir'.
-        Verify the file names of the .grb files found in ['SAS']['dynamic_ancillary_file_group']['troposphere_files']
-        are the same as the converted files that are now .nc files and are residing in scratch_dir.
+
+        Verify the file names of the .grb files found in
+        ['SAS']['dynamic_ancillary_file_group']['troposphere_files']
+        are the same as the converted files that are now .nc files and are
+        residing in scratch_dir.
         """
         starting_grb_file_names = []
         ending_grb_file_names = []
         runconfig_path = join(self.data_dir, 'test_disp_s1_config.yaml')
 
         pge = DispS1Executor(pge_name="DispS1PgeTest", runconfig_path=runconfig_path)
+
         # Pull out the starting value of the troposphere_files
         with open(runconfig_path, 'r', encoding='utf-8') as infile:
             runconfig_dict = yaml.safe_load(infile)
         starting_tropo_paths = \
             runconfig_dict['RunConfig']['Groups']['SAS']['dynamic_ancillary_file_group']['troposphere_files']
+
         # Strip the path and extension (.grb) in order to compare just the file names after
         # conversion and placement of the converted file into /scratch_dir/<fname>.nc
         for starting_path in starting_tropo_paths:
             if starting_path[-4:] == '.grb':
                 starting_grb_file_names.append(starting_path.split('/')[-1][:-4])
+
         # Run only the preprocessor and the sas_executable, so the temporary directories are created and still alive.
         pge.run_preprocessor()
         pge.run_sas_executable()
         temp_sas_runconfig = 'disp_s1_pge_test/scratch_dir/' + pge.runconfig.filename.split('/')[-1][:-5] + '_sas.yaml'
+
         # open the yaml file
         with open(temp_sas_runconfig, 'r') as file:
             data = yaml.safe_load(file)
+
         ending_tropo_paths = data['dynamic_ancillary_file_group']['troposphere_files']
-        # Verify the .grb files are in /scratch_dir
+
+        # Verify the converted .grb files are in /scratch_dir
         # Strip the path and extension (changed to .nc in this case) to allow comparison of file name only.
         for ending_path in ending_tropo_paths:
-            self.assertIn('scratch_dir', ending_path)
-            self.assertTrue(exists(ending_path))        # verify the files exist on disk
-            ending_grb_file_names.append(ending_path.split('/')[-1][:-3])
+            ending_file_name = os.path.splitext(os.path.basename(ending_path))[0]
+            if ending_file_name in starting_grb_file_names:
+                self.assertIn('scratch_dir', ending_path)
+                self.assertTrue(exists(ending_path))        # verify the files exist on disk
+                ending_grb_file_names.append(ending_file_name)
 
         self.assertEqual(starting_grb_file_names, ending_grb_file_names)
 

src/opera/util/input_validation.py

@@ -139,7 +139,7 @@ def validate_slc_s1_inputs(runconfig, logger, name):
             logger.critical(name, ErrorCode.INVALID_INPUT, error_msg)
 
 
-def get_burst_id_set(input_file_group, logger, name):
+def get_burst_id_set(input_file_group : list, logger, name) -> set:
     """
     Compiles a set of burst_ids from a list of files defined in the runconfig file.
     Each file in the list should have a burst_id in the file name.
@@ -155,7 +155,8 @@ def get_burst_id_set(input_file_group, logger, name):
 
     Returns
     -------
-    burst_ids :set
+    burst_ids : set
+        The unique set of burst IDs parsed from the list of input file names.
 
     Raises
     ------
@@ -176,7 +177,8 @@ def get_burst_id_set(input_file_group, logger, name):
     return burst_ids
 
 
-def check_disp_s1_ancillary_burst_ids(cslc_input_burst_ids, ancillary_file_list, logger, name):
+def check_disp_s1_ancillary_burst_ids(cslc_input_burst_ids : set,
+                                      ancillary_file_list : list, logger, name):
     # pylint: disable=C0103
     """
     Verify burst_ids from the ancillary input files:
@@ -207,7 +209,7 @@ def check_disp_s1_ancillary_burst_ids(cslc_input_burst_ids, ancillary_file_list,
 
     """
     nl, tab, dtab = '\n', '\t', '\t\t'   # used to format log output in fstrings.
-    ancillary_burst_ids = get_burst_id_set(ancillary_file_list, logger, name)
+    ancillary_burst_ids : set = get_burst_id_set(ancillary_file_list, logger, name)
 
     # Test none of the ancillary inputs have the same burst ID
     if len(ancillary_burst_ids) != len(ancillary_file_list):
@@ -264,8 +266,8 @@ def get_cslc_input_burst_id_set(cslc_input_file_list, logger, name):
                                              cslc_input_file_list))
     single_input_file_list = list(set(cslc_input_file_list) - set(compressed_input_file_list))
 
-    compressed_file_burst_id_set = get_burst_id_set(compressed_input_file_list, logger, name)
-    single_file_burst_id_set = get_burst_id_set(single_input_file_list, logger, name)
+    compressed_file_burst_id_set : set = get_burst_id_set(compressed_input_file_list, logger, name)
+    single_file_burst_id_set : set = get_burst_id_set(single_input_file_list, logger, name)
 
     # Case 1:  uncompressed files only in cslc inputs
     if len(compressed_file_burst_id_set) == 0:
@@ -316,43 +318,48 @@ def validate_disp_inputs(runconfig, logger, name):
         logger, name
     )
 
-    if 'amplitude_dispersion_files' in dyn_anc_file_group:
+    if ('amplitude_dispersion_files' in dyn_anc_file_group and
+            len(dyn_anc_file_group['amplitude_dispersion_files']) > 0):
         check_input_list(dyn_anc_file_group['amplitude_dispersion_files'], logger, name,
                          valid_extensions=('.tif', '.tiff'), check_zero_size=True)
         check_disp_s1_ancillary_burst_ids(cslc_burst_id_set,
                                           dyn_anc_file_group['amplitude_dispersion_files'],
                                           logger,
                                           name)
 
-    if 'amplitude_mean_files' in dyn_anc_file_group:
+    if ('amplitude_mean_files' in dyn_anc_file_group and
+            len(dyn_anc_file_group['amplitude_mean_files']) > 0):
         check_input_list(dyn_anc_file_group['amplitude_mean_files'], logger, name,
                          valid_extensions=('.tif', '.tiff'), check_zero_size=True)
         check_disp_s1_ancillary_burst_ids(cslc_burst_id_set,
                                           dyn_anc_file_group['amplitude_mean_files'],
                                           logger,
                                           name)
 
-    if 'static_layers_files' in dyn_anc_file_group:
+    if ('static_layers_files' in dyn_anc_file_group and
+            len(dyn_anc_file_group['static_layers_files']) > 0):
         check_input_list(dyn_anc_file_group['static_layers_files'], logger, name,
                          valid_extensions=('.h5',), check_zero_size=True)
         check_disp_s1_ancillary_burst_ids(cslc_burst_id_set,
                                           dyn_anc_file_group['static_layers_files'],
                                           logger,
                                           name)
 
-    if 'mask_file' in dyn_anc_file_group:
+    if 'mask_file' in dyn_anc_file_group and dyn_anc_file_group['mask_file']:
         check_input(dyn_anc_file_group['mask_file'], logger, name,
                     valid_extensions=('.tif', '.tiff', '.vrt', '.flg'), check_zero_size=True)
 
-    if 'dem_file' in dyn_anc_file_group:
+    if 'dem_file' in dyn_anc_file_group and dyn_anc_file_group['dem_file']:
         check_input(dyn_anc_file_group['dem_file'], logger, name,
                     valid_extensions=('.tif', '.tiff', '.vrt'), check_zero_size=True)
 
-    if 'ionosphere_files' in dyn_anc_file_group:
+    if ('ionosphere_files' in dyn_anc_file_group
+            and len(dyn_anc_file_group['ionosphere_files']) > 0):
         check_input_list(dyn_anc_file_group['ionosphere_files'], logger, name,
                          check_zero_size=True)
 
-    if 'troposphere_files' in dyn_anc_file_group:
+    if ('troposphere_files' in dyn_anc_file_group and
+            len(dyn_anc_file_group['troposphere_files']) > 0):
         check_input_list(dyn_anc_file_group['troposphere_files'], logger, name,
                          valid_extensions=('.nc', '.h5', '.grb'), check_zero_size=True)