##// END OF EJS Templates
schain
RSS
comentario e identacion
comentario e identacion
Alexander Valdez - - Load All Authors

File last commit:

r1674:6310ac0f91c8
r1705:9e4b98696352
Show More
jroproc_base.py
221 lines | 5.8 KiB | text/x-python | PythonLexer
/ schainpy / model / proc / jroproc_base.py
History | Source | Raw |Copy content |Copy permalink
Daniel Valdez
This is the new organization by packages and scripts for Signal Chain, this version contains new features and bugs fixed until August 2014
 r487 '''
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 Base clases to create Processing units and operations, the MPDecorator
must be used in plotting and writing operations to allow to run as an
external process.
Daniel Valdez
This is the new organization by packages and scripts for Signal Chain, this version contains new features and bugs fixed until August 2014
 r487 '''
Alexander Valdez
Update reference ISR repository
 r1674 import os
José Chávez
checking misspelled kwargs in operations/processing units
 r929 import inspect
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 import zmq
import time
import pickle
jespinoza
Errors handling and gracefully terminate main process
 r1241 import traceback
Juan C. Espinoza
Add input queues for processing units and external operations
 r1235 from threading import Thread
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 from multiprocessing import Process, Queue
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 from schainpy.utils import log
Alexander Valdez
Update reference ISR repository
 r1674 import copy
QUEUE_SIZE = int(os.environ.get('QUEUE_MAX_SIZE', '10'))
Daniel Valdez
This is the new organization by packages and scripts for Signal Chain, this version contains new features and bugs fixed until August 2014
 r487
Miguel Valdez
Merge with branch schain_julia_drifts from rev. 803 to 995....
 r568 class ProcessingUnit(object):
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 '''
Base class to create Signal Chain Units
'''
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
jespinoza
Update ParamReader to support diferent HDF5 files, fix Read/Write Madrigal files
 r1254 proc_type = 'processing'
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 def __init__(self):
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
Daniel Valdez
This is the new organization by packages and scripts for Signal Chain, this version contains new features and bugs fixed until August 2014
 r487 self.dataIn = None
Miguel Valdez
A new SendToServer Unit has been created to upload files to a remote server....
 r573 self.dataOut = None
Daniel Valdez
This is the new organization by packages and scripts for Signal Chain, this version contains new features and bugs fixed until August 2014
 r487 self.isConfig = False
Juan C. Espinoza
Review MP changes, three types of operations: self, other and external
 r1177 self.operations = []
Alexander Valdez
Update reference ISR repository
 r1674 self.name = 'Test'
self.inputs = []
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 def setInput(self, unit):
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
Alexander Valdez
Update reference ISR repository
 r1674 attr = 'dataIn'
for i, u in enumerate(unit):
if i==0:
self.dataIn = u.dataOut#.copy()
self.inputs.append('dataIn')
else:
setattr(self, 'dataIn{}'.format(i), u.dataOut)#.copy())
self.inputs.append('dataIn{}'.format(i))
José Chávez
checking misspelled kwargs in operations/processing units
 r929 def getAllowedArgs(self):
Juan C. Espinoza
Add __attrs__ attribute to Process clasess to improve CLI finder
 r1097 if hasattr(self, '__attrs__'):
return self.__attrs__
else:
return inspect.getargspec(self.run).args
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
Juan C. Espinoza
New plotting architecture with buffering/throttle capabilities
 r1187 def addOperation(self, conf, operation):
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 '''
'''
Alexander Valdez
Update reference ISR repository
 r1674
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 self.operations.append((operation, conf.type, conf.getKwargs()))
George Yong
Multiprocessing for Spectra (all operation) working
 r1171
Miguel Valdez
r577 def getOperationObj(self, objId):
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
Juan C. Espinoza
Review MP changes, three types of operations: self, other and external
 r1177 if objId not in list(self.operations.keys()):
Miguel Valdez
r577 return None
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
Juan C. Espinoza
Review MP changes, three types of operations: self, other and external
 r1177 return self.operations[objId]
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 def call(self, **kwargs):
'''
'''
try:
if self.dataIn is not None and self.dataIn.flagNoData and not self.dataIn.error:
Alexander Valdez
Update reference ISR repository
 r1674 if self.dataIn.runNextUnit:
return not self.dataIn.isReady()
else:
return self.dataIn.isReady()
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 elif self.dataIn is None or not self.dataIn.error:
self.run(**kwargs)
elif self.dataIn.error:
self.dataOut.error = self.dataIn.error
self.dataOut.flagNoData = True
except:
Alexander Valdez
Update reference ISR repository
 r1674 err = traceback.format_exc()
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 if 'SchainWarning' in err:
log.warning(err.split('SchainWarning:')[-1].split('\n')[0].strip(), self.name)
elif 'SchainError' in err:
log.error(err.split('SchainError:')[-1].split('\n')[0].strip(), self.name)
else:
Juan C. Espinoza
Fix ScpecraWriter and CohInt attribute
 r1310 log.error(err, self.name)
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 self.dataOut.error = True
for op, optype, opkwargs in self.operations:
Alexander Valdez
Update reference ISR repository
 r1674 aux = self.dataOut.copy()
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 if optype == 'other' and not self.dataOut.flagNoData:
self.dataOut = op.run(self.dataOut, **opkwargs)
elif optype == 'external' and not self.dataOut.flagNoData:
Alexander Valdez
Update reference ISR repository
 r1674 op.queue.put(aux)
