schain Commit - r1241:c3044f867269 · Jicamarca Repository

Errors handling and gracefully terminate main process

jespinoza -

r1241:c3044f867269

parent child

Context file:

r1241:c3044f867269

Collapse all files

schainpy/admin.py +1 -2

                 @staticmethod
                 def send_email(**kwargs):
                     notifier = SchainNotify()
-                    print(kwargs)
                     notifier.notify(**kwargs)
                 @staticmethod
                     msg = MIMEMultipart()
                     msg['Subject'] = subject
-                    msg['From'] = "(Python SChain API): " + email_from
+                    msg['From'] = "SChain API (v{}) <{}>".format(schainpy.__version__, email_from)
                     msg['Reply-to'] = email_from
                     msg['To'] = email_to

schainpy/controller.py +70 -42

             import math
             import time
             import zmq
-            from multiprocessing import Process, cpu_count
+            from multiprocessing import Process, Queue, cpu_count
             from threading import Thread
             from xml.etree.ElementTree import ElementTree, Element, SubElement, tostring
             from xml.dom import minidom
                     return kwargs
-                def setup(self, id, name, priority, type, project_id):
+                def setup(self, id, name, priority, type, project_id, err_queue):
                     self.id = str(id)
                     self.project_id = project_id
                     self.name = name
                     self.type = type
                     self.priority = priority
+                    self.err_queue = err_queue
                     self.parmConfObjList = []
                 def removeParameters(self):
                         opObj = className()
                     elif self.type == 'external':
                         kwargs = self.getKwargs()
-                        opObj = className(self.id, self.project_id, **kwargs)
+                        opObj = className(self.id, self.id, self.project_id, self.err_queue, 'Operation', **kwargs)
                         opObj.start()
+                        self.opObj = opObj
                     return opObj
                     return self.procUnitObj
-                def setup(self, project_id, id, name, datatype, inputId):
+                def setup(self, project_id, id, name, datatype, inputId, err_queue):
                     '''
                         id sera el topico a publicar
                         inputId sera el topico a subscribirse
                     self.name = name
                     self.datatype = datatype
                     self.inputId = inputId
+                    self.err_queue = err_queue
                     self.opConfObjList = []
                     self.addOperation(name='run', optype='self')
                     id = self.__getNewId()
                     priority = self.__getPriority() # Sin mucho sentido, pero puede usarse
                     opConfObj = OperationConf()
-                    opConfObj.setup(id, name=name, priority=priority, type=optype, project_id=self.project_id)
+                    opConfObj.setup(id, name=name, priority=priority, type=optype, project_id=self.project_id, err_queue=self.err_queue)
                     self.opConfObjList.append(opConfObj)
                     return opConfObj
                     className = eval(self.name)
                     kwargs = self.getKwargs()
-                    procUnitObj = className(self.id, self.inputId, self.project_id, **kwargs) # necesitan saber su id y su entrada por fines de ipc
+                    procUnitObj = className(self.id, self.inputId, self.project_id, self.err_queue, 'ProcUnit', **kwargs)
                     log.success('creating process...', self.name)
                     for opConfObj in self.opConfObjList:
                         else:
                             opObj = opConfObj.createObject()
-                        log.success('creating operation: {}, type:{}'.format(
+                        log.success('adding operation: {}, type:{}'.format(
                             opConfObj.name,
                             opConfObj.type), self.name)
                     return self.ELEMENTNAME
-                def setup(self, project_id, id, name, datatype, path='', startDate='', endDate='',
+                def setup(self, project_id, id, name, datatype, err_queue, path='', startDate='', endDate='',
                           startTime='', endTime='', server=None, **kwargs):
                     self.startTime = startTime
                     self.endTime = endTime
                     self.server = server
+                    self.err_queue = err_queue
                     self.addRunOperation(**kwargs)
                 def update(self, **kwargs):
                     self.email = None
                     self.alarm = None
                     self.procUnitConfObjDict = {}
+                    self.err_queue = Queue()
                 def __getNewId(self):
                     self.description = description
                     self.email = email
                     self.alarm = alarm
+                    if name:
+                        self.name = '{} ({})'.format(Process.__name__, name)
                 def update(self, **kwargs):
                         idReadUnit = str(id)
                     readUnitConfObj = ReadUnitConf()
-                    readUnitConfObj.setup(self.id, idReadUnit, name, datatype, **kwargs)
+                    readUnitConfObj.setup(self.id, idReadUnit, name, datatype, self.err_queue, **kwargs)
                     self.procUnitConfObjDict[readUnitConfObj.getId()] = readUnitConfObj
                     return readUnitConfObj
                     '''
