##// END OF EJS Templates
schain
RSS
Separación de datos antes del 17 de Julio del 2021, donde se modificó los apuntes y la lectura de estos
joabAM -
r1371:603b419b2641
parent child
Show More
Side by Side Unified
Context file:
r1371:603b419b2641 Loading diff...:
Show file before | Show file after | Hide comments
@@ -0,0 +1,1
1 <Project description="AMISR EEJ Experiment" id="11" name="Process (eej_proc)"><ReadUnit id="111" inputId="None" name="AMISRReader"><Parameter name="path" value="/media/soporte/UARS_4T_D02/AMISR_DATA/2021/" /><Parameter name="startDate" value="2021/07/11" /><Parameter name="endDate" value="2021/07/11" /><Parameter name="startTime" value="07:01:30" /><Parameter name="endTime" value="19:00:00" /><Parameter name="walk" value="1" /><Parameter name="code" value="(1, -1, -1, -1, 1, 1, 1, 1, -1, -1, -1, 1, -1, -1, -1, 1, -1, -1, -1, 1, -1, -1, 1, -1, 1, 1, -1, 1)" /><Parameter name="nCode" value="1" /><Parameter name="nBaud" value="28" /><Parameter name="timezone" value="ut" /><Parameter name="online" value="0" /></ReadUnit><ProcUnit id="112" inputId="111" name="VoltageProc"><Operation id="1121" name="setAttribute"><Parameter name="frequency" value="445090000.0" /></Operation><Operation id="1122" name="Decoder"><Parameter name="code" value="(1, -1, -1, -1, 1, 1, 1, 1, -1, -1, -1, 1, -1, -1, -1, 1, -1, -1, -1, 1, -1, -1, 1, -1, 1, 1, -1, 1)" /><Parameter name="nCode" value="1" /><Parameter name="nBaud" value="28" /><Parameter name="osamp" value="1" /></Operation></ProcUnit><ProcUnit id="113" inputId="112" name="SpectraProc"><Parameter name="nFFTPoints" value="16" /><Operation id="1131" name="IncohInt"><Parameter name="n" value="150" /></Operation><Operation id="1132" name="removeDC" /><Operation id="1133" name="SpectraPlot"><Parameter name="id" value="21" /><Parameter name="xaxis" value="velocity" /><Parameter name="ymax" value="300" /><Parameter name="showprofile" value="1" /><Parameter name="wintitle" value="AMISR Beam 0" /><Parameter name="zmin" value="45" /><Parameter name="zmax" value="65" /><Parameter name="save" value="/home/soporte/Data/EEJ/EEJ2021192/plots" /><Parameter name="colormap" value="jet" /><Parameter name="localtime" value="0" /><Parameter name="show" value="1" /></Operation><Operation id="1134" name="SpectraWriter"><Parameter name="path" value="/home/soporte/Data/EEJ/EEJ2021192" /><Parameter name="blocksPerFile" value="10" /></Operation><Operation id="1135" name="NoisePlot"><Parameter name="id" value="3" /><Parameter name="wintitle" value="title0" /><Parameter name="showprofile" value="0" /><Parameter name="xmin" value="07" /><Parameter name="xmax" value="18" /><Parameter name="ymin" value="45" /><Parameter name="ymax" value="65" /><Parameter name="save" value="/home/soporte/Data/EEJ/EEJ2021192" /><Parameter name="localtime" value="0" /><Parameter name="show" value="0" /></Operation><Operation id="1136" name="RTIPlot"><Parameter name="id" value="2" /><Parameter name="localtime" value="0" /><Parameter name="wintitle" value="RTI" /><Parameter name="xmin" value="07" /><Parameter name="xmax" value="18" /><Parameter name="ymin" value="0" /><Parameter name="zmin" value="45" /><Parameter name="zmax" value="65" /><Parameter name="showprofile" value="0" /><Parameter name="save" value="/home/soporte/Data/EEJ/EEJ2021192/plots" /><Parameter name="colormap" value="jet" /><Parameter name="show" value="1" /></Operation></ProcUnit><ProcUnit id="114" inputId="113" name="ParametersProc"><Operation id="1141" name="SpectralMoments" /><Operation id="1142" name="ParamWriter"><Parameter name="path" value="/home/soporte/Data/EEJ/EEJ2021192" /><Parameter name="blocksPerFile" value="10" /><Parameter name="metadataList" value="['type', 'inputUnit', 'heightList']" /><Parameter name="dataList" value="['moments', 'data_SNR', 'utctime']" /><Parameter name="mode" value="1" /></Operation></ProcUnit></Project> No newline at end of file
Show file before | Show file after | Hide comments
This diff has been collapsed as it changes many lines, (1577 lines changed) Show them Hide them
@@ -1,918 +1,659
1 # Copyright (c) 2012-2020 Jicamarca Radio Observatory
2 # All rights reserved.
3 #
4 # Distributed under the terms of the BSD 3-clause license.
5 """API to create signal chain projects
6
7 The API is provide through class: Project
8 """
9
10 import re
11 import sys
12 import ast
13 import datetime
14 import traceback
15 import time
16 import multiprocessing
17 from multiprocessing import Process, Queue
18 from threading import Thread
19 from xml.etree.ElementTree import ElementTree, Element, SubElement
20
21 from schainpy.admin import Alarm, SchainWarning
22 from schainpy.model import *
23 from schainpy.utils import log
24
25 if 'darwin' in sys.platform and sys.version_info[0] == 3 and sys.version_info[1] > 7:
26 multiprocessing.set_start_method('fork')
27
28 DTYPES = {
29 'Voltage': '.r',
30 'Spectra': '.pdata'
31 }
32
33
34 def MPProject(project, n=cpu_count()):
35 '''
36 Project wrapper to run schain in n processes
37 '''
38
39 rconf = project.getReadUnitObj()
40 op = rconf.getOperationObj('run')
41 dt1 = op.getParameterValue('startDate')
42 dt2 = op.getParameterValue('endDate')
43 tm1 = op.getParameterValue('startTime')
44 tm2 = op.getParameterValue('endTime')
45 days = (dt2 - dt1).days
46
47 for day in range(days + 1):
48 skip = 0
49 cursor = 0
50 processes = []
51 dt = dt1 + datetime.timedelta(day)
52 dt_str = dt.strftime('%Y/%m/%d')
53 reader = JRODataReader()
54 paths, files = reader.searchFilesOffLine(path=rconf.path,
55 startDate=dt,
56 endDate=dt,
57 startTime=tm1,
58 endTime=tm2,
59 ext=DTYPES[rconf.datatype])
60 nFiles = len(files)
61 if nFiles == 0:
62 continue
63 skip = int(math.ceil(nFiles / n))
64 while nFiles > cursor * skip:
65 rconf.update(startDate=dt_str, endDate=dt_str, cursor=cursor,
66 skip=skip)
67 p = project.clone()
68 p.start()
69 processes.append(p)
70 cursor += 1
71
72 def beforeExit(exctype, value, trace):
73 for process in processes:
74 process.terminate()
75 process.join()
76 print(traceback.print_tb(trace))
77
78 sys.excepthook = beforeExit
79
80 for process in processes:
81 process.join()
82 process.terminate()
83
84 time.sleep(3)
85
86 def wait(context):
87
88 time.sleep(1)
89 c = zmq.Context()
90 receiver = c.socket(zmq.SUB)
91 receiver.connect('ipc:///tmp/schain_{}_pub'.format(self.id))
92 receiver.setsockopt(zmq.SUBSCRIBE, self.id.encode())
93 msg = receiver.recv_multipart()[1]
94 context.terminate()
95
96 class ParameterConf():
97
98 id = None
99 name = None
100 value = None
101 format = None
102
103 __formated_value = None
104
105 ELEMENTNAME = 'Parameter'
106
107 def __init__(self):
108
109 self.format = 'str'
110
111 def getElementName(self):
112
113 return self.ELEMENTNAME
114
115 def getValue(self):
116
117 value = self.value
118 format = self.format
119
120 if self.__formated_value != None:
121
122 return self.__formated_value
123
124 if format == 'obj':
125 return value
126
127 if format == 'str':
128 self.__formated_value = str(value)
129 return self.__formated_value
130
131 if value == '':
132 raise ValueError('%s: This parameter value is empty' % self.name)
133
134 if format == 'list':
135 strList = [s.strip() for s in value.split(',')]
136 self.__formated_value = strList
137
138 return self.__formated_value
139
140 if format == 'intlist':
141 '''
142 Example:
143 value = (0,1,2)
144 '''
145
146 new_value = ast.literal_eval(value)
147
148 if type(new_value) not in (tuple, list):
149 new_value = [int(new_value)]
150
151 self.__formated_value = new_value
152
153 return self.__formated_value
154
155 if format == 'floatlist':
156 '''
157 Example:
158 value = (0.5, 1.4, 2.7)
159 '''
160
161 new_value = ast.literal_eval(value)
162
163 if type(new_value) not in (tuple, list):
164 new_value = [float(new_value)]
165
166 self.__formated_value = new_value
167
168 return self.__formated_value
169
170 if format == 'date':
171 strList = value.split('/')
172 intList = [int(x) for x in strList]
173 date = datetime.date(intList[0], intList[1], intList[2])
174
175 self.__formated_value = date
176
177 return self.__formated_value
178
179 if format == 'time':
180 strList = value.split(':')
181 intList = [int(x) for x in strList]
182 time = datetime.time(intList[0], intList[1], intList[2])
183
184 self.__formated_value = time
185
186 return self.__formated_value
187
188 if format == 'pairslist':
189 '''
190 Example:
191 value = (0,1),(1,2)
192 '''
193
194 new_value = ast.literal_eval(value)
195
196 if type(new_value) not in (tuple, list):
197 raise ValueError('%s has to be a tuple or list of pairs' % value)
198
199 if type(new_value[0]) not in (tuple, list):
200 if len(new_value) != 2:
201 raise ValueError('%s has to be a tuple or list of pairs' % value)
202 new_value = [new_value]
203
204 for thisPair in new_value:
205 if len(thisPair) != 2:
206 raise ValueError('%s has to be a tuple or list of pairs' % value)
207
208 self.__formated_value = new_value
209
210 return self.__formated_value
211
212 if format == 'multilist':
213 '''
214 Example:
215 value = (0,1,2),(3,4,5)
216 '''
217 multiList = ast.literal_eval(value)
218
219 if type(multiList[0]) == int:
220 multiList = ast.literal_eval('(' + value + ')')
221
222 self.__formated_value = multiList
223
224 return self.__formated_value
225
226 if format == 'bool':
227 value = int(value)
228
229 if format == 'int':
230 value = float(value)
231
232 format_func = eval(format)
233
234 self.__formated_value = format_func(value)
235
236 return self.__formated_value
237
238 def updateId(self, new_id):
239
240 self.id = str(new_id)
241
242 def setup(self, id, name, value, format='str'):
243 self.id = str(id)
244 self.name = name
245 if format == 'obj':
246 self.value = value
247 else:
248 self.value = str(value)
249 self.format = str.lower(format)
250
251 self.getValue()
252
253 return 1
254
255 def update(self, name, value, format='str'):
256
257 self.name = name
258 self.value = str(value)
259 self.format = format
260
261 def makeXml(self, opElement):
262 if self.name not in ('queue',):
263 parmElement = SubElement(opElement, self.ELEMENTNAME)
264 parmElement.set('id', str(self.id))
265 parmElement.set('name', self.name)
266 parmElement.set('value', self.value)
267 parmElement.set('format', self.format)
268
269 def readXml(self, parmElement):
270
271 self.id = parmElement.get('id')
272 self.name = parmElement.get('name')
273 self.value = parmElement.get('value')
274 self.format = str.lower(parmElement.get('format'))
275
276 # Compatible with old signal chain version
277 if self.format == 'int' and self.name == 'idfigure':
278 self.name = 'id'
279
280 def printattr(self):
281
282 print('Parameter[%s]: name = %s, value = %s, format = %s, project_id = %s' % (self.id, self.name, self.value, self.format, self.project_id))
283
284 class OperationConf():
285
286 ELEMENTNAME = 'Operation'
287
288 def __init__(self):
289
290 self.id = '0'
291 self.name = None
292 self.priority = None
293 self.parameters = {}
294 self.object = None
295 self.operations = []
296
297 def getId(self):
298
299 return self.id
300
301 def getNewId(self):
302
303 return int(self.id) * 10 + len(self.operations) + 1
304
305 def updateId(self, new_id):
306
307 self.id = str(new_id)
308
309 n = 1
310 for conf in self.operations:
311 conf_id = str(int(new_id) * 10 + n)
312 conf.updateId(conf_id)
313 n += 1
314
315 def getKwargs(self):
316
317 params = {}
318
319 for key, value in self.parameters.items():
320 if value not in (None, '', ' '):
321 params[key] = value
322
323 return params
324
325 def update(self, **kwargs):
326
327 for key, value in kwargs.items():
328 self.addParameter(name=key, value=value)
329
330 def addParameter(self, name, value, format=None):
331 '''
332 '''
333
334 if isinstance(value, str) and re.search(r'(\d+/\d+/\d+)', value):
335 self.parameters[name] = datetime.date(*[int(x) for x in value.split('/')])
336 elif isinstance(value, str) and re.search(r'(\d+:\d+:\d+)', value):
337 self.parameters[name] = datetime.time(*[int(x) for x in value.split(':')])
338 else:
339 try:
340 self.parameters[name] = ast.literal_eval(value)
341 except:
342 if isinstance(value, str) and ',' in value:
343 self.parameters[name] = value.split(',')
344 else:
345 self.parameters[name] = value
346
347 def getParameters(self):
348
349 params = {}
350 for key, value in self.parameters.items():
351 s = type(value).__name__
352 if s == 'date':
353 params[key] = value.strftime('%Y/%m/%d')
354 elif s == 'time':
355 params[key] = value.strftime('%H:%M:%S')
356 else:
357 params[key] = str(value)
358
359 return params
360
361 def makeXml(self, element):
362
363 xml = SubElement(element, self.ELEMENTNAME)
364 for label in self.xml_labels:
365 xml.set(label, str(getattr(self, label)))
366
367 for key, value in self.getParameters().items():
368 xml_param = SubElement(xml, 'Parameter')
369 xml_param.set('name', key)
370 xml_param.set('value', value)
371
372 for conf in self.operations:
373 conf.makeXml(xml)
374
375 def __str__(self):
376
377 if self.ELEMENTNAME == 'Operation':
378 s = ' {}[id={}]\n'.format(self.name, self.id)
379 else:
380 s = '{}[id={}, inputId={}]\n'.format(self.name, self.id, self.inputId)
381
382 for key, value in self.parameters.items():
383 if self.ELEMENTNAME == 'Operation':
384 s += ' {}: {}\n'.format(key, value)
385 else:
386 s += ' {}: {}\n'.format(key, value)
387
388 for conf in self.operations:
389 s += str(conf)
390
391 return s
392
393 class OperationConf(ConfBase):
394
395 ELEMENTNAME = 'Operation'
396 xml_labels = ['id', 'name']
397
398 def setup(self, id, name, priority, project_id, err_queue):
399
400 self.id = str(id)
401 self.project_id = project_id
402 self.name = name
403 self.type = 'other'
404 self.err_queue = err_queue
405
406 def readXml(self, element, project_id, err_queue):
407
408 self.id = element.get('id')
409 self.name = element.get('name')
410 self.type = 'other'
411 self.project_id = str(project_id)
412 self.err_queue = err_queue
413
414 for elm in element.iter('Parameter'):
415 self.addParameter(elm.get('name'), elm.get('value'))
416
417 def createObject(self):
418
419 className = eval(self.name)
420
421 if 'Plot' in self.name or 'Writer' in self.name or 'Send' in self.name or 'print' in self.name:
422 kwargs = self.getKwargs()
423 opObj = className(self.id, self.id, self.project_id, self.err_queue, **kwargs)
424 opObj.start()
425 self.type = 'external'
426 else:
427 opObj = className()
428
429 self.object = opObj
430 return opObj
431
432 class ProcUnitConf(ConfBase):
433
434 ELEMENTNAME = 'ProcUnit'
435 xml_labels = ['id', 'inputId', 'name']
436
437 def setup(self, project_id, id, name, datatype, inputId, err_queue):
438 '''
439 '''
440
441 if datatype == None and name == None:
442 raise ValueError('datatype or name should be defined')
443
444 if name == None:
445 if 'Proc' in datatype:
446 name = datatype
447 else:
448 name = '%sProc' % (datatype)
449
450 if datatype == None:
451 datatype = name.replace('Proc', '')
452
453 self.id = str(id)
454 self.project_id = project_id
455 self.name = name
456 self.datatype = datatype
457 self.inputId = inputId
458 self.err_queue = err_queue
459 self.operations = []
460 self.parameters = {}
461
462 def removeOperation(self, id):
463
464 i = [1 if x.id==id else 0 for x in self.operations]
465 self.operations.pop(i.index(1))
466
467 def getOperation(self, id):
468
469 for conf in self.operations:
470 if conf.id == id:
471 return conf
472
473 def addOperation(self, name, optype='self'):
474 '''
475 '''
476
477 id = self.getNewId()
478 conf = OperationConf()
479 conf.setup(id, name=name, priority='0', project_id=self.project_id, err_queue=self.err_queue)
480 self.operations.append(conf)
481
482 return conf
483
484 def readXml(self, element, project_id, err_queue):
485
486 self.id = element.get('id')
487 self.name = element.get('name')
488 self.inputId = None if element.get('inputId') == 'None' else element.get('inputId')
489 self.datatype = element.get('datatype', self.name.replace(self.ELEMENTNAME.replace('Unit', ''), ''))
490 self.project_id = str(project_id)
491 self.err_queue = err_queue
492 self.operations = []
493 self.parameters = {}
494
495 for elm in element:
496 if elm.tag == 'Parameter':
497 self.addParameter(elm.get('name'), elm.get('value'))
498 elif elm.tag == 'Operation':
499 conf = OperationConf()
500 conf.readXml(elm, project_id, err_queue)
501 self.operations.append(conf)
502
503 def createObjects(self):
504 '''
505 Instancia de unidades de procesamiento.
506 '''
507
508 className = eval(self.name)
509 #print(self.name)
510 kwargs = self.getKwargs()
511 procUnitObj = className()
512 procUnitObj.name = self.name
513 log.success('creating process...', self.name)
514
515 for conf in self.operations:
516
517 opObj = conf.createObject()
518
519 log.success('adding operation: {}, type:{}'.format(
520 conf.name,
521 conf.type), self.name)
522
523 procUnitObj.addOperation(conf, opObj)
524
525 self.object = procUnitObj
526
527 def run(self):
528 '''
529 '''
530
531 return self.object.call(**self.getKwargs())
532
533
534 class ReadUnitConf(ProcUnitConf):
535
536 ELEMENTNAME = 'ReadUnit'
537
538 def __init__(self):
539
540 self.id = None
541 self.datatype = None
542 self.name = None
543 self.inputId = None
544 self.operations = []
545 self.parameters = {}
546
547 def setup(self, project_id, id, name, datatype, err_queue, path='', startDate='', endDate='',
548 startTime='', endTime='', server=None, **kwargs):
549
550 if datatype == None and name == None:
551 raise ValueError('datatype or name should be defined')
552 if name == None:
553 if 'Reader' in datatype:
554 name = datatype
555 datatype = name.replace('Reader','')
556 else:
557 name = '{}Reader'.format(datatype)
558 if datatype == None:
559 if 'Reader' in name:
560 datatype = name.replace('Reader','')
561 else:
562 datatype = name
563 name = '{}Reader'.format(name)
564
565 self.id = id
566 self.project_id = project_id
567 self.name = name
568 self.datatype = datatype
569 self.err_queue = err_queue
570
571 self.addParameter(name='path', value=path)
572 self.addParameter(name='startDate', value=startDate)
573 self.addParameter(name='endDate', value=endDate)
574 self.addParameter(name='startTime', value=startTime)
575 self.addParameter(name='endTime', value=endTime)
576
577 for key, value in kwargs.items():
578 self.addParameter(name=key, value=value)
579
580
581 class Project(Process):
582 """API to create signal chain projects"""
583
584 ELEMENTNAME = 'Project'
585
586 def __init__(self, name=''):
587
588 Process.__init__(self)
589 self.id = '1'
590 if name:
591 self.name = '{} ({})'.format(Process.__name__, name)
592 self.filename = None
593 self.description = None
594 self.email = None
595 self.alarm = []
596 self.configurations = {}
597 # self.err_queue = Queue()
598 self.err_queue = None
599 self.started = False
600
601 def getNewId(self):
602
603 idList = list(self.configurations.keys())
604 id = int(self.id) * 10
605
606 while True:
607 id += 1
608
609 if str(id) in idList:
610 continue
611
612 break
613
614 return str(id)
615
616 def updateId(self, new_id):
617
618 self.id = str(new_id)
619
620 keyList = list(self.configurations.keys())
621 keyList.sort()
622
623 n = 1
624 new_confs = {}
625
626 for procKey in keyList:
627
628 conf = self.configurations[procKey]
629 idProcUnit = str(int(self.id) * 10 + n)
630 conf.updateId(idProcUnit)
631 new_confs[idProcUnit] = conf
632 n += 1
633
634 self.configurations = new_confs
635
636 def setup(self, id=1, name='', description='', email=None, alarm=[]):
637
638 self.id = str(id)
639 self.description = description
640 self.email = email
641 self.alarm = alarm
642 if name:
643 self.name = '{} ({})'.format(Process.__name__, name)
644
645 def update(self, **kwargs):
646
647 for key, value in kwargs.items():
648 setattr(self, key, value)
649
650 def clone(self):
651
652 p = Project()
653 p.id = self.id
654 p.name = self.name
655 p.description = self.description
656 p.configurations = self.configurations.copy()
657
658 return p
659
660 def addReadUnit(self, id=None, datatype=None, name=None, **kwargs):
661
662 '''
663 '''
664
665 if id is None:
666 idReadUnit = self.getNewId()
667 else:
668 idReadUnit = str(id)
669
670 conf = ReadUnitConf()
671 conf.setup(self.id, idReadUnit, name, datatype, self.err_queue, **kwargs)
672 self.configurations[conf.id] = conf
673
674 return conf
675
676 def addProcUnit(self, id=None, inputId='0', datatype=None, name=None):
677
678 '''
679 '''
680
681 if id is None:
682 idProcUnit = self.getNewId()
683 else:
684 idProcUnit = id
685
686 conf = ProcUnitConf()
687 conf.setup(self.id, idProcUnit, name, datatype, inputId, self.err_queue)
688 self.configurations[conf.id] = conf
689
690 return conf
691
692 def removeProcUnit(self, id):
693
694 if id in self.configurations:
695 self.configurations.pop(id)
696
697 def getReadUnit(self):
698
699 for obj in list(self.configurations.values()):
700 if obj.ELEMENTNAME == 'ReadUnit':
701 return obj
702
703 return None
704
705 def getProcUnit(self, id):
706
707 return self.configurations[id]
708
709 def getUnits(self):
710
711 keys = list(self.configurations)
712 keys.sort()
713
714 for key in keys:
715 yield self.configurations[key]
716
717 def updateUnit(self, id, **kwargs):
718
719 conf = self.configurations[id].update(**kwargs)
720
721 def makeXml(self):
722
723 xml = Element('Project')
724 xml.set('id', str(self.id))
725 xml.set('name', self.name)
726 xml.set('description', self.description)
727
728 for conf in self.configurations.values():
729 conf.makeXml(xml)
730
731 self.xml = xml
732
733 def writeXml(self, filename=None):
734
735 if filename == None:
736 if self.filename:
737 filename = self.filename
738 else:
739 filename = 'schain.xml'
740
741 if not filename:
742 print('filename has not been defined. Use setFilename(filename) for do it.')
743 return 0
744
745 abs_file = os.path.abspath(filename)
746
747 if not os.access(os.path.dirname(abs_file), os.W_OK):
748 print('No write permission on %s' % os.path.dirname(abs_file))
749 return 0
750
751 if os.path.isfile(abs_file) and not(os.access(abs_file, os.W_OK)):
752 print('File %s already exists and it could not be overwriten' % abs_file)
753 return 0
754
755 self.makeXml()
756
757 ElementTree(self.xml).write(abs_file, method='xml')
758
759 self.filename = abs_file
760
761 return 1
762
763 def readXml(self, filename):
764
765 abs_file = os.path.abspath(filename)
766
767 self.configurations = {}
768
769 try:
770 self.xml = ElementTree().parse(abs_file)
771 except:
772 log.error('Error reading %s, verify file format' % filename)
773 return 0
774
775 self.id = self.xml.get('id')
776 self.name = self.xml.get('name')
777 self.description = self.xml.get('description')
778
779 for element in self.xml:
780 if element.tag == 'ReadUnit':
781 conf = ReadUnitConf()
782 conf.readXml(element, self.id, self.err_queue)
783 self.configurations[conf.id] = conf
784 elif element.tag == 'ProcUnit':
785 conf = ProcUnitConf()
786 input_proc = self.configurations[element.get('inputId')]
787 conf.readXml(element, self.id, self.err_queue)
788 self.configurations[conf.id] = conf
789
790 self.filename = abs_file
791
792 return 1
793
794 def __str__(self):
795
796 text = '\nProject[id=%s, name=%s, description=%s]\n\n' % (
797 self.id,
798 self.name,
799 self.description,
800 )
801
802 for conf in self.configurations.values():
803 text += '{}'.format(conf)
804
805 return text
806
807 def createObjects(self):
808
809 keys = list(self.configurations.keys())
810 keys.sort()
811 for key in keys:
812 conf = self.configurations[key]
813 conf.createObjects()
814 if conf.inputId is not None:
815 conf.object.setInput(self.configurations[conf.inputId].object)
816
817 def monitor(self):
818
819 t = Thread(target=self._monitor, args=(self.err_queue, self.ctx))
820 t.start()
821
822 def _monitor(self, queue, ctx):
823
824 import socket
825
826 procs = 0
827 err_msg = ''
828
829 while True:
830 msg = queue.get()
831 if '#_start_#' in msg:
832 procs += 1
833 elif '#_end_#' in msg:
834 procs -=1
835 else:
836 err_msg = msg
837
838 if procs == 0 or 'Traceback' in err_msg:
839 break
840 time.sleep(0.1)
841
842 if '|' in err_msg:
843 name, err = err_msg.split('|')
844 if 'SchainWarning' in err:
845 log.warning(err.split('SchainWarning:')[-1].split('\n')[0].strip(), name)
846 elif 'SchainError' in err:
847 log.error(err.split('SchainError:')[-1].split('\n')[0].strip(), name)
848 else:
849 log.error(err, name)
850 else:
851 name, err = self.name, err_msg
852
853 time.sleep(1)
854
855 ctx.term()
856
857 message = ''.join(err)
858
859 if err_msg:
860 subject = 'SChain v%s: Error running %s\n' % (
861 schainpy.__version__, self.name)
862
863 subtitle = 'Hostname: %s\n' % socket.gethostbyname(
864 socket.gethostname())
865 subtitle += 'Working directory: %s\n' % os.path.abspath('./')
866 subtitle += 'Configuration file: %s\n' % self.filename
867 subtitle += 'Time: %s\n' % str(datetime.datetime.now())
868
869 readUnitConfObj = self.getReadUnit()
870 if readUnitConfObj:
871 subtitle += '\nInput parameters:\n'
872 subtitle += '[Data path = %s]\n' % readUnitConfObj.parameters['path']
873 subtitle += '[Start date = %s]\n' % readUnitConfObj.parameters['startDate']
874 subtitle += '[End date = %s]\n' % readUnitConfObj.parameters['endDate']
875 subtitle += '[Start time = %s]\n' % readUnitConfObj.parameters['startTime']
876 subtitle += '[End time = %s]\n' % readUnitConfObj.parameters['endTime']
877
878 a = Alarm(
879 modes=self.alarm,
880 email=self.email,
881 message=message,
882 subject=subject,
883 subtitle=subtitle,
884 filename=self.filename
885 )
886
887 a.start()
888
889 def setFilename(self, filename):
890
891 self.filename = filename
892
893 def runProcs(self):
894
895 err = False
896 n = len(self.configurations)
897
898 while not err:
899 for conf in self.getUnits():
900 ok = conf.run()
901 if ok == 'Error':
902 n -= 1
903 continue
904 elif not ok:
905 break
906 if n == 0:
907 err = True
908
909 def run(self):
910
911 log.success('\nStarting Project {} [id={}]'.format(self.name, self.id), tag='')
912 self.started = True
913 self.start_time = time.time()
914 self.createObjects()
915 self.runProcs()
916 log.success('{} Done (Time: {:4.2f}s)'.format(
917 self.name,
918 time.time()-self.start_time), '')
1 # Copyright (c) 2012-2020 Jicamarca Radio Observatory
2 # All rights reserved.
3 #
4 # Distributed under the terms of the BSD 3-clause license.
5 """API to create signal chain projects
6
7 The API is provide through class: Project
8 """
9
10 import re
11 import sys
12 import ast
13 import datetime
14 import traceback
15 import time
16 import multiprocessing
17 from multiprocessing import Process, Queue
18 from threading import Thread
19 from xml.etree.ElementTree import ElementTree, Element, SubElement
20
21 from schainpy.admin import Alarm, SchainWarning
22 from schainpy.model import *
23 from schainpy.utils import log
24
25 if 'darwin' in sys.platform and sys.version_info[0] == 3 and sys.version_info[1] > 7:
26 multiprocessing.set_start_method('fork')
27
28 class ConfBase():
29
30 def __init__(self):
31
32 self.id = '0'
33 self.name = None
34 self.priority = None
35 self.parameters = {}
36 self.object = None
37 self.operations = []
38
39 def getId(self):
40
41 return self.id
42
43 def getNewId(self):
44
45 return int(self.id) * 10 + len(self.operations) + 1
46
47 def updateId(self, new_id):
48
49 self.id = str(new_id)
50
51 n = 1
52 for conf in self.operations:
53 conf_id = str(int(new_id) * 10 + n)
54 conf.updateId(conf_id)
55 n += 1
56
57 def getKwargs(self):
58
59 params = {}
60
61 for key, value in self.parameters.items():
62 if value not in (None, '', ' '):
63 params[key] = value
64
65 return params
66
67 def update(self, **kwargs):
68
69 for key, value in kwargs.items():
70 self.addParameter(name=key, value=value)
71
72 def addParameter(self, name, value, format=None):
73 '''
74 '''
75
76 if isinstance(value, str) and re.search(r'(\d+/\d+/\d+)', value):
77 self.parameters[name] = datetime.date(*[int(x) for x in value.split('/')])
78 elif isinstance(value, str) and re.search(r'(\d+:\d+:\d+)', value):
79 self.parameters[name] = datetime.time(*[int(x) for x in value.split(':')])
80 else:
81 try:
82 self.parameters[name] = ast.literal_eval(value)
83 except:
84 if isinstance(value, str) and ',' in value:
85 self.parameters[name] = value.split(',')
86 else:
87 self.parameters[name] = value
88
89 def getParameters(self):
90
91 params = {}
92 for key, value in self.parameters.items():
93 s = type(value).__name__
94 if s == 'date':
95 params[key] = value.strftime('%Y/%m/%d')
96 elif s == 'time':
97 params[key] = value.strftime('%H:%M:%S')
98 else:
99 params[key] = str(value)
100
101 return params
102
103 def makeXml(self, element):
104
105 xml = SubElement(element, self.ELEMENTNAME)
106 for label in self.xml_labels:
107 xml.set(label, str(getattr(self, label)))
108
109 for key, value in self.getParameters().items():
110 xml_param = SubElement(xml, 'Parameter')
111 xml_param.set('name', key)
112 xml_param.set('value', value)
113
114 for conf in self.operations:
115 conf.makeXml(xml)
116
117 def __str__(self):
118
119 if self.ELEMENTNAME == 'Operation':
120 s = ' {}[id={}]\n'.format(self.name, self.id)
121 else:
122 s = '{}[id={}, inputId={}]\n'.format(self.name, self.id, self.inputId)
123
124 for key, value in self.parameters.items():
125 if self.ELEMENTNAME == 'Operation':
126 s += ' {}: {}\n'.format(key, value)
127 else:
128 s += ' {}: {}\n'.format(key, value)
129
130 for conf in self.operations:
131 s += str(conf)
132
133 return s
134
135 class OperationConf(ConfBase):
136
137 ELEMENTNAME = 'Operation'
138 xml_labels = ['id', 'name']
139
140 def setup(self, id, name, priority, project_id, err_queue):
141
142 self.id = str(id)
143 self.project_id = project_id
144 self.name = name
145 self.type = 'other'
146 self.err_queue = err_queue
147
148 def readXml(self, element, project_id, err_queue):
149
150 self.id = element.get('id')
151 self.name = element.get('name')
152 self.type = 'other'
153 self.project_id = str(project_id)
154 self.err_queue = err_queue
155
156 for elm in element.iter('Parameter'):
157 self.addParameter(elm.get('name'), elm.get('value'))
158
159 def createObject(self):
160
161 className = eval(self.name)
162
163 if 'Plot' in self.name or 'Writer' in self.name or 'Send' in self.name or 'print' in self.name:
164 kwargs = self.getKwargs()
165 opObj = className(self.id, self.id, self.project_id, self.err_queue, **kwargs)
166 opObj.start()
167 self.type = 'external'
168 else:
169 opObj = className()
170
171 self.object = opObj
172 return opObj
173
174 class ProcUnitConf(ConfBase):
175
176 ELEMENTNAME = 'ProcUnit'
177 xml_labels = ['id', 'inputId', 'name']
178
179 def setup(self, project_id, id, name, datatype, inputId, err_queue):
180 '''
181 '''
182
183 if datatype == None and name == None:
184 raise ValueError('datatype or name should be defined')
185
186 if name == None:
187 if 'Proc' in datatype:
188 name = datatype
189 else:
190 name = '%sProc' % (datatype)
191
192 if datatype == None:
193 datatype = name.replace('Proc', '')
194
195 self.id = str(id)
196 self.project_id = project_id
197 self.name = name
198 self.datatype = datatype
199 self.inputId = inputId
200 self.err_queue = err_queue
201 self.operations = []
202 self.parameters = {}
203
204 def removeOperation(self, id):
205
206 i = [1 if x.id==id else 0 for x in self.operations]
207 self.operations.pop(i.index(1))
208
209 def getOperation(self, id):
210
211 for conf in self.operations:
212 if conf.id == id:
213 return conf
214
215 def addOperation(self, name, optype='self'):
216 '''
217 '''
218
219 id = self.getNewId()
220 conf = OperationConf()
221 conf.setup(id, name=name, priority='0', project_id=self.project_id, err_queue=self.err_queue)
222 self.operations.append(conf)
223
224 return conf
225
226 def readXml(self, element, project_id, err_queue):
227
228 self.id = element.get('id')
229 self.name = element.get('name')
230 self.inputId = None if element.get('inputId') == 'None' else element.get('inputId')
231 self.datatype = element.get('datatype', self.name.replace(self.ELEMENTNAME.replace('Unit', ''), ''))
232 self.project_id = str(project_id)
233 self.err_queue = err_queue
234 self.operations = []
235 self.parameters = {}
236
237 for elm in element:
238 if elm.tag == 'Parameter':
239 self.addParameter(elm.get('name'), elm.get('value'))
240 elif elm.tag == 'Operation':
241 conf = OperationConf()
242 conf.readXml(elm, project_id, err_queue)
243 self.operations.append(conf)
244
245 def createObjects(self):
246 '''
247 Instancia de unidades de procesamiento.
248 '''
249
250 className = eval(self.name)
251 kwargs = self.getKwargs()
252 procUnitObj = className()
253 procUnitObj.name = self.name
254 log.success('creating process...', self.name)
255
256 for conf in self.operations:
257
258 opObj = conf.createObject()
259
260 log.success('adding operation: {}, type:{}'.format(
261 conf.name,
262 conf.type), self.name)
263
264 procUnitObj.addOperation(conf, opObj)
265
266 self.object = procUnitObj
267
268 def run(self):
269 '''
270 '''
271
272 return self.object.call(**self.getKwargs())
273
274
275 class ReadUnitConf(ProcUnitConf):
276
277 ELEMENTNAME = 'ReadUnit'
278
279 def __init__(self):
280
281 self.id = None
282 self.datatype = None
283 self.name = None
284 self.inputId = None
285 self.operations = []
286 self.parameters = {}
287
288 def setup(self, project_id, id, name, datatype, err_queue, path='', startDate='', endDate='',
289 startTime='', endTime='', server=None, **kwargs):
290
291 if datatype == None and name == None:
292 raise ValueError('datatype or name should be defined')
293 if name == None:
294 if 'Reader' in datatype:
295 name = datatype
296 datatype = name.replace('Reader','')
297 else:
298 name = '{}Reader'.format(datatype)
299 if datatype == None:
300 if 'Reader' in name:
301 datatype = name.replace('Reader','')
302 else:
303 datatype = name
304 name = '{}Reader'.format(name)
305
306 self.id = id
307 self.project_id = project_id
308 self.name = name
309 self.datatype = datatype
310 self.err_queue = err_queue
311
312 self.addParameter(name='path', value=path)
313 self.addParameter(name='startDate', value=startDate)
314 self.addParameter(name='endDate', value=endDate)
315 self.addParameter(name='startTime', value=startTime)
316 self.addParameter(name='endTime', value=endTime)
317
318 for key, value in kwargs.items():
319 self.addParameter(name=key, value=value)
320
321
322 class Project(Process):
323 """API to create signal chain projects"""
324
325 ELEMENTNAME = 'Project'
326
327 def __init__(self, name=''):
328
329 Process.__init__(self)
330 self.id = '1'
331 if name:
332 self.name = '{} ({})'.format(Process.__name__, name)
333 self.filename = None
334 self.description = None
335 self.email = None
336 self.alarm = []
337 self.configurations = {}
338 # self.err_queue = Queue()
339 self.err_queue = None
340 self.started = False
341
342 def getNewId(self):
343
344 idList = list(self.configurations.keys())
345 id = int(self.id) * 10
346
347 while True:
348 id += 1
349
350 if str(id) in idList:
351 continue
352
353 break
354
355 return str(id)
356
357 def updateId(self, new_id):
358
359 self.id = str(new_id)
360
361 keyList = list(self.configurations.keys())
362 keyList.sort()
363
364 n = 1
365 new_confs = {}
366
367 for procKey in keyList:
368
369 conf = self.configurations[procKey]
370 idProcUnit = str(int(self.id) * 10 + n)
371 conf.updateId(idProcUnit)
372 new_confs[idProcUnit] = conf
373 n += 1
374
375 self.configurations = new_confs
376
377 def setup(self, id=1, name='', description='', email=None, alarm=[]):
378
379 self.id = str(id)
380 self.description = description
381 self.email = email
382 self.alarm = alarm
383 if name:
384 self.name = '{} ({})'.format(Process.__name__, name)
385
386 def update(self, **kwargs):
387
388 for key, value in kwargs.items():
389 setattr(self, key, value)
390
391 def clone(self):
392
393 p = Project()
394 p.id = self.id
395 p.name = self.name
396 p.description = self.description
397 p.configurations = self.configurations.copy()
398
399 return p
400
401 def addReadUnit(self, id=None, datatype=None, name=None, **kwargs):
402
403 '''
404 '''
405
406 if id is None:
407 idReadUnit = self.getNewId()
408 else:
409 idReadUnit = str(id)
410
411 conf = ReadUnitConf()
412 conf.setup(self.id, idReadUnit, name, datatype, self.err_queue, **kwargs)
413 self.configurations[conf.id] = conf
414
415 return conf
416
417 def addProcUnit(self, id=None, inputId='0', datatype=None, name=None):
418
419 '''
420 '''
421
422 if id is None:
423 idProcUnit = self.getNewId()
424 else:
425 idProcUnit = id
426
427 conf = ProcUnitConf()
428 conf.setup(self.id, idProcUnit, name, datatype, inputId, self.err_queue)
429 self.configurations[conf.id] = conf
430
431 return conf
432
433 def removeProcUnit(self, id):
434
435 if id in self.configurations:
436 self.configurations.pop(id)
437
438 def getReadUnit(self):
439
440 for obj in list(self.configurations.values()):
441 if obj.ELEMENTNAME == 'ReadUnit':
442 return obj
443
444 return None
445
446 def getProcUnit(self, id):
447
448 return self.configurations[id]
449
450 def getUnits(self):
451
452 keys = list(self.configurations)
453 keys.sort()
454
455 for key in keys:
456 yield self.configurations[key]
457
458 def updateUnit(self, id, **kwargs):
459
460 conf = self.configurations[id].update(**kwargs)
461
462 def makeXml(self):
463
464 xml = Element('Project')
465 xml.set('id', str(self.id))
466 xml.set('name', self.name)
467 xml.set('description', self.description)
468
469 for conf in self.configurations.values():
470 conf.makeXml(xml)
471
472 self.xml = xml
473
474 def writeXml(self, filename=None):
475
476 if filename == None:
477 if self.filename:
478 filename = self.filename
479 else:
480 filename = 'schain.xml'
481
482 if not filename:
483 print('filename has not been defined. Use setFilename(filename) for do it.')
484 return 0
485
486 abs_file = os.path.abspath(filename)
487
488 if not os.access(os.path.dirname(abs_file), os.W_OK):
489 print('No write permission on %s' % os.path.dirname(abs_file))
490 return 0
491
492 if os.path.isfile(abs_file) and not(os.access(abs_file, os.W_OK)):
493 print('File %s already exists and it could not be overwriten' % abs_file)
494 return 0
495
496 self.makeXml()
497
498 ElementTree(self.xml).write(abs_file, method='xml')
499
500 self.filename = abs_file
501
502 return 1
503
504 def readXml(self, filename):
505
506 abs_file = os.path.abspath(filename)
507
508 self.configurations = {}
509
510 try:
511 self.xml = ElementTree().parse(abs_file)
512 except:
513 log.error('Error reading %s, verify file format' % filename)
514 return 0
515
516 self.id = self.xml.get('id')
517 self.name = self.xml.get('name')
518 self.description = self.xml.get('description')
519
520 for element in self.xml:
521 if element.tag == 'ReadUnit':
522 conf = ReadUnitConf()
523 conf.readXml(element, self.id, self.err_queue)
524 self.configurations[conf.id] = conf
525 elif element.tag == 'ProcUnit':
526 conf = ProcUnitConf()
527 input_proc = self.configurations[element.get('inputId')]
528 conf.readXml(element, self.id, self.err_queue)
529 self.configurations[conf.id] = conf
530
531 self.filename = abs_file
532
533 return 1
534
535 def __str__(self):
536
537 text = '\nProject[id=%s, name=%s, description=%s]\n\n' % (
538 self.id,
539 self.name,
540 self.description,
541 )
542
543 for conf in self.configurations.values():
544 text += '{}'.format(conf)
545
546 return text
547
548 def createObjects(self):
549
550 keys = list(self.configurations.keys())
551 keys.sort()
552 for key in keys:
553 conf = self.configurations[key]
554 conf.createObjects()
555 if conf.inputId is not None:
556 conf.object.setInput(self.configurations[conf.inputId].object)
557
558 def monitor(self):
559
560 t = Thread(target=self._monitor, args=(self.err_queue, self.ctx))
561 t.start()
562
563 def _monitor(self, queue, ctx):
564
565 import socket
566
567 procs = 0
568 err_msg = ''
569
570 while True:
571 msg = queue.get()
572 if '#_start_#' in msg:
573 procs += 1
574 elif '#_end_#' in msg:
575 procs -=1
576 else:
577 err_msg = msg
578
579 if procs == 0 or 'Traceback' in err_msg:
580 break
581 time.sleep(0.1)
582
583 if '|' in err_msg:
584 name, err = err_msg.split('|')
585 if 'SchainWarning' in err:
586 log.warning(err.split('SchainWarning:')[-1].split('\n')[0].strip(), name)
587 elif 'SchainError' in err:
588 log.error(err.split('SchainError:')[-1].split('\n')[0].strip(), name)
589 else:
590 log.error(err, name)
591 else:
592 name, err = self.name, err_msg
593
594 time.sleep(1)
595
596 ctx.term()
597
598 message = ''.join(err)
599
600 if err_msg:
601 subject = 'SChain v%s: Error running %s\n' % (
602 schainpy.__version__, self.name)
603
604 subtitle = 'Hostname: %s\n' % socket.gethostbyname(
605 socket.gethostname())
606 subtitle += 'Working directory: %s\n' % os.path.abspath('./')
607 subtitle += 'Configuration file: %s\n' % self.filename
608 subtitle += 'Time: %s\n' % str(datetime.datetime.now())
609
610 readUnitConfObj = self.getReadUnit()
611 if readUnitConfObj:
612 subtitle += '\nInput parameters:\n'
613 subtitle += '[Data path = %s]\n' % readUnitConfObj.parameters['path']
614 subtitle += '[Start date = %s]\n' % readUnitConfObj.parameters['startDate']
615 subtitle += '[End date = %s]\n' % readUnitConfObj.parameters['endDate']
616 subtitle += '[Start time = %s]\n' % readUnitConfObj.parameters['startTime']
617 subtitle += '[End time = %s]\n' % readUnitConfObj.parameters['endTime']
618
619 a = Alarm(
620 modes=self.alarm,
621 email=self.email,
622 message=message,
623 subject=subject,
624 subtitle=subtitle,
625 filename=self.filename
626 )
627
628 a.start()
629
630 def setFilename(self, filename):
631
632 self.filename = filename
633
634 def runProcs(self):
635
636 err = False
637 n = len(self.configurations)
638
639 while not err:
640 for conf in self.getUnits():
641 ok = conf.run()
642 if ok == 'Error':
643 n -= 1
644 continue
645 elif not ok:
646 break
647 if n == 0:
648 err = True
649
650 def run(self):
651
652 log.success('\nStarting Project {} [id={}]'.format(self.name, self.id), tag='')
653 self.started = True
654 self.start_time = time.time()
655 self.createObjects()
656 self.runProcs()
657 log.success('{} Done (Time: {:4.2f}s)'.format(
658 self.name,
659 time.time()-self.start_time), '')
Show file before | Show file after | Hide comments
@@ -1,656 +1,658
1 1 '''
2 2 Created on Set 9, 2015
3 3
4 4 @author: roj-idl71 Karim Kuyeng
5 5 '''
6 6
7 7 import os
8 8 import sys
9 9 import glob
10 10 import fnmatch
11 11 import datetime
12 12 import time
13 13 import re
14 14 import h5py
15 15 import numpy
16 16
17 17 try:
18 18 from gevent import sleep
19 19 except:
20 20 from time import sleep
21 21
22 22 from schainpy.model.data.jroheaderIO import RadarControllerHeader, SystemHeader
23 23 from schainpy.model.data.jrodata import Voltage
24 24 from schainpy.model.proc.jroproc_base import ProcessingUnit, Operation, MPDecorator
25 25 from numpy import imag
26 26
27 @MPDecorator
27
28 28 class AMISRReader(ProcessingUnit):
29 29 '''
30 30 classdocs
31 31 '''
32 32
33 33 def __init__(self):
34 34 '''
35 35 Constructor
36 36 '''
37 37
38 38 ProcessingUnit.__init__(self)
39 39
40 40 self.set = None
41 41 self.subset = None
42 42 self.extension_file = '.h5'
43 43 self.dtc_str = 'dtc'
44 44 self.dtc_id = 0
45 45 self.status = True
46 46 self.isConfig = False
47 47 self.dirnameList = []
48 48 self.filenameList = []
49 49 self.fileIndex = None
50 50 self.flagNoMoreFiles = False
51 51 self.flagIsNewFile = 0
52 52 self.filename = ''
53 53 self.amisrFilePointer = None
54 54 self.realBeamCode = []
55 55
56 56 #self.dataset = None
57 57
58 58
59 59
60 60
61 61 self.profileIndex = 0
62 62
63 63
64 64 self.beamCodeByFrame = None
65 65 self.radacTimeByFrame = None
66 66
67 67 self.dataset = None
68 68
69 69
70 70
71 71
72 72 self.__firstFile = True
73 73
74 74 self.buffer = None
75 75
76 76
77 77 self.timezone = 'ut'
78 78
79 79 self.__waitForNewFile = 20
80 80 self.__filename_online = None
81 81 #Is really necessary create the output object in the initializer
82 82 self.dataOut = Voltage()
83 83 self.dataOut.error=False
84 84
85
85 86 def setup(self,path=None,
86 87 startDate=None,
87 88 endDate=None,
88 89 startTime=None,
89 90 endTime=None,
90 91 walk=True,
91 92 timezone='ut',
92 93 all=0,
93 94 code = None,
94 95 nCode = 0,
95 96 nBaud = 0,
96 97 online=False):
97 98
98 #print ("T",path)
99
99 100
100 101 self.timezone = timezone
101 102 self.all = all
102 103 self.online = online
103 104
104 105 self.code = code
105 106 self.nCode = int(nCode)
106 107 self.nBaud = int(nBaud)
107 108
108 109
109 110
110 111 #self.findFiles()
111 112 if not(online):
112 113 #Busqueda de archivos offline
113 114 self.searchFilesOffLine(path, startDate, endDate, startTime, endTime, walk)
114 115 else:
115 116 self.searchFilesOnLine(path, startDate, endDate, startTime,endTime,walk)
116 117
117 118 if not(self.filenameList):
118 119 print("There is no files into the folder: %s"%(path))
119 120 sys.exit(-1)
120 121
121 122 self.fileIndex = -1
122 123
123 124 self.readNextFile(online)
124 125
125 126 '''
126 127 Add code
127 128 '''
128 129 self.isConfig = True
129
130 # print("Setup Done")
130 131 pass
131 132
132 133
133 134 def readAMISRHeader(self,fp):
134 135 header = 'Raw11/Data/RadacHeader'
135 136 self.beamCodeByPulse = fp.get(header+'/BeamCode') # LIST OF BEAMS PER PROFILE, TO BE USED ON REARRANGE
136 self.beamcodeFile = fp['Setup/Beamcodefile'][()].decode()
137 self.trueBeams = self.beamcodeFile.split("\n")
138 self.trueBeams.pop()#remove last
139 [self.realBeamCode.append(x) for x in self.trueBeams if x not in self.realBeamCode]
140 self.beamCode = [int(x, 16) for x in self.realBeamCode]
137 if (self.startDate> datetime.date(2021, 7, 15)): #Se cambió la forma de extracción de Apuntes el 17
138 self.beamcodeFile = fp['Setup/Beamcodefile'][()].decode()
139 self.trueBeams = self.beamcodeFile.split("\n")
140 self.trueBeams.pop()#remove last
141 [self.realBeamCode.append(x) for x in self.trueBeams if x not in self.realBeamCode]
142 self.beamCode = [int(x, 16) for x in self.realBeamCode]
143 else:
144 _beamCode= fp.get('Raw11/Data/Beamcodes') #se usa la manera previa al cambio de apuntes
145 self.beamCode = _beamCode[0,:]
146
141 147
142 #self.beamCode = fp.get('Raw11/Data/Beamcodes') # NUMBER OF CHANNELS AND IDENTIFY POSITION TO CREATE A FILE WITH THAT INFO
143 148 #self.code = fp.get(header+'/Code') # NOT USE FOR THIS
144 149 self.frameCount = fp.get(header+'/FrameCount')# NOT USE FOR THIS
145 150 self.modeGroup = fp.get(header+'/ModeGroup')# NOT USE FOR THIS
146 151 self.nsamplesPulse = fp.get(header+'/NSamplesPulse')# TO GET NSA OR USING DATA FOR THAT
147 152 self.pulseCount = fp.get(header+'/PulseCount')# NOT USE FOR THIS
148 153 self.radacTime = fp.get(header+'/RadacTime')# 1st TIME ON FILE ANDE CALCULATE THE REST WITH IPP*nindexprofile
149 154 self.timeCount = fp.get(header+'/TimeCount')# NOT USE FOR THIS
150 155 self.timeStatus = fp.get(header+'/TimeStatus')# NOT USE FOR THIS
151 156 self.rangeFromFile = fp.get('Raw11/Data/Samples/Range')
152 157 self.frequency = fp.get('Rx/Frequency')
153 158 txAus = fp.get('Raw11/Data/Pulsewidth')
154 159
155 160
156 161 self.nblocks = self.pulseCount.shape[0] #nblocks
157 162
158 163 self.nprofiles = self.pulseCount.shape[1] #nprofile
159 164 self.nsa = self.nsamplesPulse[0,0] #ngates
160 165 self.nchannels = len(self.beamCode)
161 166 self.ippSeconds = (self.radacTime[0][1] -self.radacTime[0][0]) #Ipp in seconds
162 167 #self.__waitForNewFile = self.nblocks # wait depending on the number of blocks since each block is 1 sec
163 168 self.__waitForNewFile = self.nblocks * self.nprofiles * self.ippSeconds # wait until new file is created
164 169
165 170 #filling radar controller header parameters
166 171 self.__ippKm = self.ippSeconds *.15*1e6 # in km
167 172 self.__txA = (txAus.value)*.15 #(ipp[us]*.15km/1us) in km
168 173 self.__txB = 0
169 174 nWindows=1
170 175 self.__nSamples = self.nsa
171 176 self.__firstHeight = self.rangeFromFile[0][0]/1000 #in km
172 177 self.__deltaHeight = (self.rangeFromFile[0][1] - self.rangeFromFile[0][0])/1000
173 178
174 179 #for now until understand why the code saved is different (code included even though code not in tuf file)
175 180 #self.__codeType = 0
176 181 # self.__nCode = None
177 182 # self.__nBaud = None
178 183 self.__code = self.code
179 184 self.__codeType = 0
180 185 if self.code != None:
181 186 self.__codeType = 1
182 187 self.__nCode = self.nCode
183 188 self.__nBaud = self.nBaud
184 189 #self.__code = 0
185 190
186 191 #filling system header parameters
187 192 self.__nSamples = self.nsa
188 193 self.newProfiles = self.nprofiles/self.nchannels
189 194 self.__channelList = list(range(self.nchannels))
190 195
191 196 self.__frequency = self.frequency[0][0]
192 197
193 198
194 199
195 200 def createBuffers(self):
196 201
197 202 pass
198 203
199 204 def __setParameters(self,path='', startDate='',endDate='',startTime='', endTime='', walk=''):
200 205 self.path = path
201 206 self.startDate = startDate
202 207 self.endDate = endDate
203 208 self.startTime = startTime
204 209 self.endTime = endTime
205 210 self.walk = walk
206 211
207 212 def __checkPath(self):
208 213 if os.path.exists(self.path):
209 214 self.status = 1
210 215 else:
211 216 self.status = 0
212 217 print('Path:%s does not exists'%self.path)
213 218
214 219 return
215 220
216 221
217 222 def __selDates(self, amisr_dirname_format):
218 223 try:
219 224 year = int(amisr_dirname_format[0:4])
220 225 month = int(amisr_dirname_format[4:6])
221 226 dom = int(amisr_dirname_format[6:8])
222 227 thisDate = datetime.date(year,month,dom)
223 228
224 229 if (thisDate>=self.startDate and thisDate <= self.endDate):
225 230 return amisr_dirname_format
226 231 except:
227 232 return None
228 233
229 234
230 235 def __findDataForDates(self,online=False):
231 236
232 237 if not(self.status):
233 238 return None
234 239
235 240 pat = '\d+.\d+'
236 241 dirnameList = [re.search(pat,x) for x in os.listdir(self.path)]
237 242 dirnameList = [x for x in dirnameList if x!=None]
238 243 dirnameList = [x.string for x in dirnameList]
239 244 if not(online):
240 245 dirnameList = [self.__selDates(x) for x in dirnameList]
241 246 dirnameList = [x for x in dirnameList if x!=None]
242 247 if len(dirnameList)>0:
243 248 self.status = 1
244 249 self.dirnameList = dirnameList
245 250 self.dirnameList.sort()
246 251 else:
247 252 self.status = 0
248 253 return None
249 254
250 255 def __getTimeFromData(self):
251 256 startDateTime_Reader = datetime.datetime.combine(self.startDate,self.startTime)
252 257 endDateTime_Reader = datetime.datetime.combine(self.endDate,self.endTime)
253 258
254 259 print('Filtering Files from %s to %s'%(startDateTime_Reader, endDateTime_Reader))
255 260 print('........................................')
256 261 filter_filenameList = []
257 262 self.filenameList.sort()
258 263 #for i in range(len(self.filenameList)-1):
259 264 for i in range(len(self.filenameList)):
260 265 filename = self.filenameList[i]
261 266 fp = h5py.File(filename,'r')
262 267 time_str = fp.get('Time/RadacTimeString')
263 268
264 269 startDateTimeStr_File = time_str[0][0].decode('UTF-8').split('.')[0]
265 270 #startDateTimeStr_File = "2019-12-16 09:21:11"
266 271 junk = time.strptime(startDateTimeStr_File, '%Y-%m-%d %H:%M:%S')
267 272 startDateTime_File = datetime.datetime(junk.tm_year,junk.tm_mon,junk.tm_mday,junk.tm_hour, junk.tm_min, junk.tm_sec)
268 273
269 274 #endDateTimeStr_File = "2019-12-16 11:10:11"
270 275 endDateTimeStr_File = time_str[-1][-1].decode('UTF-8').split('.')[0]
271 276 junk = time.strptime(endDateTimeStr_File, '%Y-%m-%d %H:%M:%S')
272 277 endDateTime_File = datetime.datetime(junk.tm_year,junk.tm_mon,junk.tm_mday,junk.tm_hour, junk.tm_min, junk.tm_sec)
273 278
274 279 fp.close()
275 280
276 281 #print("check time", startDateTime_File)
277 282 if self.timezone == 'lt':
278 283 startDateTime_File = startDateTime_File - datetime.timedelta(minutes = 300)
279 284 endDateTime_File = endDateTime_File - datetime.timedelta(minutes = 300)
280 285 if (endDateTime_File>=startDateTime_Reader and endDateTime_File<endDateTime_Reader):
281 286 #self.filenameList.remove(filename)
282 287 filter_filenameList.append(filename)
283 288
284 289 if (endDateTime_File>=endDateTime_Reader):
285 290 break
286 291
287 292
288 293 filter_filenameList.sort()
289 294 self.filenameList = filter_filenameList
290 295 return 1
291 296
292 297 def __filterByGlob1(self, dirName):
293 298 filter_files = glob.glob1(dirName, '*.*%s'%self.extension_file)
294 299 filter_files.sort()
295 300 filterDict = {}
296 301 filterDict.setdefault(dirName)
297 302 filterDict[dirName] = filter_files
298 303 return filterDict
299 304
300 305 def __getFilenameList(self, fileListInKeys, dirList):
301 306 for value in fileListInKeys:
302 307 dirName = list(value.keys())[0]
303 308 for file in value[dirName]:
304 309 filename = os.path.join(dirName, file)
305 310 self.filenameList.append(filename)
306 311
307 312
308 313 def __selectDataForTimes(self, online=False):
309 314 #aun no esta implementado el filtro for tiempo
310 315 if not(self.status):
311 316 return None
312 317
313 318 dirList = [os.path.join(self.path,x) for x in self.dirnameList]
314 319
315 320 fileListInKeys = [self.__filterByGlob1(x) for x in dirList]
316 321
317 322 self.__getFilenameList(fileListInKeys, dirList)
318 323 if not(online):
319 324 #filtro por tiempo
320 325 if not(self.all):
321 326 self.__getTimeFromData()
322 327
323 328 if len(self.filenameList)>0:
324 329 self.status = 1
325 330 self.filenameList.sort()
326 331 else:
327 332 self.status = 0
328 333 return None
329 334
330 335 else:
331 336 #get the last file - 1
332 337 self.filenameList = [self.filenameList[-2]]
333 338 new_dirnameList = []
334 339 for dirname in self.dirnameList:
335 340 junk = numpy.array([dirname in x for x in self.filenameList])
336 341 junk_sum = junk.sum()
337 342 if junk_sum > 0:
338 343 new_dirnameList.append(dirname)
339 344 self.dirnameList = new_dirnameList
340 345 return 1
341 346
342 347 def searchFilesOnLine(self, path, startDate, endDate, startTime=datetime.time(0,0,0),
343 348 endTime=datetime.time(23,59,59),walk=True):
344 349
345 350 if endDate ==None:
346 351 startDate = datetime.datetime.utcnow().date()
347 352 endDate = datetime.datetime.utcnow().date()
348 353
349 354 self.__setParameters(path=path, startDate=startDate, endDate=endDate,startTime = startTime,endTime=endTime, walk=walk)
350 355
351 356 self.__checkPath()
352 357
353 358 self.__findDataForDates(online=True)
354 359
355 360 self.dirnameList = [self.dirnameList[-1]]
356 361
357 362 self.__selectDataForTimes(online=True)
358 363
359 364 return
360 365
361 366
362 367 def searchFilesOffLine(self,
363 368 path,
364 369 startDate,
365 370 endDate,
366 371 startTime=datetime.time(0,0,0),
367 372 endTime=datetime.time(23,59,59),
368 373 walk=True):
369 374
370 375 self.__setParameters(path, startDate, endDate, startTime, endTime, walk)
371 376
372 377 self.__checkPath()
373 378
374 379 self.__findDataForDates()
375 380
376 381 self.__selectDataForTimes()
377 382
378 383 for i in range(len(self.filenameList)):
379 384 print("%s" %(self.filenameList[i]))
380 385
381 386 return
382 387
383 388 def __setNextFileOffline(self):
384 389 idFile = self.fileIndex
385 390
386 391 while (True):
387 392 idFile += 1
388 393 if not(idFile < len(self.filenameList)):
389 394 self.flagNoMoreFiles = 1
390 395 print("No more Files")
391 396 return 0
392 397
393 398 filename = self.filenameList[idFile]
394 399
395 400 amisrFilePointer = h5py.File(filename,'r')
396 401
397 402 break
398 403
399 404 self.flagIsNewFile = 1
400 405 self.fileIndex = idFile
401 406 self.filename = filename
402 407
403 408 self.amisrFilePointer = amisrFilePointer
404 409
405 410 print("Setting the file: %s"%self.filename)
406 411
407 412 return 1
408 413
409 414
410 415 def __setNextFileOnline(self):
411 416 filename = self.filenameList[0]
412 417 if self.__filename_online != None:
413 418 self.__selectDataForTimes(online=True)
414 419 filename = self.filenameList[0]
415 420 wait = 0
416 421 #self.__waitForNewFile=5 ## DEBUG:
417 422 while self.__filename_online == filename:
418 423 print('waiting %d seconds to get a new file...'%(self.__waitForNewFile))
419 424 if wait == 5:
420 425 self.flagNoMoreFiles = 1
421 426 return 0
422 427 sleep(self.__waitForNewFile)
423 428 self.__selectDataForTimes(online=True)
424 429 filename = self.filenameList[0]
425 430 wait += 1
426 431
427 432 self.__filename_online = filename
428 433
429 434 self.amisrFilePointer = h5py.File(filename,'r')
430 435 self.flagIsNewFile = 1
431 436 self.filename = filename
432 437 print("Setting the file: %s"%self.filename)
433 438 return 1
434 439
435 440
436 441 def readData(self):
437 442 buffer = self.amisrFilePointer.get('Raw11/Data/Samples/Data')
438 443 re = buffer[:,:,:,0]
439 444 im = buffer[:,:,:,1]
440 445 dataset = re + im*1j
441 446
442 447 self.radacTime = self.amisrFilePointer.get('Raw11/Data/RadacHeader/RadacTime')
443 448 timeset = self.radacTime[:,0]
444 449
445 450 return dataset,timeset
446 451
447 452 def reshapeData(self):
448 453 #self.beamCodeByPulse, self.beamCode, self.nblocks, self.nprofiles, self.nsa,
449 454 channels = self.beamCodeByPulse[0,:]
450 455 nchan = self.nchannels
451 456 #self.newProfiles = self.nprofiles/nchan #must be defined on filljroheader
452 457 nblocks = self.nblocks
453 458 nsamples = self.nsa
454 459
455 460 #Dimensions : nChannels, nProfiles, nSamples
456 461 new_block = numpy.empty((nblocks, nchan, numpy.int_(self.newProfiles), nsamples), dtype="complex64")
457 462 ############################################
458 463
459 464 for thisChannel in range(nchan):
460 465 new_block[:,thisChannel,:,:] = self.dataset[:,numpy.where(channels==self.beamCode[thisChannel])[0],:]
461 466
462 467
463 468 new_block = numpy.transpose(new_block, (1,0,2,3))
464 469 new_block = numpy.reshape(new_block, (nchan,-1, nsamples))
465 470
466 471 return new_block
467 472
468 473 def updateIndexes(self):
469 474
470 475 pass
471 476
472 477 def fillJROHeader(self):
473 478
474 479 #fill radar controller header
475 480 self.dataOut.radarControllerHeaderObj = RadarControllerHeader(ipp=self.__ippKm,
476 481 txA=self.__txA,
477 482 txB=0,
478 483 nWindows=1,
479 484 nHeights=self.__nSamples,
480 485 firstHeight=self.__firstHeight,
481 486 deltaHeight=self.__deltaHeight,
482 487 codeType=self.__codeType,
483 488 nCode=self.__nCode, nBaud=self.__nBaud,
484 489 code = self.__code,
485 490 fClock=1)
486 491
487 492 #fill system header
488 493 self.dataOut.systemHeaderObj = SystemHeader(nSamples=self.__nSamples,
489 494 nProfiles=self.newProfiles,
490 495 nChannels=len(self.__channelList),
491 496 adcResolution=14,
492 497 pciDioBusWidth=32)
493 498
494 499 self.dataOut.type = "Voltage"
495 500
496 501 self.dataOut.data = None
497 502
498 503 self.dataOut.dtype = numpy.dtype([('real','<i8'),('imag','<i8')])
499 504
500 505 # self.dataOut.nChannels = 0
501 506
502 507 # self.dataOut.nHeights = 0
503 508
504 509 self.dataOut.nProfiles = self.newProfiles*self.nblocks
505 510
506 511 #self.dataOut.heightList = self.__firstHeigth + numpy.arange(self.__nSamples, dtype = numpy.float)*self.__deltaHeigth
507 512 ranges = numpy.reshape(self.rangeFromFile.value,(-1))
508 513 self.dataOut.heightList = ranges/1000.0 #km
509 514
510 515
511 516 self.dataOut.channelList = self.__channelList
512 517
513 518 self.dataOut.blocksize = self.dataOut.nChannels * self.dataOut.nHeights
514 519
515 520 # self.dataOut.channelIndexList = None
516 521
517 522 self.dataOut.flagNoData = True
518 523
519 524 #Set to TRUE if the data is discontinuous
520 525 self.dataOut.flagDiscontinuousBlock = False
521 526
522 527 self.dataOut.utctime = None
523 528
524 529 #self.dataOut.timeZone = -5 #self.__timezone/60 #timezone like jroheader, difference in minutes between UTC and localtime
525 530 if self.timezone == 'lt':
526 531 self.dataOut.timeZone = time.timezone / 60. #get the timezone in minutes
527 532 else:
528 533 self.dataOut.timeZone = 0 #by default time is UTC
529 534
530 535 self.dataOut.dstFlag = 0
531 536
532 537 self.dataOut.errorCount = 0
533 538
534 539 self.dataOut.nCohInt = 1
535 540
536 541 self.dataOut.flagDecodeData = False #asumo que la data esta decodificada
537 542
538 543 self.dataOut.flagDeflipData = False #asumo que la data esta sin flip
539 544
540 545 self.dataOut.flagShiftFFT = False
541 546
542 547 self.dataOut.ippSeconds = self.ippSeconds
543 548
544 549 #Time interval between profiles
545 550 #self.dataOut.timeInterval = self.dataOut.ippSeconds * self.dataOut.nCohInt
546 551
547 552 self.dataOut.frequency = self.__frequency
548 553 self.dataOut.realtime = self.online
549 554 pass
550 555
551 556 def readNextFile(self,online=False):
552 557
553 558 if not(online):
554 559 newFile = self.__setNextFileOffline()
555 560 else:
556 561 newFile = self.__setNextFileOnline()
557 562
558 563 if not(newFile):
559 564 self.dataOut.error = True
560 565 return 0
561 566 #if self.__firstFile:
562 567 self.readAMISRHeader(self.amisrFilePointer)
563 568
564 569 self.createBuffers()
565 570
566 571 self.fillJROHeader()
567 572
568 573 #self.__firstFile = False
569 574
570 575
571 576
572 577 self.dataset,self.timeset = self.readData()
573 578
574 579 if self.endDate!=None:
575 580 endDateTime_Reader = datetime.datetime.combine(self.endDate,self.endTime)
576 581 time_str = self.amisrFilePointer.get('Time/RadacTimeString')
577 582 startDateTimeStr_File = time_str[0][0].decode('UTF-8').split('.')[0]
578 583 junk = time.strptime(startDateTimeStr_File, '%Y-%m-%d %H:%M:%S')
579 584 startDateTime_File = datetime.datetime(junk.tm_year,junk.tm_mon,junk.tm_mday,junk.tm_hour, junk.tm_min, junk.tm_sec)
580 585 if self.timezone == 'lt':
581 586 startDateTime_File = startDateTime_File - datetime.timedelta(minutes = 300)
582 587 if (startDateTime_File>endDateTime_Reader):
583 588 return 0
584 589
585 590 self.jrodataset = self.reshapeData()
586 591 #----self.updateIndexes()
587 592 self.profileIndex = 0
588 593
589 594 return 1
590 595
591 596
592 597 def __hasNotDataInBuffer(self):
593 598 if self.profileIndex >= (self.newProfiles*self.nblocks):
594 599 return 1
595 600 return 0
596 601
597 602
598 603 def getData(self):
599 604
600 605 if self.flagNoMoreFiles:
601 606 self.dataOut.flagNoData = True
602 607 return 0
603 608
604 609 if self.__hasNotDataInBuffer():
605 610 if not (self.readNextFile(self.online)):
606 611 return 0
607 612
608 613
609 614 if self.dataset is None: # setear esta condicion cuando no hayan datos por leer
610 615 self.dataOut.flagNoData = True
611 616 return 0
612 617
613 618 #self.dataOut.data = numpy.reshape(self.jrodataset[self.profileIndex,:],(1,-1))
614 619
615 620 self.dataOut.data = self.jrodataset[:,self.profileIndex,:]
616 621
617 622 #print("R_t",self.timeset)
618 623
619 624 #self.dataOut.utctime = self.jrotimeset[self.profileIndex]
620 625 #verificar basic header de jro data y ver si es compatible con este valor
621 626 #self.dataOut.utctime = self.timeset + (self.profileIndex * self.ippSeconds * self.nchannels)
622 627 indexprof = numpy.mod(self.profileIndex, self.newProfiles)
623 628 indexblock = self.profileIndex/self.newProfiles
624 629 #print (indexblock, indexprof)
625 630 diffUTC = 1.8e4 #UTC diference from peru in seconds --Joab
626 631 diffUTC = 0
627 632 t_comp = (indexprof * self.ippSeconds * self.nchannels) + diffUTC #
628 #cambio posible 18/02/2020
629
630
631 633
632 634 #print("utc :",indexblock," __ ",t_comp)
633 635 #print(numpy.shape(self.timeset))
634 636 self.dataOut.utctime = self.timeset[numpy.int_(indexblock)] + t_comp
635 637 #self.dataOut.utctime = self.timeset[self.profileIndex] + t_comp
636 638 #print(self.dataOut.utctime)
637 639 self.dataOut.profileIndex = self.profileIndex
638 640 self.dataOut.flagNoData = False
639 641 # if indexprof == 0:
640 642 # print self.dataOut.utctime
641 643
642 644 self.profileIndex += 1
643 645
644 646 return self.dataOut.data
645 647
646 648
647 649 def run(self, **kwargs):
648 650 '''
649 651 This method will be called many times so here you should put all your code
650 652 '''
651
653 #print("running kamisr")
652 654 if not self.isConfig:
653 655 self.setup(**kwargs)
654 656 self.isConfig = True
655 657
656 658 self.getData()
Show file before | Show file after | Hide comments
@@ -1,527 +1,527
1 1 '''
2 2 Created on Jul 2, 2014
3 3
4 4 @author: roj-idl71
5 5 '''
6 6 import numpy
7 7
8 8 from schainpy.model.io.jroIO_base import LOCALTIME, JRODataReader, JRODataWriter
9 9 from schainpy.model.proc.jroproc_base import ProcessingUnit, Operation, MPDecorator
10 10 from schainpy.model.data.jroheaderIO import PROCFLAG, BasicHeader, SystemHeader, RadarControllerHeader, ProcessingHeader
11 11 from schainpy.model.data.jrodata import Spectra
12 12 from schainpy.utils import log
13 13
14 14
15 15 class SpectraReader(JRODataReader, ProcessingUnit):
16 16 """
17 17 Esta clase permite leer datos de espectros desde archivos procesados (.pdata). La lectura
18 18 de los datos siempre se realiza por bloques. Los datos leidos (array de 3 dimensiones)
19 19 son almacenados en tres buffer's para el Self Spectra, el Cross Spectra y el DC Channel.
20 20
21 21 paresCanalesIguales * alturas * perfiles (Self Spectra)
22 22 paresCanalesDiferentes * alturas * perfiles (Cross Spectra)
23 23 canales * alturas (DC Channels)
24 24
25 25 Esta clase contiene instancias (objetos) de las clases BasicHeader, SystemHeader,
26 26 RadarControllerHeader y Spectra. Los tres primeros se usan para almacenar informacion de la
27 27 cabecera de datos (metadata), y el cuarto (Spectra) para obtener y almacenar un bloque de
28 28 datos desde el "buffer" cada vez que se ejecute el metodo "getData".
29 29
30 30 Example:
31 31 dpath = "/home/myuser/data"
32 32
33 33 startTime = datetime.datetime(2010,1,20,0,0,0,0,0,0)
34 34
35 35 endTime = datetime.datetime(2010,1,21,23,59,59,0,0,0)
36 36
37 37 readerObj = SpectraReader()
38 38
39 39 readerObj.setup(dpath, startTime, endTime)
40 40
41 41 while(True):
42 42
43 43 readerObj.getData()
44 44
45 45 print readerObj.data_spc
46 46
47 47 print readerObj.data_cspc
48 48
49 49 print readerObj.data_dc
50 50
51 51 if readerObj.flagNoMoreFiles:
52 52 break
53 53
54 54 """
55 55
56 56 def __init__(self):#, **kwargs):
57 57 """
58 58 Inicializador de la clase SpectraReader para la lectura de datos de espectros.
59 59
60 60 Inputs:
61 61 dataOut : Objeto de la clase Spectra. Este objeto sera utilizado para
62 62 almacenar un perfil de datos cada vez que se haga un requerimiento
63 63 (getData). El perfil sera obtenido a partir del buffer de datos,
64 64 si el buffer esta vacio se hara un nuevo proceso de lectura de un
65 65 bloque de datos.
66 66 Si este parametro no es pasado se creara uno internamente.
67 67
68 68 Affected:
69 69 self.dataOut
70 70
71 71 Return : None
72 72 """
73 73
74 74 ProcessingUnit.__init__(self)
75 75
76 76 self.pts2read_SelfSpectra = 0
77 77 self.pts2read_CrossSpectra = 0
78 self.pts2read_DCchannels = 0
78 self.pts2read_DCchannels = 0
79 79 self.ext = ".pdata"
80 80 self.optchar = "P"
81 81 self.basicHeaderObj = BasicHeader(LOCALTIME)
82 82 self.systemHeaderObj = SystemHeader()
83 83 self.radarControllerHeaderObj = RadarControllerHeader()
84 84 self.processingHeaderObj = ProcessingHeader()
85 85 self.lastUTTime = 0
86 86 self.maxTimeStep = 30
87 87 self.dataOut = Spectra()
88 88 self.profileIndex = 1
89 89 self.nRdChannels = None
90 90 self.nRdPairs = None
91 91 self.rdPairList = []
92 92
93 93 def createObjByDefault(self):
94 94
95 95 dataObj = Spectra()
96 96
97 97 return dataObj
98 98
99 99 def __hasNotDataInBuffer(self):
100 100 return 1
101 101
102 102
103 103 def getBlockDimension(self):
104 104 """
105 105 Obtiene la cantidad de puntos a leer por cada bloque de datos
106 106
107 107 Affected:
108 108 self.nRdChannels
109 109 self.nRdPairs
110 110 self.pts2read_SelfSpectra
111 111 self.pts2read_CrossSpectra
112 112 self.pts2read_DCchannels
113 113 self.blocksize
114 114 self.dataOut.nChannels
115 115 self.dataOut.nPairs
116 116
117 117 Return:
118 118 None
119 119 """
120 120 self.nRdChannels = 0
121 121 self.nRdPairs = 0
122 122 self.rdPairList = []
123 123
124 124 for i in range(0, self.processingHeaderObj.totalSpectra*2, 2):
125 125 if self.processingHeaderObj.spectraComb[i] == self.processingHeaderObj.spectraComb[i+1]:
126 126 self.nRdChannels = self.nRdChannels + 1 #par de canales iguales
127 127 else:
128 128 self.nRdPairs = self.nRdPairs + 1 #par de canales diferentes
129 129 self.rdPairList.append((self.processingHeaderObj.spectraComb[i], self.processingHeaderObj.spectraComb[i+1]))
130 130
131 131 pts2read = self.processingHeaderObj.nHeights * self.processingHeaderObj.profilesPerBlock
132 132
133 133 self.pts2read_SelfSpectra = int(self.nRdChannels * pts2read)
134 134 self.blocksize = self.pts2read_SelfSpectra
135 135
136 136 if self.processingHeaderObj.flag_cspc:
137 137 self.pts2read_CrossSpectra = int(self.nRdPairs * pts2read)
138 138 self.blocksize += self.pts2read_CrossSpectra
139 139
140 140 if self.processingHeaderObj.flag_dc:
141 141 self.pts2read_DCchannels = int(self.systemHeaderObj.nChannels * self.processingHeaderObj.nHeights)
142 142 self.blocksize += self.pts2read_DCchannels
143 143
144 144 def readBlock(self):
145 145 """
146 146 Lee el bloque de datos desde la posicion actual del puntero del archivo
147 147 (self.fp) y actualiza todos los parametros relacionados al bloque de datos
148 148 (metadata + data). La data leida es almacenada en el buffer y el contador del buffer
149 149 es seteado a 0
150 150
151 151 Return: None
152 152
153 153 Variables afectadas:
154 154
155 155 self.flagIsNewFile
156 156 self.flagIsNewBlock
157 157 self.nTotalBlocks
158 158 self.data_spc
159 159 self.data_cspc
160 160 self.data_dc
161 161
162 162 Exceptions:
163 163 Si un bloque leido no es un bloque valido
164 164 """
165
165
166 166 fpointer = self.fp.tell()
167 167
168 168 spc = numpy.fromfile( self.fp, self.dtype[0], self.pts2read_SelfSpectra )
169 169 spc = spc.reshape( (self.nRdChannels, self.processingHeaderObj.nHeights, self.processingHeaderObj.profilesPerBlock) ) #transforma a un arreglo 3D
170 170
171 171 if self.processingHeaderObj.flag_cspc:
172 172 cspc = numpy.fromfile( self.fp, self.dtype, self.pts2read_CrossSpectra )
173 173 cspc = cspc.reshape( (self.nRdPairs, self.processingHeaderObj.nHeights, self.processingHeaderObj.profilesPerBlock) ) #transforma a un arreglo 3D
174 174
175 175 if self.processingHeaderObj.flag_dc:
176 176 dc = numpy.fromfile( self.fp, self.dtype, self.pts2read_DCchannels ) #int(self.processingHeaderObj.nHeights*self.systemHeaderObj.nChannels) )
177 177 dc = dc.reshape( (self.systemHeaderObj.nChannels, self.processingHeaderObj.nHeights) ) #transforma a un arreglo 2D
178 178
179 179 if not self.processingHeaderObj.shif_fft:
180 180 #desplaza a la derecha en el eje 2 determinadas posiciones
181 181 shift = int(self.processingHeaderObj.profilesPerBlock/2)
182 182 spc = numpy.roll( spc, shift , axis=2 )
183 183
184 184 if self.processingHeaderObj.flag_cspc:
185 185 #desplaza a la derecha en el eje 2 determinadas posiciones
186 186 cspc = numpy.roll( cspc, shift, axis=2 )
187 187
188 188 #Dimensions : nChannels, nProfiles, nSamples
189 189 spc = numpy.transpose( spc, (0,2,1) )
190 190 self.data_spc = spc
191 191
192 192 if self.processingHeaderObj.flag_cspc:
193 193 cspc = numpy.transpose( cspc, (0,2,1) )
194 194 self.data_cspc = cspc['real'] + cspc['imag']*1j
195 195 else:
196 196 self.data_cspc = None
197 197
198 198 if self.processingHeaderObj.flag_dc:
199 199 self.data_dc = dc['real'] + dc['imag']*1j
200 200 else:
201 201 self.data_dc = None
202 202
203 203 self.flagIsNewFile = 0
204 204 self.flagIsNewBlock = 1
205 205
206 206 self.nTotalBlocks += 1
207 207 self.nReadBlocks += 1
208 208
209 209 return 1
210 210
211 211 def getFirstHeader(self):
212 212
213 213 self.getBasicHeader()
214 214 self.dataOut.systemHeaderObj = self.systemHeaderObj.copy()
215 215 self.dataOut.radarControllerHeaderObj = self.radarControllerHeaderObj.copy()
216 216 self.dataOut.dtype = self.dtype
217 217 self.dataOut.pairsList = self.rdPairList
218 218 self.dataOut.nProfiles = self.processingHeaderObj.profilesPerBlock
219 219 self.dataOut.nFFTPoints = self.processingHeaderObj.profilesPerBlock
220 220 self.dataOut.nCohInt = self.processingHeaderObj.nCohInt
221 221 self.dataOut.nIncohInt = self.processingHeaderObj.nIncohInt
222 222 xf = self.processingHeaderObj.firstHeight + self.processingHeaderObj.nHeights*self.processingHeaderObj.deltaHeight
223 223 self.dataOut.heightList = numpy.arange(self.processingHeaderObj.firstHeight, xf, self.processingHeaderObj.deltaHeight)
224 224 self.dataOut.channelList = list(range(self.systemHeaderObj.nChannels))
225 225 self.dataOut.flagShiftFFT = True #Data is always shifted
226 226 self.dataOut.flagDecodeData = self.processingHeaderObj.flag_decode #asumo q la data no esta decodificada
227 227 self.dataOut.flagDeflipData = self.processingHeaderObj.flag_deflip #asumo q la data esta sin flip
228 228
229 229 def getData(self):
230 230 """
231 231 First method to execute before "RUN" is called.
232 232
233 233 Copia el buffer de lectura a la clase "Spectra",
234 234 con todos los parametros asociados a este (metadata). cuando no hay datos en el buffer de
235 235 lectura es necesario hacer una nueva lectura de los bloques de datos usando "readNextBlock"
236 236
237 237 Return:
238 238 0 : Si no hay mas archivos disponibles
239 239 1 : Si hizo una buena copia del buffer
240 240
241 241 Affected:
242 242 self.dataOut
243 243 self.flagDiscontinuousBlock
244 244 self.flagIsNewBlock
245 245 """
246 246
247 247 if self.flagNoMoreFiles:
248 248 self.dataOut.flagNoData = True
249 249 return 0
250 250
251 251 self.flagDiscontinuousBlock = 0
252 252 self.flagIsNewBlock = 0
253 253
254 254 if self.__hasNotDataInBuffer():
255 255
256 256 if not( self.readNextBlock() ):
257 257 self.dataOut.flagNoData = True
258 258 return 0
259 259
260 260 #data es un numpy array de 3 dmensiones (perfiles, alturas y canales)
261 261
262 262 if self.data_spc is None:
263 263 self.dataOut.flagNoData = True
264 264 return 0
265 265
266 266 self.getBasicHeader()
267 267 self.getFirstHeader()
268 268 self.dataOut.data_spc = self.data_spc
269 269 self.dataOut.data_cspc = self.data_cspc
270 270 self.dataOut.data_dc = self.data_dc
271 271 self.dataOut.flagNoData = False
272 272 self.dataOut.realtime = self.online
273 273
274 274 return self.dataOut.data_spc
275 275
276 276
277 277 @MPDecorator
278 278 class SpectraWriter(JRODataWriter, Operation):
279 279
280 280 """
281 281 Esta clase permite escribir datos de espectros a archivos procesados (.pdata). La escritura
282 282 de los datos siempre se realiza por bloques.
283 283 """
284 284
285 285 def __init__(self):
286 286 """
287 287 Inicializador de la clase SpectraWriter para la escritura de datos de espectros.
288 288
289 289 Affected:
290 290 self.dataOut
291 291 self.basicHeaderObj
292 292 self.systemHeaderObj
293 293 self.radarControllerHeaderObj
294 294 self.processingHeaderObj
295 295
296 296 Return: None
297 297 """
298 298
299 299 Operation.__init__(self)
300 300
301 301 self.ext = ".pdata"
302 302 self.optchar = "P"
303 303 self.shape_spc_Buffer = None
304 304 self.shape_cspc_Buffer = None
305 305 self.shape_dc_Buffer = None
306 306 self.data_spc = None
307 307 self.data_cspc = None
308 308 self.data_dc = None
309 309 self.setFile = None
310 310 self.noMoreFiles = 0
311 311 self.basicHeaderObj = BasicHeader(LOCALTIME)
312 312 self.systemHeaderObj = SystemHeader()
313 313 self.radarControllerHeaderObj = RadarControllerHeader()
314 314 self.processingHeaderObj = ProcessingHeader()
315 315
316 316 def hasAllDataInBuffer(self):
317 317 return 1
318 318
319 319
320 320 def setBlockDimension(self):
321 321 """
322 322 Obtiene las formas dimensionales del los subbloques de datos que componen un bloque
323 323
324 324 Affected:
325 325 self.shape_spc_Buffer
326 326 self.shape_cspc_Buffer
327 327 self.shape_dc_Buffer
328 328
329 329 Return: None
330 330 """
331 331 self.shape_spc_Buffer = (self.dataOut.nChannels,
332 332 self.processingHeaderObj.nHeights,
333 333 self.processingHeaderObj.profilesPerBlock)
334 334
335 335 self.shape_cspc_Buffer = (self.dataOut.nPairs,
336 336 self.processingHeaderObj.nHeights,
337 337 self.processingHeaderObj.profilesPerBlock)
338 338
339 339 self.shape_dc_Buffer = (self.dataOut.nChannels,
340 340 self.processingHeaderObj.nHeights)
341 341
342 342
343 343 def writeBlock(self):
344 344 """processingHeaderObj
345 345 Escribe el buffer en el file designado
346 346
347 347 Affected:
348 348 self.data_spc
349 349 self.data_cspc
350 350 self.data_dc
351 351 self.flagIsNewFile
352 352 self.flagIsNewBlock
353 353 self.nTotalBlocks
354 354 self.nWriteBlocks
355 355
356 356 Return: None
357 357 """
358 358
359 359 spc = numpy.transpose( self.data_spc, (0,2,1) )
360 360 if not self.processingHeaderObj.shif_fft:
361 361 spc = numpy.roll( spc, int(self.processingHeaderObj.profilesPerBlock/2), axis=2 ) #desplaza a la derecha en el eje 2 determinadas posiciones
362 362 data = spc.reshape((-1))
363 363 data = data.astype(self.dtype[0])
364 364 data.tofile(self.fp)
365 365
366 366 if self.data_cspc is not None:
367
367
368 368 cspc = numpy.transpose( self.data_cspc, (0,2,1) )
369 369 data = numpy.zeros( numpy.shape(cspc), self.dtype )
370 370 #print 'data.shape', self.shape_cspc_Buffer
371 371 if not self.processingHeaderObj.shif_fft:
372 372 cspc = numpy.roll( cspc, int(self.processingHeaderObj.profilesPerBlock/2), axis=2 ) #desplaza a la derecha en el eje 2 determinadas posiciones
373 373 data['real'] = cspc.real
374 374 data['imag'] = cspc.imag
375 375 data = data.reshape((-1))
376 376 data.tofile(self.fp)
377 377
378 378 if self.data_dc is not None:
379
379
380 380 dc = self.data_dc
381 381 data = numpy.zeros( numpy.shape(dc), self.dtype )
382 382 data['real'] = dc.real
383 383 data['imag'] = dc.imag
384 384 data = data.reshape((-1))
385 385 data.tofile(self.fp)
386 386
387 387 # self.data_spc.fill(0)
388 388 #
389 389 # if self.data_dc is not None:
390 390 # self.data_dc.fill(0)
391 391 #
392 392 # if self.data_cspc is not None:
393 393 # self.data_cspc.fill(0)
394 394
395 395 self.flagIsNewFile = 0
396 396 self.flagIsNewBlock = 1
397 397 self.nTotalBlocks += 1
398 398 self.nWriteBlocks += 1
399 399 self.blockIndex += 1
400 400
401 401 # print "[Writing] Block = %d04" %self.blockIndex
402 402
403 403 def putData(self):
404 404 """
405 405 Setea un bloque de datos y luego los escribe en un file
406 406
407 407 Affected:
408 408 self.data_spc
409 409 self.data_cspc
410 410 self.data_dc
411 411
412 412 Return:
413 413 0 : Si no hay data o no hay mas files que puedan escribirse
414 414 1 : Si se escribio la data de un bloque en un file
415 415 """
416 416
417 417 if self.dataOut.flagNoData:
418 418 return 0
419 419
420 420 self.flagIsNewBlock = 0
421 421
422 422 if self.dataOut.flagDiscontinuousBlock:
423 423 self.data_spc.fill(0)
424 424 if self.dataOut.data_cspc is not None:
425 425 self.data_cspc.fill(0)
426 426 if self.dataOut.data_dc is not None:
427 427 self.data_dc.fill(0)
428 428 self.setNextFile()
429 429
430 430 if self.flagIsNewFile == 0:
431 431 self.setBasicHeader()
432 432
433 433 self.data_spc = self.dataOut.data_spc.copy()
434 434
435 435 if self.dataOut.data_cspc is not None:
436 436 self.data_cspc = self.dataOut.data_cspc.copy()
437 437
438 438 if self.dataOut.data_dc is not None:
439 439 self.data_dc = self.dataOut.data_dc.copy()
440 440
441 441 # #self.processingHeaderObj.dataBlocksPerFile)
442 442 if self.hasAllDataInBuffer():
443 443 # self.setFirstHeader()
444 444 self.writeNextBlock()
445 445
446 446 def __getBlockSize(self):
447 447 '''
448 448 Este metodos determina el cantidad de bytes para un bloque de datos de tipo Spectra
449 449 '''
450 450
451 451 dtype_width = self.getDtypeWidth()
452 452
453 453 pts2write = self.dataOut.nHeights * self.dataOut.nFFTPoints
454 454
455 455 pts2write_SelfSpectra = int(self.dataOut.nChannels * pts2write)
456 456 blocksize = (pts2write_SelfSpectra*dtype_width)
457 457
458 458 if self.dataOut.data_cspc is not None:
459 459 pts2write_CrossSpectra = int(self.dataOut.nPairs * pts2write)
460 460 blocksize += (pts2write_CrossSpectra*dtype_width*2)
461 461
462 462 if self.dataOut.data_dc is not None:
463 463 pts2write_DCchannels = int(self.dataOut.nChannels * self.dataOut.nHeights)
464 464 blocksize += (pts2write_DCchannels*dtype_width*2)
465 465
466 466 # blocksize = blocksize #* datatypeValue * 2 #CORREGIR ESTO
467 467
468 468 return blocksize
469 469
470 470 def setFirstHeader(self):
471 471
472 472 """
473 473 Obtiene una copia del First Header
474 474
475 475 Affected:
476 476 self.systemHeaderObj
477 477 self.radarControllerHeaderObj
478 478 self.dtype
479 479
480 480 Return:
481 481 None
482 482 """
483 483
484 484 self.systemHeaderObj = self.dataOut.systemHeaderObj.copy()
485 485 self.systemHeaderObj.nChannels = self.dataOut.nChannels
486 486 self.radarControllerHeaderObj = self.dataOut.radarControllerHeaderObj.copy()
487 487
488 488 self.processingHeaderObj.dtype = 1 # Spectra
489 489 self.processingHeaderObj.blockSize = self.__getBlockSize()
490 490 self.processingHeaderObj.profilesPerBlock = self.dataOut.nFFTPoints
491 491 self.processingHeaderObj.dataBlocksPerFile = self.blocksPerFile
492 492 self.processingHeaderObj.nWindows = 1 #podria ser 1 o self.dataOut.processingHeaderObj.nWindows
493 493 self.processingHeaderObj.nCohInt = self.dataOut.nCohInt# Se requiere para determinar el valor de timeInterval
494 494 self.processingHeaderObj.nIncohInt = self.dataOut.nIncohInt
495 495 self.processingHeaderObj.totalSpectra = self.dataOut.nPairs + self.dataOut.nChannels
496 496 self.processingHeaderObj.shif_fft = self.dataOut.flagShiftFFT
497 497
498 498 if self.processingHeaderObj.totalSpectra > 0:
499 499 channelList = []
500 500 for channel in range(self.dataOut.nChannels):
501 501 channelList.append(channel)
502 502 channelList.append(channel)
503 503
504 504 pairsList = []
505 505 if self.dataOut.nPairs > 0:
506 506 for pair in self.dataOut.pairsList:
507 507 pairsList.append(pair[0])
508 508 pairsList.append(pair[1])
509 509
510 510 spectraComb = channelList + pairsList
511 511 spectraComb = numpy.array(spectraComb, dtype="u1")
512 512 self.processingHeaderObj.spectraComb = spectraComb
513 513
514 514 if self.dataOut.code is not None:
515 515 self.processingHeaderObj.code = self.dataOut.code
516 516 self.processingHeaderObj.nCode = self.dataOut.nCode
517 517 self.processingHeaderObj.nBaud = self.dataOut.nBaud
518 518
519 519 if self.processingHeaderObj.nWindows != 0:
520 520 self.processingHeaderObj.firstHeight = self.dataOut.heightList[0]
521 521 self.processingHeaderObj.deltaHeight = self.dataOut.heightList[1] - self.dataOut.heightList[0]
522 522 self.processingHeaderObj.nHeights = self.dataOut.nHeights
523 523 self.processingHeaderObj.samplesWin = self.dataOut.nHeights
524 524
525 525 self.processingHeaderObj.processFlags = self.getProcessFlags()
526 526
527 self.setBasicHeader() No newline at end of file
527 self.setBasicHeader()
Show file before | Show file after | Hide comments
@@ -1,203 +1,203
1 1 '''
2 2 Base clases to create Processing units and operations, the MPDecorator
3 3 must be used in plotting and writing operations to allow to run as an
4 4 external process.
5 5 '''
6 6
7 7 import inspect
8 8 import zmq
9 9 import time
10 10 import pickle
11 11 import traceback
12 12 from threading import Thread
13 13 from multiprocessing import Process, Queue
14 14 from schainpy.utils import log
15 15
16 16
17 17 class ProcessingUnit(object):
18 18 '''
19 19 Base class to create Signal Chain Units
20 20 '''
21 21
22 22 proc_type = 'processing'
23 23
24 24 def __init__(self):
25 25
26 26 self.dataIn = None
27 27 self.dataOut = None
28 28 self.isConfig = False
29 29 self.operations = []
30 30
31 31 def setInput(self, unit):
32 32
33 33 self.dataIn = unit.dataOut
34 34
35 35 def getAllowedArgs(self):
36 36 if hasattr(self, '__attrs__'):
37 37 return self.__attrs__
38 38 else:
39 39 return inspect.getargspec(self.run).args
40 40
41 41 def addOperation(self, conf, operation):
42 42 '''
43 43 '''
44 44
45 45 self.operations.append((operation, conf.type, conf.getKwargs()))
46 46
47 47 def getOperationObj(self, objId):
48 48
49 49 if objId not in list(self.operations.keys()):
50 50 return None
51 51
52 52 return self.operations[objId]
53 53
54 54 def call(self, **kwargs):
55 55 '''
56 56 '''
57 57
58 58 try:
59 59 if self.dataIn is not None and self.dataIn.flagNoData and not self.dataIn.error:
60 60 return self.dataIn.isReady()
61 61 elif self.dataIn is None or not self.dataIn.error:
62 62 self.run(**kwargs)
63 63 elif self.dataIn.error:
64 64 self.dataOut.error = self.dataIn.error
65 65 self.dataOut.flagNoData = True
66 66 except:
67 67 err = traceback.format_exc()
68 68 if 'SchainWarning' in err:
69 69 log.warning(err.split('SchainWarning:')[-1].split('\n')[0].strip(), self.name)
70 70 elif 'SchainError' in err:
71 71 log.error(err.split('SchainError:')[-1].split('\n')[0].strip(), self.name)
72 72 else:
73 73 log.error(err, self.name)
74 74 self.dataOut.error = True
75 75
76 76 for op, optype, opkwargs in self.operations:
77 77 if optype == 'other' and not self.dataOut.flagNoData:
78 78 self.dataOut = op.run(self.dataOut, **opkwargs)
79 79 elif optype == 'external' and not self.dataOut.flagNoData:
80 80 op.queue.put(self.dataOut)
81 81 elif optype == 'external' and self.dataOut.error:
82 82 op.queue.put(self.dataOut)
83 83
84 84 return 'Error' if self.dataOut.error else self.dataOut.isReady()
85 85
86 86 def setup(self):
87 87
88 88 raise NotImplementedError
89 89
90 90 def run(self):
91 91
92 92 raise NotImplementedError
93 93
94 94 def close(self):
95 95
96 96 return
97 97
98 98
99 99 class Operation(object):
100 100
101 101 '''
102 102 '''
103 103
104 104 proc_type = 'operation'
105 105
106 106 def __init__(self):
107 107
108 108 self.id = None
109 109 self.isConfig = False
110 110
111 111 if not hasattr(self, 'name'):
112 112 self.name = self.__class__.__name__
113 113
114 114 def getAllowedArgs(self):
115 115 if hasattr(self, '__attrs__'):
116 116 return self.__attrs__
117 117 else:
118 118 return inspect.getargspec(self.run).args
119 119
120 120 def setup(self):
121 121
122 122 self.isConfig = True
123 123
124 124 raise NotImplementedError
125 125
126 126 def run(self, dataIn, **kwargs):
127 127 """
128 128 Realiza las operaciones necesarias sobre la dataIn.data y actualiza los
129 129 atributos del objeto dataIn.
130 130
131 131 Input:
132 132
133 133 dataIn : objeto del tipo JROData
134 134
135 135 Return:
136 136
137 137 None
138 138
139 139 Affected:
140 140 __buffer : buffer de recepcion de datos.
141 141
142 142 """
143 143 if not self.isConfig:
144 144 self.setup(**kwargs)
145 145
146 146 raise NotImplementedError
147 147
148 148 def close(self):
149 149
150 150 return
151 151
152 152
153 153 def MPDecorator(BaseClass):
154 154 """
155 155 Multiprocessing class decorator
156 156
157 157 This function add multiprocessing features to a BaseClass.
158 158 """
159 159
160 160 class MPClass(BaseClass, Process):
161 161
162 162 def __init__(self, *args, **kwargs):
163 163 super(MPClass, self).__init__()
164 164 Process.__init__(self)
165 165
166 166 self.args = args
167 167 self.kwargs = kwargs
168 168 self.t = time.time()
169 169 self.op_type = 'external'
170 170 self.name = BaseClass.__name__
171 171 self.__doc__ = BaseClass.__doc__
172 172
173 173 if 'plot' in self.name.lower() and not self.name.endswith('_'):
174 174 self.name = '{}{}'.format(self.CODE.upper(), 'Plot')
175 175
176 176 self.start_time = time.time()
177 self.err_queue = args[3]
177 #self.err_queue = args[2]
178 178 self.queue = Queue(maxsize=1)
179 179 self.myrun = BaseClass.run
180 180
181 181 def run(self):
182 182
183 183 while True:
184 184
185 185 dataOut = self.queue.get()
186 186
187 187 if not dataOut.error:
188 188 try:
189 189 BaseClass.run(self, dataOut, **self.kwargs)
190 190 except:
191 191 err = traceback.format_exc()
192 192 log.error(err, self.name)
193 193 else:
194 194 break
195 195
196 196 self.close()
197 197
198 198 def close(self):
199 199
200 200 BaseClass.close(self)
201 201 log.success('Done...(Time:{:4.2f} secs)'.format(time.time()-self.start_time), self.name)
202 202
203 203 return MPClass
Show file before | Show file after | Hide comments
@@ -1,200 +1,203
1 #!/usr/bin/env python
1
2 2 import os, sys
3 3 import time
4 4 import datetime
5 5
6 6 path = os.path.dirname(os.getcwd())
7 7 path = os.path.dirname(path)
8 8 sys.path.insert(0, path)
9 9
10 10 from schainpy.controller import Project
11 11
12 12 def main():
13 13
14 14
15 15 desc = "AMISR EEJ Experiment"
16 16 filename = "amisr_reader.xml"
17 17 xmin = '07'
18 18 xmax = '18' #-> el plot genera +1 en la hora, es decir aparece 18 como máximo
19 19 ymin = '0'
20 20 ymax = '300'
21 21 dbmin = '45' #'60'#'55' #'40' #noise esf eej
22 22 dbmax = '65' #'70' #'55'
23 showSPC = '0' #view plot Spectra
24 showRTI = '0' #view plot RTI
25 showNOISE = '0' #view plot NOISE
26 localtime='0' #para ajustar el horario en las gráficas '0' para dejar en utc
23 showSPC = '1' #view plot Spectra
24 showRTI = '1' #view plot RTI
25 showNOISE = '1' #view plot NOISE
26 localtime='1' #para ajustar el horario en las gráficas '0' para dejar en utc
27 27 code = '1,-1,-1,-1,1,1,1,1,-1,-1,-1,1,-1,-1,-1,1,-1,-1,-1,1,-1,-1,1,-1,1,1,-1,1'
28 28 nCode = '1'
29 29 nBaud = '28'
30 30 nosamp = '1' # oversample for EEJ
31 31 today = time.strftime("%Y/%m/%d")
32 32 #startDate=today
33 33 #endDate=today
34 startDate='2019/12/16'
35 endDate='2019/12/16'
34 startDate='2021/07/11'
35 endDate='2021/07/11'
36 36 #inPath= '/home/soporte/dataAMISR_test/'
37 37 inPath= '/home/soporte/dataAMISR/'
38 inPath= '/media/soporte/UARS_4T_D02/AMISR_DATA/2021/'
38 39 #inPath = '/mnt/data_amisr'
39 40 outPath = '/home/soporte/Data/EEJ'
40 41
41 42 ##.......................................................................................
42 43 ##.......................................................................................
43 44
44 45 #l = startDate.split('/') #adding day of the year to outPath
45 46 l = startDate.split('/')
46 47 datelist = datetime.date(int(l[0]),int(l[1]),int(l[2]))
47 48 DOY = datelist.timetuple().tm_yday
48 49 outPath= outPath+"/EEJ"+l[0]+str(DOY)
49 50 if os.path.exists(outPath):
50 51 print("outPath", outPath)
51 52 else :
52 53 os.mkdir(outPath)
53 54 print("Creating...", outPath)
54 55
55 56 ##.......................................................................................
56 57 ##.......................................................................................
57 58 controllerObj = Project()
58 59 controllerObj.setup(id = '11', name='eej_proc', description=desc)
59 60 ##.......................................................................................
60 61 ##.......................................................................................
61 62 readUnitConfObj = controllerObj.addReadUnit(datatype='AMISRReader',
62 63 path=inPath,
63 64 startDate=startDate,#startDate, #'2014/10/07',
64 65 endDate=endDate, #endDate '2014/10/07',
65 66 startTime='07:01:30',#'07:00:00',
66 67 endTime='19:00:00',#'15:00:00',
67 walk=0,
68 walk=1,
68 69 code = code,
69 70 nCode = nCode,
70 71 nBaud = nBaud,
71 timezone='ut',
72 timezone='lt',
72 73 online=0)
73 74
75
74 76 #AMISR Processing Unit
75 77 ##.......................................................................................
76 78 ##.......................................................................................
77 79 procUnitConfObj0 = controllerObj.addProcUnit(datatype='VoltageProc', inputId=readUnitConfObj.getId())
78 opObj10 = procUnitConfObj0.addOperation(name='setRadarFrequency')
79 opObj10.addParameter(name='frequency', value='445e6', format='float')
80 opObj10 = procUnitConfObj0.addOperation(name='setAttribute')
81 opObj10.addParameter(name='frequency', value='445.09e6')
82 # opObj10 = procUnitConfObj0.addOperation(name='setRadarFrequency')
83 # opObj10.addParameter(name='frequency', value='445e6', format='float')
80 84
81 85
82 86 opObj01 = procUnitConfObj0.addOperation(name='Decoder', optype='other')
83 87 opObj01.addParameter(name='code', value=code, format='floatlist')
84 88 opObj01.addParameter(name='nCode', value=nCode, format='int')
85 89 opObj01.addParameter(name='nBaud', value=nBaud, format='int')
86 90 opObj01.addParameter(name='osamp', value=nosamp, format='int')
87 91
88 92
89 opObj02 = procUnitConfObj0.addOperation(name='CohInt', optype='other')
90 opObj02.addParameter(name='n', value='2', format='int')
93 # opObj02 = procUnitConfObj0.addOperation(name='CohInt', optype='other')
94 # opObj02.addParameter(name='n', value='2', format='int')
91 95
92 96
93 97
94 98 ##.......................................................................................
95 99 ##.......................................................................................
96 100
97 101 procUnitConfObj1 = controllerObj.addProcUnit(datatype='SpectraProc', inputId=procUnitConfObj0.getId())
98 102 procUnitConfObj1.addParameter(name='nFFTPoints', value='16', format='int')
99 103
100 104
101 105 opObj11 = procUnitConfObj1.addOperation(name='IncohInt', optype='other')
102 106 opObj11.addParameter(name='n', value='150', format='int') #300?
103 107
104 108 ## Remove DC signal
105 109 opObj11 = procUnitConfObj1.addOperation(name='removeDC')
106 110 ##.......................................................................................
107 111 ##.......................................................................................
108 112
109 opObj13 = procUnitConfObj1.addOperation(name='getNoise' , optype ='self')
110 opObj13.addParameter(name='minHei', value='100', format='float')
111 opObj13.addParameter(name='maxHei', value='280', format='float')
113 # opObj13 = procUnitConfObj1.addOperation(name='getNoise' , optype ='self')
114 # opObj13.addParameter(name='minHei', value='100', format='float')
115 # opObj13.addParameter(name='maxHei', value='280', format='float')
112 116
113 117
114 118 #
115 119 opObj12 = procUnitConfObj1.addOperation(name='SpectraPlot', optype='external')
116 120 opObj12.addParameter(name='id', value='21', format='int')
117 121 opObj12.addParameter(name= 'xaxis', value='velocity')
118 122 opObj12.addParameter(name='ymax', value=ymax, format='int')
119 123 opObj12.addParameter(name='showprofile', value='1', format='int')
120 124 opObj12.addParameter(name='wintitle', value='AMISR Beam 0', format='str')
121 125 opObj12.addParameter(name='zmin', value=dbmin, format='int')
122 126 opObj12.addParameter(name='zmax', value=dbmax, format='int')
123 127 opObj12.addParameter(name='save', value=outPath+'/plots', format='str')
124 128 opObj12.addParameter(name='colormap', value='jet', format='str')
125 129 opObj12.addParameter(name='localtime', value=localtime,format='int')
126 130 opObj12.addParameter(name='show', value = showSPC, format='int')
127 131
128 132
129 133 ##Generate *.pdata from AMISR data
130 134 ##.......................................................................................
131 135 ##.......................................................................................
132 136 opObj13 = procUnitConfObj1.addOperation(name='SpectraWriter', optype='external')
133 137 opObj13.addParameter(name='path', value=outPath)
134 138 opObj13.addParameter(name='blocksPerFile', value='10', format='int')
135 139
136 140
137 141 opObj14 = procUnitConfObj1.addOperation(name='NoisePlot', optype='external')
138 142 opObj14.addParameter(name='id', value='3', format='int')
139 143 opObj14.addParameter(name='wintitle', value='title0', format='str')
140 144 opObj14.addParameter(name='showprofile', value='0', format='int')
141 opObj14.addParameter(name='xmin', value=xmin, format='int')
142 opObj14.addParameter(name='xmax', value=xmax, format='int')
145 opObj14.addParameter(name='tmin', value=xmin, format='int')
146 opObj14.addParameter(name='tmax', value=xmax, format='int')
143 147 opObj14.addParameter(name='ymin', value=dbmin, format='int')
144 148 opObj14.addParameter(name='ymax', value=dbmax, format='int')
145 149 opObj14.addParameter(name='save', value=outPath, format='str')
146 150 opObj14.addParameter(name='localtime', value=localtime,format='int')
147 151 opObj14.addParameter(name='show', value = showNOISE, format='int')
148 152
149
153 #
150 154 opObj15 = procUnitConfObj1.addOperation(name='RTIPlot', optype='external')
151 155 opObj15.addParameter(name='id', value='2', format='int')
152 156 opObj15.addParameter(name='localtime', value=localtime,format='int')
153 157 opObj15.addParameter(name='wintitle', value='RTI', format='str')
154 opObj15.addParameter(name='xmin', value=xmin, format='int')
155 opObj15.addParameter(name='xmax', value=xmax, format='int') #max value =23
158 opObj15.addParameter(name='tmin', value=xmin, format='int')
159 opObj15.addParameter(name='tmax', value=xmax, format='int') #max value =23
156 160 opObj15.addParameter(name='ymin', value=ymin, format='int')
157 161 opObj15.addParameter(name='zmin', value=dbmin, format='int')
158 162 opObj15.addParameter(name='zmax', value=dbmax, format='int')
159 163 opObj15.addParameter(name='showprofile', value='0', format='int')
160 164 opObj15.addParameter(name='save', value=outPath+'/plots', format='str')
161 165 opObj15.addParameter(name='colormap', value='jet', format='str')
162 166 opObj15.addParameter(name='show', value = showRTI, format='int')
163 167
164 168
165 169
166 170 ##.......................................................................................
167 171 ##.......................................................................................
168 172
169 173 procUnitConfObj2 = controllerObj.addProcUnit(datatype='ParametersProc', inputId=procUnitConfObj1.getId())
170 174 opObj16 = procUnitConfObj2.addOperation(name='SpectralMoments', optype='other')
171 175
172 176
173 #Using ParamWriter::::
177 #Using HDFWriter::::
174 178 ##.......................................................................................
175 179 ##.......................................................................................
176 opObj17 = procUnitConfObj2.addOperation(name='ParamWriter', optype='external')
180 opObj17 = procUnitConfObj2.addOperation(name='HDFWriter', optype='external')
177 181 opObj17.addParameter(name='path', value=outPath)
178 182 opObj17.addParameter(name='blocksPerFile', value='10', format='int')
179 183 opObj17.addParameter(name='metadataList',value='type,inputUnit,heightList',format='list')
180 184 opObj17.addParameter(name='dataList',value='moments,data_SNR,utctime',format='list')
181 opObj17.addParameter(name='mode',value='1',format='int') #'0' channels, '1' parameters, '3' table (for meteors)
182 ##opObj17.addParameter(name='setType', value ='anything', format='str')#no usar
185
183 186
184 187
185 188 ##.......................................................................................
186 189 ##.......................................................................................
187 190 #print("Escribiendo el archivo XML",controllerObj.writeXml(path +'/'+filename))
188 191
189 192 controllerObj.start()
190 193
191 194 #print("Leyendo el archivo XML",controllerObj.readXml(path +'/'+filename))
192 195
193 196 ##.......................................................................................
194 197 ##.......................................................................................
195 198
196 199 if __name__ == '__main__':
197 200 import time
198 201 start_time = time.time()
199 202 main()
200 203 print("--- %s seconds ---" % (time.time() - start_time))
Show file before | Show file after | Hide comments
1 NO CONTENT: file was removed
General Comments 0
You need to be logged in to leave comments. Login now