elif optype == 'external' and self.dataOut.error:
op.queue.put(aux)
try:
if self.dataOut.runNextUnit:
runNextUnit = self.dataOut.runNextUnit
else:
runNextUnit = self.dataOut.isReady()
except:
runNextUnit = self.dataOut.isReady()
return 'Error' if self.dataOut.error else runNextUnit# self.dataOut.isReady()
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 def setup(self):
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 raise NotImplementedError
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 def run(self):
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 raise NotImplementedError
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 def close(self):
Juan C. Espinoza
New plotting architecture with buffering/throttle capabilities
 r1187
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 return
Juan C. Espinoza
New plotting architecture with buffering/throttle capabilities
 r1187
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 class Operation(object):
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 '''
'''
Alexander Valdez
Update reference ISR repository
 r1674
jespinoza
Update ParamReader to support diferent HDF5 files, fix Read/Write Madrigal files
 r1254 proc_type = 'operation'
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 def __init__(self):
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
Juan C. Espinoza
New plotting architecture with buffering/throttle capabilities
 r1187 self.id = None
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 self.isConfig = False
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 if not hasattr(self, 'name'):
self.name = self.__class__.__name__
Juan C. Espinoza
New plotting architecture with buffering/throttle capabilities
 r1187
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 def getAllowedArgs(self):
if hasattr(self, '__attrs__'):
return self.__attrs__
else:
return inspect.getargspec(self.run).args
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 def setup(self):
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 self.isConfig = True
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 raise NotImplementedError
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 def run(self, dataIn, **kwargs):
Daniel Valdez
This is the new organization by packages and scripts for Signal Chain, this version contains new features and bugs fixed until August 2014
 r487 """
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 Realiza las operaciones necesarias sobre la dataIn.data y actualiza los
atributos del objeto dataIn.
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
Miguel Valdez
Merge with branch schain_julia_drifts from rev. 803 to 995....
 r568 Input:
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 dataIn : objeto del tipo JROData
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 Return:
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 None
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 Affected:
__buffer : buffer de recepcion de datos.
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 """
if not self.isConfig:
self.setup(**kwargs)
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 raise NotImplementedError
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 def close(self):
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
Juan C. Espinoza
New plotting architecture with buffering/throttle capabilities
 r1187 return
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
Alexander Valdez
Update reference ISR repository
 r1674
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 def MPDecorator(BaseClass):
"""
Juan C. Espinoza
Review MP changes, three types of operations: self, other and external
 r1177 Multiprocessing class decorator
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
Alexander Valdez
Update reference ISR repository
 r1674 This function add multiprocessing features to a BaseClass.
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 """
Juan C. Espinoza
New plotting architecture with buffering/throttle capabilities
 r1187
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 class MPClass(BaseClass, Process):
Juan C. Espinoza
Review MP changes, three types of operations: self, other and external
 r1177
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 def __init__(self, *args, **kwargs):
super(MPClass, self).__init__()
Process.__init__(self)
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 self.args = args
self.kwargs = kwargs
jespinoza
Use of delays instead of input queue to keep dataouts and avoid loose of them
 r1245 self.t = time.time()
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 self.op_type = 'external'
Juan C. Espinoza
Review MP changes, three types of operations: self, other and external
 r1177 self.name = BaseClass.__name__
jespinoza
Update ParamReader to support diferent HDF5 files, fix Read/Write Madrigal files
 r1254 self.__doc__ = BaseClass.__doc__
Alexander Valdez
Update reference ISR repository
 r1674
George Yong
Fix bug in CrossSpectraPlot
 r1201 if 'plot' in self.name.lower() and not self.name.endswith('_'):
self.name = '{}{}'.format(self.CODE.upper(), 'Plot')
Alexander Valdez
Update reference ISR repository
 r1674
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 self.start_time = time.time()
jespinoza
Errors handling and gracefully terminate main process
 r1241 self.err_queue = args[3]
Alexander Valdez
Update reference ISR repository
 r1674 self.queue = Queue(maxsize=QUEUE_SIZE)
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 self.myrun = BaseClass.run
George Yong
Multiprocessing for Spectra (all operation) working
 r1171
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 def run(self):
Alexander Valdez
Update reference ISR repository
 r1674
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 while True:
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 dataOut = self.queue.get()
George Yong
Multiprocessing for writing Units(Spectral, Voltage and Parameters)
 r1179
jespinoza
Errors handling and gracefully terminate main process
 r1241 if not dataOut.error:
jespinoza
Fix excessive memory RAM consumption
 r1268 try:
BaseClass.run(self, dataOut, **self.kwargs)
except:
Alexander Valdez
Update reference ISR repository
 r1674 err = traceback.format_exc()
Juan C. Espinoza
Fix ScpecraWriter and CohInt attribute
 r1310 log.error(err, self.name)
George Yong
Multiprocessing for Spectra (all operation) working
 r1171 else:
Juan C. Espinoza
Rewrite controller, remove MPDecorator to units (keep for plots an writers) use of queues for interproc comm instead of zmq, self operations are no longer supported
 r1287 break
Juan C. Valdez
Version 2.2.5 Fixed several bugs, add jro colormap for spectra/rti, add ParamWriter, TODO: Fix decimation currently disabled
 r860
Juan C. Espinoza
Review MP changes, three types of operations: self, other and external
 r1177 self.close()
def close(self):
Juan C. Espinoza
New plotting architecture with buffering/throttle capabilities
 r1187
BaseClass.close(self)
Alexander Valdez
Update reference ISR repository
 r1674 log.success('Done...(Time:{:4.2f} secs)'.format(time.time() - self.start_time), self.name)
Juan C. Espinoza
New plotting architecture with buffering/throttle capabilities
 r1187
Alexander Valdez
Update reference ISR repository
 r1674 return MPClass