topsStack: add --num-proc for step topo (#193)

* stackSentl.createParser(): align break line

+ align linebreak of help msg
+ use choices=[] for auto argument checking
+ use (default: %(default)s) to show defult value

* stackSent: bugfix for new paris if ref date is close to recent

This commit fixs the missing pair between the reference date and the new acquisitions, if the reference date is the most recent "num_conn" acquisition.

* Stack: check existing orbit with glob instead of os.path.exist(dir)

* topsStack: add --num-proc for topo step

stackSentinel: add --num-proc to pass the number of parallel processes to topo.py, so that mp.Pool() can be independent of OMP_NUM_THREADS

Co-authored-by: Zhang Yunjun <yunjunz@users.noreply.github.com>

contrib/stack/topsStack/Stack.py

@@ -75,6 +75,7 @@ class config(object):
         self.f.write('reference : ' + self.outDir + '\n')
         self.f.write('dem : ' + self.dem + '\n')
         self.f.write('geom_referenceDir : ' + self.geom_referenceDir + '\n')
+        self.f.write('numProcess : ' + str(self.numProcess) + '\n')
         self.f.write('##########################' + '\n')
 
     def geo2rdr(self,function):
@@ -315,6 +316,7 @@ class run(object):
         configObj.outDir = os.path.join(self.work_dir, 'reference')
         configObj.geom_referenceDir = os.path.join(self.work_dir, 'geom_reference')
         configObj.dem = os.path.join(self.work_dir, configObj.dem)
+        configObj.numProcess = self.numProcess
         configObj.Sentinel1_TOPS('[Function-1]')
         configObj.topo('[Function-2]')
         configObj.finalize()
@@ -940,8 +942,9 @@ class sentinelSLC(object):
            print ("downloading precise or restituted orbits ...")
 
            restitutedOrbitDir = os.path.join(workDir ,'orbits/' + self.date)
-           if os.path.exists(restitutedOrbitDir):
+           orbitFiles = glob.glob(os.path.join(restitutedOrbitDir,'*.EOF'))
-              orbitFile = glob.glob(os.path.join(restitutedOrbitDir,'*.EOF'))[0]
+           if len(orbitFiles) > 0:
+              orbitFile = orbitFiles[0]
 
               #fields = orbitFile.split('_')
               fields = os.path.basename(orbitFile).split('_')
@@ -954,7 +957,7 @@ class sentinelSLC(object):
 
            #if not os.path.exists(restitutedOrbitDir):
            else:
-              os.makedirs(restitutedOrbitDir)
+              os.makedirs(restitutedOrbitDir, exist_ok=True)
 
               cmd = 'fetchOrbit.py -i ' + self.safe_file + ' -o ' + restitutedOrbitDir
               print(cmd)

contrib/stack/topsStack/stackSentinel.py

@@ -94,81 +94,89 @@ def createParser():
     parser = argparse.ArgumentParser( description='Preparing the directory structure and config files for stack processing of Sentinel data')
 
     parser.add_argument('-H','--hh', nargs=0, action=customArgparseAction,
-                help='Display detailed help information.')
+                        help='Display detailed help information.')
 
     parser.add_argument('-s', '--slc_directory', dest='slc_dirname', type=str, required=True,
-            help='Directory with all Sentinel SLCs')
+                        help='Directory with all Sentinel SLCs')
 
     parser.add_argument('-o', '--orbit_directory', dest='orbit_dirname', type=str, required=True,
-            help='Directory with all orbits')
+                        help='Directory with all orbits')
 
     parser.add_argument('-a', '--aux_directory', dest='aux_dirname', type=str, required=True,
-            help='Directory with all aux files')
+                        help='Directory with all aux files')
 
     parser.add_argument('-w', '--working_directory', dest='work_dir', type=str, default='./',
-            help='Working directory ')
+                        help='Working directory (default: %(default)s).')
 
     parser.add_argument('-d', '--dem', dest='dem', type=str, required=True,
-            help='Directory with the DEM')
+                        help='Directory with the DEM')
 
     parser.add_argument('-m', '--reference_date', dest='reference_date', type=str, default=None,
-            help='Directory with reference acquisition')
+                        help='Directory with reference acquisition')
 
     parser.add_argument('-c','--num_connections', dest='num_connections', type=str, default = '1',
-            help='number of interferograms between each date and subsequent dates. -- Default : 1')
+                        help='number of interferograms between each date and subsequent dates (default: %(default)s).')
-
-    parser.add_argument('-O','--num_overlap_connections', dest='num_overlap_connections', type=str, default = '3',
-                help='number of overlap interferograms between each date and subsequent dates used for NESD computation (for azimuth offsets misregistration). -- Default : 3')
 
     parser.add_argument('-n', '--swath_num', dest='swath_num', type=str, default='1 2 3',
-            help="A list of swaths to be processed. -- Default : '1 2 3'")
+                        help="A list of swaths to be processed. -- Default : '1 2 3'")
 