-                    idProcUnit = self.__getNewId() #Topico para subscripcion
+                    idProcUnit = self.__getNewId()
                     procUnitConfObj = ProcUnitConf()
-                    procUnitConfObj.setup(self.id, idProcUnit, name, datatype, inputId) #topic_read, topic_write,
+                    procUnitConfObj.setup(self.id, idProcUnit, name, datatype, inputId, self.err_queue)
                     self.procUnitConfObjDict[procUnitConfObj.getId()] = procUnitConfObj
                     return procUnitConfObj
                 def __str__(self):
-                    print('Project[%s]: name = %s, description = %s, project_id = %s' % (self.id,
+                    print('Project: name = %s, description = %s, id = %s' % (
                                                                         self.name,
                                                                         self.description,
-                                                                        self.project_id))
+                                                                        self.id))
                     for procUnitConfObj in self.procUnitConfObjDict.values():
                         print(procUnitConfObj)
                     for key in keys:
                         self.procUnitConfObjDict[key].createObjects()
-                def __handleError(self, procUnitConfObj, modes=None, stdout=True):
+                def monitor(self):
+                    t = Thread(target=self.__monitor, args=(self.err_queue, self.ctx))
+                    t.start()
+                def __monitor(self, queue, ctx):
                     import socket
-                    if modes is None:
+                    procs = 0
-                        modes = self.alarm
+                    err_msg = ''
+                    while True:
+                        msg = queue.get()
+                        if '#_start_#' in msg:
+                            procs += 1
+                        elif '#_end_#' in msg:
+                            procs -=1
+                        else:
+                            err_msg = msg
+                        if procs == 0 or 'Traceback' in err_msg:
+                            break
+                        time.sleep(0.1)
+                    if '|' in err_msg:
+                        name, err = err_msg.split('|')
+                        if 'SchainWarning' in err:
+                            log.warning(err.split('SchainWarning:')[-1].split('\n')[0].strip(), name)
+                        elif 'SchainError' in err:
+                            log.error(err.split('SchainError:')[-1].split('\n')[0].strip(), name)
+                        else:
+                            log.error(err, name)
+                    else:
+                        name, err = self.name, err_msg
-                    if not self.alarm:
+                    time.sleep(2)
-                        modes = []
-                    err = traceback.format_exception(sys.exc_info()[0],
+                    for conf in self.procUnitConfObjDict.values():
-                                                     sys.exc_info()[1],
+                        for confop in conf.opConfObjList:
-                                                     sys.exc_info()[2])
+                            if confop.type == 'external':
+                                confop.opObj.terminate()
+                        conf.procUnitObj.terminate()
-                    log.error('{}'.format(err[-1]), procUnitConfObj.name)
+                    ctx.term()
                     message = ''.join(err)
