schain-jc Commit - r1171:92232440951e

Multiprocessing for Spectra (all operation) working

George Yong -

r1171:92232440951e

parent child

Context file:

r1171:92232440951e

Collapse all files

schainpy/controller.py +138 -148

             '''
+            Updated on January , 2018, for multiprocessing purposes
+            Author: Sergio Cortez
             Created on September , 2012
-            @author:
             '''
+            from platform import python_version
             import sys
             import ast
             import datetime
             import traceback
             import math
             import time
+            import zmq
             from multiprocessing import Process, cpu_count
             from xml.etree.ElementTree import ElementTree, Element, SubElement, tostring
             from xml.dom import minidom
-            import schainpy
             from schainpy.admin import Alarm, SchainWarning
-            from schainpy.model import *
+            ### Temporary imports!!!
+            # from schainpy.model import *
+            from schainpy.model.io import *
+            from schainpy.model.graphics import *
+            from schainpy.model.proc.jroproc_base import *
+            from schainpy.model.proc.bltrproc_parameters import *
+            from schainpy.model.proc.jroproc_spectra import *
+            from schainpy.model.proc.jroproc_parameters import *
+            from schainpy.model.utils.jroutils_publish import *
             from schainpy.utils import log
+            ###
             DTYPES = {
                 'Voltage': '.r',
                     time.sleep(3)
             class ParameterConf():
                 id = None
                     print('Parameter[%s]: name = %s, value = %s, format = %s' % (self.id, self.name, self.value, self.format))
             class OperationConf():
                 id = None
                     self.id = '0'
                     self.name = None
                     self.priority = None
-                    self.type = 'self'
+                    self.topic = None
                 def __getNewId(self):
                     return int(self.id) * 10 + len(self.parmConfObjList) + 1
+                def getId(self):
+                    return self.id
                 def updateId(self, new_id):
                     self.id = str(new_id)
                     self.name = name
                     self.type = type
                     self.priority = priority
                     self.parmConfObjList = []
                 def removeParameters(self):
                     for parmConfObj in self.parmConfObjList:
                         parmConfObj.printattr()
-                def createObject(self, plotter_queue=None):
+                def createObject(self):
-                    if self.type == 'self':
-                        raise ValueError('This operation type cannot be created')
-                    if self.type == 'plotter':
+                    className = eval(self.name)
-                        if not plotter_queue:
+                    kwargs = self.getKwargs()
-                            raise ValueError('plotter_queue is not defined. Use:\nmyProject = Project()\nmyProject.setPlotterQueue(plotter_queue)')
-                        opObj = Plotter(self.name, plotter_queue)
-                    if self.type == 'external' or self.type == 'other':
+                    opObj = className(self.id, **kwargs)
-                        className = eval(self.name)
+                    opObj.start()
-                        kwargs = self.getKwargs()
-                        opObj = className(**kwargs)
+                    print('    Operation created')
                     return opObj
             class ProcUnitConf():
                 id = None
                     self.id = None
                     self.datatype = None
                     self.name = None
                     self.inputId = None
                     self.opConfObjList = []
                     return self.id
                 def updateId(self, new_id, parentId=parentId):
+                    '''
                     new_id = int(parentId) * 10 + (int(self.id) % 10)
                     new_inputId = int(parentId) * 10 + (int(self.inputId) % 10)
                     # If this proc unit has not inputs
-                    if self.inputId == '0':
+                    #if self.inputId == '0':
-                        new_inputId = 0
+                        #new_inputId = 0
                     n = 1
                     for opConfObj in self.opConfObjList:
                     self.parentId = str(parentId)
                     self.id = str(new_id)
-                    self.inputId = str(new_inputId)
+                    #self.inputId = str(new_inputId)
+                    '''
+                    n = 1
                 def getInputId(self):
                     return self.inputId
                     return self.procUnitObj
                 def setup(self, id, name, datatype, inputId, parentId=None):
+                    '''
+                        id sera el topico a publicar
+                        inputId sera el topico a subscribirse
+                    '''
                     # Compatible with old signal chain version
                     if datatype == None and name == None:
                         raise ValueError('datatype or name should be defined')
+                    #Definir una condicion para inputId cuando sea 0
                     if name == None:
                         if 'Proc' in datatype:
                             name = datatype
                     self.id = str(id)
                     self.name = name
                     self.datatype = datatype
                     self.inputId = inputId
                     self.parentId = parentId
                     self.opConfObjList = []
                     self.addOperation(name='run', optype='self')
                 def removeOperations(self):
                     return opObj
-                def addOperation(self, name, optype='self'):
+                def addOperation(self, name, optype = 'self'):
+                    '''
+                        Actualizacion - > proceso comunicacion
+                        En el caso de optype='self', elminar. DEfinir comuncacion IPC -> Topic
+                        definir el tipoc de socket o comunicacion ipc++
+                    '''
                     id = self.__getNewId()
-                    priority = self.__getPriority()
+                    priority = self.__getPriority() # Sin mucho sentido, pero puede usarse
                     opConfObj = OperationConf()
                     opConfObj.setup(id, name=name, priority=priority, type=optype)
                     return kwargs
-                def createObjects(self, plotter_queue=None):
+                def createObjects(self, dictUnits):
+                    '''
+                    Instancia de unidades de procesamiento.
+                    '''
                     className = eval(self.name)
                     kwargs = self.getKwargs()
-                    procUnitObj = className(**kwargs)
+                    procUnitObj = className(self.id, self.inputId, dictUnits, **kwargs) # necesitan saber su id y su entrada por fines de ipc
                     for opConfObj in self.opConfObjList:
                             procUnitObj.addOperationKwargs(
                                 opConfObj.id, **opConfObj.getKwargs())
                             continue
+                        print("Creating operation process:", opConfObj.name, "for", self.name)
-                        opObj = opConfObj.createObject(plotter_queue)
+                        opObj = opConfObj.createObject()
-                        self.opObjDict[opConfObj.id] = opObj
+                        #self.opObjDict[opConfObj.id] = opObj.name
-                        procUnitObj.addOperation(opObj, opConfObj.id)
+                        procUnitObj.addOperation(opConfObj.name, opConfObj.id)
+                    procUnitObj.start()
                     self.procUnitObj = procUnitObj
                     return procUnitObj
                 def run(self):
-                    is_ok = False
+                    is_ok = True
+                    """
                     for opConfObj in self.opConfObjList:
                         kwargs = {}
                                                     opId=opConfObj.id)
                         is_ok = is_ok or sts
+                    """
                     return is_ok
                 def close(self):
                     for opConfObj in self.opConfObjList:
                 def getElementName(self):
                     return self.ELEMENTNAME
                 def setup(self, id, name, datatype, path='', startDate='', endDate='',
                           startTime='', endTime='', parentId=None, server=None, **kwargs):
+                    '''
+                    *****el id del proceso sera el Topico
+                    Adicion de {topic}, si no esta presente -> error
+                    kwargs deben ser trasmitidos en la instanciacion
+                    '''
                     # Compatible with old signal chain version
                     if datatype == None and name == None:
                         raise ValueError('datatype or name should be defined')
                     self.opConfObjList = []
                 def addRunOperation(self, **kwargs):
                     opObj = self.addOperation(name='run', optype='self')
                     if self.server is None:
                         opObj.addParameter(
             class Project(Process):
                 id = None
-                # name = None
                 description = None
                 filename = None
                 ELEMENTNAME = 'Project'
-                plotterQueue = None
-                def __init__(self, plotter_queue=None):
+                def __init__(self):
                     Process.__init__(self)
                     self.id = None
                     self.description = None
                     self.email = None
                     self.alarm = None
-                    self.plotterQueue = plotter_queue
                     self.procUnitConfObjDict = {}
                 def __getNewId(self):
                 def setup(self, id, name='', description='', email=None, alarm=[]):
-                    print()
+                    print(' ')
                     print('*' * 60)
-                    print('   Starting SIGNAL CHAIN PROCESSING v%s ' % schainpy.__version__)
+                    print('*  Starting SIGNAL CHAIN PROCESSING (Multiprocessing) v%s *' % schainpy.__version__)
                     print('*' * 60)
-                    print()
+                    print("*  Python " + python_version() + "  *")
+                    print('*' * 19)
+                    print(' ')
                     self.id = str(id)
                     self.description = description
                     self.email = email
                     self.alarm = alarm
                 def addReadUnit(self, id=None, datatype=None, name=None, **kwargs):
+                    '''
+                    Actualizacion:
+                        Se agrego un nuevo argumento: topic -relativo a la forma de comunicar los procesos simultaneos
+                        * El id del proceso sera el topico al que se deben subscribir los procUnits para recibir la informacion(data)
+                    '''
                     if id is None:
                         idReadUnit = self.__getNewId()
                     else:
                                           parentId=self.id, **kwargs)
                     self.procUnitConfObjDict[readUnitConfObj.getId()] = readUnitConfObj
                     return readUnitConfObj
                 def addProcUnit(self, inputId='0', datatype=None, name=None):
-                    idProcUnit = self.__getNewId()
+                    '''
+                    Actualizacion:
+                        Se agrego dos nuevos argumentos: topic_read (lee data de otro procUnit) y topic_write(escribe o envia data a otro procUnit)
+                        Deberia reemplazar a "inputId"
+                        ** A fin de mantener el inputID, este sera la representaacion del topicoal que deben subscribirse. El ID propio de la intancia
+                        (proceso) sera el topico de la publicacion, todo sera asignado de manera dinamica.
+                    '''
+                    idProcUnit = self.__getNewId() #Topico para subscripcion
                     procUnitConfObj = ProcUnitConf()
-                    procUnitConfObj.setup(idProcUnit, name, datatype,
+                    procUnitConfObj.setup(idProcUnit, name, datatype, inputId, #topic_read, topic_write,
-                                          inputId, parentId=self.id)
+                                          parentId=self.id)
                     self.procUnitConfObjDict[procUnitConfObj.getId()] = procUnitConfObj
                 def createObjects(self):
                     for procUnitConfObj in list(self.procUnitConfObjDict.values()):
-                        procUnitConfObj.createObjects(self.plotterQueue)
+                        print("Creating process:", procUnitConfObj.name)
+                        procUnitConfObj.createObjects(self.procUnitConfObjDict)
-                def __connect(self, objIN, thisObj):
-                    thisObj.setInput(objIN.getOutputObj())
-                def connectObjects(self):
-                    for thisPUConfObj in list(self.procUnitConfObjDict.values()):
-                        inputId = thisPUConfObj.getInputId()
+                    print('All processes were created')
-                        if int(inputId) == 0:
-                            continue
-                        # Get input object
-                        puConfINObj = self.procUnitConfObjDict[inputId]
-                        puObjIN = puConfINObj.getProcUnitObj()
-                        # Get current object
-                        thisPUObj = thisPUConfObj.getProcUnitObj()
-                        self.__connect(puObjIN, thisPUObj)
                 def __handleError(self, procUnitConfObj, modes=None, stdout=True):
                     err = traceback.format_exception(sys.exc_info()[0],
                                                      sys.exc_info()[1],
                                                      sys.exc_info()[2])
                     log.error('{}'.format(err[-1]), procUnitConfObj.name)
                     message = ''.join(err)
                     self.filename = filename
-                def setPlotterQueue(self, plotter_queue):
+                def setProxyCom(self):
-                    raise NotImplementedError('Use schainpy.controller_api.ControllerThread instead Project class')
+                    ctx = zmq.Context()
+                    if not os.path.exists('/tmp/socketTmp'): os.mkdir('/tmp/socketTmp')
-                def getPlotterQueue(self):
+                    xsub = ctx.socket(zmq.XSUB)
+                    xsub.bind('ipc:///tmp/socketTmp/a')
-                    raise NotImplementedError('Use schainpy.controller_api.ControllerThread instead Project class')
+                    xpub = ctx.socket(zmq.XPUB)
+                    xpub.bind('ipc:///tmp/socketTmp/b')
-                def useExternalPlotter(self):
+                    print("Controller Ready: Processes and proxy created")
+                    zmq.proxy(xsub, xpub)
-                    raise NotImplementedError('Use schainpy.controller_api.ControllerThread instead Project class')
                 def run(self):
                     log.success('Starting {}'.format(self.name), tag='')
                     self.start_time = time.time()
                     self.createObjects()
-                    self.connectObjects()
+                    self.setProxyCom()
-                    keyList = list(self.procUnitConfObjDict.keys())
-                    keyList.sort()
-                    err = None
-                    while(True):
-                        is_ok = False
-                        for procKey in keyList:
+                    # Iniciar todos los procesos .start(), monitoreo de procesos. ELiminar lo de abajo
-                            procUnitConfObj = self.procUnitConfObjDict[procKey]
-                            try:
-                                sts = procUnitConfObj.run()
-                                is_ok = is_ok or sts
-                            except SchainWarning:
-                                err = self.__handleError(procUnitConfObj, modes=[2, 3], stdout=False)
-                                is_ok = False
-                                break
-                            except KeyboardInterrupt:
-                                is_ok = False
-                                break
-                            except ValueError as e:
-                                time.sleep(0.5)
-                                err = self.__handleError(procUnitConfObj)
-                                is_ok = False
-                                break
-                            except:
-                                time.sleep(0.5)
-                                err = self.__handleError(procUnitConfObj)
-                                is_ok = False
-                                break
-                        # If every process unit finished so end process
-                        if not(is_ok):
-                            break
-                        if not self.runController():
-                            break
                     # Closing every process
-                    for procKey in keyList:
-                        procUnitConfObj = self.procUnitConfObjDict[procKey]
-                        procUnitConfObj.close()
-                    if err is not None:
-                        err.start()
-                        # err.join()
                     log.success('{} finished (time: {}s)'.format(
                         self.name,
                         time.time()-self.start_time))

schainpy/model/graphics/figure.py +4 -3

             import time, datetime
             from schainpy.model.graphics import mpldriver
-            from schainpy.model.proc.jroproc_base import Operation
+            from schainpy.model.proc.jroproc_base import MPDecorator, Operation
             def isTimeInHourRange(datatime, xmin, xmax):
                 created = False
                 parameters = {}
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Operation.__init__(self, **kwargs)
+                    Operation.__init__(self)#, **kwargs)
                 def __del__(self):

schainpy/model/graphics/jroplot_spectra.py +63 -18

             from .figure import Figure, isRealtime, isTimeInHourRange
             from .plotting_codes import *
+            from schainpy.model.proc.jroproc_base import MPDecorator
+            from schainpy.utils import log
+            @MPDecorator
             class SpectraPlot(Figure):
                 isConfig = None
                 HEIGHTPROF = None
                 PREFIX = 'spc'
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Figure.__init__(self, **kwargs)
+                    Figure.__init__(self)#, **kwargs)
                     self.isConfig = False
                     self.__nsubplots = 1
                     self.WIDTH = 250
                     self.HEIGHT = 250
                     self.WIDTHPROF = 120
                         zmin            :    None,
                         zmax            :    None
                     """
+                    if dataOut.flagNoData:
+                        return dataOut
                     if realtime:
                         if not(isRealtime(utcdatatime = dataOut.utctime)):
                             print('Skipping this plot function')
                               wr_period=wr_period,
                               thisDatetime=thisDatetime)
+                    return dataOut
+            @MPDecorator
             class CrossSpectraPlot(Figure):
                 isConfig = None
                 HEIGHTPROF = None
                 PREFIX = 'cspc'
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Figure.__init__(self, **kwargs)
+                    Figure.__init__(self)#, **kwargs)
                     self.isConfig = False
                     self.__nsubplots = 4
                     self.counter_imagwr = 0
                         zmax            :    None
                     """
+                    if dataOut.flagNoData:
+                        return dataOut
                     if pairsList == None:
                         pairsIndexList = dataOut.pairsIndexList
                     else:
                               wr_period=wr_period,
                               thisDatetime=thisDatetime)
