Task #7491: Merge intro trunk the longbaseline pipeline with skip support.

CEP/Pipeline/recipes/sip/bin/long_baseline_pipeline.py

@@ -134,7 +134,7 @@ class msss_imager_pipeline(control):
 
         # ******************************************************************
         # (1) prepare phase: copy and collect the ms
-        concat_ms_map_path, timeslice_map_path, raw_ms_per_image_map_path, \
+        concat_ms_map_path, timeslice_map_path, ms_per_image_map_path, \
             processed_ms_dir = self._long_baseline(input_mapfile,
                          target_mapfile, add_beam_tables, output_ms_mapfile)
 
@@ -210,7 +210,7 @@ class msss_imager_pipeline(control):
 
     @xml_node
     def _finalize(self, awimager_output_map, processed_ms_dir,
-                  raw_ms_per_image_map, sourcelist_map, minbaseline,
+                  ms_per_image_map, sourcelist_map, minbaseline,
                   maxbaseline, target_mapfile,
                   output_ms_mapfile, sourcedb_map, skip = False):
         """
@@ -229,7 +229,7 @@ class msss_imager_pipeline(control):
         else:
             placed_image_mapfile = self.run_task("imager_finalize",
                 target_mapfile, awimager_output_map = awimager_output_map,
-                    raw_ms_per_image_map = raw_ms_per_image_map,
+                    ms_per_image_map = ms_per_image_map,
                     sourcelist_map = sourcelist_map,
                     sourcedb_map = sourcedb_map,
                     minbaseline = minbaseline,
@@ -251,7 +251,7 @@ class msss_imager_pipeline(control):
         the time slices into a large virtual measurement set
         """
         # Create the dir where found and processed ms are placed
-        # raw_ms_per_image_map_path contains all the original ms locations:
+        # ms_per_image_map_path contains all the original ms locations:
         # this list contains possible missing files
         processed_ms_dir = os.path.join(self.scratch_directory, "subbands")
 
@@ -265,8 +265,8 @@ class msss_imager_pipeline(control):
         output_mapfile = self._write_datamap_to_file(None, "prepare_output")
         time_slices_mapfile = self._write_datamap_to_file(None,
                                                     "prepare_time_slices")
-        raw_ms_per_image_mapfile = self._write_datamap_to_file(None,
-                                                         "raw_ms_per_image")
+        ms_per_image_mapfile = self._write_datamap_to_file(None,
+                                                         "ms_per_image")
 
         # get some parameters from the imaging pipeline parset:
         subbandgroups_per_ms = self.parset.getInt("LongBaseline.subbandgroups_per_ms")
@@ -280,7 +280,7 @@ class msss_imager_pipeline(control):
                 subbands_per_subbandgroup = subbands_per_subbandgroup,
                 mapfile = output_mapfile,
                 slices_mapfile = time_slices_mapfile,
-                raw_ms_per_image_mapfile = raw_ms_per_image_mapfile,
+                ms_per_image_mapfile = ms_per_image_mapfile,
                 working_directory = self.scratch_directory,
                 processed_ms_dir = processed_ms_dir,
                 add_beam_tables = add_beam_tables,
@@ -299,15 +299,15 @@ class msss_imager_pipeline(control):
                                                         'slices_mapfile')
             self.logger.error(error_msg)
             raise PipelineException(error_msg)
-        if not ('raw_ms_per_image_mapfile' in output_keys):
+        if not ('ms_per_image_mapfile' in output_keys):
             error_msg = "The imager_prepare master script did not"\
                     "return correct data. missing: {0}".format(
-                                                'raw_ms_per_image_mapfile')
+                                                'ms_per_image_mapfile')
             self.logger.error(error_msg)
             raise PipelineException(error_msg)
 
         # Return the mapfiles paths with processed data
-        return output_mapfile, outputs["slices_mapfile"], raw_ms_per_image_mapfile, \
+        return output_mapfile, outputs["slices_mapfile"], ms_per_image_mapfile, \
             processed_ms_dir
 
 

CEP/Pipeline/recipes/sip/master/long_baseline.py

@@ -101,9 +101,9 @@ class long_baseline(BaseRecipe, RemoteCommandRecipeMixIn):
             '--slices-mapfile',
             help="Path to mapfile containing the produced subband groups"
         ),