-                    if stdout:
+                    if err_msg:
-                        sys.stderr.write(message)
                         subject = 'SChain v%s: Error running %s\n' % (
-                        schainpy.__version__, procUnitConfObj.name)
+                            schainpy.__version__, self.name)
-                    subtitle = '%s: %s\n' % (
+                        subtitle = 'Hostname: %s\n' % socket.gethostbyname(
-                        procUnitConfObj.getElementName(), procUnitConfObj.name)
-                    subtitle += 'Hostname: %s\n' % socket.gethostbyname(
                             socket.gethostname())
                         subtitle += 'Working directory: %s\n' % os.path.abspath('./')
                         subtitle += 'Configuration file: %s\n' % self.filename
                             subtitle += '[End time = %s]\n' % readUnitConfObj.endTime
                         a = Alarm(
-                        modes=modes,
+                            modes=self.alarm,
                             email=self.email,
                             message=message,
                             subject=subject,
                             filename=self.filename
                         )
-                    return a
+                        a.start()
                 def isPaused(self):
                     return 0
                     self.filename = filename
-                def setProxyCom(self):
+                def setProxy(self):
                     if not os.path.exists('/tmp/schain'):
                         os.mkdir('/tmp/schain')
                     xpub.bind('ipc:///tmp/schain/{}_pub'.format(self.id))
                     xsub = self.ctx.socket(zmq.XSUB)
                     xsub.bind('ipc:///tmp/schain/{}_sub'.format(self.id))
+                    self.monitor()
                     try:
                         zmq.proxy(xpub, xsub)
-                    except: # zmq.ContextTerminated:
+                    except zmq.ContextTerminated:
                         xpub.close()
                         xsub.close()
                     log.success('Starting {}: {}'.format(self.name, self.id), tag='')
                     self.start_time = time.time()
                     self.createObjects()
-                    # t = Thread(target=wait, args=(self.ctx, ))
+                    self.setProxy()
-                    # t.start()
+                    log.success('{} Done (Time: {}s)'.format(
-                    self.setProxyCom()
-                    # Iniciar todos los procesos .start(), monitoreo de procesos. ELiminar lo de abajo
-                    log.success('{} Done (time: {}s)'.format(
                         self.name,
-                        time.time()-self.start_time))
+                        time.time()-self.start_time), '')

schainpy/model/io/bltrIO_param.py +3 -3

             import numpy
+            import schainpy.admin
             from schainpy.model.proc.jroproc_base import ProcessingUnit, MPDecorator
             from schainpy.model.data.jrodata import Parameters
             from schainpy.model.io.jroIO_base import JRODataReader, isNumber
                     '''
                     if self.flagNoMoreFiles:
                         self.dataOut.flagNoData = True
-                        self.dataOut.error = 'No More files to read'
+                        raise schainpy.admin.SchainError('No More files to read')
-                        return
                     if not self.readNextBlock():
                         self.dataOut.flagNoData = True
-                        self.dataOut.error = 'Time for wait new file reach!!!'
+                        raise schainpy.admin.SchainError('Time for wait new file reach!!!')
                     self.set_output()

schainpy/model/io/bltrIO_spectra.py +2 -6

             SPEED_OF_LIGHT = 299792458
             SPEED_OF_LIGHT = 3e8
-            try:
-                from gevent import sleep
-            except:
-                from time import sleep
             from .utils import folder_in_range
+            import schainpy.admin
             from schainpy.model.data.jrodata import Spectra
             from schainpy.model.proc.jroproc_base import ProcessingUnit, Operation, MPDecorator
             from schainpy.utils import log
                     if self.flagNoMoreFiles:
                         self.dataOut.flagNoData = True
-                        self.dataOut.error = 'No more files'
+                        raise schainpy.admin.SchainError('No more files')
                     self.readBlock()

schainpy/model/io/jroIO_amisr.py 0 -1

                     if self.flagNoMoreFiles:
                         self.dataOut.flagNoData = True
-                        print('Process finished')
                         return 0
                     if self.__hasNotDataInBuffer():

schainpy/model/io/jroIO_base.py +33 -33

             import traceback
             import zmq
-            try:
-                from gevent import sleep
-            except:
-                from time import sleep
             from schainpy.model.data.jroheaderIO import PROCFLAG, BasicHeader, SystemHeader, RadarControllerHeader, ProcessingHeader
             from schainpy.model.data.jroheaderIO import get_dtype_index, get_numpy_dtype, get_procflag_dtype, get_dtype_width
             from schainpy.utils import log
                         idFile += 1
                         if not(idFile < len(self.filenameList)):
                             self.flagNoMoreFiles = 1
-            #                 print "[Reading] No more Files"
                             return 0
                         filename = self.filenameList[idFile]
                             for nTries in range(tries):
                                 if firstTime_flag:
-                                    print("\t[Reading] Waiting %0.2f sec for the next file: \"%s\" , try %03d ..." % (self.delay, filename, nTries + 1))
+                                    log.warning(
-                                    sleep(self.delay)
+                                        "Waiting %0.2f sec for the next file: \"%s\" , try %03d ..." % (self.delay, filename, nTries + 1),
+                                        self.name)
+                                    time.sleep(self.delay)
                                 else:
-                                    print("\t[Reading] Searching the next \"%s%04d%03d%03d%s\" file ..." % (self.optchar, self.year, self.doy, self.set, self.ext))
+                                    log.warning(
+                                        "Searching the next \"%s%04d%03d%03d%s\" file ..." % (self.optchar, self.year, self.doy, self.set, self.ext),
+                                        self.name)
                                 fullfilename, filename = checkForRealPath(
                                     self.path, self.foldercounter, self.year, self.doy, self.set, self.ext)
                             firstTime_flag = False
-                            log.warning('Skipping the file {} due to this file doesn\'t exist'.format(filename))
+                            log.warning(
+                                'Skipping the file {} due to this file doesn\'t exist'.format(filename),
+                                self.name)
                             self.set += 1
                             # si no encuentro el file buscado cambio de carpeta y busco en la siguiente carpeta
                             self.fp.close()
                         self.fp = open(fullfilename, 'rb')
                         self.flagNoMoreFiles = 0
-            #             print '[Reading] Setting the file: %s' % fullfilename
                     else:
+                        raise schainpy.admin.SchainError('Time for waiting new files reach')
                         self.fileSize = 0
                         self.filename = None
                         self.flagIsNewFile = 0
                         self.fp = None
                         self.flagNoMoreFiles = 1
-            #             print '[Reading] No more files to read'
                     return fileOk_flag
                         newFile = self.__setNextFileOffline()
                     if not(newFile):
-                        self.dataOut.error = 'No more files to read'
+                        raise schainpy.admin.SchainWarning('No more files to read')
-                        return 0
                     if self.verbose:
                         print('[Reading] Setting the file: %s' % self.filename)
                             return 0
                         print("[Reading] Waiting %0.2f seconds for the next block, try %03d ..." % (self.delay, nTries + 1))
-                        sleep(self.delay)
+                        time.sleep(self.delay)
                     return 0
                             "Waiting %0.2f seconds for the next block, try %03d ..." % (self.delay, nTries + 1),
                             self.name
                             )
-                        sleep(self.delay)
+                        time.sleep(self.delay)
                     return 0
                     # Skip block out of startTime and endTime
                     while True:
                         if not(self.__setNewBlock()):
-                            self.dataOut.error = 'No more files to read'
+                            raise schainpy.admin.SchainWarning('No more files to read')
-                            return 0
                         if not(self.readBlock()):
                             return 0
                         pattern_path = multi_path[0]
                     if path_empty:
-                        print("[Reading] No *%s files in %s for %s to %s" % (ext, pattern_path, startDate, endDate))
+                        raise schainpy.admin.SchainError("[Reading] No *%s files in %s for %s to %s" % (ext, pattern_path, startDate, endDate))
                     else:
                         if not dateList:
-                            print("[Reading] Date range selected invalid [%s - %s]: No *%s files in %s)" % (startDate, endDate, ext, path))
+                            raise schainpy.admin.SchainError("[Reading] Date range selected invalid [%s - %s]: No *%s files in %s)" % (startDate, endDate, ext, path))
                     if include_path:
                         return dateList, pathList
                           oneDDict=None,
                           twoDDict=None,
                           independentParam=None):
+                    self.online = online
+                    self.realtime = realtime
+                    self.delay = delay
+                    self.getByBlock = getblock
+                    self.nTxs = nTxs
+                    self.startTime = startTime
+                    self.endTime = endTime
+                    self.endDate = endDate
+                    self.startDate = startDate
                     if server is not None:
                         if 'tcp://' in server:
                             address = server
                                     break
                                 print('[Reading] Waiting %0.2f sec for an valid file in %s: try %02d ...' % (self.delay, path, nTries + 1))
-                                sleep(self.delay)
+                                time.sleep(self.delay)
                             if not(fullpath):
-                                self.dataOut.error = 'There isn\'t any valid file in {}'.format(path)
+                                raise schainpy.admin.SchainError('There isn\'t any valid file in {}'.format(path))
                                 return
                             self.year = year
                             basename, ext = os.path.splitext(file_name)
                             last_set = int(basename[-3:])
-                        self.online = online
-                        self.realtime = realtime
-                        self.delay = delay
                         ext = ext.lower()
                         self.ext = ext
-                        self.getByBlock = getblock
-                        self.nTxs = nTxs
-                        self.startTime = startTime
-                        self.endTime = endTime
-                        self.endDate = endDate
-                        self.startDate = startDate
                         # Added-----------------
                         self.selBlocksize = blocksize
                         self.selBlocktime = blocktime
                                 self.filenameList = []
                                 return
-                #         self.getBasicHeader()
                         if last_set != None:
                             self.dataOut.last_block = last_set * \
                                 self.processingHeaderObj.dataBlocksPerFile + self.basicHeaderObj.dataBlock

schainpy/model/io/jroIO_digitalRF.py +9 -13

             # METADATA
             import os
+            import time
             import datetime
             import numpy
             import timeit
             from fractions import Fraction
-            try:
+            import schainpy.admin
-                from gevent import sleep
-            except:
-                from time import sleep
             from schainpy.model.data.jroheaderIO import RadarControllerHeader, SystemHeader
             from schainpy.model.data.jrodata import Voltage
             from schainpy.model.proc.jroproc_base import ProcessingUnit, Operation, MPDecorator
-            from time import time
             import pickle
             try:
                     return False
                 def timeit(self, toExecute):
-                    t0 = time()
+                    t0 = time.time()
                     toExecute()
-                    self.executionTime = time() - t0
+                    self.executionTime = time.time() - t0
                     if self.oldAverage is None:
                         self.oldAverage = self.executionTime
                     self.oldAverage = (self.executionTime + self.count *
                             if self.__readNextBlock():
                                 break
                             if self.__thisUnixSample > self.__endUTCSecond * self.__sample_rate:
-                                self.dataOut.error = 'Error'
+                                raise schainpy.admin.SchainError('Error')
                                 return
                             if self.__flagDiscontinuousBlock:
-                                self.dataOut.error = 'discontinuous block found'
+                                raise schainpy.admin.SchainError('discontinuous block found')
                                 return
                             if not self.__online:
-                                self.dataOut.error = 'Online?'
+                                raise schainpy.admin.SchainError('Online?')
                                 return
                             err_counter += 1
                             if err_counter > nTries:
-                                self.dataOut.error = 'Max retrys reach'
+                                raise schainpy.admin.SchainError('Max retrys reach')
                                 return
                             print('[Reading] waiting %d seconds to read a new block' % seconds)
-                            sleep(seconds)
+                            time.sleep(seconds)
                     self.dataOut.data = self.__data_buffer[:,
                                                            self.__bufferIndex:self.__bufferIndex + self.__nSamples]

schainpy/model/io/jroIO_hf.py 0 -1

                 def getData(self):
                     if self.flagNoMoreFiles:
                         self.dataOut.flagNoData = True
-                        print('Process finished')
                         return 0
                     if self.__hasNotDataInBuffer():

schainpy/model/io/jroIO_kamisr.py 0 -1

                     if self.flagNoMoreFiles:
                         self.dataOut.flagNoData = True
-                        print('Process finished')
                         return 0
                     if self.__hasNotDataInBuffer():

schainpy/model/io/jroIO_madrigal.py +3 -1

             import numpy
             import h5py
+            import schainpy.admin
             from schainpy.model.io.jroIO_base import LOCALTIME, JRODataReader, JRODataWriter
             from schainpy.model.proc.jroproc_base import ProcessingUnit, Operation, MPDecorator
             from schainpy.model.data.jrodata import Parameters
                     '''
                     if self.flagNoMoreFiles:
                         self.dataOut.flagNoData = True
-                        self.dataOut.error = 'No file left to process'
+                        raise schainpy.admin.SchainError('No file left to process')
                         return 0
                     if not  self.readNextBlock():

schainpy/model/io/jroIO_param.py +3 -3

             import re
             import datetime
+            import schainpy.admin
             from schainpy.model.data.jrodata import *
             from schainpy.model.proc.jroproc_base import ProcessingUnit, Operation, MPDecorator
             from schainpy.model.io.jroIO_base import *
                     idFile = self.fileIndex
                     if not(idFile < len(self.filenameList)):
-                        self.dataOut.error = "No more Files"
+                        raise schainpy.admin.SchainError("No more Files")
                         return 0
                     filename = self.filenameList[idFile]
                     idFile = self.fileIndex
                     if not(idFile < len(self.filenameList)):
-                        self.dataOut.error = 'No more files'
+                        raise schainpy.admin.SchainError('No more files')
-                        return 0
                     filename = self.filenameList[idFile]
                     self.fp = h5py.File(filename, 'r')

schainpy/model/io/jroIO_spectra.py 0 -3

                     #Eliminar de la base la herencia
                     ProcessingUnit.__init__(self)#, **kwargs)
-            #         self.isConfig = False
                     self.pts2read_SelfSpectra = 0
                     self.__isFirstTimeOnline = 1
-            #         self.ippSeconds = 0
                     self.flagDiscontinuousBlock = 0
                     if self.flagNoMoreFiles:
                         self.dataOut.flagNoData = True
-                        print('Process finished')
                         return 0
                     self.flagDiscontinuousBlock = 0

schainpy/model/proc/jroproc_base.py +28 -53

             import zmq
             import time
             import pickle
+            import traceback
             from queue import Queue
             from threading import Thread
             from multiprocessing import Process
-            from zmq.utils.monitor import recv_monitor_message
             from schainpy.utils import log
                         self.receiver = None
                         self.i = 0
                         self.name = BaseClass.__name__
                         if 'plot' in self.name.lower() and not self.name.endswith('_'):
                             self.name = '{}{}'.format(self.CODE.upper(), 'Plot')
-                        self.start_time = time.time()
-                        if len(self.args) is 3:
+                        self.start_time = time.time()
-                            self.typeProc = "ProcUnit"
                         self.id = args[0]
                         self.inputId = args[1]
                         self.project_id = args[2]
-                        elif len(self.args) is 2:
+                        self.err_queue = args[3]
-                            self.id = args[0]
+                        self.typeProc = args[4]
-                            self.inputId = args[0]
-                            self.project_id = args[1]
-                            self.typeProc = "Operation"
                         self.queue = InputQueue(self.project_id, self.inputId)
+                        self.err_queue.put('#_start_#')
                     def subscribe(self):
                         '''
                         if self.inputId is None:
                             self.i += 1
-                            if self.i % 100 == 0:
+                            if self.i % 80 == 0:
                                 self.i = 0
                                 time.sleep(0.01)
                         '''
                         while True:
+                            try:
                                 BaseClass.run(self, **self.kwargs)
+                            except:
+                                err = traceback.format_exc()
+                                if 'No more files' in err:
+                                    log.warning('No more files to read', self.name)
+                                else:
+                                    self.err_queue.put('{}|{}'.format(self.name, err))
+                                self.dataOut.error = True
                             for op, optype, opId, kwargs in self.operations:
                                 if optype == 'self' and not self.dataOut.flagNoData:
                             self.publish(self.dataOut, self.id)
                             if self.dataOut.error:
-                                log.error(self.dataOut.error, self.name)
-                                # self.sender.send_multipart([str(self.project_id).encode(), 'end'.encode()])
                                 break
-                        time.sleep(1)
+                        time.sleep(0.5)
                     def runProc(self):
                         '''
                             if self.dataIn.flagNoData and self.dataIn.error is None:
                                 continue
+                            elif not self.dataIn.error:
+                                try:
                                     BaseClass.run(self, **self.kwargs)
+                                except:
-                            if self.dataIn.error:
+                                    self.err_queue.put('{}|{}'.format(self.name, traceback.format_exc()))
+                                    self.dataOut.error = True
+                            elif self.dataIn.error:
                                 self.dataOut.error = self.dataIn.error
                                 self.dataOut.flagNoData = True
                                 elif optype == 'external' and not self.dataOut.flagNoData:
                                     self.publish(self.dataOut, opId)
-                            if not self.dataOut.flagNoData or self.dataOut.error:
                             self.publish(self.dataOut, self.id)
                             for op, optype, opId, kwargs in self.operations:
-                                    if optype == 'self' and self.dataOut.error:
+                                if optype == 'external' and self.dataOut.error:
-                                        op(**kwargs)
-                                    elif optype == 'other' and self.dataOut.error:
-                                        self.dataOut = op.run(self.dataOut, **kwargs)
-                                    elif optype == 'external' and self.dataOut.error:
                                     self.publish(self.dataOut, opId)
-                            if self.dataIn.error:
+                            if self.dataOut.error:
                                 break
-                        time.sleep(1)
+                        time.sleep(0.5)
                     def runOp(self):
                         '''
                             dataOut = self.listen()
+                            if not dataOut.error:
                                 BaseClass.run(self, dataOut, **self.kwargs)
+                            else:
-                            if dataOut.error:
                                 break
-                        time.sleep(1)
                     def run(self):
                         if self.typeProc is "ProcUnit":
                             if self.inputId is not None:
                                 self.subscribe()
                             self.set_publisher()
                         self.close()
-                    def event_monitor(self, monitor):
-                        events = {}
-                        for name in dir(zmq):
-                            if name.startswith('EVENT_'):
-                                value = getattr(zmq, name)
-                                events[value] = name
-                        while monitor.poll():
-                            evt = recv_monitor_message(monitor)
-                            if evt['event'] == 32:
-                                self.connections += 1
-                            if evt['event'] == 512:
-                                pass
-                            evt.update({'description': events[evt['event']]})
-                            if evt['event'] == zmq.EVENT_MONITOR_STOPPED:
-                                break
-                        monitor.close()
-                        print('event monitor thread done!')
                     def close(self):
                         BaseClass.close(self)
+                        self.err_queue.put('#_end_#')
                         if self.sender:
                             self.sender.close()

schainpy/model/proc/jroproc_parameters.py 0 -1

                     self.dataOut.heightList = self.dataIn.getHeiRange()
                     self.dataOut.frequency = self.dataIn.frequency
                     # self.dataOut.noise = self.dataIn.noise
-                    self.dataOut.error = self.dataIn.error
                 def run(self):

General Comments 0

Write
Preview

You need to be logged in to leave comments. Login now

No TODOs yet

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages