End2end workflow (#802)

* Added end-to-end workflow in CI workflow test suite
* Added two end-to-end tests datasets: ALOS PALSAR data over Chile and Rosamond

Author: yzheng6

tools/imagesets/centos7conda/distrib_nisar/Dockerfile

@@ -11,8 +11,8 @@ RUN cd /opt \
  && git clone https://[email protected]/NISAR-ADT/QualityAssurance \
  && git clone https://[email protected]/NISAR-ADT/CFChecker \
  && git clone https://[email protected]/NISAR-ADT/calTools \
- && cd /opt/QualityAssurance && git checkout cd455fd && rm -rf .git \
- && cd /opt/CFChecker && git checkout R1.1 && rm -rf .git \
+ && cd /opt/QualityAssurance && git checkout R2 && rm -rf .git \
+ && cd /opt/CFChecker && git checkout R2 && rm -rf .git \
  && cd /opt/calTools && git checkout R2 && rm -rf .git 
 
 FROM $distrib_img

tools/imagesets/imgset.py

@@ -324,7 +324,7 @@ def distribrun(self, testdir, cmd, logfile=None, dataname=None, nisarimg=False,
             --rm -i {self.tty} {img} sh -ci"  
         run_with_logging(dockercall, cmd, logger, printlog=self.printlog)
 
-    def workflowtest(self, wfname, testname, dataname, pyname, arg=""): # hmmmmmmmmm
+    def workflowtest(self, wfname, testname, dataname, pyname, suf="", description="", arg=""): 
         """
         Run the specified workflow test using the distrib image.
         
@@ -334,22 +334,42 @@ def workflowtest(self, wfname, testname, dataname, pyname, arg=""): # hmmmmmmmmm
             Workflow name (e.g. "rslc")
         testname : str
             Workflow test name (e.g. "RSLC_REE1")
-        dataname : str
-            Test input data (e.g. "L0B_RRSD_REE1")
+        dataname : str or iterable of str or None
+            Test input dataset(s) to be mounted (e.g. "L0B_RRSD_REE1", ["L0B_RRSD_REE1", "L0B_RRSD_REE2"]). 
+            If None, no input datasets are used.
         pyname : str
             Name of the isce3 module to execute (e.g. "pybind_nisar.workflows.focus")
+        suf: str
+            Suffix in runconfig and output directory name to differentiate between
+            reference and secondary data in end-to-end tests
+        description: str
+            Extra test description to print out to differentiate between
+            reference and secondary data in end-to-end tests                     
         arg : str, optional
             Additional command line argument(s) to pass to the workflow
         """
-        print(f"\nRunning workflow test {testname}\n")
+        print(f"\nRunning workflow test {testname}{description}\n")
         testdir = os.path.abspath(pjoin(self.testdir, testname))
-        os.makedirs(pjoin(testdir, f"output_{wfname}"), exist_ok=True)
-        os.makedirs(pjoin(testdir, f"scratch_{wfname}"), exist_ok=True)
-        # copy test runconfig to test directory
-        shutil.copyfile(pjoin(runconfigdir, f"{testname}.yaml"), 
-                        pjoin(testdir, f"runconfig_{wfname}.yaml"))
-        log = pjoin(testdir, f"output_{wfname}", "stdouterr.log")
-        cmd = [f"time python3 -m {pyname} {arg} runconfig_{wfname}.yaml"]
+        # create input directories before docker volume mount to avoid root ownership 
+        # of these directories
+        if dataname is not None:
+            if type(dataname) is not list:
+                dataname = [dataname]
+            for data in dataname:
+                os.makedirs(pjoin(testdir, f"input_{data}"), exist_ok=True)
+        # create output directories
+        os.makedirs(pjoin(testdir, f"output_{wfname}{suf}"), exist_ok=True)
+        os.makedirs(pjoin(testdir, f"scratch_{wfname}{suf}"), exist_ok=True)
+        # copy test runconfig to test directory (for end-to-end testing, we need to 
+        # distinguish between the runconfig files for each individual workflow)
+        if testname.startswith("end2end"):
+            inputrunconfig = f"{testname}_{wfname}{suf}.yaml"
+        else:
+            inputrunconfig = f"{testname}{suf}.yaml"
+        shutil.copyfile(pjoin(runconfigdir, inputrunconfig), 
+                        pjoin(testdir, f"runconfig_{wfname}{suf}.yaml"))
+        log = pjoin(testdir, f"output_{wfname}{suf}", "stdouterr.log")
+        cmd = [f"time python3 -m {pyname} {arg} runconfig_{wfname}{suf}.yaml"]
         try:
             self.distribrun(testdir, cmd, logfile=log, dataname=dataname,
                             loghdlrname=f'wftest.{os.path.basename(testdir)}')
@@ -380,6 +400,32 @@ def insartest(self, tests=None):
         for testname, dataname in tests:
             self.workflowtest("insar", testname, dataname, "pybind_nisar.workflows.insar", arg="--restart")
 
+    def end2endtest(self, tests=None):
+        """
+        Run all workflows for one pair of L0B input data, including RSLC, GSLC, GCOV, RIFG, RUNW, GUNW.
+        The GSLC, GCOV, and InSAR products are generated from outputs of the RSLC workflow.
+        """
+        if tests is None:
+            tests = workflowtests['end2end'].items()
+        for testname, dataname in tests:
+            # copy runconfigs and create output direcotories
+            testdir = os.path.abspath(pjoin(self.testdir, testname))
+            for wfname in ['rslc', 'gslc', 'gcov', 'insar']:
+                if wfname == 'rslc':
+                    pyname = 'pybind_nisar.workflows.focus'
+                else:
+                    pyname = f'pybind_nisar.workflows.{wfname}'
+
+                if wfname == 'insar':
+                    self.workflowtest(wfname, testname, dataname, pyname, arg="--restart",
+                                      description=" InSAR product")
+                else:
+                    self.workflowtest(wfname, testname, dataname, pyname, suf="_ref",
+                                      description=f" {wfname.upper()} reference product")
+                    self.workflowtest(wfname, testname, dataname, pyname, suf="_sec",
+                                      description=f" {wfname.upper()} secondary product")
+
+
     def noisesttest(self, tests=None):
         if tests is None:
             tests = workflowtests['noisest'].items()
@@ -443,7 +489,7 @@ def mintests(self):
         self.ptatest(tests=list(workflowtests['pta'].items())[:1])
         self.beamformtest(tests=list(workflowtests['beamform'].items())[:1])
 
-    def workflowqa(self, wfname, testname):
+    def workflowqa(self, wfname, testname, suf="", description=""):
         """
         Run QA and CF compliance checking for the specified workflow using the NISAR distrib image.
         
@@ -453,15 +499,22 @@ def workflowqa(self, wfname, testname):
             Workflow name (e.g. "rslc")
         testname: str
             Workflow test name (e.g. "RSLC_REE1")
-        """
-        print(f"\nRunning workflow QA on test {testname}\n")
+        suf: str
+            Suffix in runconfig and output directory name to differentiate between
+            reference and secondary data in end-to-end tests
+        description: str
+            Extra test description to print out to differentiate between
+            reference and secondary data in end-to-end tests                     
+        """
+        print(f"\nRunning workflow QA on test {testname}{description}\n")
         testdir = os.path.abspath(pjoin(self.testdir, testname))
-        os.makedirs(pjoin(testdir, f"qa_{wfname}"), exist_ok=True)
-        log = pjoin(testdir, f"qa_{wfname}", "stdouterr.log")
-        cmd = [f"time cfchecks.py output_{wfname}/{wfname}.h5",
-               f"""time verify_{wfname}.py --fpdf qa_{wfname}/graphs.pdf \
-                    --fhdf qa_{wfname}/stats.h5 --flog qa_{wfname}/qa.log --validate \
-                    --quality output_{wfname}/{wfname}.h5"""]
+        os.makedirs(pjoin(testdir, f"qa_{wfname}{suf}"), exist_ok=True)
+        log = pjoin(testdir, f"qa_{wfname}{suf}", "stdouterr.log")
+        # run qa command
+        cmd = [f"time cfchecks.py output_{wfname}{suf}/{wfname}.h5",
+               f"""time verify_{wfname}.py --fpdf qa_{wfname}{suf}/graphs.pdf \
+                    --fhdf qa_{wfname}{suf}/stats.h5 --flog qa_{wfname}{suf}/qa.log --validate \
+                    --quality output_{wfname}{suf}/{wfname}.h5"""]
         try:
             self.distribrun(testdir, cmd, logfile=log, nisarimg=True,
                             loghdlrname=f'wfqa.{os.path.basename(testdir)}')
@@ -515,7 +568,7 @@ def insarqa(self, tests=None):
             testdir = os.path.abspath(pjoin(self.testdir, testname))
             # run QA for each of the InSAR products
             for product in ['rifg', 'runw', 'gunw']:
-                print(f"\nRunning workflow QA on InSAR test {testname} product {product.upper()}\n")
+                print(f"\nRunning workflow QA on test {testname} {product.upper()} product\n")
                 qadir = pjoin(testdir, f"qa_{product}")
                 os.makedirs(qadir, exist_ok=True)
                 log = pjoin(qadir,f"stdouterr.log")
@@ -529,11 +582,27 @@ def insarqa(self, tests=None):
                                     loghdlrname=f'wfqa.{os.path.basename(testdir)}.{product}')
                 except subprocess.CalledProcessError as e:
                     if product == 'gunw':
-                        raise RuntimeError(f"Workflow QA on InSAR test {testname} product {product.upper()} failed\n") from e
+                        raise RuntimeError(f"Workflow QA on test {testname} {product.upper()} product failed\n") from e
                     else:
                         # do not exit since CF checker errors are expected
-                        print(f"Found known errors running CF Checker on InSAR test {testname} product {product.upper()}\n")
-       
+                        print(f"Found known errors running CF Checker on test {testname} {product.upper()} product\n")
+      
+    def end2endqa(self, tests=None):
+        """
+        Run QA on all end2end workflow test results for one pair of L0B input data, including RSLC, GSLC, GCOV,
+        RIFG, RUNW, GUNW.
+
+        """
+        if tests is None:
+            tests = workflowtests['end2end'].items()
+        for testname, dataname in tests:
+            for wfname in ['rslc', 'gslc', 'gcov']:
+                for suf, descr in [('_ref', 'reference'), ('_sec', 'secondary')]:
+                    self.workflowqa(wfname, testname, suf=suf, description=f' {wfname.upper()} {descr} product')
+
+            self.insarqa([testname])
+
+
     def minqa(self):
         """
         Only run qa for first test in each workflow

tools/imagesets/runconfigs/end2end_pair_ALPSRP110160680_ALPSRP116870680_Rosamond_gcov_ref.yaml

@@ -0,0 +1,153 @@
+runconfig:
+    name: NISAR_L2-L-GCOV_RUNCONFIG
+
+    groups:
+        PGENameGroup:
+            PGEName: GCOV_L_PGE
+
+        InputFileGroup:
+            # REQUIRED - One NISAR L1B RSLC formatted HDF5 file
+            InputFilePath:
+            - output_rslc_ref/rslc.h5
+
+        DynamicAncillaryFileGroup:
+            # REQUIRED - Use the provided DEM as input
+            DEMFile: input_L0B_RRSD_ALPSRP110160680_Rosamond/dem.tif
+
+        ProductPathGroup:
+            # Directory where PGE will place results. Irrelevant to SAS.
+            ProductPath: output_gcov_ref
+
+            # Directory where SAS can write temporary data
+            ScratchPath: scratch_gcov_ref
+
+            # SAS writes output product to the following file. PGE may rename.
+            # NOTE: For R2 will need to handle mixed-mode case with multiple outputs of RSLC workflow.
+            SASOutputFile: output_gcov_ref/gcov.h5
+
+        PrimaryExecutable:
+            ProductType: GCOV
+
+        DebugLevelGroup:
+            DebugSwitch: false
+
+        #adt section - isce3 + pyre workflow
+        processing:
+            # OPTIONAL - Frequencies and polarisations to be processed
+            input_subset:
+                # OPTIONAL - List of frequencies to process. Default empty representing all
+                list_of_frequencies:
+                    # keys for frequency A and B are required.
+                    # valid options for polarizations
+                    # 'all' for all polarizations found in RSLC
+                    # [polarizations] for list of specific frequency(s) e.g. [HH, HV] or [HH]
+                    # empty for no polarizations
+                    A: 
+
+                # OPTIONAL - If we want full covariance instead of diagonals only. Default False
+                fullcovariance:   False
+
+            # TODO OPTIONAL - Only checked when internet access is available
+            dem_download:
+                # OPTIONAL - s3 bucket / curl URL / local file
+                source: 
+                top_left:
+                    x:
+                    y:
+                bottom_right:
+                    x:
+                    y:
+
+            # OPTIONAL - if amplitude data needs to be mulitlooked before GCOV generation
+            pre_process:   
+               azimuth_looks: 1
+               range_looks:   1
+
+            # OPTIONAL - to control behavior of RTC module
+            rtc:
+                output_type: gamma0
+
+                # OPTIONAL - Choices:
+                # "bilinear_distribution" (default)
+                # "area_projection"
+                algorithm_type: area_projection
+
+                # OPTIONAL - Choices:
+                # "beta0" (default)
+                # "sigma0"
+                input_terrain_radiometry: sigma0
+
+                # OPTIONAL - Minimum RTC area factor in dB
+                rtc_min_value_db: -30
+
+            # OPTIONAL - Mechanism to specify output posting and DEM
+            geocode:
+                # OPTIONAL -
+                algorithm_type: area_projection
+ 
+                # OPTIONAL - Choices: "single_block", "geogrid", "geogrid_radargrid", and "auto" (default)
+                memory_mode:
+
+                # OPTIONAL - Processing upsampling factor applied to input geogrid
+                geogrid_upsampling: 1
+
+                # OPTIONAL - absolute radiometric correction
+                abs_rad_cal: 1
+
+                save_nlooks: True
+                save_rtc: True
+                 
+                # OPTIONAL - Same as input DEM if not provided.
+                outputEPSG: 
+
+                # OPTIONAL - Spacing between pixels, in same units as output EPSG.
+                # If no provided, values will match spacing in provided DEM
+                output_posting:
+                    A:
+                        x_posting:  
+                        y_posting:  
+                    B:
+                        x_posting:  
+                        y_posting:  
+
+                # OPTIONAL - To control output grid in same units as output EPSG
+                y_snap: 
+
+                # OPTIONAL - To control output grid in same units as output EPSG
+                x_snap: 
+
+                # OPTIONAL - Can control with absolute values or with snap values
+                top_left:        
+                    # OPTIONAL - Set top-left y in same units as output EPSG
+                    y_abs:
+                    # OPTIONAL - Set top-left x in same units as output EPSG
+                    x_abs:
+
+                # OPTIONAL - Can control with absolute values or with snap values
+                bottom_right:
+                    y_abs:
+                    x_abs:
+
+            # OPTIONAL - if noise correction desired (for ISRO)
+            noise_correction:
+                # OPTIONAL -
+                apply_correction: False
+
+                # OPTIONAL -
+                correction_type:
+
+        # OPTIONAL - To setup type of worker
+        worker:
+            # OPTIONAL - To prevent downloading DEM / other data automatically. Default True
+            internet_access: False
+
+            # OPTIONAL - To explicitly use GPU capability if available. Default False
+            gpu_enabled: False
+            # TODO
+            #gpu_id: 0            
+
+        # OPTIONAL - Location to save logs, use runconfig*_yaml.log if not specified
+        logging: 
+            path: output_gcov_ref/gcov.log
+            write_mode: 'w'
+