##// END OF EJS Templates
schain
RSS
removed prints
José Chávez -
r930:ff6d2f615973
parent child
Show More
Show file before | Show file after | Hide comments
@@ -1,347 +1,346
1 1 '''
2 2
3 3 $Author: murco $
4 4 $Id: jroproc_base.py 1 2012-11-12 18:56:07Z murco $
5 5 '''
6 6 import inspect
7 7 from fuzzywuzzy import process
8 8
9 9 def checkKwargs(method, kwargs):
10 10 currentKwargs = kwargs
11 11 choices = inspect.getargspec(method).args
12 12 try:
13 13 choices.remove('self')
14 14 except Exception as e:
15 15 pass
16 16
17 17 try:
18 18 choices.remove('dataOut')
19 19 except Exception as e:
20 20 pass
21 21
22 22 for kwarg in kwargs:
23 23 fuzz = process.extractOne(kwarg, choices)
24 print fuzz
25 24 if fuzz is None:
26 25 continue
27 26 if fuzz[1] < 100:
28 27 raise Exception('\x1b[2;30;43mDid you mean {} instead of {} in {}? \x1b[0m'.
29 28 format(fuzz[0], kwarg, method.__self__.__class__.__name__))
30 29
31 30 class ProcessingUnit(object):
32 31
33 32 """
34 33 Esta es la clase base para el procesamiento de datos.
35 34
36 35 Contiene el metodo "call" para llamar operaciones. Las operaciones pueden ser:
37 36 - Metodos internos (callMethod)
38 37 - Objetos del tipo Operation (callObject). Antes de ser llamados, estos objetos
39 38 tienen que ser agreagados con el metodo "add".
40 39
41 40 """
42 41 # objeto de datos de entrada (Voltage, Spectra o Correlation)
43 42 dataIn = None
44 43 dataInList = []
45 44
46 45 # objeto de datos de entrada (Voltage, Spectra o Correlation)
47 46 dataOut = None
48 47
49 48 operations2RunDict = None
50 49
51 50 isConfig = False
52 51
53 52
54 53 def __init__(self, *args, **kwargs):
55 54
56 55 self.dataIn = None
57 56 self.dataInList = []
58 57
59 58 self.dataOut = None
60 59
61 60 self.operations2RunDict = {}
62 61 self.operationKwargs = {}
63 62
64 63 self.isConfig = False
65 64
66 65 self.args = args
67 66 self.kwargs = kwargs
68 67 checkKwargs(self.run, kwargs)
69 68
70 69 def getAllowedArgs(self):
71 70 return inspect.getargspec(self.run).args
72 71
73 72 def addOperationKwargs(self, objId, **kwargs):
74 73 '''
75 74 '''
76 75
77 76 self.operationKwargs[objId] = kwargs
78 77
79 78
80 79 def addOperation(self, opObj, objId):
81 80
82 81 """
83 82 Agrega un objeto del tipo "Operation" (opObj) a la lista de objetos "self.objectList" y retorna el
84 83 identificador asociado a este objeto.
85 84
86 85 Input:
87 86
88 87 object : objeto de la clase "Operation"
89 88
90 89 Return:
91 90
92 91 objId : identificador del objeto, necesario para ejecutar la operacion
93 92 """
94 93
95 94 self.operations2RunDict[objId] = opObj
96 95
97 96 return objId
98 97
99 98 def getOperationObj(self, objId):
100 99
101 100 if objId not in self.operations2RunDict.keys():
102 101 return None
103 102
104 103 return self.operations2RunDict[objId]
105 104
106 105 def operation(self, **kwargs):
107 106
108 107 """
109 108 Operacion directa sobre la data (dataOut.data). Es necesario actualizar los valores de los
110 109 atributos del objeto dataOut
111 110
112 111 Input:
113 112
114 113 **kwargs : Diccionario de argumentos de la funcion a ejecutar
115 114 """
116 115
117 116 raise NotImplementedError
118 117
119 118 def callMethod(self, name, opId):
120 119
121 120 """
122 121 Ejecuta el metodo con el nombre "name" y con argumentos **kwargs de la propia clase.
123 122
124 123 Input:
125 124 name : nombre del metodo a ejecutar
126 125
127 126 **kwargs : diccionario con los nombres y valores de la funcion a ejecutar.
128 127
129 128 """
130 129
131 130 #Checking the inputs
132 131 if name == 'run':
133 132
134 133 if not self.checkInputs():
135 134 self.dataOut.flagNoData = True
136 135 return False
137 136 else:
138 137 #Si no es un metodo RUN la entrada es la misma dataOut (interna)
139 138 if self.dataOut is not None and self.dataOut.isEmpty():
140 139 return False
141 140
142 141 #Getting the pointer to method
143 142 methodToCall = getattr(self, name)
144 143
145 144 #Executing the self method
146 145
147 146 if hasattr(self, 'mp'):
148 147 if name=='run':
149 148 if self.mp is False:
150 149 self.mp = True
151 150 self.start()
152 151 else:
153 152 methodToCall(**self.operationKwargs[opId])
154 153 else:
155 154 if name=='run':
156 155 methodToCall(**self.kwargs)
157 156 else:
158 157 methodToCall(**self.operationKwargs[opId])
159 158
160 159 if self.dataOut is None:
161 160 return False
162 161
163 162 if self.dataOut.isEmpty():
164 163 return False
165 164
166 165 return True
167 166
168 167 def callObject(self, objId):
169 168
170 169 """
171 170 Ejecuta la operacion asociada al identificador del objeto "objId"
172 171
173 172 Input:
174 173
175 174 objId : identificador del objeto a ejecutar
176 175
177 176 **kwargs : diccionario con los nombres y valores de la funcion a ejecutar.
178 177
179 178 Return:
180 179
181 180 None
182 181 """
183 182
184 183 if self.dataOut is not None and self.dataOut.isEmpty():
185 184 return False
186 185
187 186 externalProcObj = self.operations2RunDict[objId]
188 187
189 188 if hasattr(externalProcObj, 'mp'):
190 189 if externalProcObj.mp is False:
191 190 self.operationKwargs[objId] = externalProcObj.kwargs
192 191 externalProcObj.mp = True
193 192 externalProcObj.start()
194 193 else:
195 194 externalProcObj.run(self.dataOut, **externalProcObj.kwargs)
196 195 self.operationKwargs[objId] = externalProcObj.kwargs
197 196
198 197 return True
199 198
200 199 def call(self, opType, opName=None, opId=None):
201 200
202 201 """
203 202 Return True si ejecuta la operacion interna nombrada "opName" o la operacion externa
204 203 identificada con el id "opId"; con los argumentos "**kwargs".
205 204
206 205 False si la operacion no se ha ejecutado.
207 206
208 207 Input:
209 208
210 209 opType : Puede ser "self" o "external"
211 210
212 211 Depende del tipo de operacion para llamar a:callMethod or callObject:
213 212
214 213 1. If opType = "self": Llama a un metodo propio de esta clase:
215 214
216 215 name_method = getattr(self, name)
217 216 name_method(**kwargs)
218 217
219 218
220 219 2. If opType = "other" o"external": Llama al metodo "run()" de una instancia de la
221 220 clase "Operation" o de un derivado de ella:
222 221
223 222 instanceName = self.operationList[opId]
224 223 instanceName.run(**kwargs)
225 224
226 225 opName : Si la operacion es interna (opType = 'self'), entonces el "opName" sera
227 226 usada para llamar a un metodo interno de la clase Processing
228 227
229 228 opId : Si la operacion es externa (opType = 'other' o 'external), entonces el
230 229 "opId" sera usada para llamar al metodo "run" de la clase Operation
231 230 registrada anteriormente con ese Id
232 231
233 232 Exception:
234 233 Este objeto de tipo Operation debe de haber sido agregado antes con el metodo:
235 234 "addOperation" e identificado con el valor "opId" = el id de la operacion.
236 235 De lo contrario retornara un error del tipo ValueError
237 236
238 237 """
239 238
240 239 if opType == 'self':
241 240
242 241 if not opName:
243 242 raise ValueError, "opName parameter should be defined"
244 243
245 244 sts = self.callMethod(opName, opId)
246 245
247 246 elif opType == 'other' or opType == 'external' or opType == 'plotter':
248 247
249 248 if not opId:
250 249 raise ValueError, "opId parameter should be defined"
251 250
252 251 if opId not in self.operations2RunDict.keys():
253 252 raise ValueError, "Any operation with id=%s has been added" %str(opId)
254 253
255 254 sts = self.callObject(opId)
256 255
257 256 else:
258 257 raise ValueError, "opType should be 'self', 'external' or 'plotter'; and not '%s'" %opType
259 258
260 259 return sts
261 260
262 261 def setInput(self, dataIn):
263 262
264 263 self.dataIn = dataIn
265 264 self.dataInList.append(dataIn)
266 265
267 266 def getOutputObj(self):
268 267
269 268 return self.dataOut
270 269
271 270 def checkInputs(self):
272 271
273 272 for thisDataIn in self.dataInList:
274 273
275 274 if thisDataIn.isEmpty():
276 275 return False
277 276
278 277 return True
279 278
280 279 def setup(self):
281 280
282 281 raise NotImplementedError
283 282
284 283 def run(self):
285 284
286 285 raise NotImplementedError
287 286
288 287 def close(self):
289 288 #Close every thread, queue or any other object here is it is neccesary.
290 289 return
291 290
292 291 class Operation(object):
293 292
294 293 """
295 294 Clase base para definir las operaciones adicionales que se pueden agregar a la clase ProcessingUnit
296 295 y necesiten acumular informacion previa de los datos a procesar. De preferencia usar un buffer de
297 296 acumulacion dentro de esta clase
298 297
299 298 Ejemplo: Integraciones coherentes, necesita la informacion previa de los n perfiles anteriores (bufffer)
300 299
301 300 """
302 301
303 302 __buffer = None
304 303 isConfig = False
305 304
306 305 def __init__(self, **kwargs):
307 306
308 307 self.__buffer = None
309 308 self.isConfig = False
310 309 self.kwargs = kwargs
311 310 checkKwargs(self.run, kwargs)
312 311
313 312 def getAllowedArgs(self):
314 313 return inspect.getargspec(self.run).args
315 314
316 315 def setup(self):
317 316
318 317 self.isConfig = True
319 318
320 319 raise NotImplementedError
321 320
322 321 def run(self, dataIn, **kwargs):
323 322
324 323 """
325 324 Realiza las operaciones necesarias sobre la dataIn.data y actualiza los
326 325 atributos del objeto dataIn.
327 326
328 327 Input:
329 328
330 329 dataIn : objeto del tipo JROData
331 330
332 331 Return:
333 332
334 333 None
335 334
336 335 Affected:
337 336 __buffer : buffer de recepcion de datos.
338 337
339 338 """
340 339 if not self.isConfig:
341 340 self.setup(**kwargs)
342 341
343 342 raise NotImplementedError
344 343
345 344 def close(self):
346 345
347 346 pass
Show file before | Show file after | Hide comments
@@ -1,96 +1,96
1 1 import argparse
2 2
3 3 from schainpy.controller import Project, multiSchain
4 4
5 5 desc = "HF_EXAMPLE"
6 6
7 7 def fiber(cursor, skip, q, dt):
8 8
9 9 controllerObj = Project()
10 10
11 11 controllerObj.setup(id='191', name='test01', description=desc)
12 12
13 13 readUnitConfObj = controllerObj.addReadUnit(datatype='SpectraReader',
14 14 path='/home/nanosat/data/julia',
15 15 startDate=dt,
16 16 endDate=dt,
17 17 startTime="00:00:00",
18 endTie="23:59:59",
18 endTime="23:59:59",
19 19 online=0,
20 20 #set=1426485881,
21 21 delay=10,
22 22 walk=1,
23 23 queue=q,
24 24 cursor=cursor,
25 25 skip=skip,
26 26 #timezone=-5*3600
27 27 )
28 28
29 29 # #opObj11 = readUnitConfObj.addOperation(name='printNumberOfBlock')
30 30 #
31 31 procUnitConfObj2 = controllerObj.addProcUnit(datatype='Spectra', inputId=readUnitConfObj.getId())
32 32 # procUnitConfObj2.addParameter(name='nipp', value='5', format='int')
33 33
34 34 procUnitConfObj3 = controllerObj.addProcUnit(datatype='ParametersProc', inputId=readUnitConfObj.getId())
35 35 opObj11 = procUnitConfObj3.addOperation(name='SpectralMoments', optype='other')
36 36
37 37 #
38 38 # opObj11 = procUnitConfObj1.addOperation(name='SpectraPlot', optype='other')
39 39 # opObj11.addParameter(name='id', value='1000', format='int')
40 40 # opObj11.addParameter(name='wintitle', value='HF_Jicamarca_Spc', format='str')
41 41 # opObj11.addParameter(name='channelList', value='0', format='intlist')
42 42 # opObj11.addParameter(name='zmin', value='-120', format='float')
43 43 # opObj11.addParameter(name='zmax', value='-70', format='float')
44 44 # opObj11.addParameter(name='save', value='1', format='int')
45 45 # opObj11.addParameter(name='figpath', value=figpath, format='str')
46 46
47 47 opObj11 = procUnitConfObj3.addOperation(name='Parameters1Plot', optype='other')
48 48 opObj11.addParameter(name='channelList', value='0', format='intList')
49 49
50 50 opObj11.addParameter(name='id', value='2000', format='int')
51 51 # opObj11.addParameter(name='colormap', value='0', format='bool')
52 52 opObj11.addParameter(name='onlySNR', value='1', format='bool')
53 53 opObj11.addParameter(name='DOP', value='0', format='bool')
54 54 # opObj11.addParameter(name='showSNR', value='1', format='bool')
55 55 # opObj11.addParameter(name='SNRthresh', value='0', format='int')
56 56 # opObj11.addParameter(name='SNRmin', value='-10', format='int')
57 57 # opObj11.addParameter(name='SNRmax', value='30', format='int')
58 58
59 59 # opObj11.addParameter(name='showSNR', value='1', format='int')
60 60 # # opObj11.addParameter(name='channelList', value='0', format='intlist')
61 61 # # opObj11.addParameter(name='xmin', value='0', format='float')
62 62 # opObj11.addParameter(name='xmin', value='0', format='float')
63 63 # opObj11.addParameter(name='xmax', value='24', format='float')
64 64
65 65 # opObj11.addParameter(name='zmin', value='-110', format='float')
66 66 # opObj11.addParameter(name='zmax', value='-70', format='float')
67 67 # opObj11.addParameter(name='save', value='0', format='int')
68 68 # # opObj11.addParameter(name='figpath', value='/tmp/', format='str')
69 69 #
70 70 opObj12 = procUnitConfObj3.addOperation(name='PublishData', optype='other')
71 71 opObj12.addParameter(name='zeromq', value=1, format='int')
72 72
73 73
74 74 # opObj13 = procUnitConfObj3.addOperation(name='PublishData', optype='other')
75 75 # opObj13.addParameter(name='zeromq', value=1, format='int')
76 76 # opObj13.addParameter(name='server', value="juanca", format='str')
77 77
78 78 opObj12.addParameter(name='delay', value=1, format='int')
79 79
80 80
81 81 # print "Escribiendo el archivo XML"
82 82 # controllerObj.writeXml(filename)
83 83 # print "Leyendo el archivo XML"
84 84 # controllerObj.readXml(filename)
85 85
86 86
87 87 # timeit.timeit('controllerObj.run()', number=2)
88 88
89 89 controllerObj.start()
90 90
91 91
92 92 if __name__ == '__main__':
93 93 parser = argparse.ArgumentParser(description='Set number of parallel processes')
94 94 parser.add_argument('--nProcess', default=1, type=int)
95 95 args = parser.parse_args()
96 96 multiSchain(fiber, nProcess=args.nProcess, startDate='2015/09/26', endDate='2015/09/26')
Show file before | Show file after | Hide comments
@@ -1,1 +1,1
1 <Project description="HF_EXAMPLE" id="191" name="test01"><ReadUnit datatype="SpectraReader" id="1911" inputId="0" name="SpectraReader"><Operation id="19111" name="run" priority="1" type="self"><Parameter format="str" id="191111" name="datatype" value="SpectraReader" /><Parameter format="str" id="191112" name="path" value="/home/nanosat/data/julia" /><Parameter format="date" id="191113" name="startDate" value="2015/09/26" /><Parameter format="date" id="191114" name="endDate" value="2015/09/26" /><Parameter format="time" id="191115" name="startTime" value="00:00:00" /><Parameter format="time" id="191116" name="endTime" value="23:59:59" /><Parameter format="int" id="191118" name="cursor" value="0" /><Parameter format="int" id="191119" name="skip" value="0" /><Parameter format="int" id="191120" name="delay" value="10" /><Parameter format="int" id="191121" name="walk" value="1" /><Parameter format="int" id="191122" name="online" value="0" /></Operation></ReadUnit><ProcUnit datatype="ParametersProc" id="1913" inputId="1911" name="ParametersProc"><Operation id="19131" name="run" priority="1" type="self" /><Operation id="19132" name="SpectralMoments" priority="2" type="other" /><Operation id="19133" name="Parameters1Plot" priority="3" type="other"><Parameter format="intlist" id="191331" name="channelList" value="0" /><Parameter format="int" id="191332" name="id" value="2000" /><Parameter format="bool" id="191333" name="olySNR" value="1" /><Parameter format="bool" id="191334" name="DOP" value="0" /></Operation><Operation id="19134" name="PublishData" priority="4" type="other"><Parameter format="int" id="191341" name="zeromq" value="1" /><Parameter format="int" id="191342" name="delay" value="1" /></Operation></ProcUnit><ProcUnit datatype="Spectra" id="1912" inputId="1911" name="SpectraProc"><Operation id="19121" name="run" priority="1" type="self" /></ProcUnit></Project> No newline at end of file
1 <Project description="HF_EXAMPLE" id="191" name="test01"><ReadUnit datatype="SpectraReader" id="1911" inputId="0" name="SpectraReader"><Operation id="19111" name="run" priority="1" type="self"><Parameter format="str" id="191111" name="datatype" value="SpectraReader" /><Parameter format="str" id="191112" name="path" value="/home/nanosat/data/julia" /><Parameter format="date" id="191113" name="startDate" value="2015/09/26" /><Parameter format="date" id="191114" name="endDate" value="2015/09/26" /><Parameter format="time" id="191115" name="startTime" value="00:00:00" /><Parameter format="time" id="191116" name="endTime" value="23:59:59" /><Parameter format="int" id="191118" name="cursor" value="0" /><Parameter format="int" id="191119" name="skip" value="0" /><Parameter format="int" id="191120" name="delay" value="10" /><Parameter format="int" id="191121" name="walk" value="1" /><Parameter format="int" id="191122" name="online" value="0" /></Operation></ReadUnit><ProcUnit datatype="ParametersProc" id="1913" inputId="1911" name="ParametersProc"><Operation id="19131" name="run" priority="1" type="self" /><Operation id="19132" name="SpectralMoments" priority="2" type="other" /><Operation id="19133" name="Parameters1Plot" priority="3" type="other"><Parameter format="intlist" id="191331" name="channelList" value="0" /><Parameter format="int" id="191332" name="id" value="2000" /><Parameter format="bool" id="191333" name="onlySNR" value="1" /><Parameter format="bool" id="191334" name="DOP" value="0" /></Operation><Operation id="19134" name="PublishData" priority="4" type="other"><Parameter format="int" id="191341" name="zeromq" value="1" /><Parameter format="int" id="191342" name="delay" value="1" /></Operation></ProcUnit><ProcUnit datatype="Spectra" id="1912" inputId="1911" name="SpectraProc"><Operation id="19121" name="run" priority="1" type="self" /></ProcUnit></Project> No newline at end of file
General Comments 0
You need to be logged in to leave comments. Login now