[WIP] changed multiThread structure to QThreadPool

QtPyLoT.py

@@ -81,7 +81,7 @@ from pylot.core.util.widgets import FilterOptionsDialog, NewEventDlg, \
     getDataType, ComparisonDialog, TuneAutopicker, PylotParaBox
 from pylot.core.util.map_projection import map_projection
 from pylot.core.util.structure import DATASTRUCTURE
-from pylot.core.util.thread import AutoPickThread, Thread, MultiThread
+from pylot.core.util.thread import AutoPickThread, Thread, Worker
 from pylot.core.util.version import get_git_version as _getVersionString
 
 if sys.version_info.major == 3:
@@ -1885,36 +1885,22 @@ class MainWindow(QMainWindow):
         # self.addListItem('Loading default values from PyLoT-input file %s'
         #                  % self.infile)
 
-        stations = []
+        args = {'parameter': self._inputs,
-        # catch all station names
+                'station': 'all',
-        for trace in self.data.getWFData():
+                'fnames': 'None',
-            station = trace.stats.station
+                'eventid': self.get_current_event_path (),
-            if not station in stations:
+                'iplot': 0,
-                stations.append(station)
+                'fig_dict': None,
-        
+                'locflag': 0}
-        mp_args = []
+
-        # create input_dict for each station in a list for multiprocessing.Pool iteration
+        self.mp_thread = QtCore.QThreadPool()
-        for station in stations:
+        self.mp_worker = Worker(autoPyLoT, args, redirect_stdout=True)
-            args = {'parameter': self._inputs,
+        self.mp_thread.start(self.mp_worker)
-                    'station': station,
+
-                    'fnames': 'None',
-                    'eventid': self.get_current_event_path (),
-                    'iplot': 0,
-                    'fig_dict': None,
-                    'locflag': 0}
-            mp_args.append(args)
-            
-        self.mp_thread = MultiThread (self, autoPyLoT, args=mp_args,
-                                      ncores=0,
-                                      progressText='Picking event...',
-                                      pb_widget=None,
-                                      redirect_stdout=True)
-        
         self.addListItem(str(self._inputs))
 
-        self.mp_thread.message.connect(self.addListItem)
+        self.mp_worker.signals.message.connect(self.addListItem)
-        self.mp_thread.start()
+        #self.mp_thread.finished.connect(self.finalizeAutoPick)
-        self.mp_thread.finished.connect(self.finalizeAutoPick)
 
     def finalizeAutoPick(self):
         self.drawPicks(picktype='auto')

pylot/core/util/thread.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
-import sys, os
+import sys, os, traceback
 import multiprocessing
-from PySide.QtCore import QThread, Signal, Qt
+from PySide.QtCore import QThread, Signal, Qt, Slot, QRunnable, QObject
 from PySide.QtGui import QDialog, QProgressBar, QLabel, QHBoxLayout, QPushButton
 
 
@@ -70,8 +70,8 @@ class Thread(QThread):
             print('Exception: {}, file: {}, line: {}'.format(exc_type, fname, exc_tb.tb_lineno))
         sys.stdout = sys.__stdout__        
 
-    # def __del__(self):
+    def __del__(self):
-    #     self.wait()
+        self.wait()
 
     def showProgressbar(self):
         if self.progressText:
@@ -103,13 +103,63 @@ class Thread(QThread):
 
     def flush(self):
         pass
-    
+
-    
+
+class Worker(QRunnable):
+    '''
+
+    '''
+    def __init__(self, fun, args,
+                 progressText=None,
+                 pb_widget=None,
+                 redirect_stdout=False):
+        super(Worker, self).__init__()
+        self.fun = fun
+        self.args = args
+        #self.kwargs = kwargs
+        self.signals = WorkerSignals()
+        self.progressText = progressText
+        self.pb_widget = pb_widget
+        self.redirect_stdout = redirect_stdout
+
+    @Slot()
+    def run(self):
+        if self.redirect_stdout:
+            sys.stdout = self
+
+        try:
+            result = self.fun(self.args)
+        except:
+            traceback.print_exc()
+            exctype, value = sys.exc_info ()[:2]
+            print(exctype, value, traceback.format_exc())
+            #self.signals.error.emit ((exctype, value, traceback.format_exc ()))
+        else:
+            self.signals.result.emit(result)
+        finally:
+            self.signals.finished.emit()
+
+    def write(self, text):
+        self.signals.message.emit(text)
+
+    def flush(self):
+        pass
+
+
+class WorkerSignals(QObject):
+    '''
+    '''
+    finished = Signal(str)
+    message = Signal(str)
+    error = Signal(tuple)
+    result = Signal(object)
+
+
 class MultiThread(QThread):
     finished = Signal(str)
     message = Signal(str)    
 
-    def __init__(self, parent, func, args, ncores=0,
+    def __init__(self, parent, func, args, ncores=1,
                  progressText=None, pb_widget=None, redirect_stdout=False):
         QThread.__init__(self, parent)
         self.func = func
@@ -122,22 +172,22 @@ class MultiThread(QThread):
         self.showProgressbar()
         
     def run(self):
-        # if self.redirect_stdout:
+        if self.redirect_stdout:
-        #     sys.stdout = self        
+             sys.stdout = self
-        #     #try:
+        try:
-        if not self.ncores:
+            if not self.ncores:
-            self.ncores = multiprocessing.cpu_count()
+                self.ncores = multiprocessing.cpu_count()
-        pool = multiprocessing.Pool(self.ncores)
+            pool = multiprocessing.Pool(self.ncores)
-        self.data = pool.map_async(self.func, self.args, callback=self.emitDone)
+            self.data = pool.map_async(self.func, self.args, callback=self.emitDone)
-        #self.data = pool.apply_async(self.func, self.shotlist, callback=self.emitDone) #emit each time returned
+            #self.data = pool.apply_async(self.func, self.shotlist, callback=self.emitDone) #emit each time returned
-        pool.close()
+            pool.close()
-        self._executed = True
+            self._executed = True
-        # except Exception as e:
+        except Exception as e:
-        #     self._executed = False
+            self._executed = False
-        #     self._executedError = e
+            self._executedError = e
-        #     exc_type, exc_obj, exc_tb = sys.exc_info()
+            exc_type, exc_obj, exc_tb = sys.exc_info()
-        #     fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
+            fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
-        #     print('Exception: {}, file: {}, line: {}'.format(exc_type, fname, exc_tb.tb_lineno))
+            print('Exception: {}, file: {}, line: {}'.format(exc_type, fname, exc_tb.tb_lineno))
         sys.stdout = sys.__stdout__        
 
     def __del__(self):
@@ -166,7 +216,7 @@ class MultiThread(QThread):
 
     def flush(self):
         pass
-    
+
     def emitDone(self, result):
         print('emitDone!')
         self.finished.emit('Done thread!')