bug from merge corrected

pipelet/pipeline.py

@@ -434,7 +434,7 @@ class Pipeline:
         """ 
         return self._curr_dirs[seg]
 
-    def get_output_file(self, seg):
+    def get_output_fn(self, seg):
         """ Return the segment output file
         
         Parameters

pipelet/repository.py

@@ -261,8 +261,8 @@ class LocalRepository(Repository):
         ['../test/seg_dbm_preproc.py']
         """
         try:
-            f = filter(self._ext_filter,
-                       glob(path.join(self.src_path, 'seg_%s_%s.*'%(seg,hook))))[0]
+            f = [filter(self._ext_filter,
+                       glob(path.join(self.src_path, 'seg_%s_%s.*'%(seg,hook))))[0]]
         except:
             f = []
         if len(f) == 0:

pipelet/worker.py

@@ -26,6 +26,20 @@ import traceback
 import signal
 from contextlib import closing
 import gc
+import logging
+
+
+class NullHandler(logging.Handler):
+    """ Extension of the logging handler class. 
+    """
+    def emit(self, record):
+        """ Avoid warnings. 
+        """
+        pass
+
+logger = logging.getLogger("scheduler")
+h = NullHandler()
+logger.addHandler(h)
 
 class AbortError(Exception):
     """ Extension of the Exception class. 
@@ -143,14 +157,15 @@ class Worker(object):
             except KeyError:
                 print 'Fail to load object %s from file %s'%(k,filename) 
 
-    def write_res(self, task_output):
+    def write_res(self, seg, task_output):
         """ Pickle the result of the segment.
         
         Parameters
         ----------
+        seg: string, segment name
         task_output: task result. 
         """
-        fn = self.pipe.get_output_fn()  
+        fn = self.pipe.get_output_fn(seg)  
         with closing (file(fn,'w')) as f:
             r = pickle.dump(task_output,f)
 
@@ -465,7 +480,7 @@ class InteractiveWorker(Worker):
         except:
             res = None
         task.task_output = res
-        self.write_res(res)
+        self.write_res(seg, res)
         try: # Save params
             var_key = glo['var_key']
             self.save_param(seg, glo, param_name=var_key)
@@ -530,7 +545,7 @@ class ThreadWorker(Worker, threading.Thread):
         except:
             res = None
         task.task_output = res
-        self.write_res(res)
+        self.write_res(seg, res)
         if task.status == "failed":
             return task
         try:  # Save params
@@ -610,7 +625,7 @@ class ProcessWorker(Worker, Process):
         except:
             res = None
         task.task_output = res
-        self.write_res(res)
+        self.write_res(seg, res)
         if task.status == "failed":
             return task
         try:  # Save params