##// END OF EJS Templates
schain-jc
RSS
Multiprocessing for Spectra (all operation) working
George Yong -
r1171:92232440951e
parent child
Show More
Side by Side Unified
Context file:
r1171:92232440951e Loading diff...:
Show file before | Show file after | Hide comments
@@ -1,23 +1,35
1 1 '''
2 Updated on January , 2018, for multiprocessing purposes
3 Author: Sergio Cortez
2 4 Created on September , 2012
3 @author:
4 5 '''
5
6 from platform import python_version
6 7 import sys
7 8 import ast
8 9 import datetime
9 10 import traceback
10 11 import math
11 12 import time
13 import zmq
12 14 from multiprocessing import Process, cpu_count
13 15
14 16 from xml.etree.ElementTree import ElementTree, Element, SubElement, tostring
15 17 from xml.dom import minidom
16 18
17 import schainpy
19
18 20 from schainpy.admin import Alarm, SchainWarning
19 from schainpy.model import *
21
22 ### Temporary imports!!!
23 # from schainpy.model import *
24 from schainpy.model.io import *
25 from schainpy.model.graphics import *
26 from schainpy.model.proc.jroproc_base import *
27 from schainpy.model.proc.bltrproc_parameters import *
28 from schainpy.model.proc.jroproc_spectra import *
29 from schainpy.model.proc.jroproc_parameters import *
30 from schainpy.model.utils.jroutils_publish import *
20 31 from schainpy.utils import log
32 ###
21 33
22 34 DTYPES = {
23 35 'Voltage': '.r',
@@ -77,7 +89,6 def MPProject(project, n=cpu_count()):
77 89
78 90 time.sleep(3)
79 91
80
81 92 class ParameterConf():
82 93
83 94 id = None
@@ -267,7 +278,6 class ParameterConf():
267 278
268 279 print('Parameter[%s]: name = %s, value = %s, format = %s' % (self.id, self.name, self.value, self.format))
269 280
270
271 281 class OperationConf():
272 282
273 283 id = None
@@ -284,12 +294,15 class OperationConf():
284 294 self.id = '0'
285 295 self.name = None
286 296 self.priority = None
287 self.type = 'self'
297 self.topic = None
288 298
289 299 def __getNewId(self):
290 300
291 301 return int(self.id) * 10 + len(self.parmConfObjList) + 1
292 302
303 def getId(self):
304 return self.id
305
293 306 def updateId(self, new_id):
294 307
295 308 self.id = str(new_id)
@@ -361,7 +374,6 class OperationConf():
361 374 self.name = name
362 375 self.type = type
363 376 self.priority = priority
364
365 377 self.parmConfObjList = []
366 378
367 379 def removeParameters(self):
@@ -443,26 +455,18 class OperationConf():
443 455 for parmConfObj in self.parmConfObjList:
444 456 parmConfObj.printattr()
445 457
446 def createObject(self, plotter_queue=None):
447
448 if self.type == 'self':
449 raise ValueError('This operation type cannot be created')
450
451 if self.type == 'plotter':
452 if not plotter_queue:
453 raise ValueError('plotter_queue is not defined. Use:\nmyProject = Project()\nmyProject.setPlotterQueue(plotter_queue)')
454
455 opObj = Plotter(self.name, plotter_queue)
456
457 if self.type == 'external' or self.type == 'other':
458 def createObject(self):
458 459
459 460 className = eval(self.name)
460 461 kwargs = self.getKwargs()
461 462
462 opObj = className(**kwargs)
463 opObj = className(self.id, **kwargs)
463 464
464 return opObj
465 opObj.start()
465 466
467 print(' Operation created')
468
469 return opObj
466 470
467 471 class ProcUnitConf():
468 472
@@ -508,13 +512,13 class ProcUnitConf():
508 512 return self.id
509 513
510 514 def updateId(self, new_id, parentId=parentId):
511
515 '''
512 516 new_id = int(parentId) * 10 + (int(self.id) % 10)
513 517 new_inputId = int(parentId) * 10 + (int(self.inputId) % 10)
514 518
515 519 # If this proc unit has not inputs
516 if self.inputId == '0':
517 new_inputId = 0
520 #if self.inputId == '0':
521 #new_inputId = 0
518 522
519 523 n = 1
520 524 for opConfObj in self.opConfObjList:
@@ -526,8 +530,9 class ProcUnitConf():
526 530
527 531 self.parentId = str(parentId)
528 532 self.id = str(new_id)
529 self.inputId = str(new_inputId)
530
533 #self.inputId = str(new_inputId)
534 '''
535 n = 1
531 536 def getInputId(self):
532 537
533 538 return self.inputId
@@ -560,11 +565,17 class ProcUnitConf():
560 565 return self.procUnitObj
561 566
562 567 def setup(self, id, name, datatype, inputId, parentId=None):
568 '''
569 id sera el topico a publicar
570 inputId sera el topico a subscribirse
571 '''
563 572
564 573 # Compatible with old signal chain version
565 574 if datatype == None and name == None:
566 575 raise ValueError('datatype or name should be defined')
567 576
577 #Definir una condicion para inputId cuando sea 0
578
568 579 if name == None:
569 580 if 'Proc' in datatype:
570 581 name = datatype
@@ -579,7 +590,6 class ProcUnitConf():
579 590 self.datatype = datatype
580 591 self.inputId = inputId
581 592 self.parentId = parentId
582
583 593 self.opConfObjList = []
584 594
585 595 self.addOperation(name='run', optype='self')
@@ -603,9 +613,15 class ProcUnitConf():
603 613 return opObj
604 614
605 615 def addOperation(self, name, optype='self'):
616 '''
617 Actualizacion - > proceso comunicacion
618 En el caso de optype='self', elminar. DEfinir comuncacion IPC -> Topic
619 definir el tipoc de socket o comunicacion ipc++
620
621 '''
606 622
607 623 id = self.__getNewId()
608 priority = self.__getPriority()
624 priority = self.__getPriority() # Sin mucho sentido, pero puede usarse
609 625
610 626 opConfObj = OperationConf()
611 627 opConfObj.setup(id, name=name, priority=priority, type=optype)
@@ -668,11 +684,15 class ProcUnitConf():
668 684
669 685 return kwargs
670 686
671 def createObjects(self, plotter_queue=None):
687 def createObjects(self, dictUnits):
688 '''
689 Instancia de unidades de procesamiento.
672 690
691 '''
673 692 className = eval(self.name)
674 693 kwargs = self.getKwargs()
675 procUnitObj = className(**kwargs)
694 procUnitObj = className(self.id, self.inputId, dictUnits, **kwargs) # necesitan saber su id y su entrada por fines de ipc
695
676 696
677 697 for opConfObj in self.opConfObjList:
678 698
@@ -682,21 +702,25 class ProcUnitConf():
682 702 procUnitObj.addOperationKwargs(
683 703 opConfObj.id, **opConfObj.getKwargs())
684 704 continue
705 print("Creating operation process:", opConfObj.name, "for", self.name)
706 opObj = opConfObj.createObject()
685 707
686 opObj = opConfObj.createObject(plotter_queue)
687 708
688 self.opObjDict[opConfObj.id] = opObj
709 #self.opObjDict[opConfObj.id] = opObj.name
689 710
690 procUnitObj.addOperation(opObj, opConfObj.id)
711 procUnitObj.addOperation(opConfObj.name, opConfObj.id)
712
713 procUnitObj.start()
691 714
692 715 self.procUnitObj = procUnitObj
693 716
717
694 718 return procUnitObj
695 719
696 720 def run(self):
697 721
698 is_ok = False
699
722 is_ok = True
723 """
700 724 for opConfObj in self.opConfObjList:
701 725
702 726 kwargs = {}
@@ -712,8 +736,10 class ProcUnitConf():
712 736
713 737 is_ok = is_ok or sts
714 738
739 """
715 740 return is_ok
716 741
742
717 743 def close(self):
718 744
719 745 for opConfObj in self.opConfObjList:
@@ -757,6 +783,15 class ReadUnitConf(ProcUnitConf):
757 783 def setup(self, id, name, datatype, path='', startDate='', endDate='',
758 784 startTime='', endTime='', parentId=None, server=None, **kwargs):
759 785
786
787 '''
788 *****el id del proceso sera el Topico
789
790 Adicion de {topic}, si no esta presente -> error
791 kwargs deben ser trasmitidos en la instanciacion
792
793 '''
794
760 795 # Compatible with old signal chain version
761 796 if datatype == None and name == None:
762 797 raise ValueError('datatype or name should be defined')
@@ -892,7 +927,6 class ReadUnitConf(ProcUnitConf):
892 927 class Project(Process):
893 928
894 929 id = None
895 # name = None
896 930 description = None
897 931 filename = None
898 932
@@ -900,16 +934,15 class Project(Process):
900 934
901 935 ELEMENTNAME = 'Project'
902 936
903 plotterQueue = None
904 937
905 def __init__(self, plotter_queue=None):
938
939 def __init__(self):
906 940
907 941 Process.__init__(self)
908 942 self.id = None
909 943 self.description = None
910 944 self.email = None
911 945 self.alarm = None
912 self.plotterQueue = plotter_queue
913 946 self.procUnitConfObjDict = {}
914 947
915 948 def __getNewId(self):
@@ -958,11 +991,13 class Project(Process):
958 991
959 992 def setup(self, id, name='', description='', email=None, alarm=[]):
960 993
961 print()
994 print(' ')
962 995 print('*' * 60)
963 print(' Starting SIGNAL CHAIN PROCESSING v%s ' % schainpy.__version__)
996 print('* Starting SIGNAL CHAIN PROCESSING (Multiprocessing) v%s *' % schainpy.__version__)
964 997 print('*' * 60)
965 print()
998 print("* Python " + python_version() + " *")
999 print('*' * 19)
1000 print(' ')
966 1001 self.id = str(id)
967 1002 self.description = description
968 1003 self.email = email
@@ -981,6 +1016,14 class Project(Process):
981 1016
982 1017 def addReadUnit(self, id=None, datatype=None, name=None, **kwargs):
983 1018
1019 '''
1020 Actualizacion:
1021 Se agrego un nuevo argumento: topic -relativo a la forma de comunicar los procesos simultaneos
1022
1023 * El id del proceso sera el topico al que se deben subscribir los procUnits para recibir la informacion(data)
1024
1025 '''
1026
984 1027 if id is None:
985 1028 idReadUnit = self.__getNewId()
986 1029 else:
@@ -996,11 +1039,21 class Project(Process):
996 1039
997 1040 def addProcUnit(self, inputId='0', datatype=None, name=None):
998 1041
999 idProcUnit = self.__getNewId()
1042 '''
1043 Actualizacion:
1044 Se agrego dos nuevos argumentos: topic_read (lee data de otro procUnit) y topic_write(escribe o envia data a otro procUnit)
1045 Deberia reemplazar a "inputId"
1046
1047 ** A fin de mantener el inputID, este sera la representaacion del topicoal que deben subscribirse. El ID propio de la intancia
1048 (proceso) sera el topico de la publicacion, todo sera asignado de manera dinamica.
1049
1050 '''
1051
1052 idProcUnit = self.__getNewId() #Topico para subscripcion
1000 1053
1001 1054 procUnitConfObj = ProcUnitConf()
1002 procUnitConfObj.setup(idProcUnit, name, datatype,
1003 inputId, parentId=self.id)
1055 procUnitConfObj.setup(idProcUnit, name, datatype, inputId, #topic_read, topic_write,
1056 parentId=self.id)
1004 1057
1005 1058 self.procUnitConfObjDict[procUnitConfObj.getId()] = procUnitConfObj
1006 1059
@@ -1156,29 +1209,11 class Project(Process):
1156 1209 def createObjects(self):
1157 1210
1158 1211 for procUnitConfObj in list(self.procUnitConfObjDict.values()):
1159 procUnitConfObj.createObjects(self.plotterQueue)
1160
1161 def __connect(self, objIN, thisObj):
1162
1163 thisObj.setInput(objIN.getOutputObj())
1212 print("Creating process:", procUnitConfObj.name)
1213 procUnitConfObj.createObjects(self.procUnitConfObjDict)
1164 1214
1165 def connectObjects(self):
1166 1215
1167 for thisPUConfObj in list(self.procUnitConfObjDict.values()):
1168
1169 inputId = thisPUConfObj.getInputId()
1170
1171 if int(inputId) == 0:
1172 continue
1173
1174 # Get input object
1175 puConfINObj = self.procUnitConfObjDict[inputId]
1176 puObjIN = puConfINObj.getProcUnitObj()
1177
1178 # Get current object
1179 thisPUObj = thisPUConfObj.getProcUnitObj()
1180
1181 self.__connect(puObjIN, thisPUObj)
1216 print('All processes were created')
1182 1217
1183 1218 def __handleError(self, procUnitConfObj, modes=None, stdout=True):
1184 1219
@@ -1268,75 +1303,30 class Project(Process):
1268 1303
1269 1304 self.filename = filename
1270 1305
1271 def setPlotterQueue(self, plotter_queue):
1306 def setProxyCom(self):
1272 1307
1273 raise NotImplementedError('Use schainpy.controller_api.ControllerThread instead Project class')
1308 ctx = zmq.Context()
1309 if not os.path.exists('/tmp/socketTmp'): os.mkdir('/tmp/socketTmp')
1310 xsub = ctx.socket(zmq.XSUB)
1311 xsub.bind('ipc:///tmp/socketTmp/a')
1312 xpub = ctx.socket(zmq.XPUB)
1313 xpub.bind('ipc:///tmp/socketTmp/b')
1274 1314
1275 def getPlotterQueue(self):
1315 print("Controller Ready: Processes and proxy created")
1316 zmq.proxy(xsub, xpub)
1276 1317
1277 raise NotImplementedError('Use schainpy.controller_api.ControllerThread instead Project class')
1278 1318
1279 def useExternalPlotter(self):
1280
1281 raise NotImplementedError('Use schainpy.controller_api.ControllerThread instead Project class')
1282 1319
1283 1320 def run(self):
1284 1321
1285 1322 log.success('Starting {}'.format(self.name), tag='')
1286 1323 self.start_time = time.time()
1287 1324 self.createObjects()
1288 self.connectObjects()
1289
1290 keyList = list(self.procUnitConfObjDict.keys())
1291 keyList.sort()
1292
1293 err = None
1325 self.setProxyCom()
1294 1326
1295 while(True):
1296
1297 is_ok = False
1298
1299 for procKey in keyList:
1300
1301 procUnitConfObj = self.procUnitConfObjDict[procKey]
1302
1303 try:
1304 sts = procUnitConfObj.run()
1305 is_ok = is_ok or sts
1306 except SchainWarning:
1307 err = self.__handleError(procUnitConfObj, modes=[2, 3], stdout=False)
1308 is_ok = False
1309 break
1310 except KeyboardInterrupt:
1311 is_ok = False
1312 break
1313 except ValueError as e:
1314 time.sleep(0.5)
1315 err = self.__handleError(procUnitConfObj)
1316 is_ok = False
1317 break
1318 except:
1319 time.sleep(0.5)
1320 err = self.__handleError(procUnitConfObj)
1321 is_ok = False
1322 break
1323
1324 # If every process unit finished so end process
1325 if not(is_ok):
1326 break
1327
1328 if not self.runController():
1329 break
1327 # Iniciar todos los procesos .start(), monitoreo de procesos. ELiminar lo de abajo
1330 1328
1331 1329 # Closing every process
1332 for procKey in keyList:
1333 procUnitConfObj = self.procUnitConfObjDict[procKey]
1334 procUnitConfObj.close()
1335
1336 if err is not None:
1337 err.start()
1338 # err.join()
1339
1340 1330 log.success('{} finished (time: {}s)'.format(
1341 1331 self.name,
1342 1332 time.time()-self.start_time)) No newline at end of file
Show file before | Show file after | Hide comments
@@ -3,7 +3,8 import numpy
3 3 import time, datetime
4 4 from schainpy.model.graphics import mpldriver
5 5
6 from schainpy.model.proc.jroproc_base import Operation
6 from schainpy.model.proc.jroproc_base import MPDecorator, Operation
7
7 8
8 9 def isTimeInHourRange(datatime, xmin, xmax):
9 10
@@ -62,9 +63,9 class Figure(Operation):
62 63
63 64 created = False
64 65 parameters = {}
65 def __init__(self, **kwargs):
66 def __init__(self):#, **kwargs):
66 67
67 Operation.__init__(self, **kwargs)
68 Operation.__init__(self)#, **kwargs)
68 69
69 70 def __del__(self):
70 71
Show file before | Show file after | Hide comments
@@ -9,8 +9,11 import numpy
9 9
10 10 from .figure import Figure, isRealtime, isTimeInHourRange
11 11 from .plotting_codes import *
12 from schainpy.model.proc.jroproc_base import MPDecorator
12 13
14 from schainpy.utils import log
13 15
16 @MPDecorator
14 17 class SpectraPlot(Figure):
15 18
16 19 isConfig = None
@@ -20,11 +23,10 class SpectraPlot(Figure):
20 23 HEIGHTPROF = None
21 24 PREFIX = 'spc'
22 25
23 def __init__(self, **kwargs):
24 Figure.__init__(self, **kwargs)
26 def __init__(self):#, **kwargs):
27 Figure.__init__(self)#, **kwargs)
25 28 self.isConfig = False
26 29 self.__nsubplots = 1
27
28 30 self.WIDTH = 250
29 31 self.HEIGHT = 250
30 32 self.WIDTHPROF = 120
@@ -104,6 +106,9 class SpectraPlot(Figure):
104 106 zmin : None,
105 107 zmax : None
106 108 """
109 if dataOut.flagNoData:
110 return dataOut
111
107 112 if realtime:
108 113 if not(isRealtime(utcdatatime = dataOut.utctime)):
109 114 print('Skipping this plot function')
@@ -219,6 +224,8 class SpectraPlot(Figure):
219 224 wr_period=wr_period,
220 225 thisDatetime=thisDatetime)
221 226
227 return dataOut
228 @MPDecorator
222 229 class CrossSpectraPlot(Figure):
223 230
224 231 isConfig = None
@@ -230,8 +237,8 class CrossSpectraPlot(Figure):
230 237 HEIGHTPROF = None
231 238 PREFIX = 'cspc'
232 239
233 def __init__(self, **kwargs):
234 Figure.__init__(self, **kwargs)
240 def __init__(self):#, **kwargs):
241 Figure.__init__(self)#, **kwargs)
235 242 self.isConfig = False
236 243 self.__nsubplots = 4
237 244 self.counter_imagwr = 0
@@ -301,6 +308,9 class CrossSpectraPlot(Figure):
301 308 zmax : None
302 309 """
303 310
311 if dataOut.flagNoData:
312 return dataOut
313
304 314 if pairsList == None:
305 315 pairsIndexList = dataOut.pairsIndexList
306 316 else:
@@ -440,7 +450,9 class CrossSpectraPlot(Figure):
440 450 wr_period=wr_period,
441 451 thisDatetime=thisDatetime)
442 452
453 return dataOut
443 454
455 @MPDecorator
444 456 class RTIPlot(Figure):
445 457
446 458 __isConfig = None
@@ -450,9 +462,9 class RTIPlot(Figure):
450 462 HEIGHTPROF = None
451 463 PREFIX = 'rti'
452 464
453 def __init__(self, **kwargs):
465 def __init__(self):#, **kwargs):
454 466
455 Figure.__init__(self, **kwargs)
467 Figure.__init__(self)#, **kwargs)
456 468 self.timerange = None
457 469 self.isConfig = False
458 470 self.__nsubplots = 1
@@ -540,6 +552,8 class RTIPlot(Figure):
540 552 zmin : None,
541 553 zmax : None
542 554 """
555 if dataOut.flagNoData:
556 return dataOut
543 557
544 558 #colormap = kwargs.get('colormap', 'jet')
545 559 if HEIGHT is not None:
@@ -650,7 +664,9 class RTIPlot(Figure):
650 664 wr_period=wr_period,
651 665 thisDatetime=thisDatetime,
652 666 update_figfile=update_figfile)
667 return dataOut
653 668
669 @MPDecorator
654 670 class CoherenceMap(Figure):
655 671 isConfig = None
656 672 __nsubplots = None
@@ -659,8 +675,8 class CoherenceMap(Figure):
659 675 HEIGHTPROF = None
660 676 PREFIX = 'cmap'
661 677
662 def __init__(self, **kwargs):
663 Figure.__init__(self, **kwargs)
678 def __init__(self):#, **kwargs):
679 Figure.__init__(self)#, **kwargs)
664 680 self.timerange = 2*60*60
665 681 self.isConfig = False
666 682 self.__nsubplots = 1
@@ -723,6 +739,10 class CoherenceMap(Figure):
723 739 server=None, folder=None, username=None, password=None,
724 740 ftp_wei=0, exp_code=0, sub_exp_code=0, plot_pos=0):
725 741
742
743 if dataOut.flagNoData:
744 return dataOut
745
726 746 if not isTimeInHourRange(dataOut.datatime, xmin, xmax):
727 747 return
728 748
@@ -855,6 +875,9 class CoherenceMap(Figure):
855 875 thisDatetime=thisDatetime,
856 876 update_figfile=update_figfile)
857 877
878 return dataOut
879
880 @MPDecorator
858 881 class PowerProfilePlot(Figure):
859 882
860 883 isConfig = None
@@ -864,8 +887,8 class PowerProfilePlot(Figure):
864 887 HEIGHTPROF = None
865 888 PREFIX = 'spcprofile'
866 889
867 def __init__(self, **kwargs):
868 Figure.__init__(self, **kwargs)
890 def __init__(self):#, **kwargs):
891 Figure.__init__(self)#, **kwargs)
869 892 self.isConfig = False
870 893 self.__nsubplots = 1
871 894
@@ -907,6 +930,9 class PowerProfilePlot(Figure):
907 930 ftp=False, wr_period=1, server=None,
908 931 folder=None, username=None, password=None):
909 932
933 if dataOut.flagNoData:
934 return dataOut
935
910 936
911 937 if channelList == None:
912 938 channelIndexList = dataOut.channelIndexList
@@ -979,6 +1005,9 class PowerProfilePlot(Figure):
979 1005 wr_period=wr_period,
980 1006 thisDatetime=thisDatetime)
981 1007
1008 return dataOut
1009
1010 @MPDecorator
982 1011 class SpectraCutPlot(Figure):
983 1012
984 1013 isConfig = None
@@ -988,8 +1017,8 class SpectraCutPlot(Figure):
988 1017 HEIGHTPROF = None
989 1018 PREFIX = 'spc_cut'
990 1019
991 def __init__(self, **kwargs):
992 Figure.__init__(self, **kwargs)
1020 def __init__(self):#, **kwargs):
1021 Figure.__init__(self)#, **kwargs)
993 1022 self.isConfig = False
994 1023 self.__nsubplots = 1
995 1024
@@ -1032,6 +1061,8 class SpectraCutPlot(Figure):
1032 1061 folder=None, username=None, password=None,
1033 1062 xaxis="frequency"):
1034 1063
1064 if dataOut.flagNoData:
1065 return dataOut
1035 1066
1036 1067 if channelList == None:
1037 1068 channelIndexList = dataOut.channelIndexList
@@ -1111,6 +1142,9 class SpectraCutPlot(Figure):
1111 1142 wr_period=wr_period,
1112 1143 thisDatetime=thisDatetime)
1113 1144
1145 return dataOut
1146
1147 @MPDecorator
1114 1148 class Noise(Figure):
1115 1149
1116 1150 isConfig = None
@@ -1119,8 +1153,8 class Noise(Figure):
1119 1153 PREFIX = 'noise'
1120 1154
1121 1155
1122 def __init__(self, **kwargs):
1123 Figure.__init__(self, **kwargs)
1156 def __init__(self):#, **kwargs):
1157 Figure.__init__(self)#, **kwargs)
1124 1158 self.timerange = 24*60*60
1125 1159 self.isConfig = False
1126 1160 self.__nsubplots = 1
@@ -1209,6 +1243,9 class Noise(Figure):
1209 1243 server=None, folder=None, username=None, password=None,
1210 1244 ftp_wei=0, exp_code=0, sub_exp_code=0, plot_pos=0):
1211 1245
1246 if dataOut.flagNoData:
1247 return dataOut
1248
1212 1249 if not isTimeInHourRange(dataOut.datatime, xmin, xmax):
1213 1250 return
1214 1251
@@ -1312,6 +1349,9 class Noise(Figure):
1312 1349 if save:
1313 1350 self.save_data(self.filename_noise, noisedB, thisDatetime)
1314 1351
1352 return dataOut
1353
1354 @MPDecorator
1315 1355 class BeaconPhase(Figure):
1316 1356
1317 1357 __isConfig = None
@@ -1319,8 +1359,8 class BeaconPhase(Figure):
1319 1359
1320 1360 PREFIX = 'beacon_phase'
1321 1361
1322 def __init__(self, **kwargs):
1323 Figure.__init__(self, **kwargs)
1362 def __init__(self):#, **kwargs):
1363 Figure.__init__(self)#, **kwargs)
1324 1364 self.timerange = 24*60*60
1325 1365 self.isConfig = False
1326 1366 self.__nsubplots = 1
@@ -1399,6 +1439,9 class BeaconPhase(Figure):
1399 1439 server=None, folder=None, username=None, password=None,
1400 1440 ftp_wei=0, exp_code=0, sub_exp_code=0, plot_pos=0):
1401 1441
1442 if dataOut.flagNoData:
1443 return dataOut
1444
1402 1445 if not isTimeInHourRange(dataOut.datatime, xmin, xmax):
1403 1446 return
1404 1447
@@ -1540,3 +1583,5 class BeaconPhase(Figure):
1540 1583 wr_period=wr_period,
1541 1584 thisDatetime=thisDatetime,
1542 1585 update_figfile=update_figfile)
1586
1587 return dataOut #Yong No newline at end of file
Show file before | Show file after | Hide comments
@@ -20,7 +20,6 try:
20 20 except:
21 21 from time import sleep
22 22
23 import schainpy.admin
24 23 from schainpy.model.data.jroheaderIO import PROCFLAG, BasicHeader, SystemHeader, RadarControllerHeader, ProcessingHeader
25 24 from schainpy.model.data.jroheaderIO import get_dtype_index, get_numpy_dtype, get_procflag_dtype, get_dtype_width
26 25 from schainpy.utils import log
@@ -885,6 +884,7 class JRODataReader(JRODataIO):
885 884 self.flagIsNewFile = 0
886 885 self.fp = None
887 886 self.flagNoMoreFiles = 1
887 # print '[Reading] No more files to read'
888 888
889 889 return fileOk_flag
890 890
@@ -898,7 +898,7 class JRODataReader(JRODataIO):
898 898 newFile = self.__setNextFileOffline()
899 899
900 900 if not(newFile):
901 raise schainpy.admin.SchainWarning('No more files to read')
901 raise(schainpy.admin.SchainWarning('No more files to read'))
902 902 return 0
903 903
904 904 if self.verbose:
@@ -1052,7 +1052,7 class JRODataReader(JRODataIO):
1052 1052 # Skip block out of startTime and endTime
1053 1053 while True:
1054 1054 if not(self.__setNewBlock()):
1055 raise schainpy
1055 raise(schainpy.admin.SchainWarning('No more files'))
1056 1056 return 0
1057 1057
1058 1058 if not(self.readBlock()):
@@ -1320,11 +1320,11 class JRODataReader(JRODataIO):
1320 1320 if fullpath:
1321 1321 break
1322 1322
1323 print('[Reading] Waiting %0.2f sec for an valid file in %s: try %02d ...' % (delay, path, nTries + 1))
1324 sleep(delay)
1323 print('[Reading] Waiting %0.2f sec for an valid file in %s: try %02d ...' % (self.delay, path, nTries + 1))
1324 sleep(self.delay)
1325 1325
1326 1326 if not(fullpath):
1327 raise schainpy.admin.SchainWarning('There isn\'t any valid file in {}'.format(path))
1327 raise(schainpy.admin.SchainWarning('There isn\'t any valid file in {}'.format(path)))
1328 1328 return
1329 1329
1330 1330 self.year = year
Show file before | Show file after | Hide comments
@@ -6,10 +6,11 Created on Jul 2, 2014
6 6 import numpy
7 7
8 8 from schainpy.model.io.jroIO_base import LOCALTIME, JRODataReader, JRODataWriter
9 from schainpy.model.proc.jroproc_base import ProcessingUnit, Operation
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
13 @MPDecorator
13 14 class SpectraReader(JRODataReader, ProcessingUnit):
14 15 """
15 16 Esta clase permite leer datos de espectros desde archivos procesados (.pdata). La lectura
@@ -69,7 +70,7 class SpectraReader(JRODataReader, ProcessingUnit):
69 70
70 71 rdPairList = []
71 72
72 def __init__(self, **kwargs):
73 def __init__(self):#, **kwargs):
73 74 """
74 75 Inicializador de la clase SpectraReader para la lectura de datos de espectros.
75 76
@@ -88,7 +89,7 class SpectraReader(JRODataReader, ProcessingUnit):
88 89 """
89 90
90 91 #Eliminar de la base la herencia
91 ProcessingUnit.__init__(self, **kwargs)
92 ProcessingUnit.__init__(self)#, **kwargs)
92 93
93 94 # self.isConfig = False
94 95
@@ -510,7 +511,7 class SpectraWriter(JRODataWriter, Operation):
510 511 """
511 512
512 513 spc = numpy.transpose( self.data_spc, (0,2,1) )
513 if self.processingHeaderObj.shif_fft:
514 if not( self.processingHeaderObj.shif_fft ):
514 515 spc = numpy.roll( spc, self.processingHeaderObj.profilesPerBlock/2, axis=2 ) #desplaza a la derecha en el eje 2 determinadas posiciones
515 516 data = spc.reshape((-1))
516 517 data = data.astype(self.dtype[0])
@@ -519,7 +520,7 class SpectraWriter(JRODataWriter, Operation):
519 520 if self.data_cspc is not None:
520 521 data = numpy.zeros( self.shape_cspc_Buffer, self.dtype )
521 522 cspc = numpy.transpose( self.data_cspc, (0,2,1) )
522 if self.processingHeaderObj.shif_fft:
523 if not( self.processingHeaderObj.shif_fft ):
523 524 cspc = numpy.roll( cspc, self.processingHeaderObj.profilesPerBlock/2, axis=2 ) #desplaza a la derecha en el eje 2 determinadas posiciones
524 525 data['real'] = cspc.real
525 526 data['imag'] = cspc.imag
Show file before | Show file after | Hide comments
This diff has been collapsed as it changes many lines, (550 lines changed) Show them Hide them
@@ -1,41 +1,39
1 1 '''
2
2 Updated for multiprocessing
3 Author : Sergio Cortez
4 Jan 2018
5 Abstract:
6 Base class for processing units and operations. A decorator provides multiprocessing features and interconnect the processes created.
7 The argument (kwargs) sent from the controller is parsed and filtered via the decorator for each processing unit or operation instantiated.
8 The decorator handle also the methods inside the processing unit to be called from the main script (not as operations) (OPERATION -> type ='self').
9
10 Based on:
3 11 $Author: murco $
4 12 $Id: jroproc_base.py 1 2012-11-12 18:56:07Z murco $
5 13 '''
14 from platform import python_version
6 15 import inspect
7 from fuzzywuzzy import process
8
9 def checkKwargs(method, kwargs):
10 currentKwargs = kwargs
11 choices = inspect.getargspec(method).args
12 try:
13 choices.remove('self')
14 except Exception as e:
15 pass
16 import zmq
17 import time
18 import pickle
19 import os
20 from multiprocessing import Process
16 21
17 try:
18 choices.remove('dataOut')
19 except Exception as e:
20 pass
22 from schainpy.utils import log
21 23
22 for kwarg in kwargs:
23 fuzz = process.extractOne(kwarg, choices)
24 if fuzz is None:
25 continue
26 if fuzz[1] < 100:
27 raise Exception('\x1b[0;32;40mDid you mean {} instead of {} in {}? \x1b[0m'.
28 format(fuzz[0], kwarg, method.__self__.__class__.__name__))
29 24
30 25 class ProcessingUnit(object):
31 26
32 27 """
33 Esta es la clase base para el procesamiento de datos.
28 Update - Jan 2018 - MULTIPROCESSING
29 All the "call" methods present in the previous base were removed.
30 The majority of operations are independant processes, thus
31 the decorator is in charge of communicate the operation processes
32 with the proccessing unit via IPC.
33
34 The constructor does not receive any argument. The remaining methods
35 are related with the operations to execute.
34 36
35 Contiene el metodo "call" para llamar operaciones. Las operaciones pueden ser:
36 - Metodos internos (callMethod)
37 - Objetos del tipo Operation (callObject). Antes de ser llamados, estos objetos
38 tienen que ser agreagados con el metodo "add".
39 37
40 38 """
41 39 # objeto de datos de entrada (Voltage, Spectra o Correlation)
@@ -43,33 +41,25 class ProcessingUnit(object):
43 41 dataInList = []
44 42
45 43 # objeto de datos de entrada (Voltage, Spectra o Correlation)
44
45 id = None
46 inputId = None
47
46 48 dataOut = None
47 49
50 dictProcs = None
51
48 52 operations2RunDict = None
49 53
50 54 isConfig = False
51 55
52
53 def __init__(self, *args, **kwargs):
56 def __init__(self):
54 57
55 58 self.dataIn = None
56 self.dataInList = []
57
58 59 self.dataOut = None
59 60
60 self.operations2RunDict = {}
61 self.operationKwargs = {}
62
63 61 self.isConfig = False
64 62
65 self.args = args
66 self.kwargs = kwargs
67
68 if not hasattr(self, 'name'):
69 self.name = self.__class__.__name__
70
71 checkKwargs(self.run, kwargs)
72
73 63 def getAllowedArgs(self):
74 64 if hasattr(self, '__attrs__'):
75 65 return self.__attrs__
@@ -82,10 +72,12 class ProcessingUnit(object):
82 72
83 73 self.operationKwargs[objId] = kwargs
84 74
85
86 75 def addOperation(self, opObj, objId):
87 76
88 77 """
78 This method is used in the controller, and update the dictionary containing the operations to execute. The dict
79 posses the id of the operation process (IPC purposes)
80
89 81 Agrega un objeto del tipo "Operation" (opObj) a la lista de objetos "self.objectList" y retorna el
90 82 identificador asociado a este objeto.
91 83
@@ -95,13 +87,14 class ProcessingUnit(object):
95 87
96 88 Return:
97 89
98 objId : identificador del objeto, necesario para ejecutar la operacion
90 objId : identificador del objeto, necesario para comunicar con master(procUnit)
99 91 """
100 92
101 93 self.operations2RunDict[objId] = opObj
102 94
103 95 return objId
104 96
97
105 98 def getOperationObj(self, objId):
106 99
107 100 if objId not in list(self.operations2RunDict.keys()):
@@ -122,239 +115,424 class ProcessingUnit(object):
122 115
123 116 raise NotImplementedError
124 117
125 def callMethod(self, name, opId):
118 def setup(self):
126 119
127 """
128 Ejecuta el metodo con el nombre "name" y con argumentos **kwargs de la propia clase.
120 raise NotImplementedError
129 121
130 Input:
131 name : nombre del metodo a ejecutar
122 def run(self):
123
124 raise NotImplementedError
125
126 def close(self):
127 #Close every thread, queue or any other object here is it is neccesary.
128 return
132 129
133 **kwargs : diccionario con los nombres y valores de la funcion a ejecutar.
130 class Operation(object):
134 131
135 132 """
133 Update - Jan 2018 - MULTIPROCESSING
136 134
137 #Checking the inputs
138 if name == 'run':
135 Most of the methods remained the same. The decorator parse the arguments and executed the run() method for each process.
136 The constructor doe snot receive any argument, neither the baseclass.
139 137
140 if not self.checkInputs():
141 self.dataOut.flagNoData = True
142 return False
143 else:
144 #Si no es un metodo RUN la entrada es la misma dataOut (interna)
145 if self.dataOut is not None and self.dataOut.isEmpty():
146 return False
147 138
148 #Getting the pointer to method
149 methodToCall = getattr(self, name)
139 Clase base para definir las operaciones adicionales que se pueden agregar a la clase ProcessingUnit
140 y necesiten acumular informacion previa de los datos a procesar. De preferencia usar un buffer de
141 acumulacion dentro de esta clase
150 142
151 #Executing the self method
143 Ejemplo: Integraciones coherentes, necesita la informacion previa de los n perfiles anteriores (bufffer)
152 144
153 if hasattr(self, 'mp'):
154 if name=='run':
155 if self.mp is False:
156 self.mp = True
157 self.start()
158 else:
159 self.operationKwargs[opId]['parent'] = self.kwargs
160 methodToCall(**self.operationKwargs[opId])
161 else:
162 if name=='run':
163 methodToCall(**self.kwargs)
145 """
146 id = None
147 __buffer = None
148 dest = None
149 isConfig = False
150 readyFlag = None
151
152 def __init__(self):
153
154 self.buffer = None
155 self.dest = None
156 self.isConfig = False
157 self.readyFlag = False
158
159 if not hasattr(self, 'name'):
160 self.name = self.__class__.__name__
161
162 def getAllowedArgs(self):
163 if hasattr(self, '__attrs__'):
164 return self.__attrs__
164 165 else:
165 methodToCall(**self.operationKwargs[opId])
166 return inspect.getargspec(self.run).args
166 167
167 if self.dataOut is None:
168 return False
168 def setup(self):
169 169
170 if self.dataOut.isEmpty():
171 return False
170 self.isConfig = True
171
172 raise NotImplementedError
172 173
173 return True
174 174
175 def callObject(self, objId):
175 def run(self, dataIn, **kwargs):
176 176
177 177 """
178 Ejecuta la operacion asociada al identificador del objeto "objId"
178 Realiza las operaciones necesarias sobre la dataIn.data y actualiza los
179 atributos del objeto dataIn.
179 180
180 181 Input:
181 182
182 objId : identificador del objeto a ejecutar
183
184 **kwargs : diccionario con los nombres y valores de la funcion a ejecutar.
183 dataIn : objeto del tipo JROData
185 184
186 185 Return:
187 186
188 187 None
188
189 Affected:
190 __buffer : buffer de recepcion de datos.
191
189 192 """
193 if not self.isConfig:
194 self.setup(**kwargs)
190 195
191 if self.dataOut is not None and self.dataOut.isEmpty():
192 return False
196 raise NotImplementedError
193 197
194 externalProcObj = self.operations2RunDict[objId]
198 def close(self):
195 199
196 if hasattr(externalProcObj, 'mp'):
197 if externalProcObj.mp is False:
198 externalProcObj.kwargs['parent'] = self.kwargs
199 self.operationKwargs[objId] = externalProcObj.kwargs
200 externalProcObj.mp = True
201 externalProcObj.start()
202 else:
203 externalProcObj.run(self.dataOut, **externalProcObj.kwargs)
204 self.operationKwargs[objId] = externalProcObj.kwargs
200 pass
205 201
206 202
207 return True
203 ######### Decorator #########
208 204
209 def call(self, opType, opName=None, opId=None):
210 """
211 Return True si ejecuta la operacion interna nombrada "opName" o la operacion externa
212 identificada con el id "opId"; con los argumentos "**kwargs".
213 205
214 False si la operacion no se ha ejecutado.
206 def MPDecorator(BaseClass):
215 207
216 Input:
208 """
209 "Multiprocessing class decorator"
217 210
218 opType : Puede ser "self" o "external"
211 This function add multiprocessing features to the base class. Also,
212 it handle the communication beetween processes (readers, procUnits and operations).
213 Receive the arguments at the moment of instantiation. According to that, discriminates if it
214 is a procUnit or an operation
215 """
219 216
220 Depende del tipo de operacion para llamar a:callMethod or callObject:
217 class MPClass(BaseClass, Process):
221 218
222 1. If opType = "self": Llama a un metodo propio de esta clase:
219 "This is the overwritten class"
220 operations2RunDict = None
221 socket_l = None
222 socket_p = None
223 socketOP = None
224 socket_router = None
225 dictProcs = None
226 typeProc = None
227 def __init__(self, *args, **kwargs):
228 super(MPClass, self).__init__()
229 Process.__init__(self)
223 230
224 name_method = getattr(self, name)
225 name_method(**kwargs)
226 231
232 self.operationKwargs = {}
233 self.args = args
227 234
228 2. If opType = "other" o"external": Llama al metodo "run()" de una instancia de la
229 clase "Operation" o de un derivado de ella:
230 235
231 instanceName = self.operationList[opId]
232 instanceName.run(**kwargs)
236 self.operations2RunDict = {}
237 self.kwargs = kwargs
233 238
234 opName : Si la operacion es interna (opType = 'self'), entonces el "opName" sera
235 usada para llamar a un metodo interno de la clase Processing
239 # The number of arguments (args) determine the type of process
236 240
237 opId : Si la operacion es externa (opType = 'other' o 'external), entonces el
238 "opId" sera usada para llamar al metodo "run" de la clase Operation
239 registrada anteriormente con ese Id
241 if len(self.args) is 3:
242 self.typeProc = "ProcUnit"
243 self.id = args[0] #topico de publicacion
244 self.inputId = args[1] #topico de subcripcion
245 self.dictProcs = args[2] #diccionario de procesos globales
246 else:
247 self.id = args[0]
248 self.typeProc = "Operation"
240 249
241 Exception:
242 Este objeto de tipo Operation debe de haber sido agregado antes con el metodo:
243 "addOperation" e identificado con el valor "opId" = el id de la operacion.
244 De lo contrario retornara un error del tipo ValueError
250 def addOperationKwargs(self, objId, **kwargs):
245 251
246 """
252 self.operationKwargs[objId] = kwargs
247 253
248 if opType == 'self':
254 def getAllowedArgs(self):
249 255
250 if not opName:
251 raise ValueError("opName parameter should be defined")
256 if hasattr(self, '__attrs__'):
257 return self.__attrs__
258 else:
259 return inspect.getargspec(self.run).args
252 260
253 sts = self.callMethod(opName, opId)
254 261
255 elif opType == 'other' or opType == 'external' or opType == 'plotter':
262 def sockListening(self, topic):
256 263
257 if not opId:
258 raise ValueError("opId parameter should be defined")
264 """
265 This function create a socket to receive objects.
266 The 'topic' argument is related to the publisher process from which the self process is
267 listening (data).
268 In the case were the self process is listening to a Reader (proc Unit),
269 special conditions are introduced to maximize parallelism.
270 """
259 271
260 if opId not in list(self.operations2RunDict.keys()):
261 raise ValueError("Any operation with id=%s has been added" %str(opId))
272 cont = zmq.Context()
273 zmq_socket = cont.socket(zmq.SUB)
274 if not os.path.exists('/tmp/socketTmp'):
275 os.mkdir('/tmp/socketTmp')
262 276
263 sts = self.callObject(opId)
277 if 'Reader' in self.dictProcs[self.inputId].name:
278 zmq_socket.connect('ipc:///tmp/socketTmp/b')
264 279
265 280 else:
266 raise ValueError("opType should be 'self', 'external' or 'plotter'; and not '%s'" %opType)
281 zmq_socket.connect('ipc:///tmp/socketTmp/%s' % self.inputId)
267 282
268 return sts
283 #log.error('RECEIVING FROM {} {}'.format(self.inputId, str(topic).encode()))
284 zmq_socket.setsockopt(zmq.SUBSCRIBE, str(topic).encode()) #yong
269 285
270 def setInput(self, dataIn):
286 return zmq_socket
271 287
272 self.dataIn = dataIn
273 self.dataInList.append(dataIn)
274 288
275 def getOutputObj(self):
289 def listenProc(self, sock):
276 290
277 return self.dataOut
291 """
292 This function listen to a ipc addres until a message is recovered. To serialize the
293 data (object), pickle has been use.
294 The 'sock' argument is the socket previously connect to an ipc address and with a topic subscription.
295 """
278 296
279 def checkInputs(self):
297 a = sock.recv_multipart()
298 a = pickle.loads(a[1])
299 return a
280 300
281 for thisDataIn in self.dataInList:
301 def sockPublishing(self):
302
303 """
304 This function create a socket for publishing purposes.
305 Depending on the process type from where is created, it binds or connect
306 to special IPC addresses.
307 """
308 time.sleep(4) #yong
309 context = zmq.Context()
310 zmq_socket = context.socket(zmq.PUB)
311 if not os.path.exists('/tmp/socketTmp'): os.mkdir('/tmp/socketTmp')
312 if 'Reader' in self.dictProcs[self.id].name:
313 zmq_socket.connect('ipc:///tmp/socketTmp/a')
314 else:
315 zmq_socket.bind('ipc:///tmp/socketTmp/%s' % self.id)
316
317 return zmq_socket
318
319 def publishProc(self, sock, data):
320
321 """
322 This function publish a python object (data) under a specific topic in a socket (sock).
323 Usually, the topic is the self id of the process.
324 """
282 325
283 if thisDataIn.isEmpty():
284 return False
326 sock.send_multipart([str(self.id).encode(), pickle.dumps(data)]) #yong
285 327
286 328 return True
287 329
288 def setup(self):
330 def sockOp(self):
289 331
290 raise NotImplementedError
332 """
333 This function create a socket for communication purposes with operation processes.
334 """
291 335
292 def run(self):
336 cont = zmq.Context()
337 zmq_socket = cont.socket(zmq.DEALER)
293 338
294 raise NotImplementedError
339 if python_version()[0] == '2':
340 zmq_socket.setsockopt(zmq.IDENTITY, self.id)
341 if python_version()[0] == '3':
342 zmq_socket.setsockopt_string(zmq.IDENTITY, self.id)
295 343
296 def close(self):
297 #Close every thread, queue or any other object here is it is neccesary.
298 return
299 344
300 class Operation(object):
345 return zmq_socket
301 346
302 """
303 Clase base para definir las operaciones adicionales que se pueden agregar a la clase ProcessingUnit
304 y necesiten acumular informacion previa de los datos a procesar. De preferencia usar un buffer de
305 acumulacion dentro de esta clase
306 347
307 Ejemplo: Integraciones coherentes, necesita la informacion previa de los n perfiles anteriores (bufffer)
348 def execOp(self, socket, opId, dataObj):
308 349
309 350 """
351 This function 'execute' an operation main routine by establishing a
352 connection with it and sending a python object (dataOut).
353 """
354 if not os.path.exists('/tmp/socketTmp'): os.mkdir('/tmp/socketTmp')
355 socket.connect('ipc:///tmp/socketTmp/%s' %opId)
310 356
311 __buffer = None
312 isConfig = False
313 357
314 def __init__(self, **kwargs):
358 socket.send(pickle.dumps(dataObj)) #yong
315 359
316 self.__buffer = None
317 self.isConfig = False
318 self.kwargs = kwargs
319 if not hasattr(self, 'name'):
320 self.name = self.__class__.__name__
321 checkKwargs(self.run, kwargs)
360 argument = socket.recv_multipart()[0]
322 361
323 def getAllowedArgs(self):
324 if hasattr(self, '__attrs__'):
325 return self.__attrs__
326 else:
327 return inspect.getargspec(self.run).args
362 argument = pickle.loads(argument)
328 363
329 def setup(self):
364 return argument
330 365
331 self.isConfig = True
366 def sockIO(self):
332 367
333 raise NotImplementedError
368 """
369 Socket defined for an operation process. It is able to recover the object sent from another process as well as a
370 identifier of who sent it.
371 """
334 372
335 def run(self, dataIn, **kwargs):
373 cont = zmq.Context()
374 if not os.path.exists('/tmp/socketTmp'): os.mkdir('/tmp/socketTmp')
375 socket = cont.socket(zmq.ROUTER)
376 socket.bind('ipc:///tmp/socketTmp/%s' % self.id)
377
378 return socket
379
380 def funIOrec(self, socket):
336 381
337 382 """
338 Realiza las operaciones necesarias sobre la dataIn.data y actualiza los
339 atributos del objeto dataIn.
383 Operation method, recover the id of the process who sent a python object.
384 The 'socket' argument is the socket binded to a specific process ipc.
385 """
340 386
341 Input:
387 #id_proc = socket.recv()
342 388
343 dataIn : objeto del tipo JROData
389 #dataObj = socket.recv_pyobj()
344 390
345 Return:
391 dataObj = socket.recv_multipart()
346 392
347 None
393 dataObj[1] = pickle.loads(dataObj[1])
394 return dataObj[0], dataObj[1]
348 395
349 Affected:
350 __buffer : buffer de recepcion de datos.
396 def funIOsen(self, socket, data, dest):
351 397
352 398 """
353 if not self.isConfig:
354 self.setup(**kwargs)
399 Operation method, send a python object to a specific destination.
400 The 'dest' argument is the id of a proccesinf unit.
401 """
355 402
356 raise NotImplementedError
403 socket.send_multipart([dest, pickle.dumps(data)]) #yong
357 404
358 def close(self):
405 return True
359 406
360 pass No newline at end of file
407
408 def runReader(self):
409
410 # time.sleep(3)
411 while True:
412
413 BaseClass.run(self, **self.kwargs)
414
415
416 keyList = list(self.operations2RunDict.keys())
417 keyList.sort()
418
419 for key in keyList:
420 self.socketOP = self.sockOp()
421 self.dataOut = self.execOp(self.socketOP, key, self.dataOut)
422
423
424 if self.flagNoMoreFiles: #Usar un objeto con flags para saber si termino el proc o hubo un error
425 self.publishProc(self.socket_p, "Finish")
426 break
427
428 if self.dataOut.flagNoData:
429 continue
430
431 print("Publishing data...")
432 self.publishProc(self.socket_p, self.dataOut)
433 # time.sleep(2)
434
435
436 print("%s done" %BaseClass.__name__)
437 return 0
438
439 def runProc(self):
440
441 # All the procUnits with kwargs that require a setup initialization must be defined here.
442
443 if self.setupReq:
444 BaseClass.setup(self, **self.kwargs)
445
446 while True:
447 self.dataIn = self.listenProc(self.socket_l)
448 print("%s received data" %BaseClass.__name__)
449
450 if self.dataIn == "Finish":
451 break
452
453 m_arg = list(self.kwargs.keys())
454 num_arg = list(range(1,int(BaseClass.run.__code__.co_argcount)))
455
456 run_arg = {}
457
458 for var in num_arg:
459 if BaseClass.run.__code__.co_varnames[var] in m_arg:
460 run_arg[BaseClass.run.__code__.co_varnames[var]] = self.kwargs[BaseClass.run.__code__.co_varnames[var]]
461
462 #BaseClass.run(self, **self.kwargs)
463 BaseClass.run(self, **run_arg)
464
465 ## Iterar sobre una serie de data que podrias aplicarse
466
467 for m_name in BaseClass.METHODS:
468
469 met_arg = {}
470
471 for arg in m_arg:
472 if arg in BaseClass.METHODS[m_name]:
473 for att in BaseClass.METHODS[m_name]:
474 met_arg[att] = self.kwargs[att]
475
476 method = getattr(BaseClass, m_name)
477 method(self, **met_arg)
478 break
479
480 if self.dataOut.flagNoData:
481 continue
482
483 keyList = list(self.operations2RunDict.keys())
484 keyList.sort()
485
486 for key in keyList:
487
488 self.socketOP = self.sockOp()
489 self.dataOut = self.execOp(self.socketOP, key, self.dataOut)
490
491
492 self.publishProc(self.socket_p, self.dataOut)
493
494
495 print("%s done" %BaseClass.__name__)
496
497 return 0
498
499 def runOp(self):
500
501 while True:
502
503 [self.dest ,self.buffer] = self.funIOrec(self.socket_router)
504
505 self.buffer = BaseClass.run(self, self.buffer, **self.kwargs)
506
507 self.funIOsen(self.socket_router, self.buffer, self.dest)
508
509 print("%s done" %BaseClass.__name__)
510 return 0
511
512
513 def run(self):
514
515 if self.typeProc is "ProcUnit":
516
517 self.socket_p = self.sockPublishing()
518
519 if 'Reader' not in self.dictProcs[self.id].name:
520 self.socket_l = self.sockListening(self.inputId)
521 self.runProc()
522
523 else:
524
525 self.runReader()
526
527 elif self.typeProc is "Operation":
528
529 self.socket_router = self.sockIO()
530
531 self.runOp()
532
533 else:
534 raise ValueError("Unknown type")
535
536 return 0
537
538 return MPClass No newline at end of file
Show file before | Show file after | Hide comments
@@ -2,16 +2,24 import itertools
2 2
3 3 import numpy
4 4
5 from .jroproc_base import ProcessingUnit, Operation
5 from schainpy.model.proc.jroproc_base import ProcessingUnit, MPDecorator, Operation
6 6 from schainpy.model.data.jrodata import Spectra
7 7 from schainpy.model.data.jrodata import hildebrand_sekhon
8 from schainpy.utils import log #yong
8 from schainpy.utils import log
9 9
10 @MPDecorator
10 11 class SpectraProc(ProcessingUnit):
11 12
12 def __init__(self, **kwargs):
13 METHODS = {'selectHeights' : ['minHei', 'maxHei'],
14 'selectChannels' : 'channelList',
15 'selectChannelsByIndex': 'channelIndexList',
16 'getBeaconSignal' : ['tauindex', 'channelindex', 'hei_ref'],
17 'selectHeightsByIndex' : ['minIndex', 'maxIndex']
18 }
13 19
14 ProcessingUnit.__init__(self, **kwargs)
20 def __init__(self):#, **kwargs):
21
22 ProcessingUnit.__init__(self)#, **kwargs)
15 23
16 24 self.buffer = None
17 25 self.firstdatatime = None
@@ -19,6 +27,7 class SpectraProc(ProcessingUnit):
19 27 self.dataOut = Spectra()
20 28 self.id_min = None
21 29 self.id_max = None
30 self.setupReq = False #Agregar a todas las unidades de proc
22 31
23 32 def __updateSpecFromVoltage(self):
24 33
@@ -774,7 +783,7 class SpectraProc(ProcessingUnit):
774 783
775 784 return 1
776 785
777
786 @MPDecorator
778 787 class IncohInt(Operation):
779 788
780 789 __profIndex = 0
@@ -795,9 +804,11 class IncohInt(Operation):
795 804
796 805 n = None
797 806
798 def __init__(self, **kwargs):
807 def __init__(self):#, **kwargs):
808
809 Operation.__init__(self)#, **kwargs)
810
799 811
800 Operation.__init__(self, **kwargs)
801 812 # self.isConfig = False
802 813
803 814 def setup(self, n=None, timeInterval=None, overlapping=False):
@@ -951,3 +962,5 class IncohInt(Operation):
951 962 dataOut.nIncohInt *= self.n
952 963 dataOut.utctime = avgdatatime
953 964 dataOut.flagNoData = False
965
966 return dataOut No newline at end of file
General Comments 0
You need to be logged in to leave comments. Login now