-    parser.add_argument('-b', '--bbox', dest='bbox', type=str, default=None, help="Lat/Lon Bounding SNWE. -- Example : '19 20 -99.5 -98.5' -- Default : common overlap between stack")
+    parser.add_argument('-b', '--bbox', dest='bbox', type=str, default=None,
+                        help="Lat/Lon Bounding SNWE. -- Example : '19 20 -99.5 -98.5' -- Default : common overlap between stack")
 
-    parser.add_argument('-t', '--text_cmd', dest='text_cmd', type=str, default=''
+    parser.add_argument('-t', '--text_cmd', dest='text_cmd', type=str, default='', 
-       , help="text command to be added to the beginning of each line of the run files. -- Example : 'source ~/.bash_profile;' -- Default : ''")
+                        help="text command to be added to the beginning of each line of the run files (default: '%(default)s'). "
+                             "Example : 'source ~/.bash_profile;'")
 
-    parser.add_argument('-x', '--exclude_dates', dest='exclude_dates', type=str, default=None
+    parser.add_argument('-x', '--exclude_dates', dest='exclude_dates', type=str, default=None, 
-       , help="List of the dates to be excluded for processing. -- Example : '20141007,20141031' -- Default: No dates excluded")
+                        help="List of the dates to be excluded for processing. -- Example : '20141007,20141031' (default: %(default)s).")
 
-    parser.add_argument('-i', '--include_dates', dest='include_dates', type=str, default=None
+    parser.add_argument('-i', '--include_dates', dest='include_dates', type=str, default=None, 
-       , help="List of the dates to be included for processing. -- Example : '20141007,20141031' -- Default: No dates excluded")
+                        help="List of the dates to be included for processing. -- Example : '20141007,20141031' (default: %(default)s).")
 
-    parser.add_argument('-z', '--azimuth_looks', dest='azimuthLooks', type=str, default='3'
+    parser.add_argument('--start_date', dest='startDate', type=str, default=None, 
-       , help='Number of looks in azimuth for interferogram multi-looking. -- Default : 3')
+                        help='Start date for stack processing. Acquisitions before start date are ignored. '
+                             'format should be YYYY-MM-DD e.g., 2015-01-23')
 
-    parser.add_argument('-r', '--range_looks', dest='rangeLooks', type=str, default='9'
+    parser.add_argument('--stop_date', dest='stopDate', type=str, default=None, 
-       , help='Number of looks in range for interferogram multi-looking. -- Default : 9')
+                        help='Stop date for stack processing. Acquisitions after stop date are ignored. '
+                             'format should be YYYY-MM-DD e.g., 2017-02-26')
 
-    parser.add_argument('-f', '--filter_strength', dest='filtStrength', type=str, default='0.5'
+    parser.add_argument('-z', '--azimuth_looks', dest='azimuthLooks', type=str, default='3', 
-       , help='Filter strength for interferogram filtering. -- Default : 0.5')
+                        help='Number of looks in azimuth for interferogram multi-looking (default: %(default)s).')
 
-    parser.add_argument('-e', '--esd_coherence_threshold', dest='esdCoherenceThreshold', type=str, default='0.85'
+    parser.add_argument('-r', '--range_looks', dest='rangeLooks', type=str, default='9', 
-           , help='Coherence threshold for estimating azimuth misregistration using enhanced spectral diversity. -- Default : 0.85')
+                        help='Number of looks in range for interferogram multi-looking (default: %(default)s).')
 
-    parser.add_argument('--snr_misreg_threshold', dest='snrThreshold', type=str, default='10'
+    parser.add_argument('-f', '--filter_strength', dest='filtStrength', type=str, default='0.5', 
-               , help='SNR threshold for estimating range misregistration using cross correlation. -- Default : 10')
+                        help='Filter strength for interferogram filtering (default: %(default)s).')
 
-    parser.add_argument('-u', '--unw_method', dest='unwMethod', type=str, default='snaphu'
+    parser.add_argument('--snr_misreg_threshold', dest='snrThreshold', type=str, default='10', 
-       , help='Unwrapping method (icu or snaphu). -- Default : snaphu')
+                        help='SNR threshold for estimating range misregistration using cross correlation (default: %(default)s).')
 
-    parser.add_argument('-p', '--polarization', dest='polarization', type=str, default='vv'
+    parser.add_argument('-p', '--polarization', dest='polarization', type=str, default='vv', 
-       , help='SAR data polarization -- Default : vv')
+                        help='SAR data polarization (default: %(default)s).')
 
