schain-jc Commit - r1175:7e36d2e90f1d

Merge branch 'v3.0-devel' of http://jro-dev.igp.gob.pe/rhodecode/schain into v3.0-devel

Juan C. Espinoza -

r1175:7e36d2e90f1d

parent child

Context file:

r1175:7e36d2e90f1d

Collapse all files

schainpy/controller.py +114 -123

             '''
+            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_voltage 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':
-                        if not plotter_queue:
-                            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':
                     className = eval(self.name)
                     kwargs = self.getKwargs()
-                        opObj = className(**kwargs)
+                    opObj = className(self.id, **kwargs)
-                    return opObj
+                    opObj.start()
+                    print('    Operation created')
+                    return opObj
             class ProcUnitConf():
                     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.datatype = datatype
                     self.inputId = inputId
                     self.parentId = parentId
                     self.opConfObjList = []
                     self.addOperation(name='run', optype='self')
                     return opObj
                 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()
-                        opObj = opConfObj.createObject(plotter_queue)
-                        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 = {}
                         is_ok = is_ok or sts
+                    """
                     return is_ok
                 def close(self):
                     for opConfObj in self.opConfObjList:
                 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')
             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
                 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:
                 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()):
+                    print('All processes were created')
-                        inputId = thisPUConfObj.getInputId()
-                        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):
                     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')
+                    xsub = ctx.socket(zmq.XSUB)
+                    xsub.bind('ipc:///tmp/socketTmp/a')
+                    xpub = ctx.socket(zmq.XPUB)
+                    xpub.bind('ipc:///tmp/socketTmp/b')
-                def getPlotterQueue(self):
+                    print("Controller Ready: Processes and proxy created")
+                    zmq.proxy(xsub, xpub)
-                    raise NotImplementedError('Use schainpy.controller_api.ControllerThread instead Project class')
-                def useExternalPlotter(self):
-                    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):
+                    # Iniciar todos los procesos .start(), monitoreo de procesos. ELiminar lo de abajo
-                        is_ok = False
-                        for procKey in keyList:
-                            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 +62 -17

             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
                               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
                               wr_period=wr_period,
                               thisDatetime=thisDatetime,
                               update_figfile=update_figfile)
+                    return dataOut  #Yong
  No newline at end of file

schainpy/model/graphics/jroplot_voltage.py +10 -3

             import os
             import datetime
             import numpy
+            from schainpy.model.proc.jroproc_base import ProcessingUnit, Operation, MPDecorator  #YONG
+            from schainpy.utils import log
             from .figure import Figure
+            @MPDecorator
             class Scope(Figure):
                 isConfig = None
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):          #YONG
-                    Figure.__init__(self, **kwargs)
+                    Figure.__init__(self)#, **kwargs)
                     self.isConfig = False
                     self.WIDTH = 300
                     self.HEIGHT = 200
                         ymin            :    None,
                         ymax            :    None,
                     """
+                    if dataOut.flagNoData:
+                        return dataOut
                     if channelList == None:
                         channelIndexList = dataOut.channelIndexList
                               ftp=ftp,
                               wr_period=wr_period,
                               thisDatetime=thisDatetime)
+                    return dataOut                
  No newline at end of file

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

             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
                         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 +4 -3

             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

schainpy/model/io/jroIO_voltage.py +5 -4

             import numpy
             from .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 Voltage
             import zmq
             from io import StringIO
             # from _sha import blocksize
+            @MPDecorator
             class VoltageReader(JRODataReader, ProcessingUnit):
                 """
                 Esta clase permite leer datos de voltage desde archivos en formato rawdata (.r). La lectura
                 optchar = "D"
                 dataOut = None
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
                     """
                     Inicializador de la clase VoltageReader para la lectura de datos de voltage.
                         None
                     """
-                    ProcessingUnit.__init__(self, **kwargs)
+                    ProcessingUnit.__init__(self)#, **kwargs)
                     self.isConfig = False
                     self.processingHeaderObj.processFlags = self.getProcessFlags()
                     self.setBasicHeader()
+  No newline at end of file

schainpy/model/proc/jroproc_base.py +364 -186

This diff has been collapsed as it changes many lines, (550 lines changed) Show them Hide them
	@@ -1,41 +1,39
	1	'''	1	'''
	2		2	Updated for multiprocessing
			3	Author : Sergio Cortez
			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:
	3	$Author: murco $	11	$Author: murco $
	4	$Id: jroproc_base.py 1 2012-11-12 18:56:07Z 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.
			33
			34	The constructor does not receive any argument. The remaining methods
			35	are related with the operations to execute.
	34		36
	35	Contiene el metodo "call" para llamar operaciones. Las operaciones pueden ser:
	36	- Metodos internos (callMethod)
	37	- Objetos del tipo Operation (callObject). Antes de ser llamados, estos objetos
	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__
	@@ -82,10 +72,12 class ProcessingUnit(object):
	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	"""
			78	This method is used in the controller, and update the dictionary containing the operations to execute. The dict
			79	posses the id of the operation process (IPC purposes)
			80
	89	Agrega un objeto del tipo "Operation" (opObj) a la lista de objetos "self.objectList" y retorna el	81	Agrega un objeto del tipo "Operation" (opObj) a la lista de objetos "self.objectList" y retorna el
	90	identificador asociado a este objeto.	82	identificador asociado a este objeto.
	91		83
	@@ -95,13 +87,14 class ProcessingUnit(object):
	95		87
	96	Return:	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()):
	@@ -122,239 +115,424 class ProcessingUnit(object):
	122		115
	123	raise NotImplementedError	116	raise NotImplementedError
	124		117
	125	def callMethod(self, name, opId):	118	def setup(self):
	126		119
	127	"""	120	raise NotImplementedError
	128	Ejecuta el metodo con el nombre "name" y con argumentos **kwargs de la propia clase.
	129		121
	130	Input:	122	def run(self):
	131	name : nombre del metodo a ejecutar	123
			124	raise NotImplementedError
			125
			126	def close(self):
			127	#Close every thread, queue or any other object here is it is neccesary.
			128	return
	132		129
	133	**kwargs : diccionario con los nombres y valores de la funcion a ejecutar.	130	class Operation(object):
	134		131
	135	~~"""~~	132	"""
			133	Update - Jan 2018 - MULTIPROCESSING
	136		134
	137	#Checking the inputs	135	Most of the methods remained the same. The decorator parse the arguments and executed the run() method for each process.
	138	if name == 'run':	136	The constructor doe snot receive any argument, neither the baseclass.
	139		137
	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		138
	148	#Getting the pointer to method	139	Clase base para definir las operaciones adicionales que se pueden agregar a la clase ProcessingUnit
	149	methodToCall = getattr(self, name)	140	y necesiten acumular informacion previa de los datos a procesar. De preferencia usar un buffer de
			141	acumulacion dentro de esta clase
	150		142
	151	#Executing the self method	143	Ejemplo: Integraciones coherentes, necesita la informacion previa de los n perfiles anteriores (bufffer)
	152		144
	153	if hasattr(self, 'mp'):	145	"""
	154	if name=='run':	146	id = None
	155	if self.mp is False:	147	__buffer = None
	156	self.mp = True	148	dest = None
	157	self.start()	149	isConfig = False
	158	else:	150	readyFlag = None
	159	self.operationKwargs[opId]['parent'] = self.kwargs	151
	160	methodToCall(**self.operationKwargs[opId])	152	def __init__(self):
	161	else:	153
	162	if name=='run':	154	self.buffer = None
	163	methodToCall(**self.kwargs)	155	self.dest = None
			156	self.isConfig = False
			157	self.readyFlag = False
			158
			159	if not hasattr(self, 'name'):
			160	self.name = self.__class__.__name__
			161
			162	def getAllowedArgs(self):
			163	if hasattr(self, '__attrs__'):
			164	return self.__attrs__
	164	else:	165	else:
	165	methodToCall(**self.operationKwargs[opId])	166	return inspect.getargspec(self.run).args
	166		167
	167	if self.dataOut is None:	168	def setup(self):
	168	return False
	169		169
	170	if self.dataOut.isEmpty():	170	self.isConfig = True
	171	return False	171
			172	raise NotImplementedError
	172		173
	173	return True
	174		174
	175	def callObject(self, objId):	175	def run(self, dataIn, **kwargs):
	176		176
	177	"""	177	"""
	178	Ejecuta la operacion asociada al identificador del objeto "objId"	178	Realiza las operaciones necesarias sobre la dataIn.data y actualiza los
			179	atributos del objeto dataIn.
	179		180
	180	Input:	181	Input:
	181		182
	182	objId : identificador del objeto a ejecutar	183	dataIn : objeto del tipo JROData
	183
	184	**kwargs : diccionario con los nombres y valores de la funcion a ejecutar.
	185		184
	186	Return:	185	Return:
	187		186
	188	None	187	None
			188
			189	Affected:
			190	__buffer : buffer de recepcion de datos.
			191
	189	"""	192	"""
			193	if not self.isConfig:
			194	self.setup(**kwargs)
	190		195
	191	if self.dataOut is not None and self.dataOut.isEmpty():	196	raise NotImplementedError
	192	return False
	193		197
	194	externalProcObj = self.operations2RunDict[objId]	198	def close(self):
	195		199
	196	if hasattr(externalProcObj, 'mp'):	200	pass
	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		201
	206		202
	207	return True	203	######### Decorator #########
	208		204
	209	def call(self, opType, opName=None, opId=None):
	210	"""
	211	Return True si ejecuta la operacion interna nombrada "opName" o la operacion externa
	212	identificada con el id "opId"; con los argumentos "**kwargs".
	213		205
	214	False si la operacion no se ha ejecutado.	206	def MPDecorator(BaseClass):
	215		207
	216	Input:	208	"""
			209	"Multiprocessing class decorator"
	217		210
	218	opType : Puede ser "self" o "external"	211	This function add multiprocessing features to the base class. Also,
			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	"""
	219		216
	220	Depende del tipo de operacion para llamar a:callMethod or callObject:	217	class MPClass(BaseClass, Process):
	221		218
	222	1. If opType = "self": Llama a un metodo propio de esta clase:	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)
	223		230
	224	name_method = getattr(self, name)
	225	name_method(**kwargs)
	226		231
			232	self.operationKwargs = {}
			233	self.args = args
	227		234
	228	2. If opType = "other" o"external": Llama al metodo "run()" de una instancia de la
	229	clase "Operation" o de un derivado de ella:
	230		235
	231	instanceName = self.operationList[opId]	236	self.operations2RunDict = {}
	232	instanceName.run(**kwargs)	237	self.kwargs = kwargs
	233		238
	234	opName : Si la operacion es interna (opType = 'self'), entonces el "opName" sera	239	# The number of arguments (args) determine the type of process
	235	usada para llamar a un metodo interno de la clase Processing
	236		240
	237	opId : Si la operacion es externa (opType = 'other' o 'external), entonces el	241	if len(self.args) is 3:
	238	"opId" sera usada para llamar al metodo "run" de la clase Operation	242	self.typeProc = "ProcUnit"
	239	registrada anteriormente con ese Id	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"
	240		249
	241	Exception:	250	def addOperationKwargs(self, objId, **kwargs):
	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		251
	246	"""	252	self.operationKwargs[objId] = kwargs
	247		253
	248	if opType == 'self':	254	def getAllowedArgs(self):
	249		255
	250	if not opName:	256	if hasattr(self, '__attrs__'):
	251	raise ValueError("opName parameter should be defined")	257	return self.__attrs__
			258	else:
			259	return inspect.getargspec(self.run).args
	252		260
	253	sts = self.callMethod(opName, opId)
	254		261
	255	elif opType == 'other' or opType == 'external' or opType == 'plotter':	262	def sockListening(self, topic):
	256		263
	257	if ~~not~~ ~~opId~~:	264	"""
	258	raise ValueError("opId parameter should be defined")	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	"""
	259		271
	260	if opId not in list(self.operations2RunDict.keys()):	272	cont = zmq.Context()
	261	raise ValueError("Any operation with id=%s has been added" %str(opId))	273	zmq_socket = cont.socket(zmq.SUB)
			274	if not os.path.exists('/tmp/socketTmp'):
			275	os.mkdir('/tmp/socketTmp')
	262		276
	263	sts = self.callObject(opId)	277	if 'Reader' in self.dictProcs[self.inputId].name:
			278	zmq_socket.connect('ipc:///tmp/socketTmp/b')
	264		279
	265	else:	280	else:
	266	raise ValueError("opType should be 'self', 'external' or 'plotter'; and not '%s'" %opType)	281	zmq_socket.connect('ipc:///tmp/socketTmp/%s' % self.inputId)
	267		282
	268	return sts	283	#log.error('RECEIVING FROM {} {}'.format(self.inputId, str(topic).encode()))
			284	zmq_socket.setsockopt(zmq.SUBSCRIBE, str(topic).encode()) #yong
	269		285
	270	def setInput(self, dataIn):	286	return zmq_socket
	271		287
	272	self.dataIn = dataIn
	273	self.dataInList.append(dataIn)
	274		288
	275	def getOutputObj(self):	289	def listenProc(self, sock):
	276		290
	277	return self.dataOut	291	"""
			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	"""
	278		296
	279	def checkInputs(self):	297	a = sock.recv_multipart()
			298	a = pickle.loads(a[1])
			299	return a
	280		300
	281	for thisDataIn in self.dataInList:	301	def sockPublishing(self):
			302
			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)
			316
			317	return zmq_socket
			318
			319	def publishProc(self, sock, data):
			320
			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	"""
	282		325
	283	if thisDataIn.isEmpty():	326	sock.send_multipart([str(self.id).encode(), pickle.dumps(data)]) #yong
	284	return False
	285		327
	286	return True	328	return True
	287		329
	288	def s~~etu~~p(self):	330	def sockOp(self):
	289		331
	290	raise NotImplementedError	332	"""
			333	This function create a socket for communication purposes with operation processes.
			334	"""
	291		335
	292	def run(self):	336	cont = zmq.Context()
			337	zmq_socket = cont.socket(zmq.DEALER)
	293		338
	294	raise NotImplementedError	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)
	295		343
	296	def close(self):
	297	#Close every thread, queue or any other object here is it is neccesary.
	298	return
	299		344
	300	class Operation(object):	345	return zmq_socket
	301		346
	302	"""
	303	Clase base para definir las operaciones adicionales que se pueden agregar a la clase ProcessingUnit
	304	y necesiten acumular informacion previa de los datos a procesar. De preferencia usar un buffer de
	305	acumulacion dentro de esta clase
	306		347
	307	Ejemplo: Integraciones coherentes, necesita la informacion previa de los n perfiles anteriores (bufffer)	348	def execOp(self, socket, opId, dataObj):
	308		349
	309	~~"""~~	350	"""
			351	This function 'execute' an operation main routine by establishing a
			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)
	310		356
	311	__buffer = None
	312	isConfig = False
	313		357
	314	def __init__(self, **kwargs):	358	socket.send(pickle.dumps(dataObj)) #yong
	315		359
	316	self.__buffer = None	360	argument = socket.recv_multipart()[0]
	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		361
	323	def getAllowedArgs(self):	362	argument = pickle.loads(argument)
	324	if hasattr(self, '__attrs__'):
	325	return self.__attrs__
	326	else:
	327	return inspect.getargspec(self.run).args
	328		363
	329	def setup(self):	364	return argument
	330		365
	331	self.isConfig = True	366	def sockIO(self):
	332		367
	333	raise NotImplementedError	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	"""
	334		372
	335	def run(self, dataIn, **kwargs):	373	cont = zmq.Context()
			374	if not os.path.exists('/tmp/socketTmp'): os.mkdir('/tmp/socketTmp')
			375	socket = cont.socket(zmq.ROUTER)
			376	socket.bind('ipc:///tmp/socketTmp/%s' % self.id)
			377
			378	return socket
			379
			380	def funIOrec(self, socket):
	336		381
	337	"""	382	"""
	338	Realiza las operaciones necesarias sobre la dataIn.data y actualiza los	383	Operation method, recover the id of the process who sent a python object.
	339	atributos del objeto dataIn.	384	The 'socket' argument is the socket binded to a specific process ipc.
			385	"""
	340		386
	341	Input:	387	#id_proc = socket.recv()
	342		388
	343	dataIn : objeto del tipo JROData	389	#dataObj = socket.recv_pyobj()
	344		390
	345	Return:	391	dataObj = socket.recv_multipart()
	346		392
	347	None	393	dataObj[1] = pickle.loads(dataObj[1])
			394	return dataObj[0], dataObj[1]
	348		395
	349	Affected:	396	def funIOsen(self, socket, data, dest):
	350	__buffer : buffer de recepcion de datos.
	351		397
	352	~~"""~~	398	"""
	353	if not self.isConfig:	399	Operation method, send a python object to a specific destination.
	354	self.setup(**kwargs)	400	The 'dest' argument is the id of a proccesinf unit.
			401	"""
	355		402
	356	raise NotImplementedError	403	socket.send_multipart([dest, pickle.dumps(data)]) #yong
	357		404
	358	def close(self):	405	return True
	359		406
	360	pass No newline at end of file	407
			408	def runReader(self):
			409
			410	# time.sleep(3)
			411	while True:
			412
			413	BaseClass.run(self, **self.kwargs)
			414
			415
			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)
			422
			423
			424	if self.flagNoMoreFiles: #Usar un objeto con flags para saber si termino el proc o hubo un error
			425	self.publishProc(self.socket_p, "Finish")
			426	break
			427
			428	if self.dataOut.flagNoData:
			429	continue
			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):
			440
			441	# All the procUnits with kwargs that require a setup initialization must be defined here.
			442
			443	if self.setupReq:
			444	BaseClass.setup(self, **self.kwargs)
			445
			446	while True:
			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()
			532
			533	else:
			534	raise ValueError("Unknown type")
			535
			536	return 0
			537
			538	return MPClass No newline at end of file

schainpy/model/proc/jroproc_spectra.py +20 -7

             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):
                     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):
                         dataOut.nIncohInt *= self.n
                         dataOut.utctime = avgdatatime
                         dataOut.flagNoData = False
+                    return dataOut
  No newline at end of file

schainpy/model/proc/jroproc_voltage.py +42 -28

             from scipy import interpolate
             #TODO
             #from schainpy import cSchain
-            from .jroproc_base import ProcessingUnit, Operation
+            from schainpy.model.proc.jroproc_base import ProcessingUnit, MPDecorator, Operation
             from schainpy.model.data.jrodata import Voltage
-            from time import time
             from schainpy.utils import log
+            from time import time
+            @MPDecorator
             class VoltageProc(ProcessingUnit):
+                METHODS = {} #yong
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):    #yong
-                    ProcessingUnit.__init__(self, **kwargs)
+                    ProcessingUnit.__init__(self)#, **kwargs)
                     #         self.objectDict = {}
                     self.dataOut = Voltage()
                     self.flip = 1
+                    self.setupReq = False  #yong
                 def run(self):
                     if self.dataIn.type == 'AMISR':
                         self.__updateObjFromAmisrInput()
                         self.dataOut.data[:,:,botLim:topLim+1]  = ynew
                 # import collections
+            @MPDecorator
             class CohInt(Operation):
                 isConfig = False
                 __dataToPutStride = False
                 n = None
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Operation.__init__(self, **kwargs)
+                    Operation.__init__(self)#, **kwargs)
                     #   self.isConfig = False
                     return avgdata, avgdatatime
                 def run(self, dataOut, n=None, timeInterval=None, stride=None, overlapping=False, byblock=False, **kwargs):
                     if not self.isConfig:
                         self.setup(n=n, stride=stride, timeInterval=timeInterval, overlapping=overlapping, byblock=byblock, **kwargs)
                         self.isConfig = True
                         # raise
                         #   dataOut.timeInterval = dataOut.ippSeconds * dataOut.nCohInt
                         dataOut.flagNoData = False
+                    return dataOut
+            @MPDecorator
             class Decoder(Operation):
                 isConfig = False
                 nCode = None
                 nBaud = None
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Operation.__init__(self, **kwargs)
+                    Operation.__init__(self)#, **kwargs)
                     self.times = None
                     self.osamp = None
                 #         self.__setValues = False
-                    self.isConfig = False
+                #    self.isConfig = False
+                    self.setupReq = False
                 def setup(self, code, osamp, dataOut):
                     self.__profIndex = 0
                     if self.__profIndex == self.nCode-1:
                         self.__profIndex = 0
-                        return 1
+                        return dataOut
                     self.__profIndex += 1
-                    return 1
+                    return dataOut
                 #        dataOut.flagDeflipData = True #asumo q la data no esta sin flip
+            @MPDecorator
             class ProfileConcat(Operation):
                 isConfig = False
                 buffer = None
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Operation.__init__(self, **kwargs)
+                    Operation.__init__(self)#, **kwargs)
                     self.profileIndex = 0
                 def reset(self):
                             xf = dataOut.heightList[0] + dataOut.nHeights * deltaHeight * m
                             dataOut.heightList = numpy.arange(dataOut.heightList[0], xf, deltaHeight)
                             dataOut.ippSeconds *= m
+                    return dataOut
+            @MPDecorator
             class ProfileSelector(Operation):
                 profileIndex = None
                 # Tamanho total de los perfiles
                 nProfiles = None
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Operation.__init__(self, **kwargs)
+                    Operation.__init__(self)#, **kwargs)
                     self.profileIndex = 0
                 def incProfileIndex(self):
                     raise ValueError("ProfileSelector needs profileList, profileRangeList or rangeList parameter")
-                    return False
+                    #return False
+                    return dataOut
+            @MPDecorator
             class Reshaper(Operation):
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Operation.__init__(self, **kwargs)
+                    Operation.__init__(self)#, **kwargs)
                     self.__buffer = None
                     self.__nitems = 0
                     dataOut.ippSeconds /= self.__nTxs
+                    return dataOut
+            @MPDecorator
             class SplitProfiles(Operation):
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Operation.__init__(self, **kwargs)
+                    Operation.__init__(self)#, **kwargs)
                 def run(self, dataOut, n):
                             raise ValueError("Could not split the data, n=%d has to be multiple of %d" %(n, shape[2]))
                         new_shape = shape[0], shape[1]*n, int(shape[2]/n)
                         dataOut.data = numpy.reshape(dataOut.data, new_shape)
                         dataOut.flagNoData = False
                     dataOut.ippSeconds /= n
+                    return dataOut
+            @MPDecorator
             class CombineProfiles(Operation):
-                def __init__(self, **kwargs):
+                def __init__(self):#, **kwargs):
-                    Operation.__init__(self, **kwargs)
+                    Operation.__init__(self)#, **kwargs)
                     self.__remData = None
                     self.__profileIndex = 0
                     dataOut.ippSeconds *= n
+                    return dataOut
             # import collections
             # from scipy.stats import mode
             #

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