Merge pull request #70 from LSSTDESC/issue/69/fix_interactive

fix up logic for interactive stuff

src/rail/core/stage.py

@@ -331,9 +331,9 @@ def input_iterator(self, tag, **kwargs):
             self.config.hdf5_groupname
         except:
             self.config.hdf5_groupname = None
-        self._input_length = handle.size(groupname=self.config.hdf5_groupname)
 
         if handle.path and handle.path!='None':
+            self._input_length = handle.size(groupname=self.config.hdf5_groupname)
             total_chunks_needed = ceil(self._input_length/self.config.chunk_size)
             # If the number of process is larger than we need, we wemove some of them
             if total_chunks_needed < self.size:  #pragma: no cover
@@ -356,6 +356,7 @@ def input_iterator(self, tag, **kwargs):
                 test_data = self.get_data('input')[self.config.hdf5_groupname]
             else:
                 test_data = self.get_data('input')
+            self._input_length = len(test_data)
             s = 0
             iterator=[[s, self._input_length, test_data]]
             return iterator

src/rail/core/util_stages.py

@@ -35,6 +35,7 @@ def run(self):
         if self.config.inplace:  #pragma: no cover
             out_data = data
         self.add_data('output', out_data)
+        self.finalize()
 
     def __repr__(self):  # pragma: no cover
         printMsg = "Stage that applies remaps the following column names in a pandas DataFrame:\n"
@@ -83,6 +84,7 @@ def run(self):
         data = self.get_data('input', allow_missing=True)
         out_data = data.iloc[self.config.start:self.config.stop]
         self.add_data('output', out_data)
+        self.finalize()
 
     def __repr__(self):  # pragma: no cover
         printMsg = "Stage that applies remaps the following column names in a pandas DataFrame:\n"
@@ -127,6 +129,7 @@ def run(self):
         out_fmt = tables_io.types.TABULAR_FORMAT_NAMES[self.config.output_format]
         out_data = tables_io.convert(data, out_fmt)
         self.add_data('output', out_data)
+        self.finalize()
 
     def __call__(self, data):
         """Return a converted table