-    parser.add_argument('-C', '--coregistration', dest='coregistration', type=str, default='NESD'
+    parser.add_argument('-C', '--coregistration', dest='coregistration', type=str, default='NESD', choices=['geometry', 'NESD'],
-           , help='Coregistration options: a) geometry b) NESD -- Default : NESD')
+                        help='Coregistration options (default: %(default)s).')
 
-    parser.add_argument('-W', '--workflow', dest='workflow', type=str, default='interferogram'
+    parser.add_argument('-O','--num_overlap_connections', dest='num_overlap_connections', type=str, default = '3',
-       , help='The InSAR processing workflow : (interferogram, offset, slc, correlation) -- Default : interferogram')
+                        help='number of overlap interferograms between each date and subsequent dates used for NESD computation '
+                             '(for azimuth offsets misregistration) (default: %(default)s).')
 
-    parser.add_argument('--start_date', dest='startDate', type=str, default=None
+    parser.add_argument('-e', '--esd_coherence_threshold', dest='esdCoherenceThreshold', type=str, default='0.85', 
-       , help='Start date for stack processing. Acquisitions before start date are ignored. format should be YYYY-MM-DD e.g., 2015-01-23')
+                        help='Coherence threshold for estimating azimuth misregistration using enhanced spectral diversity (default: %(default)s).')
 
-    parser.add_argument('--stop_date', dest='stopDate', type=str, default=None
+    parser.add_argument('-W', '--workflow', dest='workflow', type=str, default='interferogram', choices=['slc', 'correlation', 'interferogram', 'offset'],
-       , help='Stop date for stack processing. Acquisitions after stop date are ignored. format should be YYYY-MM-DD e.g., 2017-02-26')
+                        help='The InSAR processing workflow (default: %(default)s).')
 
     parser.add_argument('-useGPU', '--useGPU', dest='useGPU',action='store_true', default=False,
-        help='Allow App to use GPU when available')
+                        help='Allow App to use GPU when available')
+
+    parser.add_argument('--num-proc', '--num-process', dest='numProcess', type=int, default=1,
+                        help='number of parallel processes (for topo only) (default: %(default)s).')
+
+    parser.add_argument('-u', '--unw_method', dest='unwMethod', type=str, default='snaphu', choices=['icu', 'snaphu'],
+                        help='Unwrapping method (default: %(default)s).')
 
     parser.add_argument('-rmFilter', '--rmFilter', dest='rmFilter', action='store_true', default=False,
                         help='Make an extra unwrap file in which filtering effect is removed')
@@ -229,7 +237,7 @@ def get_dates(inps):
     if inps.startDate is not None:
         stackStartDate = datetime.datetime(*time.strptime(inps.startDate, "%Y-%m-%d")[0:6])
     else:
-        #if startDate is None let's fix it to first JPL's staellite lunch date :)
+        #if startDate is None let's fix it to first JPL's satellite lunch date :)
         stackStartDate = datetime.datetime(*time.strptime("1958-01-31", "%Y-%m-%d")[0:6])
 
     if inps.stopDate is not None:
@@ -742,7 +750,8 @@ def main(iargs=None):
         print('')
         print('Updating an existing stack ...')
         print('')
-        pairs = selectNeighborPairs(secondaryDates, inps.num_connections,updateStack)   # will be change later
+        dates4NewPairs = sorted(secondaryDates + [stackReferenceDate])[1:]
+        pairs = selectNeighborPairs(dates4NewPairs, inps.num_connections,updateStack)   # will be change later
     else:
         pairs = selectNeighborPairs(acquisitionDates, inps.num_connections,updateStack)
 

contrib/stack/topsStack/topo.py

@@ -21,6 +21,8 @@ def createParser():
             help='DEM to use for coregistration')
     parser.add_argument('-g', '--geom_referenceDir', type=str, dest='geom_referenceDir', default='geom_reference',
             help='Directory for geometry files of the reference')
+    parser.add_argument('-n', '--numProcess', type=int, dest='numProcess', default=1,
+            help='Number of parallel processes (default: %(default)s).')
 
     return parser
 
@@ -98,11 +100,8 @@ def main(iargs=None):
             inputs.append((dirname, demImage, reference, ind))
 
     # parallel processing
-    numThread = int(os.environ.get('OMP_NUM_THREADS', mp.cpu_count()))
+    print('running in parallel with {} processes'.format(inps.numProcess))
-    numThread = min(numThread, mp.cpu_count())
+    pool = mp.Pool(inps.numProcess)
-    print('running in parallel with {} processes'.format(numThread))
-
-    pool = mp.Pool(numThread)
     results = pool.map(call_topo, inputs)
     pool.close()