schain-jc Commit - r1177:b013e28003fc

Review MP changes, three types of operations: self, other and external

Juan C. Espinoza -

r1177:b013e28003fc

parent child

Context file:

r1177:b013e28003fc

Collapse all files

schainpy/admin.py +5 -2

             import time
             import traceback
             import smtplib
-            import configparser
+            if sys.version[0] == '3':
+                from configparser import ConfigParser
+            else:
+                from ConfigParser import ConfigParser
             import io
             from threading import Thread
             from multiprocessing import Process
                         return
                     # create Parser using standard module ConfigParser
-                    self.__parser = configparser.ConfigParser()
+                    self.__parser = ConfigParser()
                     # read conf file into a StringIO with "[madrigal]\n" section heading prepended
                     strConfFile = io.StringIO("[schain]\n" + self.__confFile.read())

schainpy/controller.py +74 -148

             import time
             import zmq
             from multiprocessing import Process, cpu_count
+            from threading import Thread
             from xml.etree.ElementTree import ElementTree, Element, SubElement, tostring
             from xml.dom import minidom
                     time.sleep(3)
+            def wait(context):
+                time.sleep(1)
+                c = zmq.Context()
+                receiver = c.socket(zmq.SUB)
+                receiver.connect('ipc:///tmp/schain_{}_pub'.format(self.id))
+                receiver.setsockopt(zmq.SUBSCRIBE, self.id.encode())
+                log.error('startinggg')
+                msg = receiver.recv_multipart()[1]
+                #log.error(msg)
+                context.terminate()
             class ParameterConf():
                 id = None
             class OperationConf():
-                id = None
-                name = None
-                priority = None
-                type = None
-                parmConfObjList = []
                 ELEMENTNAME = 'Operation'
                 def __init__(self):
                     return kwargs
-                def setup(self, id, name, priority, type):
+                def setup(self, id, name, priority, type, project_id):
                     self.id = str(id)
+                    self.project_id = project_id
                     self.name = name
                     self.type = type
                     self.priority = priority
                 def createObject(self):
                     className = eval(self.name)
-                    kwargs = self.getKwargs()
+                    if self.type == 'other':
-                    opObj = className(self.id, **kwargs)
+                        opObj = className()
+                    elif self.type == 'external':
-                    opObj.start()
+                        kwargs = self.getKwargs()
+                        opObj = className(self.id, self.project_id, **kwargs)
-                    print('    Operation created')
+                        opObj.start()
                     return opObj
             class ProcUnitConf():
-                id = None
-                name = None
-                datatype = None
-                inputId = None
-                parentId = None
-                opConfObjList = []
-                procUnitObj = None
-                opObjList = []
                 ELEMENTNAME = 'ProcUnit'
                 def __init__(self):
                     self.datatype = None
                     self.name = None
                     self.inputId = None
                     self.opConfObjList = []
                     self.procUnitObj = None
                     self.opObjDict = {}
                     return self.id
-                def updateId(self, new_id, parentId=parentId):
+                def updateId(self, new_id):
                     '''
                     new_id = int(parentId) * 10 + (int(self.id) % 10)
                     new_inputId = int(parentId) * 10 + (int(self.inputId) % 10)
                     #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):
+                def setup(self, project_id, id, name, datatype, inputId):
                     '''
                         id sera el topico a publicar
                         inputId sera el topico a subscribirse
                         datatype = name.replace('Proc', '')
                     self.id = str(id)
+                    self.project_id = project_id
                     self.name = name
                     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'):
+                def addOperation(self, name, optype='self'):
                     '''
                         Actualizacion - > proceso comunicacion
                         En el caso de optype='self', elminar. DEfinir comuncacion IPC -> Topic
                     id = self.__getNewId()
                     priority = self.__getPriority() # Sin mucho sentido, pero puede usarse
                     opConfObj = OperationConf()
-                    opConfObj.setup(id, name=name, priority=priority, type=optype)
+                    opConfObj.setup(id, name=name, priority=priority, type=optype, project_id=self.project_id)
                     self.opConfObjList.append(opConfObj)
                     return opConfObj
                     return kwargs
-                def createObjects(self, dictUnits):
+                def createObjects(self):
                     '''
                     Instancia de unidades de procesamiento.
                     '''
                     className = eval(self.name)
                     kwargs = self.getKwargs()
-                    procUnitObj = className(self.id, self.inputId, dictUnits, **kwargs) # necesitan saber su id y su entrada por fines de ipc
+                    procUnitObj = className(self.id, self.inputId, self.project_id, **kwargs) # necesitan saber su id y su entrada por fines de ipc
+                    log.success('creating process...', self.name)
                     for opConfObj in self.opConfObjList:
-                        if opConfObj.type == 'self' and self.name == 'run':
+                        if opConfObj.type == 'self' and opConfObj.name == 'run':
                             continue
                         elif opConfObj.type == 'self':
-                            procUnitObj.addOperationKwargs(
+                            opObj = getattr(procUnitObj, opConfObj.name)
-                                opConfObj.id, **opConfObj.getKwargs())
+                        else:
-                            continue
+                            opObj = opConfObj.createObject()
-                        print("Creating operation process:", opConfObj.name, "for", self.name)
-                        opObj = opConfObj.createObject()
-                        #self.opObjDict[opConfObj.id] = opObj.name
+                        log.success('creating operation: {}, type:{}'.format(
+                            opConfObj.name,
+                            opConfObj.type), self.name)
-                        procUnitObj.addOperation(opConfObj.name, opConfObj.id)
+                        procUnitObj.addOperation(opConfObj, opObj)
                     procUnitObj.start()
                     self.procUnitObj = procUnitObj
-                    return procUnitObj
-                def run(self):
-                    is_ok = True
-                    """
-                    for opConfObj in self.opConfObjList:
-                        kwargs = {}
-                        for parmConfObj in opConfObj.getParameterObjList():
-                            if opConfObj.name == 'run' and parmConfObj.name == 'datatype':
-                                continue
-                            kwargs[parmConfObj.name] = parmConfObj.getValue()
-                        sts = self.procUnitObj.call(opType=opConfObj.type,
-                                                    opName=opConfObj.name,
-                                                    opId=opConfObj.id)
-                        is_ok = is_ok or sts
-                    """
-                    return is_ok
                 def close(self):
                     for opConfObj in self.opConfObjList:
             class ReadUnitConf(ProcUnitConf):
-                path = None
-                startDate = None
-                endDate = None
-                startTime = None
-                endTime = None
                 ELEMENTNAME = 'ReadUnit'
                 def __init__(self):
                     self.datatype = None
                     self.name = None
                     self.inputId = None
-                    self.parentId = None
                     self.opConfObjList = []
-                    self.opObjList = []
                 def getElementName(self):
                     return self.ELEMENTNAME
-                def setup(self, id, name, datatype, path='', startDate='', endDate='',
+                def setup(self, project_id, id, name, datatype, path='', startDate='', endDate='',
-                          startTime='', endTime='', parentId=None, server=None, **kwargs):
+                          startTime='', endTime='', server=None, **kwargs):
                     '''
                             name = '{}Reader'.format(name)
                     self.id = id
+                    self.project_id = project_id
                     self.name = name
                     self.datatype = datatype
                     if path != '':
                     self.endDate = endDate
                     self.startTime = startTime
                     self.endTime = endTime
-                    self.inputId = '0'
-                    self.parentId = parentId
                     self.server = server
                     self.addRunOperation(**kwargs)
                         self.datatype = self.name.replace('Reader', '')
                     attrs = ('path', 'startDate', 'endDate',
-                             'startTime', 'endTime', 'parentId')
+                             'startTime', 'endTime')
                     for attr in attrs:
                         if attr in kwargs:
                             setattr(self, attr, kwargs.pop(attr))
-                    self.inputId = '0'
                     self.updateRunOperation(**kwargs)
                 def removeOperations(self):
                     self.id = upElement.get('id')
                     self.name = upElement.get('name')
                     self.datatype = upElement.get('datatype')
-                    self.inputId = upElement.get('inputId')
                     if self.ELEMENTNAME == 'ReadUnit':
                         self.datatype = self.datatype.replace('Reader', '')
-                    if self.inputId == 'None':
-                        self.inputId = '0'
                     self.opConfObjList = []
                     opElementList = upElement.iter(OperationConf().getElementName())
             class Project(Process):
-                id = None
-                description = None
-                filename = None
-                procUnitConfObjDict = None
                 ELEMENTNAME = 'Project'
                 def __init__(self):
                     Process.__init__(self)
                     self.id = None
+                    self.filename = None
                     self.description = None
                     self.email = None
                     self.alarm = None
                 def __getNewId(self):
                     idList = list(self.procUnitConfObjDict.keys())
                     id = int(self.id) * 10
                     while True:
                         procUnitConfObj = self.procUnitConfObjDict[procKey]
                         idProcUnit = str(int(self.id) * 10 + n)
-                        procUnitConfObj.updateId(idProcUnit, parentId=self.id)
+                        procUnitConfObj.updateId(idProcUnit)
                         newProcUnitConfObjDict[idProcUnit] = procUnitConfObj
                         n += 1
                     self.procUnitConfObjDict = newProcUnitConfObjDict
-                def setup(self, id, name='', description='', email=None, alarm=[]):
+                def setup(self, id=1, name='', description='', email=None, alarm=[]):
                     print(' ')
                     print('*' * 60)
                         idReadUnit = str(id)
                     readUnitConfObj = ReadUnitConf()
-                    readUnitConfObj.setup(idReadUnit, name, datatype,
+                    readUnitConfObj.setup(self.id, idReadUnit, name, datatype, **kwargs)
-                                          parentId=self.id, **kwargs)
                     self.procUnitConfObjDict[readUnitConfObj.getId()] = readUnitConfObj
                     return readUnitConfObj
                     '''
                     idProcUnit = self.__getNewId() #Topico para subscripcion
                     procUnitConfObj = ProcUnitConf()
-                    procUnitConfObj.setup(idProcUnit, name, datatype, inputId, #topic_read, topic_write,
+                    procUnitConfObj.setup(self.id, idProcUnit, name, datatype, inputId) #topic_read, topic_write,
-                                          parentId=self.id)
                     self.procUnitConfObjDict[procUnitConfObj.getId()] = procUnitConfObj
                     return procUnitConfObj
                     for readUnitElement in readUnitElementList:
                         readUnitConfObj = ReadUnitConf()
                         readUnitConfObj.readXml(readUnitElement)
-                        if readUnitConfObj.parentId == None:
-                            readUnitConfObj.parentId = self.id
                         self.procUnitConfObjDict[readUnitConfObj.getId()] = readUnitConfObj
                     procUnitElementList = self.projectElement.iter(
                     for procUnitElement in procUnitElementList:
                         procUnitConfObj = ProcUnitConf()
                         procUnitConfObj.readXml(procUnitElement)
-                        if procUnitConfObj.parentId == None:
-                            procUnitConfObj.parentId = self.id
                         self.procUnitConfObjDict[procUnitConfObj.getId()] = procUnitConfObj
                     self.filename = abs_file
                     return 1
-                def printattr(self):
+                def __str__(self):
                     print('Project[%s]: name = %s, description = %s' % (self.id,
                                                                         self.name,
                                                                         self.description))
-                    for procUnitConfObj in list(self.procUnitConfObjDict.values()):
+                    for procUnitConfObj in self.procUnitConfObjDict.values():
-                        procUnitConfObj.printattr()
+                        print(procUnitConfObj)
                 def createObjects(self):
-                    for procUnitConfObj in list(self.procUnitConfObjDict.values()):
+                    for procUnitConfObj in self.procUnitConfObjDict.values():
-                        print("Creating process:", procUnitConfObj.name)
+                        procUnitConfObj.createObjects()
-                        procUnitConfObj.createObjects(self.procUnitConfObjDict)
-                    print('All processes were created')
                 def __handleError(self, procUnitConfObj, modes=None, stdout=True):
                     self.filename = filename
                 def setProxyCom(self):
+                    if not os.path.exists('/tmp/schain'):
+                        os.mkdir('/tmp/schain')
-                    ctx = zmq.Context()
+                    self.ctx = zmq.Context()
-                    if not os.path.exists('/tmp/socketTmp'): os.mkdir('/tmp/socketTmp')
+                    xpub = self.ctx.socket(zmq.XPUB)
-                    xsub = ctx.socket(zmq.XSUB)
+                    xpub.bind('ipc:///tmp/schain/{}_pub'.format(self.id))
-                    xsub.bind('ipc:///tmp/socketTmp/a')
+                    xsub = self.ctx.socket(zmq.XSUB)
-                    xpub = ctx.socket(zmq.XPUB)
+                    xsub.bind('ipc:///tmp/schain/{}_sub'.format(self.id))
-                    xpub.bind('ipc:///tmp/socketTmp/b')
-                    print("Controller Ready: Processes and proxy created")
+                    try:
-                    zmq.proxy(xsub, xpub)
+                        zmq.proxy(xpub, xsub)
+                    except zmq.ContextTerminated:
+                        xpub.close()
+                        xsub.close()
                 def run(self):
-                    log.success('Starting {}'.format(self.name), tag='')
+                    log.success('Starting {}: {}'.format(self.name, self.id), tag='')
                     self.start_time = time.time()
                     self.createObjects()
+                    # t = Thread(target=wait, args=(self.ctx, ))
+                    # t.start()
                     self.setProxyCom()
                     # Iniciar todos los procesos .start(), monitoreo de procesos. ELiminar lo de abajo
-                    # Closing every process
                     log.success('{} finished (time: {}s)'.format(
                         self.name,
-                        time.time()-self.start_time))
  No newline at end of file
+                        time.time()-self.start_time))

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

                         newFile = self.__setNextFileOffline()
                     if not(newFile):
-                        raise(schainpy.admin.SchainWarning('No more files to read'))
+                        self.dataOut.error = (-1, '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.admin.SchainWarning('No more files'))
+                            self.dataOut.error = (-1, 'No more files to read')
                             return 0
                         if not(self.readBlock()):
                                 sleep(self.delay)
                             if not(fullpath):
-                                raise(schainpy.admin.SchainWarning('There isn\'t any valid file in {}'.format(path)))
+                                self.dataOut.error = (-1, 'There isn\'t any valid file in {}'.format(path))
                                 return
                             self.year = year

schainpy/model/proc/jroproc_base.py +115 -286

@@ -18,7 +18,6 import time
18	import pickle	18	import pickle
19	import os	19	import os
20	from multiprocessing import Process	20	from multiprocessing import Process
21
22	from schainpy.utils import log	21	from schainpy.utils import log
23		22
24		23
@@ -36,43 +35,30 class ProcessingUnit(object):
36		35
37		36
38	"""	37	"""
39	# objeto de datos de entrada (Voltage, Spectra o Correlation)	38
		39	METHODS = {}
40	dataIn = None	40	dataIn = None
41	dataInList = []	41	dataInList = []
42
43	# objeto de datos de entrada (Voltage, Spectra o Correlation)
44
45	id = None	42	id = None
46	inputId = None	43	inputId = None
47
48	dataOut = None	44	dataOut = None
49
50	dictProcs = None	45	dictProcs = None
51
52	operations2RunDict = None
53
54	isConfig = False	46	isConfig = False
55		47
56	def __init__(self):	48	def __init__(self):
57		49
58	self.dataIn = None	50	self.dataIn = None
59	self.dataOut = None	51	self.dataOut = None
60
61	self.isConfig = False	52	self.isConfig = False
		53	self.operations = []
62		54
63	def getAllowedArgs(self):	55	def getAllowedArgs(self):
64	if hasattr(self, '__attrs__'):	56	if hasattr(self, '__attrs__'):
65	return self.__attrs__	57	return self.__attrs__
66	else:	58	else:
67	return inspect.getargspec(self.run).args	59	return inspect.getargspec(self.run).args
68
69	def addOperationKwargs(self, objId, **kwargs):
70	'''
71	'''
72
73	self.operationKwargs[objId] = kwargs
74		60
75	def addOperation(self, ~~opObj~~, o~~bjId~~):	61	def addOperation(self, conf, operation):
76		62
77	"""	63	"""
78	This method is used in the controller, and update the dictionary containing the operations to execute. The dict	64	This method is used in the controller, and update the dictionary containing the operations to execute. The dict
@@ -90,17 +76,14 class ProcessingUnit(object):
90	objId : identificador del objeto, necesario para comunicar con master(procUnit)	76	objId : identificador del objeto, necesario para comunicar con master(procUnit)
91	"""	77	"""
92		78
93	self.operations2RunDict[objId] = opObj	79	self.operations.append((operation, conf.type, conf.id, conf.getKwargs()))
94
95	return objId
96
97		80
98	def getOperationObj(self, objId):	81	def getOperationObj(self, objId):
99		82
100	if objId not in list(self.operations~~2RunDict~~.keys()):	83	if objId not in list(self.operations.keys()):
101	return None	84	return None
102		85
103	return self.operations~~2RunDict~~[objId]	86	return self.operations[objId]
104		87
105	def operation(self, **kwargs):	88	def operation(self, **kwargs):
106		89
@@ -200,339 +183,185 class Operation(object):
200	pass	183	pass
201		184
202		185
203	######### Decorator #########
204
205
206	def MPDecorator(BaseClass):	186	def MPDecorator(BaseClass):
207		187
208	"""	188	"""
209	"Multiprocessing class decorator"	189	Multiprocessing class decorator
210		190
211	This function add multiprocessing features to ~~the base c~~lass. Also,	191	This function add multiprocessing features to a BaseClass. Also, it handle
212	~~it handle~~ the communication beetween processes (readers, procUnits and operations).	192	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	"""	193	"""
216		194
217	class MPClass(BaseClass, Process):	195	class MPClass(BaseClass, Process):
218		196
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):	197	def __init__(self, args, *kwargs):
228	super(MPClass, self).__init__()	198	super(MPClass, self).__init__()
229	Process.__init__(self)	199	Process.__init__(self)
230
231
232	self.operationKwargs = {}	200	self.operationKwargs = {}
233	self.args = args	201	self.args = args
234
235
236	self.operations2RunDict = {}
237	self.kwargs = kwargs	202	self.kwargs = kwargs
238		203	self.sender = None
239	# The number of arguments (args) determine the type of process	204	self.receiver = None
		205	self.name = BaseClass.__name__
240		206
241	if len(self.args) is 3:	207	if len(self.args) is 3:
242	self.typeProc = "ProcUnit"	208	self.typeProc = "ProcUnit"
243	self.id = args[0] ~~#topico de publicacion~~	209	self.id = args[0]
244	self.inputId = args[1] ~~#topico de subcripcion~~	210	self.inputId = args[1]
245	self.~~dictProcs~~ = args[2] ~~#diccionario de procesos globales~~	211	self.project_id = args[2]
246	else:	212	else:
247	self.id = args[0]	213	self.id = args[0]
		214	self.inputId = args[0]
		215	self.project_id = args[1]
248	self.typeProc = "Operation"	216	self.typeProc = "Operation"
249
250	def addOperationKwargs(self, objId, **kwargs):
251
252	self.operationKwargs[objId] = kwargs
253		217
254	def getAllowedArgs(self):	218	def getAllowedArgs(self):
255		219
256	if hasattr(self, '__attrs__'):	220	if hasattr(self, '__attrs__'):
257	return self.__attrs__	221	return self.__attrs__
258	else:	222	else:
259	return inspect.getargspec(~~self~~.run).args	223	return inspect.getargspec(BaseClass.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
285
286	return zmq_socket
287
288
289	def listenProc(self, sock):
290
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	"""
296
297	a = sock.recv_multipart()
298	a = pickle.loads(a[1])
299	return a
300
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	"""
325
326	sock.send_multipart([str(self.id).encode(), pickle.dumps(data)]) #yong
327
328	return True
329
330	def sockOp(self):
331
332	"""
333	This function create a socket for communication purposes with operation processes.
334	"""
335
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)
343
344
345	return zmq_socket
346
347
348	def execOp(self, socket, opId, dataObj):
349
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)
356
357
358	socket.send(pickle.dumps(dataObj)) #yong
359		224
360	argument = socket.recv_multipart()[0]	225	def subscribe(self):
361		226	'''
362	argument = pickle.loads(argument)	227	This function create a socket to receive objects from the
363		228	topic `inputId`.
364	return argument	229	'''
365
366	def sockIO(self):
367
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	"""
372
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):
381
382	"""
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	"""
386
387	#id_proc = socket.recv()
388
389	#dataObj = socket.recv_pyobj()
390		230
391	dataObj = socket.recv_multipart()	231	c = zmq.Context()
		232	self.receiver = c.socket(zmq.SUB)
		233	self.receiver.connect('ipc:///tmp/schain/{}_pub'.format(self.project_id))
		234	self.receiver.setsockopt(zmq.SUBSCRIBE, self.inputId.encode())
		235
		236	def listen(self):
		237	'''
		238	This function waits for objects and deserialize using pickle
		239	'''
392		240
393	data~~Obj~~[1] = pickle.loads(~~dataObj~~[1])	241	data = pickle.loads(self.receiver.recv_multipart()[1])
394	return data~~Obj~~[0], ~~dataObj~~[1]	242	return data
395		243
396	def funIOsen(self, socket, data, dest):	244	def set_publisher(self):
		245	'''
		246	This function create a socket for publishing purposes.
		247	'''
397		248
398	~~"""~~	249	time.sleep(1)
399	Operation method, send a python object to a specific destination.	250	c = zmq.Context()
400	The 'dest' argument is the id of a proccesinf unit.	251	self.sender = c.socket(zmq.PUB)
401	"""	252	self.sender.connect('ipc:///tmp/schain/{}_sub'.format(self.project_id))
402
403	socket.send_multipart([dest, pickle.dumps(data)]) #yong
404		253
405	return True	254	def publish(self, data, id):
		255	'''
		256	This function publish an object, to a specific topic.
		257	'''
406		258
		259	self.sender.send_multipart([str(id).encode(), pickle.dumps(data)])
407		260
408	def runReader(self):	261	def runReader(self):
409		262	'''
410	# time.sleep(3)	263	Run fuction for read units
		264	'''
411	while True:	265	while True:
412		266
413	BaseClass.run(self, **self.kwargs)	267	BaseClass.run(self, **self.kwargs)
414		268
415		269	if self.dataOut.error[0] == -1:
416	keyList = list(self.operations2RunDict.keys())	270	log.error(self.dataOut.error[1])
417	keyList.sort()	271	self.publish('end', self.id)
418		272	#self.sender.send_multipart([str(self.project_id).encode(), 'end'.encode()])
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	273	break
427		274
		275	for op, optype, id, kwargs in self.operations:
		276	if optype=='self':
		277	op(**kwargs)
		278	elif optype=='other':
		279	self.dataOut = op.run(self.dataOut, **self.kwargs)
		280	elif optype=='external':
		281	self.publish(self.dataOut, opId)
		282
428	if self.dataOut.flagNoData:	283	if self.dataOut.flagNoData:
429	continue	284	continue
430		285
431	#print("Publishing data...")	286	self.publish(self.dataOut, self.id)
432	self.publishProc(self.socket_p, self.dataOut)	287
433	# time.sleep(2)
434
435
436	print("%s done" %BaseClass.__name__)
437	return 0
438
439	def runProc(self):	288	def runProc(self):
440		289	'''
441	# All the procUnits with kwargs that require a setup initialization must be defined here.	290	Run function for proccessing units
442		291	'''
443	if self.setupReq:
444	BaseClass.setup(self, **self.kwargs)
445		292
446	while True:	293	while True:
447	self.dataIn = self.listen~~Proc~~(~~self~~.~~socket_l~~)	294	self.dataIn = self.listen()
448	#print("%s received data" %BaseClass.__name__)
449
450	if self.dataIn == "Finish":
451	break
452		295
453	m_arg = list(self.kwargs.keys())	296	if self.dataIn == 'end':
454	num_arg = list(range(1,int(BaseClass.run.__code__.co_argcount)))	297	self.publish('end', self.id)
455		298	for op, optype, opId, kwargs in self.operations:
456	run_arg = {}	299	if optype == 'external':
457		300	self.publish('end', opId)
458	~~for~~ ~~var~~ in ~~num_arg~~:	301	break
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		302
469	met_arg = {}	303	if self.dataIn.flagNoData:
		304	continue
470		305
471	for arg in m_arg:	306	BaseClass.run(self, **self.kwargs)
472	if arg in BaseClass.METHODS[m_name]:
473	for att in BaseClass.METHODS[m_name]:
474	met_arg[att] = self.kwargs[att]
475		307
476	method = getattr(BaseClass, m_name)	308	for op, optype, opId, kwargs in self.operations:
477	method(self, **met_arg)	309	if optype=='self':
478	~~break~~	310	op(**kwargs)
		311	elif optype=='other':
		312	self.dataOut = op.run(self.dataOut, **kwargs)
		313	elif optype=='external':
		314	self.publish(self.dataOut, opId)
479		315
480	if self.dataOut.flagNoData:	316	if self.dataOut.flagNoData:
481	continue	317	continue
482		318
483	keyList = list(self.operations2RunDict.keys())	319	self.publish(self.dataOut, self.id)
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		320
499	def runOp(self):	321	def runOp(self):
		322	'''
		323	Run function for operations
		324	'''
500		325
501	while True:	326	while True:
502		327
503	[self.dest ,self.buffer] = self.funIOrec(self.socket_router)	328	dataOut = self.listen()
504
505	self.buffer = BaseClass.run(self, self.buffer, **self.kwargs)
506		329
507	self.funIOsen(self.socket_router, self.buffer, self.dest)	330	if dataOut == 'end':
508		331	break
509	print("%s done" %BaseClass.__name__)	332
510	return 0	333	BaseClass.run(self, dataOut, **self.kwargs)
511		334
512
513	def run(self):	335	def run(self):
514
515	if self.typeProc is "ProcUnit":
516
517	self.socket_p = self.sockPublishing()
518		336
519	if 'Reader' not in self.dictProcs[self.id].name:	337	if self.typeProc is "ProcUnit":
520	self.socket_l = self.sockListening(self.inputId)	338
521	self.runProc()	339	if self.inputId is not None:
		340	self.subscribe()
		341	self.set_publisher()
522		342
		343	if 'Reader' not in BaseClass.__name__:
		344	self.runProc()
523	else:	345	else:
524
525	self.runReader()	346	self.runReader()
526		347
527	elif self.typeProc is "Operation":	348	elif self.typeProc is "Operation":
528		349
529	self.s~~ocket_router~~ = ~~self~~.~~sockIO~~()	350	self.subscribe()
530
531	self.runOp()	351	self.runOp()
532		352
533	else:	353	else:
534	raise ValueError("Unknown type")	354	raise ValueError("Unknown type")
535		355
536	return 0	356	print("%s done" % BaseClass.__name__)
537		357	self.close()
		358
		359	def close(self):
		360
		361	if self.sender:
		362	self.sender.close()
		363
		364	if self.receiver:
		365	self.receiver.close()
		366
538	return MPClass No newline at end of file	367	return MPClass

schainpy/model/proc/jroproc_spectra.py +2 -4

                 def run(self, nProfiles=None, nFFTPoints=None, pairsList=[], ippFactor=None, shift_fft=False):
-                    self.dataOut.flagNoData = True
                     if self.dataIn.type == "Spectra":
                         self.dataOut.copy(self.dataIn)
                         # if not pairsList:
                     return 1
-            @MPDecorator
             class IncohInt(Operation):
                 __profIndex = 0
                         dataOut.nIncohInt *= self.n
                         dataOut.utctime = avgdatatime
                         dataOut.flagNoData = False
                     return dataOut

schainpy/model/proc/jroproc_voltage.py +11 -14

             import sys
             import numpy
             from scipy import interpolate
-            from schainpy.model.proc.jroproc_base import ProcessingUnit, MPDecorator, Operation
+            from schainpy.model.proc.jroproc_base import ProcessingUnit,, Operation
             from schainpy.model.data.jrodata import Voltage
             from schainpy.utils import log
             from time import time
             @MPDecorator
             class VoltageProc(ProcessingUnit):
-                METHODS = {} #yong
+                def __init__(self):
-                def __init__(self):#, **kwargs):    #yong
+                    ProcessingUnit.__init__(self)
-                    ProcessingUnit.__init__(self)#, **kwargs)
-                    #         self.objectDict = {}
                     self.dataOut = Voltage()
                     self.flip = 1
-                    self.setupReq = False  #yong
+                    self.setupReq = False
                 def run(self):
                         self.dataOut.data[:,:,botLim:topLim+1]  = ynew
                 # import collections
-            @MPDecorator
             class CohInt(Operation):
                 isConfig = False
                         #   dataOut.timeInterval = dataOut.ippSeconds * dataOut.nCohInt
                         dataOut.flagNoData = False
                     return dataOut
-            @MPDecorator
             class Decoder(Operation):
                 isConfig = False
                     return dataOut
                 #        dataOut.flagDeflipData = True #asumo q la data no esta sin flip
-            @MPDecorator
             class ProfileConcat(Operation):
                 isConfig = False
                             dataOut.heightList = numpy.arange(dataOut.heightList[0], xf, deltaHeight)
                             dataOut.ippSeconds *= m
                     return dataOut
-            @MPDecorator
             class ProfileSelector(Operation):
                 profileIndex = None
                     #return False
                     return dataOut
-            @MPDecorator
             class Reshaper(Operation):
                 def __init__(self):#, **kwargs):
                     dataOut.ippSeconds /= self.__nTxs
                     return dataOut
-            @MPDecorator
             class SplitProfiles(Operation):
                 def __init__(self):#, **kwargs):
                     dataOut.ippSeconds /= n
                     return dataOut
-            @MPDecorator
             class CombineProfiles(Operation):
                 def __init__(self):#, **kwargs):

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