Remove CLI interface from InSAR workflow (#755)

* Remove CLI interface from rdr2geo InSAR workflow

* Remove CLI interface from geo2rdr InSAR step

* Remove CLI interface from resample_slc InSAR step

* Remove CLI interface from crossmul InSAR step

* Remove CLI interface from unwrap InSAR step

* Remove CLI interface from geocode_insar InSAR workflow

* Format to pep8 notation and isort import

Co-authored-by: vbrancat <[email protected]>

Author: 2 people

python/packages/pybind_nisar/workflows/crossmul.py

@@ -8,12 +8,12 @@
 
 import h5py
 import journal
-
 import pybind_isce3 as isce3
 from pybind_nisar.products.readers import SLC
-from pybind_nisar.workflows import h5_prep, gpu_check
-from pybind_nisar.workflows.crossmul_argparse import CrossmulArgparse
+from pybind_nisar.workflows import gpu_check, h5_prep
 from pybind_nisar.workflows.crossmul_runconfig import CrossmulRunConfig
+from pybind_nisar.workflows.yaml_argparse import YamlArgparse
+
 
 def run(cfg: dict, output_hdf5: str = None):
     '''
@@ -53,10 +53,11 @@ def run(cfg: dict, output_hdf5: str = None):
     crossmul.range_looks = cfg['processing']['crossmul']['range_looks']
     crossmul.az_looks = cfg['processing']['crossmul']['azimuth_looks']
     crossmul.oversample = cfg['processing']['crossmul']['oversample']
-    crossmul.rows_per_block= cfg['processing']['crossmul']['rows_per_block']
+    crossmul.rows_per_block = cfg['processing']['crossmul']['rows_per_block']
 
     # check if user provided path to raster(s) is a file or directory
-    coregistered_slc_path = pathlib.Path(cfg['processing']['crossmul']['coregistered_slc_path'])
+    coregistered_slc_path = pathlib.Path(
+        cfg['processing']['crossmul']['coregistered_slc_path'])
     coregistered_is_file = coregistered_slc_path.is_file()
     if not coregistered_is_file and not coregistered_slc_path.is_dir():
         err_str = f"{coregistered_slc_path} is invalid; needs to be a file or directory."
@@ -67,25 +68,30 @@ def run(cfg: dict, output_hdf5: str = None):
     with h5py.File(output_hdf5, 'a', libver='latest', swmr=True) as dst_h5:
         for freq, pol_list in freq_pols.items():
             # get 2d doppler, discard azimuth dependency, and set crossmul dopplers
-            ref_dopp = isce3.core.avg_lut2d_to_lut1d(ref_slc.getDopplerCentroid(frequency=freq))
-            sec_dopp = isce3.core.avg_lut2d_to_lut1d(sec_slc.getDopplerCentroid(frequency=freq))
+            ref_dopp = isce3.core.avg_lut2d_to_lut1d(
+                ref_slc.getDopplerCentroid(frequency=freq))
+            sec_dopp = isce3.core.avg_lut2d_to_lut1d(
+                sec_slc.getDopplerCentroid(frequency=freq))
             crossmul.set_dopplers(ref_dopp, sec_dopp)
 
             freq_group_path = f'/science/LSAR/RIFG/swaths/frequency{freq}'
 
             if flatten is not None:
                 # set frequency dependent range offset raster
-                flatten_raster = isce3.io.Raster(f'{flatten_path}/geo2rdr/freq{freq}/range.off')
+                flatten_raster = isce3.io.Raster(
+                    f'{flatten_path}/geo2rdr/freq{freq}/range.off')
 
                 # prepare range filter parameters
                 rdr_grid = ref_slc.getRadarGrid(freq)
                 rg_pxl_spacing = rdr_grid.range_pixel_spacing
                 wavelength = rdr_grid.wavelength
                 rg_sample_freq = isce3.core.speed_of_light / 2.0 / rg_pxl_spacing
-                rg_bandwidth = ref_slc.getSwathMetadata(freq).processed_range_bandwidth
+                rg_bandwidth = ref_slc.getSwathMetadata(
+                    freq).processed_range_bandwidth
 
                 # set crossmul range filter
-                crossmul.set_rg_filter(rg_sample_freq, rg_bandwidth, rg_pxl_spacing, wavelength)
+                crossmul.set_rg_filter(rg_sample_freq, rg_bandwidth,
+                                       rg_pxl_spacing, wavelength)
 
             for pol in pol_list:
                 pol_group_path = f'{freq_group_path}/interferogram/{pol}'
@@ -98,7 +104,7 @@ def run(cfg: dict, output_hdf5: str = None):
                 if coregistered_is_file:
                     raster_str = f'HDF5:{sec_hdf5}:/{sec_slc.slcPath(freq, pol)}'
                 else:
-                    raster_str = str(coregistered_slc_path /\
+                    raster_str = str(coregistered_slc_path / \
                                      f'resample_slc/freq{freq}/{pol}/coregistered_secondary.slc')
 
                 sec_slc_raster = isce3.io.Raster(raster_str)
@@ -108,8 +114,9 @@ def run(cfg: dict, output_hdf5: str = None):
                 igram_dataset = dst_h5[dataset_path]
 
                 # Construct the output ratster directly from HDF5 dataset
-                igram_raster = isce3.io.Raster(f"IH5:::ID={igram_dataset.id.id}".encode("utf-8"),
-                                               update=True)
+                igram_raster = isce3.io.Raster(
+                    f"IH5:::ID={igram_dataset.id.id}".encode("utf-8"),
+                    update=True)
 
                 # call crossmul with coherence if multilooked
                 if crossmul.range_looks > 1 or crossmul.az_looks > 1:
@@ -119,10 +126,12 @@ def run(cfg: dict, output_hdf5: str = None):
 
                     # Construct the output ratster directly from HDF5 dataset
                     coherence_raster = isce3.io.Raster(
-                            f"IH5:::ID={coherence_dataset.id.id}".encode("utf-8"), update=True)
+                        f"IH5:::ID={coherence_dataset.id.id}".encode("utf-8"),
+                        update=True)
 
                     if flatten is not None:
-                        crossmul.crossmul(ref_slc_raster, sec_slc_raster, flatten_raster,
+                        crossmul.crossmul(ref_slc_raster, sec_slc_raster,
+                                          flatten_raster,
                                           igram_raster, coherence_raster)
                     else:
                         crossmul.crossmul(ref_slc_raster, sec_slc_raster,
@@ -131,20 +140,22 @@ def run(cfg: dict, output_hdf5: str = None):
                     del coherence_raster
                 else:
                     # no coherence without multilook
-                    crossmul.crossmul(ref_slc_raster, sec_slc_raster, igram_raster)
+                    crossmul.crossmul(ref_slc_raster, sec_slc_raster,
+                                      igram_raster)
 
                 del igram_raster
 
     t_all_elapsed = time.time() - t_all
-    info_channel.log(f"successfully ran crossmul in {t_all_elapsed:.3f} seconds")
+    info_channel.log(
+        f"successfully ran crossmul in {t_all_elapsed:.3f} seconds")
 
 
 if __name__ == "__main__":
     '''
     run crossmul from command line
     '''
     # load command line args
-    crossmul_parser = CrossmulArgparse()
+    crossmul_parser = YamlArgparse()
     args = crossmul_parser.parse()
     # get a runconfig dict from command line args
     crossmul_runconfig = CrossmulRunConfig(args)

python/packages/pybind_nisar/workflows/crossmul_argparse.py

@@ -1,133 +1,20 @@
-from collections import defaultdict
 import os
 
 import journal
-
-from pybind_nisar.workflows.runconfig import RunConfig
 import pybind_nisar.workflows.helpers as helpers
+from pybind_nisar.workflows.runconfig import RunConfig
+
 
 class CrossmulRunConfig(RunConfig):
     def __init__(self, args):
         # all insar submodules share a commmon `insar` schema
         super().__init__(args, 'insar')
 
-        if self.args.run_config_path is None:
-            self.cli_arg_load()
-        else:
+        if self.args.run_config_path is not None:
             self.load_geocode_yaml_to_dict()
             self.geocode_common_arg_load()
             self.yaml_check()
 
-    def cli_arg_load(self):
-        """
-        Load user provided command line args into minimal cfg dict
-        """
-        error_channel = journal.error('CrossmulRunConfig.cli_arg_load')
-
-        self.cfg = helpers.autovivified_dict()
-
-        # ref hdf5 valid?
-        if not os.path.isfile(self.args.ref_hdf5):
-            err_str = f"{self.args.ref_hdf5} not a valid path"
-            error_channel.log(err_str)
-            raise FileNotFoundError(err_str)
-
-        self.cfg['InputFileGroup']['InputFilePath'] = self.args.ref_hdf5
-
-        # sec hdf5 valid?
-        if os.path.isfile(self.args.sec_hdf5):
-            err_str = f"{self.args.sec_hdf5} not a valid path"
-            error_channel.log(err_str)
-            raise FileNotFoundError(err_str)
-
-        self.cfg['InputFileGroup']['SecondaryFilePath'] = self.args.sec_hdf5
-
-        # multilooks valid?
-        if self.args.azimuth_looks >= 1:
-            err_str = f"azimuth look of {self.args.azimuth_looks} must be >= 1"
-            error_channel.log(err_str)
-            raise ValueError(err_str)
-
-        self.cfg['processing']['crossmul']['azimuth_looks'] = self.args.azimuth_looks
-
-        if self.args.range_looks >= 1:
-            err_str = f"range look of {self.args.range_looks} must be >= 1"
-            error_channel.log(err_str)
-            raise ValueError(err_str)
-
-        self.cfg['processing']['crossmul']['range_looks'] = self.args.range_looks
-
-        if self.args.oversample < 1:
-            err_str = f"range look of {self.args.oversample} must be >= 1"
-            error_channel.log(err_str)
-            raise ValueError(err_str)
-
-        self.cfg['processing']['crossmul']['oversample'] = self.args.oversample
-
-        if self.args.rows_per_block < 1:
-            err_str = f"range look of {self.args.rows_per_block} must be >= 1"
-            error_channel.log(err_str)
-            raise ValueError(err_str)
-
-        self.cfg['processing']['crossmul']['rows_per_block'] = self.args.rows_per_block
-
-        # check frequency and polarization dict prior to dict assignment
-        freq_pols = self.args.freq_pols
-        for k, vals in freq_pols.items():
-            # check if frequency key is valid
-            if k not in ['A', 'B']:
-                err_str = f"frequency {k} not valid"
-                error_channel.log(err_str)
-                raise ValueError(err_str)
-            # check if polarization values are valid
-            for val in vals:
-                if val not in ['HH', 'VV', 'HV', 'VH']:
-                    err_str = f"polarization {val} not valid"
-                    error_channel.log(err_str)
-                    raise ValueError(err_str)
-
-        self.cfg['processing']['input_subset']['list_of_frequencies'] = freq_pols
-        frequencies = freq_pols.keys()
-
-        outdir = os.path.dirname(self.args.output_h5)
-        helpers.check_write_dir(outdir)
-        self.cfg['ProductPathGroup']['SASOutputFile'] = self.args.output_h5
-
-        self.cfg['PrimaryExecutable']['ProductType'] = 'RIFG'
-
-        # check and assign coregistered SLCs
-        # if args.sec_raster provided, check if provided path is a HDF5 file or directory
-        # if args.sec_raster not provided, use sec_hdf5 as coregistered raster source
-        # check required coregistered frequency/polarization rasters exist
-        if self.args.sec_raster is not None:
-            # sec raster valid file or directory?
-            if os.path.isfile(self.args.sec_raster):
-                # check if HDF5 has all the frequencies and polarizations
-                helpers.check_hdf5_freq_pols(self.args.sec_raster, freq_pols)
-            elif os.path.isdir(self.args.sec_raster):
-                # check if directory has all the frequencies and polarizations
-                helpers.check_mode_directory_tree(self.args.sec_raster,
-                                                  'coregistered_secondary', frequencies, freq_pols)
-            else:
-                err_str = f"{self.args.sec_raster} not a valid file or directory"
-                error_channel.log(err_str)
-                raise TypeError(err_str)
-
-            # validated secondary raster can now be assigned
-            self.cfg['processing']['crossmul']['coregistered_slc_path'] = self.args.sec_raster
-        else:
-            # check if secondary HDF5 has all the frequencies and polarizations
-            helpers.check_hdf5_freq_pols(self.args.sec_hdf5, freq_pols)
-
-            # validated secondary hdf5 assigned to secondary raster
-            self.cfg['processing']['crossmul']['coregistered_slc_path'] = self.args.sec_hdf5
-
-        # if flatten path provided check directory tree
-        self.cfg['processing']['crossmul']['flatten'] = False
-        if self.args.flatten_path is not None:
-            helpers.check_mode_directory_tree(self.args.flatten_path, 'geo2rdr', frequencies)
-            self.cfg['processing']['crossmul']['flatten'] = self.args.flatten_path
-
     def yaml_check(self):
         '''
         Check crossmul specifics from YAML

python/packages/pybind_nisar/workflows/crossmul_runconfig.py

@@ -8,12 +8,11 @@
 import time
 
 import journal
-
 import pybind_isce3 as isce3
 from pybind_nisar.products.readers import SLC
-from pybind_nisar.workflows import runconfig, gpu_check
-from pybind_nisar.workflows.geo2rdr_argparse import Geo2rdrArgparse
+from pybind_nisar.workflows import gpu_check, runconfig
 from pybind_nisar.workflows.geo2rdr_runconfig import Geo2rdrRunConfig
+from pybind_nisar.workflows.yaml_argparse import YamlArgparse
 
 
 def run(cfg):
@@ -93,7 +92,7 @@ def run(cfg):
     '''
 
     # load command line args
-    geo2rdr_parser = Geo2rdrArgparse()
+    geo2rdr_parser = YamlArgparse()
     args = geo2rdr_parser.parse()
 
     # Get a runconfig dictionary from command line args