-        'raw_ms_per_image_mapfile': ingredient.StringField(
-            '--raw-ms-per-image-mapfile',
-            help="Path to mapfile containing the raw ms for each produced"
+        'ms_per_image_mapfile': ingredient.StringField(
+            '--ms-per-image-mapfile',
+            help="Path to mapfile containing the ms for each produced"
                 "image"
         ),
         'processed_ms_dir': ingredient.StringField(
@@ -129,8 +129,8 @@ class long_baseline(BaseRecipe, RemoteCommandRecipeMixIn):
         'slices_mapfile': ingredient.FileField(
             help="Path to mapfile containing the produced subband groups"),
 
-        'raw_ms_per_image_mapfile': ingredient.FileField(
-            help="Path to mapfile containing the raw ms for each produced"
+        'ms_per_image_mapfile': ingredient.FileField(
+            help="Path to mapfile containing the ms for each produced"
                 "image")
     }
 
@@ -167,7 +167,8 @@ class long_baseline(BaseRecipe, RemoteCommandRecipeMixIn):
         n_subband_groups = len(output_map)
 
         output_map.iterator = final_output_map.iterator = DataMap.SkipIterator
-        for idx_sb_group, (output_item, final_item) in enumerate(zip(output_map, final_output_map)):
+        for idx_sb_group, (output_item, final_item) in enumerate(zip(output_map, 
+                                                            final_output_map)):
             #create the input files for this node
             self.logger.debug("Creating input data subset for processing"
                               "on: {0}".format(output_item.host))
@@ -220,8 +221,19 @@ class long_baseline(BaseRecipe, RemoteCommandRecipeMixIn):
         concat_ms = copy.deepcopy(output_map)
         slices = []
         finished_runs = 0
-        #scan the return dict for completed key
-        for (item, job) in zip(concat_ms, jobs):
+        # If we have a skipped item, add the item to the slices with skip set
+        jobs_idx = 0
+        for item in concat_ms: 
+            # If this is an item that is skipped via the skip parameter in 
+            # the parset, append a skipped             
+            if item.skip:    
+                slices.append(tuple([item.host, [], True]))
+                continue
+
+            # we cannot use the skip iterator so we need to manually get the
+            # current job from the list
+            job = jobs[jobs_idx]
+
             # only save the slices if the node has completed succesfull
             if job.results["returncode"] == 0:
                 finished_runs += 1
@@ -230,11 +242,14 @@ class long_baseline(BaseRecipe, RemoteCommandRecipeMixIn):
             else:
                 # Set the dataproduct to skipped!!
                 item.skip = True
-                slices.append(tuple([item.host, ["/Failed"], True]))
+                slices.append(tuple([item.host, [], True]))
                 msg = "Failed run on {0}. NOT Created: {1} ".format(
                     item.host, item.file)
                 self.logger.warn(msg)
 
+            # we have a non skipped workitem, increase the job idx
+            jobs_idx += 1
+
         if finished_runs == 0:
             self.logger.error("None of the started compute node finished:"
                 "The current recipe produced no output, aborting")
@@ -252,15 +267,15 @@ class long_baseline(BaseRecipe, RemoteCommandRecipeMixIn):
                 self.inputs['slices_mapfile']))
 
         #map with actual input mss.
-        self._store_data_map(self.inputs["raw_ms_per_image_mapfile"],
+        self._store_data_map(self.inputs["ms_per_image_mapfile"],
             DataMap(paths_to_image_mapfiles),
-                "mapfile containing (raw) input ms per image:")
+                "mapfile containing input ms per image:")
 
         # Set the return values
         self.outputs['mapfile'] = output_ms_mapfile_path
         self.outputs['slices_mapfile'] = self.inputs['slices_mapfile']
-        self.outputs['raw_ms_per_image_mapfile'] = \
-            self.inputs["raw_ms_per_image_mapfile"]
+        self.outputs['ms_per_image_mapfile'] = \
+            self.inputs["ms_per_image_mapfile"]
 
         return 0
 