+                    return dataOut
+            @MPDecorator
             class RTIPlot(Figure):
                 __isConfig = None
                 HEIGHTPROF = None
                 PREFIX = 'rti'
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Figure.__init__(self, **kwargs)
+                    Figure.__init__(self)#, **kwargs)
                     self.timerange = None
                     self.isConfig = False
                     self.__nsubplots = 1
                         zmin            :    None,
                         zmax            :    None
                     """
+                    if dataOut.flagNoData:
+                        return dataOut
                     #colormap = kwargs.get('colormap', 'jet')
                     if HEIGHT is not None:
                               wr_period=wr_period,
                               thisDatetime=thisDatetime,
                               update_figfile=update_figfile)
+                    return dataOut
+            @MPDecorator
             class CoherenceMap(Figure):
                 isConfig = None
                 __nsubplots = None
                 HEIGHTPROF = None
                 PREFIX = 'cmap'
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Figure.__init__(self, **kwargs)
+                    Figure.__init__(self)#, **kwargs)
                     self.timerange = 2*60*60
                     self.isConfig = False
                     self.__nsubplots = 1
                         server=None, folder=None, username=None, password=None,
                         ftp_wei=0, exp_code=0, sub_exp_code=0, plot_pos=0):
+                    if dataOut.flagNoData:
+                        return dataOut
                     if not isTimeInHourRange(dataOut.datatime, xmin, xmax):
                         return
                               thisDatetime=thisDatetime,
                               update_figfile=update_figfile)
+                    return dataOut
+            @MPDecorator
             class PowerProfilePlot(Figure):
                 isConfig = None
                 HEIGHTPROF = None
                 PREFIX = 'spcprofile'
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Figure.__init__(self, **kwargs)
+                    Figure.__init__(self)#, **kwargs)
                     self.isConfig = False
                     self.__nsubplots = 1
                         ftp=False, wr_period=1, server=None,
                         folder=None, username=None, password=None):
+                    if dataOut.flagNoData:
+                        return dataOut
                     if channelList == None:
                         channelIndexList = dataOut.channelIndexList
                               ftp=ftp,
                               wr_period=wr_period,
                               thisDatetime=thisDatetime)
+                    return dataOut
+            @MPDecorator
             class SpectraCutPlot(Figure):
                 isConfig = None
                 HEIGHTPROF = None
                 PREFIX = 'spc_cut'
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Figure.__init__(self, **kwargs)
+                    Figure.__init__(self)#, **kwargs)
                     self.isConfig = False
                     self.__nsubplots = 1
                         folder=None, username=None, password=None,
                         xaxis="frequency"):
+                    if dataOut.flagNoData:
+                        return dataOut
                     if channelList == None:
                         channelIndexList = dataOut.channelIndexList
                               wr_period=wr_period,
                               thisDatetime=thisDatetime)
+                    return dataOut
+            @MPDecorator
             class Noise(Figure):
                 isConfig = None
                 PREFIX = 'noise'
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Figure.__init__(self, **kwargs)
+                    Figure.__init__(self)#, **kwargs)
                     self.timerange = 24*60*60
                     self.isConfig = False
                     self.__nsubplots = 1
                         server=None, folder=None, username=None, password=None,
                         ftp_wei=0, exp_code=0, sub_exp_code=0, plot_pos=0):
+                    if dataOut.flagNoData:
+                        return dataOut
                     if not isTimeInHourRange(dataOut.datatime, xmin, xmax):
                         return
                     if save:
                         self.save_data(self.filename_noise, noisedB, thisDatetime)
+                    return dataOut
+            @MPDecorator
             class BeaconPhase(Figure):
                 __isConfig = None
                 PREFIX = 'beacon_phase'
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Figure.__init__(self, **kwargs)
+                    Figure.__init__(self)#, **kwargs)
                     self.timerange = 24*60*60
                     self.isConfig = False
                     self.__nsubplots = 1
                         server=None, folder=None, username=None, password=None,
                         ftp_wei=0, exp_code=0, sub_exp_code=0, plot_pos=0):
+                    if dataOut.flagNoData:
+                        return dataOut
                     if not isTimeInHourRange(dataOut.datatime, xmin, xmax):
                         return
                               ftp=ftp,
                               wr_period=wr_period,
                               thisDatetime=thisDatetime,
-                              update_figfile=update_figfile)
  No newline at end of file
+                              update_figfile=update_figfile)
+                    return dataOut  #Yong
  No newline at end of file

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

             except:
                 from time import sleep
-            import schainpy.admin
             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
                         self.flagIsNewFile = 0
                         self.fp = None
                         self.flagNoMoreFiles = 1
+            #             print '[Reading] No more files to read'
                     return fileOk_flag
                     else:
                         newFile = self.__setNextFileOffline()
                     if not(newFile):
-                        raise schainpy.admin.SchainWarning('No more files to read')
+                        raise(schainpy.admin.SchainWarning('No more files to read'))
                         return 0
                     if self.verbose:
                     # Skip block out of startTime and endTime
                     while True:
                         if not(self.__setNewBlock()):
-                            raise schainpy
+                            raise(schainpy.admin.SchainWarning('No more files'))
                             return 0
                         if not(self.readBlock()):
                                 if fullpath:
                                     break
-                                print('[Reading] Waiting %0.2f sec for an valid file in %s: try %02d ...' % (delay, path, nTries + 1))
+                                print('[Reading] Waiting %0.2f sec for an valid file in %s: try %02d ...' % (self.delay, path, nTries + 1))
-                                sleep(delay)
+                                sleep(self.delay)
                             if not(fullpath):
-                                raise schainpy.admin.SchainWarning('There isn\'t any valid file in {}'.format(path))
+                                raise(schainpy.admin.SchainWarning('There isn\'t any valid file in {}'.format(path)))
                                 return
                             self.year = year

schainpy/model/io/jroIO_spectra.py +6 -5

             import numpy
             from schainpy.model.io.jroIO_base import LOCALTIME, JRODataReader, JRODataWriter
-            from schainpy.model.proc.jroproc_base import ProcessingUnit, Operation
+            from schainpy.model.proc.jroproc_base import ProcessingUnit, Operation, MPDecorator
             from schainpy.model.data.jroheaderIO import PROCFLAG, BasicHeader, SystemHeader, RadarControllerHeader, ProcessingHeader
             from schainpy.model.data.jrodata import Spectra
+            @MPDecorator
             class SpectraReader(JRODataReader, ProcessingUnit):
                 """
                 Esta clase permite leer datos de espectros desde archivos procesados (.pdata). La lectura
                 rdPairList = []
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
                     """
                     Inicializador de la clase SpectraReader para la lectura de datos de espectros.
                     """
                     #Eliminar de la base la herencia
