jroproc_base.py
390 lines
| 11.5 KiB
| text/x-python
|
PythonLexer
Daniel Valdez
|
r487 | ''' | ||
George Yong
|
r1171 | Updated for multiprocessing | ||
| Author : Sergio Cortez | ||||
| Jan 2018 | ||||
| Abstract: | ||||
| Base class for processing units and operations. A decorator provides multiprocessing features and interconnect the processes created. | ||||
| The argument (kwargs) sent from the controller is parsed and filtered via the decorator for each processing unit or operation instantiated. | ||||
| The decorator handle also the methods inside the processing unit to be called from the main script (not as operations) (OPERATION -> type ='self'). | ||||
| Based on: | ||||
| $Author: murco $ | ||||
| $Id: jroproc_base.py 1 2012-11-12 18:56:07Z murco $ | ||||
|
Daniel Valdez
|
r487 | ''' | ||
Juan C. Espinoza
|
r1187 | |||
José Chávez
|
r929 | import inspect | ||
|
George Yong
|
r1171 | import zmq | ||
| import time | ||||
| import pickle | ||||
| import os | ||||
| from multiprocessing import Process | ||||
|
Juan C. Espinoza
|
r1187 | from zmq.utils.monitor import recv_monitor_message | ||
|
George Yong
|
r1171 | from schainpy.utils import log | ||
|
José Chávez
|
r929 | |||
|
Daniel Valdez
|
r487 | |||
Miguel Valdez
|
r568 | class ProcessingUnit(object): | ||
|
Juan C. Valdez
|
r860 | |||
|
Daniel Valdez
|
r487 | """ | ||
|
George Yong
|
r1171 | Update - Jan 2018 - MULTIPROCESSING | ||
| All the "call" methods present in the previous base were removed. | ||||
| The majority of operations are independant processes, thus | ||||
| the decorator is in charge of communicate the operation processes | ||||
| with the proccessing unit via IPC. | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | The constructor does not receive any argument. The remaining methods | ||
| are related with the operations to execute. | ||||
|
Juan C. Valdez
|
r860 | |||
|
Daniel Valdez
|
r487 | """ | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def __init__(self): | ||
|
Juan C. Valdez
|
r860 | |||
|
Daniel Valdez
|
r487 | self.dataIn = None | ||
|
Miguel Valdez
|
r573 | self.dataOut = None | ||
|
Daniel Valdez
|
r487 | self.isConfig = False | ||
|
Juan C. Espinoza
|
r1177 | self.operations = [] | ||
|
Juan C. Espinoza
|
r1187 | self.plots = [] | ||
|
Juan C. Valdez
|
r860 | |||
|
José Chávez
|
r929 | def getAllowedArgs(self): | ||
|
Juan C. Espinoza
|
r1097 | if hasattr(self, '__attrs__'): | ||
| return self.__attrs__ | ||||
| else: | ||||
| return inspect.getargspec(self.run).args | ||||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1187 | def addOperation(self, conf, operation): | ||
|
Daniel Valdez
|
r487 | """ | ||
|
George Yong
|
r1171 | This method is used in the controller, and update the dictionary containing the operations to execute. The dict | ||
| posses the id of the operation process (IPC purposes) | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | Agrega un objeto del tipo "Operation" (opObj) a la lista de objetos "self.objectList" y retorna el | ||
| identificador asociado a este objeto. | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | Input: | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | object : objeto de la clase "Operation" | ||
| Return: | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | objId : identificador del objeto, necesario para comunicar con master(procUnit) | ||
|
Daniel Valdez
|
r487 | """ | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1187 | self.operations.append( | ||
| (operation, conf.type, conf.id, conf.getKwargs())) | ||||
| if 'plot' in self.name.lower(): | ||||
| self.plots.append(operation.CODE) | ||||
|
George Yong
|
r1171 | |||
|
Miguel Valdez
|
r577 | def getOperationObj(self, objId): | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1177 | if objId not in list(self.operations.keys()): | ||
|
Miguel Valdez
|
r577 | return None | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1177 | return self.operations[objId] | ||
|
Juan C. Valdez
|
r860 | |||
|
Daniel Valdez
|
r487 | def operation(self, **kwargs): | ||
| """ | ||||
| Operacion directa sobre la data (dataOut.data). Es necesario actualizar los valores de los | ||||
| atributos del objeto dataOut | ||||
|
Juan C. Valdez
|
r860 | |||
|
Daniel Valdez
|
r487 | Input: | ||
|
Juan C. Valdez
|
r860 | |||
|
Daniel Valdez
|
r487 | **kwargs : Diccionario de argumentos de la funcion a ejecutar | ||
| """ | ||||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1187 | raise NotImplementedError | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def setup(self): | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | raise NotImplementedError | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def run(self): | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | raise NotImplementedError | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def close(self): | ||
|
Juan C. Espinoza
|
r1187 | |||
|
George Yong
|
r1171 | return | ||
|
Juan C. Espinoza
|
r1187 | |||
|
George Yong
|
r1171 | class Operation(object): | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | """ | ||
| Update - Jan 2018 - MULTIPROCESSING | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | Most of the methods remained the same. The decorator parse the arguments and executed the run() method for each process. | ||
| The constructor doe snot receive any argument, neither the baseclass. | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | Clase base para definir las operaciones adicionales que se pueden agregar a la clase ProcessingUnit | ||
| y necesiten acumular informacion previa de los datos a procesar. De preferencia usar un buffer de | ||||
| acumulacion dentro de esta clase | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | Ejemplo: Integraciones coherentes, necesita la informacion previa de los n perfiles anteriores (bufffer) | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | """ | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def __init__(self): | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1187 | self.id = None | ||
|
George Yong
|
r1171 | self.isConfig = False | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | if not hasattr(self, 'name'): | ||
| self.name = self.__class__.__name__ | ||||
|
Juan C. Espinoza
|
r1187 | |||
|
George Yong
|
r1171 | def getAllowedArgs(self): | ||
| if hasattr(self, '__attrs__'): | ||||
| return self.__attrs__ | ||||
| else: | ||||
| return inspect.getargspec(self.run).args | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def setup(self): | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | self.isConfig = True | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | raise NotImplementedError | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def run(self, dataIn, **kwargs): | ||
|
Daniel Valdez
|
r487 | """ | ||
|
George Yong
|
r1171 | Realiza las operaciones necesarias sobre la dataIn.data y actualiza los | ||
| atributos del objeto dataIn. | ||||
|
Juan C. Valdez
|
r860 | |||
|
Miguel Valdez
|
r568 | Input: | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | dataIn : objeto del tipo JROData | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | Return: | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | None | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | Affected: | ||
| __buffer : buffer de recepcion de datos. | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | """ | ||
| if not self.isConfig: | ||||
| self.setup(**kwargs) | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | raise NotImplementedError | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def close(self): | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1187 | return | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def MPDecorator(BaseClass): | ||
| """ | ||||
|
Juan C. Espinoza
|
r1177 | Multiprocessing class decorator | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1177 | This function add multiprocessing features to a BaseClass. Also, it handle | ||
| the communication beetween processes (readers, procUnits and operations). | ||||
|
George Yong
|
r1171 | """ | ||
|
Juan C. Espinoza
|
r1187 | |||
|
George Yong
|
r1171 | class MPClass(BaseClass, Process): | ||
|
Juan C. Espinoza
|
r1177 | |||
|
George Yong
|
r1171 | def __init__(self, *args, **kwargs): | ||
| super(MPClass, self).__init__() | ||||
| Process.__init__(self) | ||||
| self.operationKwargs = {} | ||||
| self.args = args | ||||
| self.kwargs = kwargs | ||||
|
Juan C. Espinoza
|
r1177 | self.sender = None | ||
| self.receiver = None | ||||
| self.name = BaseClass.__name__ | ||||
|
George Yong
|
r1201 | if 'plot' in self.name.lower() and not self.name.endswith('_'): | ||
| self.name = '{}{}'.format(self.CODE.upper(), 'Plot') | ||||
|
Juan C. Espinoza
|
r1187 | self.start_time = time.time() | ||
|
George Yong
|
r1171 | |||
| if len(self.args) is 3: | ||||
| self.typeProc = "ProcUnit" | ||||
|
Juan C. Espinoza
|
r1187 | self.id = args[0] | ||
| self.inputId = args[1] | ||||
| self.project_id = args[2] | ||||
| elif len(self.args) is 2: | ||||
|
George Yong
|
r1171 | self.id = args[0] | ||
|
Juan C. Espinoza
|
r1177 | self.inputId = args[0] | ||
| self.project_id = args[1] | ||||
|
George Yong
|
r1171 | self.typeProc = "Operation" | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1177 | def subscribe(self): | ||
| ''' | ||||
| This function create a socket to receive objects from the | ||||
| topic `inputId`. | ||||
| ''' | ||||
|
Juan C. Espinoza
|
r1187 | |||
|
Juan C. Espinoza
|
r1177 | c = zmq.Context() | ||
| self.receiver = c.socket(zmq.SUB) | ||||
|
Juan C. Espinoza
|
r1187 | self.receiver.connect( | ||
| 'ipc:///tmp/schain/{}_pub'.format(self.project_id)) | ||||
|
Juan C. Espinoza
|
r1177 | self.receiver.setsockopt(zmq.SUBSCRIBE, self.inputId.encode()) | ||
|
George Yong
|
r1191 | |||
|
Juan C. Espinoza
|
r1177 | def listen(self): | ||
| ''' | ||||
| This function waits for objects and deserialize using pickle | ||||
| ''' | ||||
|
George Yong
|
r1191 | |||
|
Juan C. Espinoza
|
r1187 | data = pickle.loads(self.receiver.recv_multipart()[1]) | ||
|
George Yong
|
r1191 | |||
|
Juan C. Espinoza
|
r1177 | return data | ||
|
George Yong
|
r1171 | |||
|
Juan C. Espinoza
|
r1177 | def set_publisher(self): | ||
| ''' | ||||
| This function create a socket for publishing purposes. | ||||
| ''' | ||||
|
George Yong
|
r1171 | |||
|
Juan C. Espinoza
|
r1177 | time.sleep(1) | ||
| c = zmq.Context() | ||||
|
Juan C. Espinoza
|
r1187 | self.sender = c.socket(zmq.PUB) | ||
| self.sender.connect( | ||||
| 'ipc:///tmp/schain/{}_sub'.format(self.project_id)) | ||||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1187 | def publish(self, data, id): | ||
|
Juan C. Espinoza
|
r1177 | ''' | ||
| This function publish an object, to a specific topic. | ||||
| ''' | ||||
| self.sender.send_multipart([str(id).encode(), pickle.dumps(data)]) | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | def runReader(self): | ||
|
Juan C. Espinoza
|
r1177 | ''' | ||
| Run fuction for read units | ||||
| ''' | ||||
|
George Yong
|
r1171 | while True: | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1187 | BaseClass.run(self, **self.kwargs) | ||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1187 | for op, optype, opId, kwargs in self.operations: | ||
|
Juan C. Espinoza
|
r1198 | if optype == 'self' and not self.dataOut.flagNoData: | ||
|
Juan C. Espinoza
|
r1177 | op(**kwargs) | ||
|
Juan C. Espinoza
|
r1198 | elif optype == 'other' and not self.dataOut.flagNoData: | ||
|
Juan C. Espinoza
|
r1177 | self.dataOut = op.run(self.dataOut, **self.kwargs) | ||
|
Juan C. Espinoza
|
r1187 | elif optype == 'external': | ||
|
Juan C. Espinoza
|
r1177 | self.publish(self.dataOut, opId) | ||
|
Juan C. Espinoza
|
r1193 | if self.dataOut.flagNoData and not self.dataOut.error: | ||
|
George Yong
|
r1171 | continue | ||
|
Juan C. Espinoza
|
r1187 | |||
| self.publish(self.dataOut, self.id) | ||||
| if self.dataOut.error: | ||||
|
Juan C. Espinoza
|
r1193 | log.error(self.dataOut.error, self.name) | ||
|
Juan C. Espinoza
|
r1187 | # self.sender.send_multipart([str(self.project_id).encode(), 'end'.encode()]) | ||
| break | ||||
| time.sleep(1) | ||||
|
George Yong
|
r1171 | def runProc(self): | ||
|
Juan C. Espinoza
|
r1177 | ''' | ||
| Run function for proccessing units | ||||
| ''' | ||||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | while True: | ||
|
George Yong
|
r1191 | self.dataIn = self.listen() | ||
|
George Yong
|
r1171 | |||
|
Juan C. Espinoza
|
r1187 | if self.dataIn.flagNoData and self.dataIn.error is None: | ||
|
Juan C. Espinoza
|
r1177 | continue | ||
|
George Yong
|
r1191 | |||
|
Juan C. Espinoza
|
r1177 | BaseClass.run(self, **self.kwargs) | ||
|
George Yong
|
r1171 | |||
|
Juan C. Espinoza
|
r1193 | if self.dataIn.error: | ||
| self.dataOut.error = self.dataIn.error | ||||
|
George Yong
|
r1206 | self.dataOut.flagNoData = True | ||
|
Juan C. Espinoza
|
r1177 | for op, optype, opId, kwargs in self.operations: | ||
|
Juan C. Espinoza
|
r1198 | if optype == 'self' and not self.dataOut.flagNoData: | ||
|
Juan C. Espinoza
|
r1177 | op(**kwargs) | ||
|
Juan C. Espinoza
|
r1198 | elif optype == 'other' and not self.dataOut.flagNoData: | ||
|
Juan C. Espinoza
|
r1177 | self.dataOut = op.run(self.dataOut, **kwargs) | ||
|
George Yong
|
r1206 | elif optype == 'external' and not self.dataOut.flagNoData: | ||
| self.publish(self.dataOut, opId) | ||||
|
George Yong
|
r1171 | |||
|
Juan C. Espinoza
|
r1198 | if not self.dataOut.flagNoData or self.dataOut.error: | ||
| self.publish(self.dataOut, self.id) | ||||
|
George Yong
|
r1206 | for op, optype, opId, kwargs in self.operations: | ||
| if optype == 'self' and self.dataOut.error: | ||||
| op(**kwargs) | ||||
| elif optype == 'other' and self.dataOut.error: | ||||
| self.dataOut = op.run(self.dataOut, **kwargs) | ||||
| elif optype == 'external' and self.dataOut.error: | ||||
| self.publish(self.dataOut, opId) | ||||
|
Juan C. Espinoza
|
r1198 | |||
|
Juan C. Espinoza
|
r1187 | if self.dataIn.error: | ||
| break | ||||
| time.sleep(1) | ||||
|
George Yong
|
r1171 | |||
| def runOp(self): | ||||
|
Juan C. Espinoza
|
r1177 | ''' | ||
|
Juan C. Espinoza
|
r1187 | Run function for external operations (this operations just receive data | ||
| ex: plots, writers, publishers) | ||||
|
Juan C. Espinoza
|
r1177 | ''' | ||
|
George Yong
|
r1191 | |||
|
George Yong
|
r1171 | while True: | ||
|
Juan C. Espinoza
|
r1177 | dataOut = self.listen() | ||
|
George Yong
|
r1179 | |||
|
Juan C. Espinoza
|
r1177 | BaseClass.run(self, dataOut, **self.kwargs) | ||
|
Juan C. Espinoza
|
r1187 | |||
| if dataOut.error: | ||||
| break | ||||
|
Juan C. Espinoza
|
r1193 | |||
|
Juan C. Espinoza
|
r1187 | time.sleep(1) | ||
|
George Yong
|
r1171 | def run(self): | ||
|
Juan C. Espinoza
|
r1177 | if self.typeProc is "ProcUnit": | ||
|
Juan C. Espinoza
|
r1187 | |||
|
Juan C. Espinoza
|
r1177 | if self.inputId is not None: | ||
|
George Yong
|
r1191 | |||
|
Juan C. Espinoza
|
r1177 | self.subscribe() | ||
|
George Yong
|
r1191 | |||
|
Juan C. Espinoza
|
r1177 | self.set_publisher() | ||
|
George Yong
|
r1171 | |||
|
Juan C. Espinoza
|
r1177 | if 'Reader' not in BaseClass.__name__: | ||
| self.runProc() | ||||
|
George Yong
|
r1171 | else: | ||
| self.runReader() | ||||
| elif self.typeProc is "Operation": | ||||
|
Juan C. Espinoza
|
r1187 | |||
|
Juan C. Espinoza
|
r1177 | self.subscribe() | ||
|
George Yong
|
r1171 | self.runOp() | ||
|
Juan C. Valdez
|
r860 | |||
|
George Yong
|
r1171 | else: | ||
| raise ValueError("Unknown type") | ||||
|
Juan C. Valdez
|
r860 | |||
|
Juan C. Espinoza
|
r1177 | self.close() | ||
|
Juan C. Espinoza
|
r1187 | def event_monitor(self, monitor): | ||
| events = {} | ||||
| for name in dir(zmq): | ||||
| if name.startswith('EVENT_'): | ||||
| value = getattr(zmq, name) | ||||
| events[value] = name | ||||
| while monitor.poll(): | ||||
| evt = recv_monitor_message(monitor) | ||||
| if evt['event'] == 32: | ||||
| self.connections += 1 | ||||
| if evt['event'] == 512: | ||||
| pass | ||||
| evt.update({'description': events[evt['event']]}) | ||||
| if evt['event'] == zmq.EVENT_MONITOR_STOPPED: | ||||
| break | ||||
| monitor.close() | ||||
| print('event monitor thread done!') | ||||
|
Juan C. Espinoza
|
r1177 | def close(self): | ||
|
Juan C. Espinoza
|
r1187 | |||
| BaseClass.close(self) | ||||
|
Juan C. Espinoza
|
r1177 | if self.sender: | ||
| self.sender.close() | ||||
|
Juan C. Espinoza
|
r1187 | |||
|
Juan C. Espinoza
|
r1177 | if self.receiver: | ||
| self.receiver.close() | ||||
|
Juan C. Espinoza
|
r1187 | log.success('Done...(Time:{:4.2f} secs)'.format(time.time()-self.start_time), self.name) | ||
| return MPClass | ||||