@@ -301,7 +316,7 @@ class long_baseline(BaseRecipe, RemoteCommandRecipeMixIn):
         """
         # The output_map contains a number of path/node pairs. The final data 
         # dataproduct of the prepare phase: The 'input' for each of these pairs
-        # is a number of raw measurement sets: The number of time slices times
+        # is a number of measurement sets: The number of time slices times
         # the number of subbands collected into each of these time slices.
         # The total length of the input map should match this.
         if len(input_map) != len(output_map) * \

CEP/Pipeline/recipes/sip/nodes/long_baseline.py

@@ -43,7 +43,7 @@ class long_baseline(LOFARnodeTCP):
     """
     def run(self, environment, parset, working_dir, processed_ms_dir,
              ndppp_executable, output_measurement_set,
-            subbandgroups_per_ms, subbands_per_subbandgroup, raw_ms_mapfile,
+            subbandgroups_per_ms, subbands_per_subbandgroup, ms_mapfile,
             asciistat_executable, statplot_executable, msselect_executable,
             rficonsole_executable, add_beam_tables, final_output_path):
         """
@@ -52,7 +52,7 @@ class long_baseline(LOFARnodeTCP):
         self.environment.update(environment)
 
         with log_time(self.logger):
-            input_map = DataMap.load(raw_ms_mapfile)
+            input_map = DataMap.load(ms_mapfile)
             #******************************************************************
             # I. Create the directories used in this recipe
             create_directory(processed_ms_dir)
@@ -71,14 +71,14 @@ class long_baseline(LOFARnodeTCP):
 
             #******************************************************************
             # 1. Copy the input files
-            copied_ms_map = self._copy_input_files(
+            processed_ms_map = self._copy_input_files(
                             processed_ms_dir, input_map)
 
             #******************************************************************
             # 2. run dppp: collect frequencies into larger group
             time_slices_path_list = \
                 self._run_dppp(working_dir, time_slice_dir,
-                    subbandgroups_per_ms, copied_ms_map, subbands_per_subbandgroup,
+                    subbandgroups_per_ms, processed_ms_map, subbands_per_subbandgroup,
                     processed_ms_dir, parset, ndppp_executable)
 
             # If no timeslices were created, bail out with exit status 1
@@ -138,6 +138,10 @@ class long_baseline(LOFARnodeTCP):
             self._deep_copy_to_output_location(output_measurement_set,
                                                final_output_path)
 
+            # Write the actually used ms for the created dataset to the input 
+            # mapfile
+            processed_ms_map.save(ms_mapfile)
+
 
 
             #******************************************************************
@@ -167,46 +171,49 @@ class long_baseline(LOFARnodeTCP):
         This function collects all the file in the input map in the
         processed_ms_dir Return value is a set of missing files
         """
-        copied_ms_map = copy.deepcopy(input_map)
+        processed_ms_map = copy.deepcopy(input_map)
+
         # loop all measurement sets
-        for input_item, copied_item in zip(input_map, copied_ms_map):
+        for input_item, copied_item in zip(input_map, processed_ms_map):
             # fill the copied item with the correct data
             copied_item.host = self.host
             copied_item.file = os.path.join(
                     processed_ms_dir, os.path.basename(input_item.file))
 
+            stderrdata = None
             # If we have to skip this ms
             if input_item.skip == True:
-                exit_status = 1  #
-
-            # skip the copy if machine is the same (execution on localhost) 
-            # make sure data is in the correct directory. for now: working_dir/[jobname]/subbands
-            #if input_item.host == "localhost":           
-            #    continue
-            # construct copy command
-            command = ["rsync", "-r", "{0}:{1}".format(
-                            input_item.host, input_item.file),
-                               "{0}".format(processed_ms_dir)]
-            if input_item.host == "localhost":
-                command = ["cp", "-r", "{0}".format(input_item.file),
-                                       "{0}".format(processed_ms_dir)]
-
-            self.logger.debug("executing: " + " ".join(command))
-
-            # Spawn a subprocess and connect the pipes
-            # The copy step is performed 720 at once in that case which might
-            # saturate the cluster.
-            copy_process = subprocess.Popen(
-                        command,
-                        stdin = subprocess.PIPE,
-                        stdout = subprocess.PIPE,
-                        stderr = subprocess.PIPE)
-
-            # Wait for finish of copy inside the loop: enforce single tread
-            # copy
-            (stdoutdata, stderrdata) = copy_process.communicate()
-
-            exit_status = copy_process.returncode
+                exit_status = 1  
+                stderrdata = "SKIPPED_FILE"
+
+            else:
+              # skip the copy if machine is the same (execution on localhost) 
+              # make sure data is in the correct directory. 
+              # for now: working_dir/[jobname]/subbands
+              # construct copy command
+              command = ["rsync", "-r", "{0}:{1}".format(
+                              input_item.host, input_item.file),
+                                 "{0}".format(processed_ms_dir)]
+              if input_item.host == "localhost":
+                  command = ["cp", "-r", "{0}".format(input_item.file),
+                                         "{0}".format(processed_ms_dir)]
+
+              self.logger.debug("executing: " + " ".join(command))
+
+              # Spawn a subprocess and connect the pipes
+              # The copy step is performed 720 at once in that case which might
+              # saturate the cluster.
+              copy_process = subprocess.Popen(
+                          command,
+                          stdin = subprocess.PIPE,
+                          stdout = subprocess.PIPE,
+                          stderr = subprocess.PIPE)
+
+              # Wait for finish of copy inside the loop: enforce single tread
+              # copy
+              (stdoutdata, stderrdata) = copy_process.communicate()
+
+              exit_status = copy_process.returncode
 
             # if copy failed log the missing file and update the skip fields
             if  exit_status != 0:
@@ -218,7 +225,7 @@ class long_baseline(LOFARnodeTCP):
 
             self.logger.debug(stdoutdata)
 
-        return copied_ms_map
+        return processed_ms_map
 
 
     def _dppp_call(self, working_dir, ndppp, cmd, environment):
@@ -233,7 +240,7 @@ class long_baseline(LOFARnodeTCP):
                                   logger, cleanup = None)
 
     def _run_dppp(self, working_dir, time_slice_dir_path, slices_per_image,
-                  copied_ms_map, subbands_per_image, collected_ms_dir_name, parset,
+                  processed_ms_map, subbands_per_image, collected_ms_dir_name, parset,
                   ndppp):
         """
         Run NDPPP:
@@ -247,9 +254,6 @@ class long_baseline(LOFARnodeTCP):
             end_slice_range = (idx_time_slice + 1) * subbands_per_image
             # Get the subset of ms that are part of the current timeslice,
             # cast to datamap
-            input_map_subgroup = DataMap(
-                            copied_ms_map[start_slice_range:end_slice_range])
-
             output_ms_name = "time_slice_{0}.dppp.ms".format(idx_time_slice)
 
             # construct time slice name
@@ -258,7 +262,21 @@ class long_baseline(LOFARnodeTCP):
 
             # convert the datamap to a file list: Do not remove skipped files:
             # ndppp needs the incorrect files there to allow filling with zeros
-            ndppp_input_ms = [item.file for item in input_map_subgroup]
+
+            ndppp_input_ms = []
+            for item in processed_ms_map[start_slice_range:end_slice_range]:
+                if item.skip:
+                    ndppp_input_ms.append("SKIPPEDSUBBAND")
+                    # We need an entry in the list: ndppp will add zeros to
+                    # pad missing subbands
+                else:
+                    ndppp_input_ms.append(item.file)
+
+            # if none of the input files was valid, skip the creation of the 
+            # timeslice all together, it will not show up in the timeslice 
+            # mapfile
+            if len(ndppp_input_ms) == 0:
+                continue
 
             # Join into a single list of paths.
             msin = "['{0}']".format("', '".join(ndppp_input_ms))
@@ -300,11 +318,9 @@ class long_baseline(LOFARnodeTCP):
                 time_slice_path_list.append(time_slice_path)
 
             # On error the current timeslice should be skipped
-            except subprocess.CalledProcessError, exception:
-                self.logger.warning(str(exception))
-                continue
-
             except Exception, exception:
+                for item in processed_ms_map[start_slice_range:end_slice_range]:
+                    item.skip = True
                 self.logger.warning(str(exception))
                 continue
 

CEP/Pipeline/test/regression_tests/target_pipeline.py

@@ -14,6 +14,8 @@ def load_and_compare_data_sets(ms1, ms2):
     # create a target array with the same length as the datacolumn
     div_array = numpy.zeros((n_row, 1, n_complex_vis), dtype=numpy.complex64)
     ms1_array = ms1.getcol('DATA')
+    # TODO: WHy are different collomns compared?
+    # is this an issue in the test dataset??
     ms2_array = ms2.getcol('CORRECTED_DATA')
 
     div_max = 0

CEP/Pipeline/visual_studio/Pipeline.pyproj

@@ -32,6 +32,9 @@
     <Content Include="mac\CMakeLists.txt" />
     <Content Include="recipes\CMakeLists.txt" />
     <Content Include="test\CMakeLists.txt" />
+    <Compile Include="recipes\sip\bin\long_baseline_pipeline.py" />
+    <Compile Include="recipes\sip\master\long_baseline.py" />
+    <Compile Include="recipes\sip\nodes\long_baseline.py" />
     <Compile Include="test\__init__.py" />
     <Compile Include="deploy\deprecated\fabfile.py" />
     <Compile Include="deploy\deprecated\start_cluster.py" />