-                    ProcessingUnit.__init__(self, **kwargs)
+                    ProcessingUnit.__init__(self)#, **kwargs)
             #         self.isConfig = False
                     """
                     spc = numpy.transpose( self.data_spc, (0,2,1) )
-                    if self.processingHeaderObj.shif_fft:
+                    if not( self.processingHeaderObj.shif_fft ):
                         spc = numpy.roll( spc, self.processingHeaderObj.profilesPerBlock/2, axis=2 ) #desplaza a la derecha en el eje 2 determinadas posiciones
                     data = spc.reshape((-1))
                     data = data.astype(self.dtype[0])
                     if self.data_cspc is not None:
                         data = numpy.zeros( self.shape_cspc_Buffer, self.dtype )
                         cspc = numpy.transpose( self.data_cspc, (0,2,1) )
-                        if self.processingHeaderObj.shif_fft:
+                        if not( self.processingHeaderObj.shif_fft ):
                             cspc = numpy.roll( cspc, self.processingHeaderObj.profilesPerBlock/2, axis=2 ) #desplaza a la derecha en el eje 2 determinadas posiciones
                         data['real'] = cspc.real
                         data['imag'] = cspc.imag

schainpy/model/proc/jroproc_base.py +392 -214

This diff has been collapsed as it changes many lines, (606 lines changed) Show them Hide them
	@@ -1,41 +1,39
	1	'''	1	'''
	2		2	Updated for multiprocessing
	3	$Author: murco $	3	Author : Sergio Cortez
	4	$Id: jroproc_base.py 1 2012-11-12 18:56:07Z murco $	4	Jan 2018
			5	Abstract:
			6	Base class for processing units and operations. A decorator provides multiprocessing features and interconnect the processes created.
			7	The argument (kwargs) sent from the controller is parsed and filtered via the decorator for each processing unit or operation instantiated.
			8	The decorator handle also the methods inside the processing unit to be called from the main script (not as operations) (OPERATION -> type ='self').
			9
			10	Based on:
			11	$Author: murco $
			12	$Id: jroproc_base.py 1 2012-11-12 18:56:07Z murco $
	5	'''	13	'''
			14	from platform import python_version
	6	import inspect	15	import inspect
	7	from fuzzywuzzy import process	16	import zmq
	8		17	import time
	9	def checkKwargs(method, kwargs):	18	import pickle
	10	currentKwargs = kwargs	19	import os
	11	choices = inspect.getargspec(method).args	20	from multiprocessing import Process
	12	try:
	13	choices.remove('self')
	14	except Exception as e:
	15	pass
	16		21
	17	try:	22	from schainpy.utils import log
	18	choices.remove('dataOut')
	19	except Exception as e:
	20	pass
	21		23
	22	for kwarg in kwargs:
	23	fuzz = process.extractOne(kwarg, choices)
	24	if fuzz is None:
	25	continue
	26	if fuzz[1] < 100:
	27	raise Exception('\x1b[0;32;40mDid you mean {} instead of {} in {}? \x1b[0m'.
	28	format(fuzz[0], kwarg, method.__self__.__class__.__name__))
	29		24
	30	class ProcessingUnit(object):	25	class ProcessingUnit(object):
	31		26
	32	"""	27	"""
	33	Esta es la clase base para el procesamiento de datos.	28	Update - Jan 2018 - MULTIPROCESSING
			29	All the "call" methods present in the previous base were removed.
			30	The majority of operations are independant processes, thus
			31	the decorator is in charge of communicate the operation processes
			32	with the proccessing unit via IPC.
	34		33
	35	Contiene el metodo "call" para llamar operaciones. Las operaciones pueden ser:	34	The constructor does not receive any argument. The remaining methods
	36	- Metodos internos (callMethod)	35	are related with the operations to execute.
	37	- Objetos del tipo Operation (callObject). Antes de ser llamados, estos objetos	36
	38	tienen que ser agreagados con el metodo "add".
	39		37
	40	"""	38	"""
	41	# objeto de datos de entrada (Voltage, Spectra o Correlation)	39	# objeto de datos de entrada (Voltage, Spectra o Correlation)
	@@ -43,33 +41,25 class ProcessingUnit(object):
	43	dataInList = []	41	dataInList = []
	44		42
	45	# objeto de datos de entrada (Voltage, Spectra o Correlation)	43	# objeto de datos de entrada (Voltage, Spectra o Correlation)
			44
			45	id = None
			46	inputId = None
			47
	46	dataOut = None	48	dataOut = None
	47		49
			50	dictProcs = None
			51
	48	operations2RunDict = None	52	operations2RunDict = None
	49		53
	50	isConfig = False	54	isConfig = False
	51		55
	52		56	def __init__(self):
	53	def __init__(self, args, *kwargs):
	54		57
	55	self.dataIn = None	58	self.dataIn = None
	56	self.dataInList = []
	57
	58	self.dataOut = None	59	self.dataOut = None
	59		60
	60	self.operations2RunDict = {}
	61	self.operationKwargs = {}
	62
	63	self.isConfig = False	61	self.isConfig = False
	64		62
	65	self.args = args
	66	self.kwargs = kwargs
	67
	68	if not hasattr(self, 'name'):
	69	self.name = self.__class__.__name__
	70
	71	checkKwargs(self.run, kwargs)
	72
	73	def getAllowedArgs(self):	63	def getAllowedArgs(self):
	74	if hasattr(self, '__attrs__'):	64	if hasattr(self, '__attrs__'):
	75	return self.__attrs__	65	return self.__attrs__
	@@ -81,27 +71,30 class ProcessingUnit(object):
	81	'''	71	'''
	82		72
	83	self.operationKwargs[objId] = kwargs	73	self.operationKwargs[objId] = kwargs
	84		74
	85
	86	def addOperation(self, opObj, objId):	75	def addOperation(self, opObj, objId):
	87		76
	88	"""	77	"""
	89	Agrega un objeto del tipo "Operation" (opObj) a la lista de objetos "self.objectList" y retorna el	78	This method is used in the controller, and update the dictionary containing the operations to execute. The dict
	90	identificador asociado a este objeto.	79	posses the id of the operation process (IPC purposes)
	91		80
	92	Input:	81	Agrega un objeto del tipo "Operation" (opObj) a la lista de objetos "self.objectList" y retorna el
			82	identificador asociado a este objeto.
	93		83
	94	object : objeto de la clase "Operation"	84	Input:
	95		85
	96	Return:	86	object : objeto de la clase "Operation"
			87
			88	Return:
	97		89
	98	objId : identificador del objeto, necesario para ~~ejecutar la operacion~~	90	objId : identificador del objeto, necesario para comunicar con master(procUnit)
	99	"""	91	"""
	100		92
	101	self.operations2RunDict[objId] = opObj	93	self.operations2RunDict[objId] = opObj
	102		94
	103	return objId	95	return objId
	104		96
			97
	105	def getOperationObj(self, objId):	98	def getOperationObj(self, objId):
	106		99
	107	if objId not in list(self.operations2RunDict.keys()):	100	if objId not in list(self.operations2RunDict.keys()):
	@@ -120,241 +113,426 class ProcessingUnit(object):
	120	**kwargs : Diccionario de argumentos de la funcion a ejecutar	113	**kwargs : Diccionario de argumentos de la funcion a ejecutar
	121	"""	114	"""
	122		115
	123	raise NotImplementedError	116	raise NotImplementedError
	124
	125	def callMethod(self, name, opId):
	126
	127	"""
	128	Ejecuta el metodo con el nombre "name" y con argumentos **kwargs de la propia clase.
	129
	130	Input:
	131	name : nombre del metodo a ejecutar
	132
	133	**kwargs : diccionario con los nombres y valores de la funcion a ejecutar.
	134
	135	"""
	136		117
	137	#Checking the inputs	118	def setup(self):
	138	if name == 'run':
	139
	140	if not self.checkInputs():
	141	self.dataOut.flagNoData = True
	142	return False
	143	else:
	144	#Si no es un metodo RUN la entrada es la misma dataOut (interna)
	145	if self.dataOut is not None and self.dataOut.isEmpty():
	146	return False
	147		119
	148	#Getting the pointer to method	120	raise NotImplementedError
	149	methodToCall = getattr(self, name)
	150		121
	151	#Executing the self method	122	def run(self):
	152		123
	153	if hasattr(self, 'mp'):	124	raise NotImplementedError
	154	if name=='run':
	155	if self.mp is False:
	156	self.mp = True
	157	self.start()
	158	else:
	159	self.operationKwargs[opId]['parent'] = self.kwargs
	160	methodToCall(**self.operationKwargs[opId])
	161	else:
	162	if name=='run':
	163	methodToCall(**self.kwargs)
	164	else:
	165	methodToCall(**self.operationKwargs[opId])
	166		125
	167	if self.dataOut is None:	126	def close(self):
	168	return False	127	#Close every thread, queue or any other object here is it is neccesary.
			128	return
			129
			130	class Operation(object):
	169		131
	170	if self.dataOut.isEmpty():	132	"""
	171	return False	133	Update - Jan 2018 - MULTIPROCESSING
	172		134
	173	return True	135	Most of the methods remained the same. The decorator parse the arguments and executed the run() method for each process.
			136	The constructor doe snot receive any argument, neither the baseclass.
	174		137
	175	def callObject(self, objId):
	176		138
	177	"""	139	Clase base para definir las operaciones adicionales que se pueden agregar a la clase ProcessingUnit
	178	Ejecuta la operacion asociada al identificador del objeto "objId"	140	y necesiten acumular informacion previa de los datos a procesar. De preferencia usar un buffer de
			141	acumulacion dentro de esta clase
	179		142
	180	Input:	143	Ejemplo: Integraciones coherentes, necesita la informacion previa de los n perfiles anteriores (bufffer)
	181		144
	182	objId : identificador del objeto a ejecutar	145	"""
			146	id = None
			147	__buffer = None
			148	dest = None
			149	isConfig = False
			150	readyFlag = None
	183		151
	184	**kwargs : diccionario con los nombres y valores de la funcion a ejecutar.	152	def __init__(self):
	185		153
	186	Return:	154	self.buffer = None
			155	self.dest = None
			156	self.isConfig = False
			157	self.readyFlag = False
	187		158
	188	None	159	if not hasattr(self, 'name'):
	189	"""	160	self.name = self.__class__.__name__
			161
			162	def getAllowedArgs(self):
			163	if hasattr(self, '__attrs__'):
			164	return self.__attrs__
			165	else:
			166	return inspect.getargspec(self.run).args
	190		167
	191	if self.dataOut is not None and self.dataOut.isEmpty():	168	def setup(self):
	192	return False
	193		169
	194	externalProcObj = self.operations2RunDict[objId]	170	self.isConfig = True
	195		171
	196	if hasattr(externalProcObj, 'mp'):	172	raise NotImplementedError
	197	if externalProcObj.mp is False:
	198	externalProcObj.kwargs['parent'] = self.kwargs
	199	self.operationKwargs[objId] = externalProcObj.kwargs
	200	externalProcObj.mp = True
	201	externalProcObj.start()
	202	else:
	203	externalProcObj.run(self.dataOut, **externalProcObj.kwargs)
	204	self.operationKwargs[objId] = externalProcObj.kwargs
	205		173
	206		174
	207	return True	175	def run(self, dataIn, **kwargs):
	208		176
	209	def call(self, opType, opName=None, opId=None):
	210	"""	177	"""
	211	Return True si ejecuta la operacion interna nombrada "opName" o la operacion externa	178	Realiza las operaciones necesarias sobre la dataIn.data y actualiza los
	212	identificada con el id "opId"; con los argumentos "**kwargs".	179	atributos del objeto dataIn.
	213
	214	False si la operacion no se ha ejecutado.
	215		180
	216	Input:	181	Input:
	217		182
	218	opType : Puede ser "self" o "external"	183	dataIn : objeto del tipo JROData
	219		184
	220	Depende del tipo de operacion para llamar a:callMethod or callObject:	185	Return:
	221		186
	222	1. If opType = "self": Llama a un metodo propio de esta clase:	187	None
	223		188
	224	name_method = getattr(self, name)	189	Affected:
	225	name_method(**kwargs)	190	__buffer : buffer de recepcion de datos.
	226		191
			192	"""
			193	if not self.isConfig:
			194	self.setup(**kwargs)
	227		195
	228	2. If opType = "other" o"external": Llama al metodo "run()" de una instancia de la	196	raise NotImplementedError
	229	clase "Operation" o de un derivado de ella:
	230		197
	231	instanceName = self.operationList[opId]	198	def close(self):
	232	instanceName.run(**kwargs)
	233		199
	234	opName : Si la operacion es interna (opType = 'self'), entonces el "opName" sera	200	pass
	235	usada para llamar a un metodo interno de la clase Processing
	236		201
	237	opId : Si la operacion es externa (opType = 'other' o 'external), entonces el
	238	"opId" sera usada para llamar al metodo "run" de la clase Operation
	239	registrada anteriormente con ese Id
	240		202
	241	Exception:	203	######### Decorator #########
	242	Este objeto de tipo Operation debe de haber sido agregado antes con el metodo:
	243	"addOperation" e identificado con el valor "opId" = el id de la operacion.
	244	De lo contrario retornara un error del tipo ValueError
	245		204
	246	"""
	247		205
	248	if opType == 'self':	206	def MPDecorator(BaseClass):
			207
			208	"""
			209	"Multiprocessing class decorator"
	249		210
	250	if not opName:	211	This function add multiprocessing features to the base class. Also,
	251	raise ValueError("opName parameter should be defined")	212	it handle the communication beetween processes (readers, procUnits and operations).
			213	Receive the arguments at the moment of instantiation. According to that, discriminates if it
			214	is a procUnit or an operation
			215	"""
			216
			217	class MPClass(BaseClass, Process):
			218
			219	"This is the overwritten class"
			220	operations2RunDict = None
			221	socket_l = None
			222	socket_p = None
			223	socketOP = None
			224	socket_router = None
			225	dictProcs = None
			226	typeProc = None
			227	def __init__(self, args, *kwargs):
			228	super(MPClass, self).__init__()
			229	Process.__init__(self)
			230
			231
			232	self.operationKwargs = {}
			233	self.args = args
			234
			235
			236	self.operations2RunDict = {}
			237	self.kwargs = kwargs
			238
			239	# The number of arguments (args) determine the type of process
			240
			241	if len(self.args) is 3:
			242	self.typeProc = "ProcUnit"
			243	self.id = args[0] #topico de publicacion
			244	self.inputId = args[1] #topico de subcripcion
			245	self.dictProcs = args[2] #diccionario de procesos globales
			246	else:
			247	self.id = args[0]
			248	self.typeProc = "Operation"
			249
			250	def addOperationKwargs(self, objId, **kwargs):
			251
			252	self.operationKwargs[objId] = kwargs
	252		253
	253	sts = self.callMethod(opName, opId)	254	def getAllowedArgs(self):
	254		255
	255	elif opType == 'other' or opType == 'external' or opType == 'plotter':	256	if hasattr(self, '__attrs__'):
			257	return self.__attrs__
			258	else:
			259	return inspect.getargspec(self.run).args
			260
			261
			262	def sockListening(self, topic):
			263
			264	"""
			265	This function create a socket to receive objects.
			266	The 'topic' argument is related to the publisher process from which the self process is
			267	listening (data).
			268	In the case were the self process is listening to a Reader (proc Unit),
			269	special conditions are introduced to maximize parallelism.
			270	"""
			271
			272	cont = zmq.Context()
			273	zmq_socket = cont.socket(zmq.SUB)
			274	if not os.path.exists('/tmp/socketTmp'):
			275	os.mkdir('/tmp/socketTmp')
			276
			277	if 'Reader' in self.dictProcs[self.inputId].name:
			278	zmq_socket.connect('ipc:///tmp/socketTmp/b')
			279
			280	else:
			281	zmq_socket.connect('ipc:///tmp/socketTmp/%s' % self.inputId)
			282
			283	#log.error('RECEIVING FROM {} {}'.format(self.inputId, str(topic).encode()))
			284	zmq_socket.setsockopt(zmq.SUBSCRIBE, str(topic).encode()) #yong
	256		285
	257	if not opId:	286	return zmq_socket
	258	raise ValueError("opId parameter should be defined")
	259		287
	260	if opId not in list(self.operations2RunDict.keys()):
	261	raise ValueError("Any operation with id=%s has been added" %str(opId))
	262		288
	263	sts = self.callObject(opId)	289	def listenProc(self, sock):
	264		290
	265	else:	291	"""
	266	raise ValueError("opType should be 'self', 'external' or 'plotter'; and not '%s'" %opType)	292	This function listen to a ipc addres until a message is recovered. To serialize the
			293	data (object), pickle has been use.
			294	The 'sock' argument is the socket previously connect to an ipc address and with a topic subscription.
			295	"""
			296
			297	a = sock.recv_multipart()
			298	a = pickle.loads(a[1])
			299	return a
	267		300
	268	return sts	301	def sockPublishing(self):
	269		302
	270	def setInput(self, dataIn):	303	"""
			304	This function create a socket for publishing purposes.
			305	Depending on the process type from where is created, it binds or connect
			306	to special IPC addresses.
			307	"""
			308	time.sleep(4) #yong
			309	context = zmq.Context()
			310	zmq_socket = context.socket(zmq.PUB)
			311	if not os.path.exists('/tmp/socketTmp'): os.mkdir('/tmp/socketTmp')
			312	if 'Reader' in self.dictProcs[self.id].name:
			313	zmq_socket.connect('ipc:///tmp/socketTmp/a')
			314	else:
			315	zmq_socket.bind('ipc:///tmp/socketTmp/%s' % self.id)
	271		316
	272	self.dataIn = dataIn	317	return zmq_socket
	273	self.dataInList.append(dataIn)
	274		318
	275	def getOutputObj(self):	319	def publishProc(self, sock, data):
	276		320
	277	return self.dataOut	321	"""
			322	This function publish a python object (data) under a specific topic in a socket (sock).
			323	Usually, the topic is the self id of the process.
			324	"""
	278		325
	279	def checkInputs(self):	326	sock.send_multipart([str(self.id).encode(), pickle.dumps(data)]) #yong
			327
			328	return True
	280		329
	281	for thisDataIn in self.dataInList:	330	def sockOp(self):
	282		331
	283	if thisDataIn.isEmpty():	332	"""
	284	return False	333	This function create a socket for communication purposes with operation processes.
			334	"""
	285		335
	286	return True	336	cont = zmq.Context()
			337	zmq_socket = cont.socket(zmq.DEALER)
			338
			339	if python_version()[0] == '2':
			340	zmq_socket.setsockopt(zmq.IDENTITY, self.id)
			341	if python_version()[0] == '3':
			342	zmq_socket.setsockopt_string(zmq.IDENTITY, self.id)
	287		343
	288	def setup(self):
	289		344
	290	raise NotImplementedError	345	return zmq_socket
	291		346
	292	def run(self):
	293		347
	294	raise NotImplementedError	348	def execOp(self, socket, opId, dataObj):
	295		349
	296	def close(self):	350	"""
	297	#Close every thread, queue or any other object here is it is neccesary.	351	This function 'execute' an operation main routine by establishing a
	298	return	352	connection with it and sending a python object (dataOut).
			353	"""
			354	if not os.path.exists('/tmp/socketTmp'): os.mkdir('/tmp/socketTmp')
			355	socket.connect('ipc:///tmp/socketTmp/%s' %opId)
			356
			357
			358	socket.send(pickle.dumps(dataObj)) #yong
			359
			360	argument = socket.recv_multipart()[0]
			361
			362	argument = pickle.loads(argument)
			363
			364	return argument
			365
			366	def sockIO(self):
	299		367
	300	class Operation(object):	368	"""
			369	Socket defined for an operation process. It is able to recover the object sent from another process as well as a
			370	identifier of who sent it.
			371	"""
	301		372
	302	"""	373	cont = zmq.Context()
	303	Clase base para definir las operaciones adicionales que se pueden agregar a la clase ProcessingUnit	374	if not os.path.exists('/tmp/socketTmp'): os.mkdir('/tmp/socketTmp')
	304	y necesiten acumular informacion previa de los datos a procesar. De preferencia usar un buffer de	375	socket = cont.socket(zmq.ROUTER)
	305	acumulacion dentro de esta clase	376	socket.bind('ipc:///tmp/socketTmp/%s' % self.id)
	306		377
	307	Ejemplo: Integraciones coherentes, necesita la informacion previa de los n perfiles anteriores (bufffer)	378	return socket
	308		379
	309	"""	380	def funIOrec(self, socket):
	310		381
	311	__buffer = None	382	"""
	312	isConfig = False	383	Operation method, recover the id of the process who sent a python object.
			384	The 'socket' argument is the socket binded to a specific process ipc.
			385	"""
	313		386
	314	def __init__(self, **kwargs):	387	#id_proc = socket.recv()
			388
			389	#dataObj = socket.recv_pyobj()
			390
			391	dataObj = socket.recv_multipart()
			392
			393	dataObj[1] = pickle.loads(dataObj[1])
			394	return dataObj[0], dataObj[1]
			395
			396	def funIOsen(self, socket, data, dest):
			397
			398	"""
			399	Operation method, send a python object to a specific destination.
			400	The 'dest' argument is the id of a proccesinf unit.
			401	"""
			402
			403	socket.send_multipart([dest, pickle.dumps(data)]) #yong
	315		404
	316	self.__buffer = None	405	return True
	317	self.isConfig = False
	318	self.kwargs = kwargs
	319	if not hasattr(self, 'name'):
	320	self.name = self.__class__.__name__
	321	checkKwargs(self.run, kwargs)
	322		406
	323	def getAllowedArgs(self):
	324	if hasattr(self, '__attrs__'):
	325	return self.__attrs__
	326	else:
	327	return inspect.getargspec(self.run).args
	328		407
	329	def ~~setup~~(self):	408	def runReader(self):
	330		409
	331	self.isConfig = True	410	# time.sleep(3)
			411	while True:
			412
			413	BaseClass.run(self, **self.kwargs)
	332		414
	333	raise NotImplementedError
	334		415
	335	def run(self, dataIn, **kwargs):	416	keyList = list(self.operations2RunDict.keys())
			417	keyList.sort()
			418
			419	for key in keyList:
			420	self.socketOP = self.sockOp()
			421	self.dataOut = self.execOp(self.socketOP, key, self.dataOut)
	336		422
	337	"""	423
	338	Realiza las operaciones necesarias sobre la dataIn.data y actualiza los	424	if self.flagNoMoreFiles: #Usar un objeto con flags para saber si termino el proc o hubo un error
	339	atributos del objeto dataIn.	425	self.publishProc(self.socket_p, "Finish")
			426	break
	340		427
	341	Input:	428	if self.dataOut.flagNoData:
	342		429	continue
	343	dataIn : objeto del tipo JROData	430
			431	print("Publishing data...")
			432	self.publishProc(self.socket_p, self.dataOut)
			433	# time.sleep(2)
			434
			435
			436	print("%s done" %BaseClass.__name__)
			437	return 0
			438
			439	def runProc(self):
	344		440
	345	Return:	441	# All the procUnits with kwargs that require a setup initialization must be defined here.
	346		442
	347	None	443	if self.setupReq:
			444	BaseClass.setup(self, **self.kwargs)
	348		445
	349	Affected:	446	while True:
	350	__buffer : buffer de recepcion de datos.	447	self.dataIn = self.listenProc(self.socket_l)
			448	print("%s received data" %BaseClass.__name__)
			449
			450	if self.dataIn == "Finish":
			451	break
			452
			453	m_arg = list(self.kwargs.keys())
			454	num_arg = list(range(1,int(BaseClass.run.__code__.co_argcount)))
			455
			456	run_arg = {}
			457
			458	for var in num_arg:
			459	if BaseClass.run.__code__.co_varnames[var] in m_arg:
			460	run_arg[BaseClass.run.__code__.co_varnames[var]] = self.kwargs[BaseClass.run.__code__.co_varnames[var]]
			461
			462	#BaseClass.run(self, **self.kwargs)
			463	BaseClass.run(self, **run_arg)
			464
			465	## Iterar sobre una serie de data que podrias aplicarse
			466
			467	for m_name in BaseClass.METHODS:
			468
			469	met_arg = {}
			470
			471	for arg in m_arg:
			472	if arg in BaseClass.METHODS[m_name]:
			473	for att in BaseClass.METHODS[m_name]:
			474	met_arg[att] = self.kwargs[att]
			475
			476	method = getattr(BaseClass, m_name)
			477	method(self, **met_arg)
			478	break
			479
			480	if self.dataOut.flagNoData:
			481	continue
			482
			483	keyList = list(self.operations2RunDict.keys())
			484	keyList.sort()
			485
			486	for key in keyList:
			487
			488	self.socketOP = self.sockOp()
			489	self.dataOut = self.execOp(self.socketOP, key, self.dataOut)
			490
			491
			492	self.publishProc(self.socket_p, self.dataOut)
			493
			494
			495	print("%s done" %BaseClass.__name__)
			496
			497	return 0
			498
			499	def runOp(self):
			500
			501	while True:
			502
			503	[self.dest ,self.buffer] = self.funIOrec(self.socket_router)
			504
			505	self.buffer = BaseClass.run(self, self.buffer, **self.kwargs)
			506
			507	self.funIOsen(self.socket_router, self.buffer, self.dest)
			508
			509	print("%s done" %BaseClass.__name__)
			510	return 0
			511
			512
			513	def run(self):
			514
			515	if self.typeProc is "ProcUnit":
			516
			517	self.socket_p = self.sockPublishing()
			518
			519	if 'Reader' not in self.dictProcs[self.id].name:
			520	self.socket_l = self.sockListening(self.inputId)
			521	self.runProc()
			522
			523	else:
			524
			525	self.runReader()
			526
			527	elif self.typeProc is "Operation":
			528
			529	self.socket_router = self.sockIO()
			530
			531	self.runOp()
	351		532
	352	"""	533	else:
	353	if not self.isConfig:	534	raise ValueError("Unknown type")
	354	self.setup(**kwargs)
	355
	356	raise NotImplementedError
	357
	358	def close(self):
	359		535
	360	pass No newline at end of file	536	return 0
			537
			538	return MPClass No newline at end of file

schainpy/model/proc/jroproc_spectra.py +23 -10

             import numpy
-            from .jroproc_base import ProcessingUnit, Operation
+            from schainpy.model.proc.jroproc_base import ProcessingUnit, MPDecorator, Operation
             from schainpy.model.data.jrodata import Spectra
             from schainpy.model.data.jrodata import hildebrand_sekhon
-            from schainpy.utils import log #yong
+            from schainpy.utils import log
+            @MPDecorator
             class SpectraProc(ProcessingUnit):
-                def __init__(self, **kwargs):
+                METHODS = {'selectHeights'        : ['minHei', 'maxHei'],
+                           'selectChannels'       : 'channelList',
+                           'selectChannelsByIndex': 'channelIndexList',
+                           'getBeaconSignal'      : ['tauindex', 'channelindex', 'hei_ref'],
+                           'selectHeightsByIndex' : ['minIndex', 'maxIndex']
+                           }
-                    ProcessingUnit.__init__(self, **kwargs)
+                def __init__(self):#, **kwargs):
+                    ProcessingUnit.__init__(self)#, **kwargs)
                     self.buffer = None
                     self.firstdatatime = None
                     self.dataOut = Spectra()
                     self.id_min = None
                     self.id_max = None
+                    self.setupReq = False #Agregar a todas las unidades de proc
                 def __updateSpecFromVoltage(self):
                             if self.dataOut.data_cspc is not None:
                                 #desplaza a la derecha en el eje 2 determinadas posiciones
                                 self.dataOut.data_cspc = numpy.roll(self.dataOut.data_cspc, shift, axis=1)
                         return True
                     if self.dataIn.type == "Voltage":
                     return 1
+            @MPDecorator
             class IncohInt(Operation):
                 __profIndex = 0
                 n = None
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
+                    Operation.__init__(self)#, **kwargs)
-                    Operation.__init__(self, **kwargs)
             #         self.isConfig = False
                 def setup(self, n=None, timeInterval=None, overlapping=False):
                 def run(self, dataOut, n=None, timeInterval=None, overlapping=False):
                     if n == 1:
                         return
                     dataOut.flagNoData = True
                     if not self.isConfig:
                         dataOut.nIncohInt *= self.n
                         dataOut.utctime = avgdatatime
-                        dataOut.flagNoData = False
  No newline at end of file
+                        dataOut.flagNoData = False
+                    return dataOut
  No newline at end of file

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