controller.py
1332 lines
| 37.2 KiB
| text/x-python
|
PythonLexer
/ schainpy / controller.py
Miguel Valdez
|
r197 | ''' | |
George Yong
|
r1171 | Updated on January , 2018, for multiprocessing purposes | |
| Author: Sergio Cortez | |||
|
Miguel Valdez
|
r197 | Created on September , 2012 | |
| ''' | |||
|
George Yong
|
r1171 | from platform import python_version | |
|
Miguel Valdez
|
r672 | import sys | |
Julio Valdez
|
r514 | import ast | |
|
Miguel Valdez
|
r687 | import datetime | |
|
Miguel Valdez
|
r672 | import traceback | |
José Chávez
|
r898 | import math | |
|
José Chávez
|
r931 | import time | |
|
George Yong
|
r1171 | import zmq | |
Juan C. Espinoza
|
r1052 | from multiprocessing import Process, cpu_count | |
|
Miguel Valdez
|
r672 | ||
|
Miguel Valdez
|
r681 | from xml.etree.ElementTree import ElementTree, Element, SubElement, tostring | |
| from xml.dom import minidom | |||
|
George Yong
|
r1171 | ||
| r1129 | from schainpy.admin import Alarm, SchainWarning | ||
|
George Yong
|
r1171 | ||
| ### Temporary imports!!! | |||
| # from schainpy.model import * | |||
| from schainpy.model.io import * | |||
| from schainpy.model.graphics import * | |||
| from schainpy.model.proc.jroproc_base import * | |||
| from schainpy.model.proc.bltrproc_parameters import * | |||
| from schainpy.model.proc.jroproc_spectra import * | |||
|
George Yong
|
r1173 | from schainpy.model.proc.jroproc_voltage import * | |
|
George Yong
|
r1171 | from schainpy.model.proc.jroproc_parameters import * | |
| from schainpy.model.utils.jroutils_publish import * | |||
|
Juan C. Espinoza
|
r1052 | from schainpy.utils import log | |
|
George Yong
|
r1171 | ### | |
|
Juan C. Espinoza
|
r1052 | ||
| DTYPES = { | |||
| 'Voltage': '.r', | |||
| 'Spectra': '.pdata' | |||
| } | |||
|
José Chávez
|
r1082 | ||
|
Juan C. Espinoza
|
r1052 | def MPProject(project, n=cpu_count()): | |
| ''' | |||
| Project wrapper to run schain in n processes | |||
| ''' | |||
| rconf = project.getReadUnitObj() | |||
| op = rconf.getOperationObj('run') | |||
| dt1 = op.getParameterValue('startDate') | |||
| dt2 = op.getParameterValue('endDate') | |||
|
Juan C. Espinoza
|
r1112 | tm1 = op.getParameterValue('startTime') | |
| tm2 = op.getParameterValue('endTime') | |||
| r892 | days = (dt2 - dt1).days | ||
|
José Chávez
|
r1082 | ||
| for day in range(days + 1): | |||
| r892 | skip = 0 | ||
| cursor = 0 | |||
| processes = [] | |||
|
Juan C. Espinoza
|
r1052 | dt = dt1 + datetime.timedelta(day) | |
| dt_str = dt.strftime('%Y/%m/%d') | |||
| reader = JRODataReader() | |||
| paths, files = reader.searchFilesOffLine(path=rconf.path, | |||
|
José Chávez
|
r1082 | startDate=dt, | |
| endDate=dt, | |||
|
Juan C. Espinoza
|
r1112 | startTime=tm1, | |
| endTime=tm2, | |||
|
José Chávez
|
r1082 | ext=DTYPES[rconf.datatype]) | |
|
Juan C. Espinoza
|
r1052 | nFiles = len(files) | |
| if nFiles == 0: | |||
|
Juan C. Espinoza
|
r999 | continue | |
|
Juan C. Espinoza
|
r1112 | skip = int(math.ceil(nFiles / n)) | |
|
José Chávez
|
r1082 | while nFiles > cursor * skip: | |
| rconf.update(startDate=dt_str, endDate=dt_str, cursor=cursor, | |||
| skip=skip) | |||
| p = project.clone() | |||
|
Juan C. Espinoza
|
r1052 | p.start() | |
| processes.append(p) | |||
|
José Chávez
|
r924 | cursor += 1 | |
| def beforeExit(exctype, value, trace): | |||
| r892 | for process in processes: | ||
| process.terminate() | |||
| process.join() | |||
|
George Yong
|
r1167 | print(traceback.print_tb(trace)) | |
|
Juan C. Espinoza
|
r957 | ||
|
José Chávez
|
r924 | sys.excepthook = beforeExit | |
| r892 | |||
| for process in processes: | |||
| process.join() | |||
|
José Chávez
|
r924 | process.terminate() | |
J Gomez
|
r958 | ||
|
José Chávez
|
r931 | time.sleep(3) | |
| r892 | |||
|
Miguel Valdez
|
r197 | class ParameterConf(): | |
| r889 | |||
|
Miguel Valdez
|
r197 | id = None | |
| name = None | |||
| value = None | |||
|
Miguel Valdez
|
r199 | format = None | |
| r889 | |||
|
Miguel Valdez
|
r529 | __formated_value = None | |
| r889 | |||
|
Miguel Valdez
|
r197 | ELEMENTNAME = 'Parameter' | |
| r889 | |||
|
Miguel Valdez
|
r197 | def __init__(self): | |
| r889 | |||
|
Miguel Valdez
|
r199 | self.format = 'str' | |
| r889 | |||
|
Miguel Valdez
|
r197 | def getElementName(self): | |
| r889 | |||
|
Miguel Valdez
|
r197 | return self.ELEMENTNAME | |
| r889 | |||
|
Miguel Valdez
|
r197 | def getValue(self): | |
|
Miguel Valdez
|
r600 | ||
| value = self.value | |||
| format = self.format | |||
| r889 | |||
|
Miguel Valdez
|
r529 | if self.__formated_value != None: | |
| r889 | |||
|
Miguel Valdez
|
r529 | return self.__formated_value | |
| r889 | |||
| r892 | if format == 'obj': | ||
| return value | |||
|
Miguel Valdez
|
r600 | if format == 'str': | |
|
Miguel Valdez
|
r596 | self.__formated_value = str(value) | |
| return self.__formated_value | |||
| r889 | |||
|
Miguel Valdez
|
r596 | if value == '': | |
|
George Yong
|
r1167 | raise ValueError('%s: This parameter value is empty' % self.name) | |
| r889 | |||
|
Miguel Valdez
|
r600 | if format == 'list': | |
|
Miguel Valdez
|
r224 | strList = value.split(',') | |
| r889 | |||
|
Miguel Valdez
|
r529 | self.__formated_value = strList | |
| r889 | |||
|
Miguel Valdez
|
r529 | return self.__formated_value | |
| r889 | |||
|
Miguel Valdez
|
r600 | if format == 'intlist': | |
|
Juan C. Espinoza
|
r1052 | ''' | |
|
Miguel Valdez
|
r535 | Example: | |
| value = (0,1,2) | |||
|
Juan C. Espinoza
|
r1052 | ''' | |
| r889 | |||
|
Miguel Valdez
|
r735 | new_value = ast.literal_eval(value) | |
| r889 | |||
|
Miguel Valdez
|
r735 | if type(new_value) not in (tuple, list): | |
| new_value = [int(new_value)] | |||
| r889 | |||
|
Miguel Valdez
|
r735 | self.__formated_value = new_value | |
| r889 | |||
|
Miguel Valdez
|
r529 | return self.__formated_value | |
| r889 | |||
|
Miguel Valdez
|
r600 | if format == 'floatlist': | |
|
Juan C. Espinoza
|
r1052 | ''' | |
|
Miguel Valdez
|
r535 | Example: | |
| value = (0.5, 1.4, 2.7) | |||
|
Juan C. Espinoza
|
r1052 | ''' | |
| r889 | |||
|
Miguel Valdez
|
r735 | new_value = ast.literal_eval(value) | |
| r889 | |||
|
Miguel Valdez
|
r735 | if type(new_value) not in (tuple, list): | |
| new_value = [float(new_value)] | |||
| r889 | |||
|
Miguel Valdez
|
r741 | self.__formated_value = new_value | |
| r889 | |||
|
Miguel Valdez
|
r529 | return self.__formated_value | |
| r889 | |||
|
Miguel Valdez
|
r600 | if format == 'date': | |
|
Miguel Valdez
|
r224 | strList = value.split('/') | |
|
Miguel Valdez
|
r197 | intList = [int(x) for x in strList] | |
| date = datetime.date(intList[0], intList[1], intList[2]) | |||
| r889 | |||
|
Miguel Valdez
|
r529 | self.__formated_value = date | |
| r889 | |||
|
Miguel Valdez
|
r529 | return self.__formated_value | |
| r889 | |||
|
Miguel Valdez
|
r600 | if format == 'time': | |
|
Miguel Valdez
|
r224 | strList = value.split(':') | |
|
Miguel Valdez
|
r197 | intList = [int(x) for x in strList] | |
| time = datetime.time(intList[0], intList[1], intList[2]) | |||
| r889 | |||
|
Miguel Valdez
|
r529 | self.__formated_value = time | |
| r889 | |||
|
Miguel Valdez
|
r529 | return self.__formated_value | |
| r889 | |||
|
Miguel Valdez
|
r600 | if format == 'pairslist': | |
|
Juan C. Espinoza
|
r1052 | ''' | |
Daniel Valdez
|
r226 | Example: | |
| value = (0,1),(1,2) | |||
|
Juan C. Espinoza
|
r1052 | ''' | |
|
Daniel Valdez
|
r226 | ||
|
Miguel Valdez
|
r735 | new_value = ast.literal_eval(value) | |
| r889 | |||
|
Miguel Valdez
|
r735 | if type(new_value) not in (tuple, list): | |
|
George Yong
|
r1167 | raise ValueError('%s has to be a tuple or list of pairs' % value) | |
| r889 | |||
|
Miguel Valdez
|
r735 | if type(new_value[0]) not in (tuple, list): | |
| if len(new_value) != 2: | |||
|
George Yong
|
r1167 | raise ValueError('%s has to be a tuple or list of pairs' % value) | |
|
Miguel Valdez
|
r735 | new_value = [new_value] | |
| r889 | |||
|
Miguel Valdez
|
r735 | for thisPair in new_value: | |
| if len(thisPair) != 2: | |||
|
George Yong
|
r1167 | raise ValueError('%s has to be a tuple or list of pairs' % value) | |
| r889 | |||
|
Miguel Valdez
|
r735 | self.__formated_value = new_value | |
| r889 | |||
|
Miguel Valdez
|
r529 | return self.__formated_value | |
| r889 | |||
|
Miguel Valdez
|
r600 | if format == 'multilist': | |
|
Juan C. Espinoza
|
r1052 | ''' | |
|
Julio Valdez
|
r514 | Example: | |
| value = (0,1,2),(3,4,5) | |||
|
Juan C. Espinoza
|
r1052 | ''' | |
|
Julio Valdez
|
r514 | multiList = ast.literal_eval(value) | |
| r889 | |||
|
Miguel Valdez
|
r600 | if type(multiList[0]) == int: | |
|
Juan C. Espinoza
|
r1052 | multiList = ast.literal_eval('(' + value + ')') | |
| r889 | |||
|
Miguel Valdez
|
r529 | self.__formated_value = multiList | |
| r889 | |||
|
Miguel Valdez
|
r529 | return self.__formated_value | |
| r889 | |||
|
Miguel Valdez
|
r677 | if format == 'bool': | |
| value = int(value) | |||
| r889 | |||
|
Miguel Valdez
|
r677 | if format == 'int': | |
| value = float(value) | |||
| r889 | |||
|
Miguel Valdez
|
r600 | format_func = eval(format) | |
| r889 | |||
|
Miguel Valdez
|
r529 | self.__formated_value = format_func(value) | |
| r889 | |||
|
Miguel Valdez
|
r529 | return self.__formated_value | |
|
Miguel Valdez
|
r596 | ||
| def updateId(self, new_id): | |||
| r889 | |||
|
Miguel Valdez
|
r596 | self.id = str(new_id) | |
| r889 | |||
|
Miguel Valdez
|
r199 | def setup(self, id, name, value, format='str'): | |
|
Miguel Valdez
|
r596 | self.id = str(id) | |
|
Miguel Valdez
|
r197 | self.name = name | |
| r892 | if format == 'obj': | ||
| self.value = value | |||
| else: | |||
| self.value = str(value) | |||
|
Miguel Valdez
|
r535 | self.format = str.lower(format) | |
| r889 | |||
|
Miguel Valdez
|
r735 | self.getValue() | |
| r889 | |||
|
Miguel Valdez
|
r643 | return 1 | |
| r889 | |||
|
Miguel Valdez
|
r577 | def update(self, name, value, format='str'): | |
| r889 | |||
|
Miguel Valdez
|
r577 | self.name = name | |
| self.value = str(value) | |||
| self.format = format | |||
| r889 | |||
|
Miguel Valdez
|
r197 | def makeXml(self, opElement): | |
|
José Chávez
|
r898 | if self.name not in ('queue',): | |
| parmElement = SubElement(opElement, self.ELEMENTNAME) | |||
| parmElement.set('id', str(self.id)) | |||
| parmElement.set('name', self.name) | |||
| parmElement.set('value', self.value) | |||
| parmElement.set('format', self.format) | |||
| r889 | |||
|
Miguel Valdez
|
r197 | def readXml(self, parmElement): | |
| r889 | |||
|
Miguel Valdez
|
r197 | self.id = parmElement.get('id') | |
| self.name = parmElement.get('name') | |||
| self.value = parmElement.get('value') | |||
|
Miguel Valdez
|
r568 | self.format = str.lower(parmElement.get('format')) | |
| r889 | |||
|
José Chávez
|
r1082 | # Compatible with old signal chain version | |
|
Miguel Valdez
|
r568 | if self.format == 'int' and self.name == 'idfigure': | |
| self.name = 'id' | |||
| r889 | |||
|
Miguel Valdez
|
r197 | def printattr(self): | |
| r889 | |||
|
George Yong
|
r1167 | print('Parameter[%s]: name = %s, value = %s, format = %s' % (self.id, self.name, self.value, self.format)) | |
|
José Chávez
|
r1082 | ||
|
Juan C. Espinoza
|
r1052 | class OperationConf(): | |
| r889 | |||
|
Miguel Valdez
|
r197 | id = None | |
| name = None | |||
| priority = None | |||
| type = None | |||
| r889 | |||
|
Miguel Valdez
|
r197 | parmConfObjList = [] | |
| r889 | |||
|
Miguel Valdez
|
r197 | ELEMENTNAME = 'Operation' | |
| r889 | |||
|
Miguel Valdez
|
r197 | def __init__(self): | |
| r889 | |||
|
Miguel Valdez
|
r589 | self.id = '0' | |
|
Miguel Valdez
|
r568 | self.name = None | |
| self.priority = None | |||
|
George Yong
|
r1171 | self.topic = None | |
| r889 | |||
|
Miguel Valdez
|
r197 | def __getNewId(self): | |
| r889 | |||
|
José Chávez
|
r1082 | return int(self.id) * 10 + len(self.parmConfObjList) + 1 | |
|
Miguel Valdez
|
r596 | ||
|
George Yong
|
r1171 | def getId(self): | |
| return self.id | |||
|
Miguel Valdez
|
r596 | def updateId(self, new_id): | |
| r889 | |||
|
Miguel Valdez
|
r596 | self.id = str(new_id) | |
| r889 | |||
|
Miguel Valdez
|
r596 | n = 1 | |
| for parmObj in self.parmConfObjList: | |||
| r889 | |||
|
José Chávez
|
r1082 | idParm = str(int(new_id) * 10 + n) | |
|
Miguel Valdez
|
r596 | parmObj.updateId(idParm) | |
| r889 | |||
|
Miguel Valdez
|
r596 | n += 1 | |
| r889 | |||
|
Miguel Valdez
|
r197 | def getElementName(self): | |
| r889 | |||
|
Miguel Valdez
|
r197 | return self.ELEMENTNAME | |
| r889 | |||
|
Miguel Valdez
|
r197 | def getParameterObjList(self): | |
| r889 | |||
|
Miguel Valdez
|
r197 | return self.parmConfObjList | |
| r889 | |||
|
Miguel Valdez
|
r577 | def getParameterObj(self, parameterName): | |
| r889 | |||
|
Miguel Valdez
|
r577 | for parmConfObj in self.parmConfObjList: | |
| r889 | |||
|
Miguel Valdez
|
r577 | if parmConfObj.name != parameterName: | |
| continue | |||
| r889 | |||
|
Miguel Valdez
|
r577 | return parmConfObj | |
| r889 | |||
|
Miguel Valdez
|
r577 | return None | |
|
Miguel Valdez
|
r606 | def getParameterObjfromValue(self, parameterValue): | |
| r889 | |||
|
Miguel Valdez
|
r577 | for parmConfObj in self.parmConfObjList: | |
| r889 | |||
|
Miguel Valdez
|
r577 | if parmConfObj.getValue() != parameterValue: | |
| continue | |||
| r889 | |||
|
Miguel Valdez
|
r577 | return parmConfObj.getValue() | |
| r889 | |||
|
Miguel Valdez
|
r577 | return None | |
| r889 | |||
|
Miguel Valdez
|
r577 | def getParameterValue(self, parameterName): | |
| r889 | |||
|
Miguel Valdez
|
r577 | parameterObj = self.getParameterObj(parameterName) | |
|
José Chávez
|
r1082 | ||
|
José Chávez
|
r1004 | # if not parameterObj: | |
|
José Chávez
|
r1011 | # return None | |
|
José Chávez
|
r1082 | ||
|
Miguel Valdez
|
r577 | value = parameterObj.getValue() | |
| r889 | |||
|
Miguel Valdez
|
r577 | return value | |
| r889 | |||
| def getKwargs(self): | |||
| kwargs = {} | |||
| for parmConfObj in self.parmConfObjList: | |||
| if self.name == 'run' and parmConfObj.name == 'datatype': | |||
| continue | |||
| kwargs[parmConfObj.name] = parmConfObj.getValue() | |||
| return kwargs | |||
|
Miguel Valdez
|
r197 | def setup(self, id, name, priority, type): | |
| r889 | |||
|
Miguel Valdez
|
r596 | self.id = str(id) | |
|
Miguel Valdez
|
r197 | self.name = name | |
| self.type = type | |||
| self.priority = priority | |||
| self.parmConfObjList = [] | |||
| r889 | |||
|
Miguel Valdez
|
r577 | def removeParameters(self): | |
| r889 | |||
|
Miguel Valdez
|
r577 | for obj in self.parmConfObjList: | |
| del obj | |||
| r889 | |||
|
Miguel Valdez
|
r577 | self.parmConfObjList = [] | |
| r889 | |||
|
Miguel Valdez
|
r199 | def addParameter(self, name, value, format='str'): | |
|
José Chávez
|
r1082 | ||
|
Juan C. Espinoza
|
r1052 | if value is None: | |
| return None | |||
|
Miguel Valdez
|
r197 | id = self.__getNewId() | |
| r889 | |||
|
Miguel Valdez
|
r197 | parmConfObj = ParameterConf() | |
|
Miguel Valdez
|
r643 | if not parmConfObj.setup(id, name, value, format): | |
| return None | |||
| r889 | |||
|
Miguel Valdez
|
r197 | self.parmConfObjList.append(parmConfObj) | |
| r889 | |||
|
Miguel Valdez
|
r197 | return parmConfObj | |
| r889 | |||
|
Miguel Valdez
|
r577 | def changeParameter(self, name, value, format='str'): | |
| r889 | |||
|
Miguel Valdez
|
r577 | parmConfObj = self.getParameterObj(name) | |
| parmConfObj.update(name, value, format) | |||
| r889 | |||
|
Miguel Valdez
|
r577 | return parmConfObj | |
| r889 | |||
|
Miguel Valdez
|
r681 | def makeXml(self, procUnitElement): | |
| r889 | |||
|
Miguel Valdez
|
r681 | opElement = SubElement(procUnitElement, self.ELEMENTNAME) | |
|
Miguel Valdez
|
r197 | opElement.set('id', str(self.id)) | |
| opElement.set('name', self.name) | |||
| opElement.set('type', self.type) | |||
| opElement.set('priority', str(self.priority)) | |||
| r889 | |||
|
Miguel Valdez
|
r197 | for parmConfObj in self.parmConfObjList: | |
| parmConfObj.makeXml(opElement) | |||
| r889 | |||
|
Miguel Valdez
|
r197 | def readXml(self, opElement): | |
| r889 | |||
|
Miguel Valdez
|
r197 | self.id = opElement.get('id') | |
| self.name = opElement.get('name') | |||
| self.type = opElement.get('type') | |||
| self.priority = opElement.get('priority') | |||
| r889 | |||
|
José Chávez
|
r1082 | # Compatible with old signal chain version | |
| # Use of 'run' method instead 'init' | |||
|
Miguel Valdez
|
r568 | if self.type == 'self' and self.name == 'init': | |
| self.name = 'run' | |||
| r889 | |||
|
Miguel Valdez
|
r197 | self.parmConfObjList = [] | |
| r889 | |||
|
Juan C. Valdez
|
r860 | parmElementList = opElement.iter(ParameterConf().getElementName()) | |
| r889 | |||
|
Miguel Valdez
|
r197 | for parmElement in parmElementList: | |
| parmConfObj = ParameterConf() | |||
| parmConfObj.readXml(parmElement) | |||
| r889 | |||
|
José Chávez
|
r1082 | # Compatible with old signal chain version | |
| # If an 'plot' OPERATION is found, changes name operation by the value of its type PARAMETER | |||
|
Miguel Valdez
|
r568 | if self.type != 'self' and self.name == 'Plot': | |
| if parmConfObj.format == 'str' and parmConfObj.name == 'type': | |||
| self.name = parmConfObj.value | |||
| continue | |||
| r889 | |||
|
Miguel Valdez
|
r568 | self.parmConfObjList.append(parmConfObj) | |
| r889 | |||
|
Miguel Valdez
|
r197 | def printattr(self): | |
| r889 | |||
|
George Yong
|
r1167 | print('%s[%s]: name = %s, type = %s, priority = %s' % (self.ELEMENTNAME, | |
|
José Chávez
|
r1082 | self.id, | |
| self.name, | |||
| self.type, | |||
|
George Yong
|
r1167 | self.priority)) | |
| r889 | |||
|
Miguel Valdez
|
r197 | for parmConfObj in self.parmConfObjList: | |
| parmConfObj.printattr() | |||
| r889 | |||
|
George Yong
|
r1171 | def createObject(self): | |
| r889 | |||
|
George Yong
|
r1171 | className = eval(self.name) | |
| kwargs = self.getKwargs() | |||
| r889 | |||
|
George Yong
|
r1171 | opObj = className(self.id, **kwargs) | |
|
José Chávez
|
r924 | ||
|
George Yong
|
r1171 | opObj.start() | |
|
José Chávez
|
r924 | ||
|
George Yong
|
r1171 | print(' Operation created') | |
| r889 | |||
|
Miguel Valdez
|
r197 | return opObj | |
| r889 | |||
|
Miguel Valdez
|
r197 | class ProcUnitConf(): | |
| r889 | |||
|
Miguel Valdez
|
r197 | id = None | |
| name = None | |||
|
Miguel Valdez
|
r199 | datatype = None | |
|
Miguel Valdez
|
r197 | inputId = None | |
|
Miguel Valdez
|
r577 | parentId = None | |
| r889 | |||
|
Miguel Valdez
|
r197 | opConfObjList = [] | |
| r889 | |||
|
Miguel Valdez
|
r197 | procUnitObj = None | |
| opObjList = [] | |||
| r889 | |||
|
Miguel Valdez
|
r197 | ELEMENTNAME = 'ProcUnit' | |
| r889 | |||
|
Miguel Valdez
|
r197 | def __init__(self): | |
| r889 | |||
|
Miguel Valdez
|
r197 | self.id = None | |
|
Miguel Valdez
|
r199 | self.datatype = None | |
|
Miguel Valdez
|
r197 | self.name = None | |
|
George Yong
|
r1171 | self.inputId = None | |
| r889 | |||
|
Miguel Valdez
|
r197 | self.opConfObjList = [] | |
| r889 | |||
|
Miguel Valdez
|
r197 | self.procUnitObj = None | |
| self.opObjDict = {} | |||
| r889 | |||
|
Miguel Valdez
|
r197 | def __getPriority(self): | |
| r889 | |||
|
José Chávez
|
r1082 | return len(self.opConfObjList) + 1 | |
| r889 | |||
|
Miguel Valdez
|
r197 | def __getNewId(self): | |
| r889 | |||
|
José Chávez
|
r1082 | return int(self.id) * 10 + len(self.opConfObjList) + 1 | |
| r889 | |||
|
Miguel Valdez
|
r197 | def getElementName(self): | |
| r889 | |||
|
Miguel Valdez
|
r197 | return self.ELEMENTNAME | |
| r889 | |||
|
Miguel Valdez
|
r197 | def getId(self): | |
| r889 | |||
|
Miguel Valdez
|
r589 | return self.id | |
|
Miguel Valdez
|
r596 | ||
|
George Yong
|
r1171 | def updateId(self, new_id, parentId=parentId): | |
| ''' | |||
|
José Chávez
|
r1082 | new_id = int(parentId) * 10 + (int(self.id) % 10) | |
| new_inputId = int(parentId) * 10 + (int(self.inputId) % 10) | |||
| r889 | |||
|
José Chávez
|
r1082 | # If this proc unit has not inputs | |
|
George Yong
|
r1171 | #if self.inputId == '0': | |
| #new_inputId = 0 | |||
| r889 | |||
|
Miguel Valdez
|
r596 | n = 1 | |
| for opConfObj in self.opConfObjList: | |||
| r889 | |||
|
José Chávez
|
r1082 | idOp = str(int(new_id) * 10 + n) | |
|
Miguel Valdez
|
r596 | opConfObj.updateId(idOp) | |
| r889 | |||
|
Miguel Valdez
|
r596 | n += 1 | |
| r889 | |||
|
Miguel Valdez
|
r596 | self.parentId = str(parentId) | |
| self.id = str(new_id) | |||
|
George Yong
|
r1171 | #self.inputId = str(new_inputId) | |
| ''' | |||
| n = 1 | |||
|
Miguel Valdez
|
r197 | def getInputId(self): | |
| r889 | |||
|
Miguel Valdez
|
r589 | return self.inputId | |
| r889 | |||
|
Miguel Valdez
|
r197 | def getOperationObjList(self): | |
| r889 | |||
|
Miguel Valdez
|
r197 | return self.opConfObjList | |
| r889 | |||
|
Miguel Valdez
|
r577 | def getOperationObj(self, name=None): | |
| r889 | |||
|
Miguel Valdez
|
r577 | for opConfObj in self.opConfObjList: | |
| r889 | |||
|
Miguel Valdez
|
r577 | if opConfObj.name != name: | |
| continue | |||
| r889 | |||
|
Miguel Valdez
|
r577 | return opConfObj | |
| r889 | |||
|
Miguel Valdez
|
r577 | return None | |
| r889 | |||
|
Miguel Valdez
|
r606 | def getOpObjfromParamValue(self, value=None): | |
| r889 | |||
|
Miguel Valdez
|
r577 | for opConfObj in self.opConfObjList: | |
| if opConfObj.getParameterObjfromValue(parameterValue=value) != value: | |||
| continue | |||
| return opConfObj | |||
| return None | |||
| r889 | |||
|
Miguel Valdez
|
r197 | def getProcUnitObj(self): | |
| r889 | |||
|
Miguel Valdez
|
r197 | return self.procUnitObj | |
| r889 | |||
|
Miguel Valdez
|
r577 | def setup(self, id, name, datatype, inputId, parentId=None): | |
|
George Yong
|
r1171 | ''' | |
| id sera el topico a publicar | |||
| inputId sera el topico a subscribirse | |||
| ''' | |||
|
José Chávez
|
r1082 | # Compatible with old signal chain version | |
| if datatype == None and name == None: | |||
|
George Yong
|
r1167 | raise ValueError('datatype or name should be defined') | |
| r889 | |||
|
George Yong
|
r1171 | #Definir una condicion para inputId cuando sea 0 | |
|
José Chávez
|
r1082 | if name == None: | |
|
Miguel Valdez
|
r596 | if 'Proc' in datatype: | |
| name = datatype | |||
| else: | |||
|
José Chávez
|
r1082 | name = '%sProc' % (datatype) | |
| r889 | |||
|
José Chávez
|
r1082 | if datatype == None: | |
| datatype = name.replace('Proc', '') | |||
| r889 | |||
|
Miguel Valdez
|
r596 | self.id = str(id) | |
|
Miguel Valdez
|
r197 | self.name = name | |
|
Miguel Valdez
|
r199 | self.datatype = datatype | |
|
George Yong
|
r1171 | self.inputId = inputId | |
|
Miguel Valdez
|
r577 | self.parentId = parentId | |
|
Miguel Valdez
|
r197 | self.opConfObjList = [] | |
| r889 | |||
|
George Yong
|
r1171 | self.addOperation(name='run', optype='self') | |
| r889 | |||
|
Miguel Valdez
|
r577 | def removeOperations(self): | |
| r889 | |||
|
Miguel Valdez
|
r577 | for obj in self.opConfObjList: | |
| del obj | |||
| r889 | |||
|
Miguel Valdez
|
r577 | self.opConfObjList = [] | |
| self.addOperation(name='run') | |||
| r889 | |||
|
Miguel Valdez
|
r219 | def addParameter(self, **kwargs): | |
|
Miguel Valdez
|
r580 | ''' | |
|
Juan C. Espinoza
|
r1052 | Add parameters to 'run' operation | |
|
Miguel Valdez
|
r580 | ''' | |
|
Miguel Valdez
|
r219 | opObj = self.opConfObjList[0] | |
| r889 | |||
|
Miguel Valdez
|
r219 | opObj.addParameter(**kwargs) | |
| r889 | |||
|
Miguel Valdez
|
r219 | return opObj | |
| r889 | |||
|
George Yong
|
r1171 | def addOperation(self, name, optype = 'self'): | |
| ''' | |||
| Actualizacion - > proceso comunicacion | |||
| En el caso de optype='self', elminar. DEfinir comuncacion IPC -> Topic | |||
| definir el tipoc de socket o comunicacion ipc++ | |||
| ''' | |||
| r889 | |||
|
Miguel Valdez
|
r197 | id = self.__getNewId() | |
|
George Yong
|
r1171 | priority = self.__getPriority() # Sin mucho sentido, pero puede usarse | |
| r889 | |||
|
Miguel Valdez
|
r197 | opConfObj = OperationConf() | |
| opConfObj.setup(id, name=name, priority=priority, type=optype) | |||
| r889 | |||
|
Miguel Valdez
|
r197 | self.opConfObjList.append(opConfObj) | |
| r889 | |||
|
Miguel Valdez
|
r197 | return opConfObj | |
| r889 | |||
|
Miguel Valdez
|
r681 | def makeXml(self, projectElement): | |
| r889 | |||
|
Miguel Valdez
|
r681 | procUnitElement = SubElement(projectElement, self.ELEMENTNAME) | |
| procUnitElement.set('id', str(self.id)) | |||
| procUnitElement.set('name', self.name) | |||
| procUnitElement.set('datatype', self.datatype) | |||
| procUnitElement.set('inputId', str(self.inputId)) | |||
| r889 | |||
|
Miguel Valdez
|
r197 | for opConfObj in self.opConfObjList: | |
|
Miguel Valdez
|
r681 | opConfObj.makeXml(procUnitElement) | |
| r889 | |||
|
Miguel Valdez
|
r197 | def readXml(self, upElement): | |
| r889 | |||
|
Miguel Valdez
|
r197 | self.id = upElement.get('id') | |
| self.name = upElement.get('name') | |||
|
Miguel Valdez
|
r199 | self.datatype = upElement.get('datatype') | |
|
Miguel Valdez
|
r197 | self.inputId = upElement.get('inputId') | |
| r889 | |||
|
Juan C. Espinoza
|
r1052 | if self.ELEMENTNAME == 'ReadUnit': | |
| self.datatype = self.datatype.replace('Reader', '') | |||
| r889 | |||
|
Juan C. Espinoza
|
r1052 | if self.ELEMENTNAME == 'ProcUnit': | |
| self.datatype = self.datatype.replace('Proc', '') | |||
| r889 | |||
|
Miguel Valdez
|
r589 | if self.inputId == 'None': | |
| self.inputId = '0' | |||
| r889 | |||
|
Miguel Valdez
|
r197 | self.opConfObjList = [] | |
| r889 | |||
|
Juan C. Valdez
|
r860 | opElementList = upElement.iter(OperationConf().getElementName()) | |
| r889 | |||
|
Miguel Valdez
|
r197 | for opElement in opElementList: | |
| opConfObj = OperationConf() | |||
| opConfObj.readXml(opElement) | |||
| self.opConfObjList.append(opConfObj) | |||
| r889 | |||
|
Miguel Valdez
|
r197 | def printattr(self): | |
| r889 | |||
|
George Yong
|
r1167 | print('%s[%s]: name = %s, datatype = %s, inputId = %s' % (self.ELEMENTNAME, | |
|
José Chávez
|
r1082 | self.id, | |
| self.name, | |||
| self.datatype, | |||
|
George Yong
|
r1167 | self.inputId)) | |
|
José Chávez
|
r1082 | ||
|
Miguel Valdez
|
r197 | for opConfObj in self.opConfObjList: | |
| opConfObj.printattr() | |||
| r889 | |||
| def getKwargs(self): | |||
| opObj = self.opConfObjList[0] | |||
| kwargs = opObj.getKwargs() | |||
| return kwargs | |||
|
George Yong
|
r1171 | def createObjects(self, dictUnits): | |
| ''' | |||
| Instancia de unidades de procesamiento. | |||
| r889 | |||
|
George Yong
|
r1171 | ''' | |
|
Miguel Valdez
|
r197 | className = eval(self.name) | |
| r889 | kwargs = self.getKwargs() | ||
|
George Yong
|
r1171 | procUnitObj = className(self.id, self.inputId, dictUnits, **kwargs) # necesitan saber su id y su entrada por fines de ipc | |
| r889 | |||
|
José Chávez
|
r924 | for opConfObj in self.opConfObjList: | |
|
José Chávez
|
r1082 | if opConfObj.type == 'self' and self.name == 'run': | |
|
Juan C. Espinoza
|
r906 | continue | |
|
José Chávez
|
r1082 | elif opConfObj.type == 'self': | |
| procUnitObj.addOperationKwargs( | |||
| opConfObj.id, **opConfObj.getKwargs()) | |||
|
Miguel Valdez
|
r197 | continue | |
|
George Yong
|
r1171 | print("Creating operation process:", opConfObj.name, "for", self.name) | |
| opObj = opConfObj.createObject() | |||
| #self.opObjDict[opConfObj.id] = opObj.name | |||
| procUnitObj.addOperation(opConfObj.name, opConfObj.id) | |||
| procUnitObj.start() | |||
| r889 | |||
|
Miguel Valdez
|
r197 | self.procUnitObj = procUnitObj | |
|
George Yong
|
r1171 | ||
| r889 | |||
|
Miguel Valdez
|
r197 | return procUnitObj | |
| r889 | |||
|
Miguel Valdez
|
r197 | def run(self): | |
|
George Yong
|
r1171 | ||
| is_ok = True | |||
| """ | |||
|
Miguel Valdez
|
r197 | for opConfObj in self.opConfObjList: | |
| r889 | |||
|
Miguel Valdez
|
r197 | kwargs = {} | |
| for parmConfObj in opConfObj.getParameterObjList(): | |||
|
Miguel Valdez
|
r577 | if opConfObj.name == 'run' and parmConfObj.name == 'datatype': | |
| continue | |||
| r889 | |||
|
Miguel Valdez
|
r197 | kwargs[parmConfObj.name] = parmConfObj.getValue() | |
| r889 | |||
|
José Chávez
|
r1082 | sts = self.procUnitObj.call(opType=opConfObj.type, | |
| opName=opConfObj.name, | |||
| opId=opConfObj.id) | |||
|
Miguel Valdez
|
r672 | is_ok = is_ok or sts | |
|
George Yong
|
r1171 | ||
| """ | |||
|
Miguel Valdez
|
r672 | return is_ok | |
|
George Yong
|
r1171 | ||
|
Miguel Valdez
|
r573 | def close(self): | |
| r889 | |||
|
Miguel Valdez
|
r577 | for opConfObj in self.opConfObjList: | |
| if opConfObj.type == 'self': | |||
| continue | |||
| r889 | |||
|
Miguel Valdez
|
r577 | opObj = self.procUnitObj.getOperationObj(opConfObj.id) | |
| opObj.close() | |||
| r889 | |||
|
Miguel Valdez
|
r577 | self.procUnitObj.close() | |
| r889 | |||
|
Miguel Valdez
|
r573 | return | |
| r889 | |||
|
José Chávez
|
r1082 | ||
|
Miguel Valdez
|
r197 | class ReadUnitConf(ProcUnitConf): | |
| r889 | |||
|
Miguel Valdez
|
r197 | path = None | |
| startDate = None | |||
| endDate = None | |||
| startTime = None | |||
| endTime = None | |||
| r889 | |||
|
Miguel Valdez
|
r197 | ELEMENTNAME = 'ReadUnit' | |
| r889 | |||
|
Miguel Valdez
|
r197 | def __init__(self): | |
| r889 | |||
|
Miguel Valdez
|
r197 | self.id = None | |
|
Miguel Valdez
|
r199 | self.datatype = None | |
|
Miguel Valdez
|
r197 | self.name = None | |
|
Miguel Valdez
|
r589 | self.inputId = None | |
| r889 | |||
|
Miguel Valdez
|
r589 | self.parentId = None | |
| r889 | |||
|
Miguel Valdez
|
r197 | self.opConfObjList = [] | |
| self.opObjList = [] | |||
| r889 | |||
|
Miguel Valdez
|
r197 | def getElementName(self): | |
| r889 | |||
|
George Yong
|
r1171 | return self.ELEMENTNAME | |
|
Juan C. Espinoza
|
r1052 | def setup(self, id, name, datatype, path='', startDate='', endDate='', | |
| startTime='', endTime='', parentId=None, server=None, **kwargs): | |||
|
George Yong
|
r1171 | ||
| ''' | |||
| *****el id del proceso sera el Topico | |||
| Adicion de {topic}, si no esta presente -> error | |||
| kwargs deben ser trasmitidos en la instanciacion | |||
| ''' | |||
|
José Chávez
|
r1082 | # Compatible with old signal chain version | |
| if datatype == None and name == None: | |||
|
George Yong
|
r1167 | raise ValueError('datatype or name should be defined') | |
|
Juan C. Espinoza
|
r1088 | if name == None: | |
|
Miguel Valdez
|
r596 | if 'Reader' in datatype: | |
| name = datatype | |||
|
Juan C. Espinoza
|
r1088 | datatype = name.replace('Reader','') | |
|
Miguel Valdez
|
r596 | else: | |
|
Juan C. Espinoza
|
r1088 | name = '{}Reader'.format(datatype) | |
| if datatype == None: | |||
| if 'Reader' in name: | |||
| datatype = name.replace('Reader','') | |||
| else: | |||
| datatype = name | |||
| name = '{}Reader'.format(name) | |||
| r889 | |||
|
Miguel Valdez
|
r197 | self.id = id | |
| self.name = name | |||
|
Miguel Valdez
|
r199 | self.datatype = datatype | |
|
José Chávez
|
r963 | if path != '': | |
| self.path = os.path.abspath(path) | |||
|
Miguel Valdez
|
r197 | self.startDate = startDate | |
| self.endDate = endDate | |||
| self.startTime = startTime | |||
| self.endTime = endTime | |||
|
Miguel Valdez
|
r589 | self.inputId = '0' | |
| self.parentId = parentId | |||
|
José Chávez
|
r963 | self.server = server | |
|
Miguel Valdez
|
r224 | self.addRunOperation(**kwargs) | |
| r889 | |||
|
Juan C. Espinoza
|
r1052 | def update(self, **kwargs): | |
|
Miguel Valdez
|
r589 | ||
|
Juan C. Espinoza
|
r1052 | if 'datatype' in kwargs: | |
| datatype = kwargs.pop('datatype') | |||
|
Miguel Valdez
|
r589 | if 'Reader' in datatype: | |
|
Juan C. Espinoza
|
r1052 | self.name = datatype | |
|
Miguel Valdez
|
r589 | else: | |
|
José Chávez
|
r1082 | self.name = '%sReader' % (datatype) | |
|
Juan C. Espinoza
|
r1052 | self.datatype = self.name.replace('Reader', '') | |
| r889 | |||
|
José Chávez
|
r1082 | attrs = ('path', 'startDate', 'endDate', | |
| 'startTime', 'endTime', 'parentId') | |||
|
Juan C. Espinoza
|
r1052 | for attr in attrs: | |
| if attr in kwargs: | |||
| setattr(self, attr, kwargs.pop(attr)) | |||
|
José Chávez
|
r1082 | ||
|
Miguel Valdez
|
r596 | self.inputId = '0' | |
|
Miguel Valdez
|
r577 | self.updateRunOperation(**kwargs) | |
| r889 | |||
|
Miguel Valdez
|
r687 | def removeOperations(self): | |
| r889 | |||
|
Miguel Valdez
|
r687 | for obj in self.opConfObjList: | |
| del obj | |||
| r889 | |||
|
Miguel Valdez
|
r687 | self.opConfObjList = [] | |
| r889 | |||
|
George Yong
|
r1171 | def addRunOperation(self, **kwargs): | |
| r889 | |||
|
George Yong
|
r1171 | opObj = self.addOperation(name='run', optype='self') | |
| r889 | |||
|
José Chávez
|
r963 | if self.server is None: | |
|
José Chávez
|
r1082 | opObj.addParameter( | |
| name='datatype', value=self.datatype, format='str') | |||
|
Juan C. Espinoza
|
r1052 | opObj.addParameter(name='path', value=self.path, format='str') | |
|
José Chávez
|
r1082 | opObj.addParameter( | |
| name='startDate', value=self.startDate, format='date') | |||
| opObj.addParameter( | |||
| name='endDate', value=self.endDate, format='date') | |||
| opObj.addParameter( | |||
| name='startTime', value=self.startTime, format='time') | |||
| opObj.addParameter( | |||
| name='endTime', value=self.endTime, format='time') | |||
|
George Yong
|
r1167 | for key, value in list(kwargs.items()): | |
|
José Chávez
|
r1082 | opObj.addParameter(name=key, value=value, | |
| format=type(value).__name__) | |||
|
José Chávez
|
r963 | else: | |
|
José Chávez
|
r1082 | opObj.addParameter(name='server', value=self.server, format='str') | |
| r889 | |||
|
Miguel Valdez
|
r197 | return opObj | |
| r889 | |||
|
Miguel Valdez
|
r577 | def updateRunOperation(self, **kwargs): | |
| r889 | |||
|
Juan C. Espinoza
|
r1052 | opObj = self.getOperationObj(name='run') | |
|
Miguel Valdez
|
r577 | opObj.removeParameters() | |
| r889 | |||
|
Juan C. Espinoza
|
r1052 | opObj.addParameter(name='datatype', value=self.datatype, format='str') | |
| opObj.addParameter(name='path', value=self.path, format='str') | |||
|
José Chávez
|
r1082 | opObj.addParameter( | |
| name='startDate', value=self.startDate, format='date') | |||
|
Juan C. Espinoza
|
r1052 | opObj.addParameter(name='endDate', value=self.endDate, format='date') | |
|
José Chávez
|
r1082 | opObj.addParameter( | |
| name='startTime', value=self.startTime, format='time') | |||
|
Juan C. Espinoza
|
r1052 | opObj.addParameter(name='endTime', value=self.endTime, format='time') | |
|
José Chávez
|
r1082 | ||
|
George Yong
|
r1167 | for key, value in list(kwargs.items()): | |
|
José Chávez
|
r1082 | opObj.addParameter(name=key, value=value, | |
| format=type(value).__name__) | |||
| r889 | |||
|
Miguel Valdez
|
r577 | return opObj | |
|
José Chávez
|
r1082 | ||
|
Miguel Valdez
|
r681 | def readXml(self, upElement): | |
| r889 | |||
|
Miguel Valdez
|
r681 | self.id = upElement.get('id') | |
| self.name = upElement.get('name') | |||
| self.datatype = upElement.get('datatype') | |||
| self.inputId = upElement.get('inputId') | |||
| r889 | |||
|
Juan C. Espinoza
|
r1052 | if self.ELEMENTNAME == 'ReadUnit': | |
| self.datatype = self.datatype.replace('Reader', '') | |||
| r889 | |||
|
Miguel Valdez
|
r681 | if self.inputId == 'None': | |
| self.inputId = '0' | |||
| r889 | |||
|
Miguel Valdez
|
r681 | self.opConfObjList = [] | |
| r889 | |||
|
Juan C. Valdez
|
r860 | opElementList = upElement.iter(OperationConf().getElementName()) | |
| r889 | |||
|
Miguel Valdez
|
r681 | for opElement in opElementList: | |
| opConfObj = OperationConf() | |||
| opConfObj.readXml(opElement) | |||
| self.opConfObjList.append(opConfObj) | |||
| r889 | |||
|
Miguel Valdez
|
r681 | if opConfObj.name == 'run': | |
| self.path = opConfObj.getParameterValue('path') | |||
| self.startDate = opConfObj.getParameterValue('startDate') | |||
| self.endDate = opConfObj.getParameterValue('endDate') | |||
| self.startTime = opConfObj.getParameterValue('startTime') | |||
| self.endTime = opConfObj.getParameterValue('endTime') | |||
| r889 | |||
|
José Chávez
|
r1082 | ||
|
José Chávez
|
r1040 | class Project(Process): | |
|
Juan C. Espinoza
|
r1052 | ||
|
Miguel Valdez
|
r197 | id = None | |
| description = None | |||
|
Miguel Valdez
|
r681 | filename = None | |
| r889 | |||
|
Miguel Valdez
|
r197 | procUnitConfObjDict = None | |
| r889 | |||
|
Miguel Valdez
|
r224 | ELEMENTNAME = 'Project' | |
| r889 | |||
|
George Yong
|
r1171 | ||
| r889 | |||
|
George Yong
|
r1171 | def __init__(self): | |
|
Juan C. Espinoza
|
r1052 | ||
|
José Chávez
|
r1040 | Process.__init__(self) | |
|
George Yong
|
r1171 | self.id = None | |
|
Miguel Valdez
|
r197 | self.description = None | |
|
Juan C. Espinoza
|
r1126 | self.email = None | |
| r1130 | self.alarm = None | ||
|
Miguel Valdez
|
r197 | self.procUnitConfObjDict = {} | |
|
Juan C. Espinoza
|
r1052 | ||
|
Miguel Valdez
|
r197 | def __getNewId(self): | |
| r889 | |||
|
George Yong
|
r1167 | idList = list(self.procUnitConfObjDict.keys()) | |
| r889 | |||
|
José Chávez
|
r1082 | id = int(self.id) * 10 | |
| r889 | |||
|
Miguel Valdez
|
r764 | while True: | |
| id += 1 | |||
| r889 | |||
|
Miguel Valdez
|
r764 | if str(id) in idList: | |
| continue | |||
| r889 | |||
|
Miguel Valdez
|
r764 | break | |
| r889 | |||
|
Miguel Valdez
|
r197 | return str(id) | |
| r889 | |||
|
Miguel Valdez
|
r197 | def getElementName(self): | |
| r889 | |||
|
Miguel Valdez
|
r197 | return self.ELEMENTNAME | |
|
Miguel Valdez
|
r577 | ||
| def getId(self): | |||
| r889 | |||
|
Miguel Valdez
|
r577 | return self.id | |
| r889 | |||
|
Miguel Valdez
|
r596 | def updateId(self, new_id): | |
| r889 | |||
|
Miguel Valdez
|
r596 | self.id = str(new_id) | |
| r889 | |||
|
George Yong
|
r1167 | keyList = list(self.procUnitConfObjDict.keys()) | |
|
Miguel Valdez
|
r596 | keyList.sort() | |
| r889 | |||
|
Miguel Valdez
|
r596 | n = 1 | |
| newProcUnitConfObjDict = {} | |||
| r889 | |||
|
Miguel Valdez
|
r596 | for procKey in keyList: | |
| r889 | |||
|
Miguel Valdez
|
r596 | procUnitConfObj = self.procUnitConfObjDict[procKey] | |
|
José Chávez
|
r1082 | idProcUnit = str(int(self.id) * 10 + n) | |
| procUnitConfObj.updateId(idProcUnit, parentId=self.id) | |||
|
Miguel Valdez
|
r596 | newProcUnitConfObjDict[idProcUnit] = procUnitConfObj | |
| n += 1 | |||
| r889 | |||
|
Miguel Valdez
|
r596 | self.procUnitConfObjDict = newProcUnitConfObjDict | |
| r889 | |||
|
Juan C. Espinoza
|
r1163 | def setup(self, id, name='', description='', email=None, alarm=[]): | |
| r889 | |||
|
George Yong
|
r1171 | print(' ') | |
|
George Yong
|
r1167 | print('*' * 60) | |
|
George Yong
|
r1171 | print('* Starting SIGNAL CHAIN PROCESSING (Multiprocessing) v%s *' % schainpy.__version__) | |
|
George Yong
|
r1167 | print('*' * 60) | |
|
George Yong
|
r1171 | print("* Python " + python_version() + " *") | |
| print('*' * 19) | |||
| print(' ') | |||
|
Miguel Valdez
|
r596 | self.id = str(id) | |
|
George Yong
|
r1171 | self.description = description | |
|
Juan C. Espinoza
|
r1126 | self.email = email | |
| self.alarm = alarm | |||
|
Miguel Valdez
|
r577 | ||
|
Juan C. Espinoza
|
r1126 | def update(self, **kwargs): | |
| r889 | |||
|
George Yong
|
r1167 | for key, value in list(kwargs.items()): | |
|
Juan C. Espinoza
|
r1126 | setattr(self, key, value) | |
| r889 | |||
|
Juan C. Espinoza
|
r1052 | def clone(self): | |
| p = Project() | |||
| p.procUnitConfObjDict = self.procUnitConfObjDict | |||
| return p | |||
|
Miguel Valdez
|
r687 | def addReadUnit(self, id=None, datatype=None, name=None, **kwargs): | |
|
Juan C. Espinoza
|
r1052 | ||
|
George Yong
|
r1171 | ''' | |
| Actualizacion: | |||
| Se agrego un nuevo argumento: topic -relativo a la forma de comunicar los procesos simultaneos | |||
| * El id del proceso sera el topico al que se deben subscribir los procUnits para recibir la informacion(data) | |||
| ''' | |||
|
Miguel Valdez
|
r687 | if id is None: | |
| idReadUnit = self.__getNewId() | |||
| else: | |||
| idReadUnit = str(id) | |||
| r889 | |||
|
Miguel Valdez
|
r197 | readUnitConfObj = ReadUnitConf() | |
|
José Chávez
|
r1082 | readUnitConfObj.setup(idReadUnit, name, datatype, | |
| parentId=self.id, **kwargs) | |||
| r889 | |||
|
Miguel Valdez
|
r197 | self.procUnitConfObjDict[readUnitConfObj.getId()] = readUnitConfObj | |
|
George Yong
|
r1171 | ||
|
Miguel Valdez
|
r197 | return readUnitConfObj | |
| r889 | |||
|
Miguel Valdez
|
r589 | def addProcUnit(self, inputId='0', datatype=None, name=None): | |
| r889 | |||
|
George Yong
|
r1171 | ''' | |
| Actualizacion: | |||
| Se agrego dos nuevos argumentos: topic_read (lee data de otro procUnit) y topic_write(escribe o envia data a otro procUnit) | |||
| Deberia reemplazar a "inputId" | |||
| ** A fin de mantener el inputID, este sera la representaacion del topicoal que deben subscribirse. El ID propio de la intancia | |||
| (proceso) sera el topico de la publicacion, todo sera asignado de manera dinamica. | |||
| ''' | |||
| idProcUnit = self.__getNewId() #Topico para subscripcion | |||
| r889 | |||
|
Miguel Valdez
|
r197 | procUnitConfObj = ProcUnitConf() | |
|
George Yong
|
r1171 | procUnitConfObj.setup(idProcUnit, name, datatype, inputId, #topic_read, topic_write, | |
| parentId=self.id) | |||
| r889 | |||
|
Miguel Valdez
|
r197 | self.procUnitConfObjDict[procUnitConfObj.getId()] = procUnitConfObj | |
| r889 | |||
|
Miguel Valdez
|
r197 | return procUnitConfObj | |
| r889 | |||
|
Miguel Valdez
|
r587 | def removeProcUnit(self, id): | |
| r889 | |||
|
George Yong
|
r1167 | if id in list(self.procUnitConfObjDict.keys()): | |
|
Miguel Valdez
|
r587 | self.procUnitConfObjDict.pop(id) | |
| r889 | |||
|
Miguel Valdez
|
r577 | def getReadUnitId(self): | |
| r889 | |||
|
Miguel Valdez
|
r577 | readUnitConfObj = self.getReadUnitObj() | |
| r889 | |||
|
Miguel Valdez
|
r577 | return readUnitConfObj.id | |
| r889 | |||
|
Miguel Valdez
|
r577 | def getReadUnitObj(self): | |
| r889 | |||
|
George Yong
|
r1167 | for obj in list(self.procUnitConfObjDict.values()): | |
|
Juan C. Espinoza
|
r1052 | if obj.getElementName() == 'ReadUnit': | |
|
Miguel Valdez
|
r577 | return obj | |
| r889 | |||
|
Miguel Valdez
|
r577 | return None | |
| r889 | |||
|
Miguel Valdez
|
r606 | def getProcUnitObj(self, id=None, name=None): | |
| r889 | |||
|
Miguel Valdez
|
r606 | if id != None: | |
| return self.procUnitConfObjDict[id] | |||
| r889 | |||
|
Miguel Valdez
|
r606 | if name != None: | |
| return self.getProcUnitObjByName(name) | |||
| r889 | |||
|
Miguel Valdez
|
r606 | return None | |
| r889 | |||
|
Miguel Valdez
|
r581 | def getProcUnitObjByName(self, name): | |
| r889 | |||
|
George Yong
|
r1167 | for obj in list(self.procUnitConfObjDict.values()): | |
|
Miguel Valdez
|
r581 | if obj.name == name: | |
| return obj | |||
| r889 | |||
|
Miguel Valdez
|
r581 | return None | |
| r889 | |||
|
Miguel Valdez
|
r606 | def procUnitItems(self): | |
| r889 | |||
|
George Yong
|
r1167 | return list(self.procUnitConfObjDict.items()) | |
| r889 | |||
| def makeXml(self): | |||
|
Miguel Valdez
|
r224 | projectElement = Element('Project') | |
|
Miguel Valdez
|
r197 | projectElement.set('id', str(self.id)) | |
| projectElement.set('name', self.name) | |||
| projectElement.set('description', self.description) | |||
| r889 | |||
|
George Yong
|
r1167 | for procUnitConfObj in list(self.procUnitConfObjDict.values()): | |
|
Miguel Valdez
|
r197 | procUnitConfObj.makeXml(projectElement) | |
| r889 | |||
|
Miguel Valdez
|
r197 | self.projectElement = projectElement | |
| r889 | |||
|
Miguel Valdez
|
r702 | def writeXml(self, filename=None): | |
| r889 | |||
|
Miguel Valdez
|
r702 | if filename == None: | |
|
Miguel Valdez
|
r708 | if self.filename: | |
| filename = self.filename | |||
| else: | |||
|
Juan C. Espinoza
|
r1052 | filename = 'schain.xml' | |
| r889 | |||
|
Miguel Valdez
|
r702 | if not filename: | |
|
George Yong
|
r1167 | print('filename has not been defined. Use setFilename(filename) for do it.') | |
|
Miguel Valdez
|
r702 | return 0 | |
| r889 | |||
|
Miguel Valdez
|
r687 | abs_file = os.path.abspath(filename) | |
| r889 | |||
|
Miguel Valdez
|
r687 | if not os.access(os.path.dirname(abs_file), os.W_OK): | |
|
George Yong
|
r1167 | print('No write permission on %s' % os.path.dirname(abs_file)) | |
|
Miguel Valdez
|
r681 | return 0 | |
| r889 | |||
|
Miguel Valdez
|
r687 | if os.path.isfile(abs_file) and not(os.access(abs_file, os.W_OK)): | |
|
George Yong
|
r1167 | print('File %s already exists and it could not be overwriten' % abs_file) | |
|
Miguel Valdez
|
r681 | return 0 | |
| r889 | |||
|
Miguel Valdez
|
r681 | self.makeXml() | |
| r889 | |||
|
Miguel Valdez
|
r687 | ElementTree(self.projectElement).write(abs_file, method='xml') | |
| r889 | |||
|
Miguel Valdez
|
r708 | self.filename = abs_file | |
| r889 | |||
|
Miguel Valdez
|
r681 | return 1 | |
|
Miguel Valdez
|
r197 | ||
|
José Chávez
|
r1082 | def readXml(self, filename=None): | |
| r889 | |||
|
Miguel Valdez
|
r830 | if not filename: | |
|
George Yong
|
r1167 | print('filename is not defined') | |
|
Miguel Valdez
|
r830 | return 0 | |
| r889 | |||
|
Miguel Valdez
|
r687 | abs_file = os.path.abspath(filename) | |
| r889 | |||
|
Miguel Valdez
|
r687 | if not os.path.isfile(abs_file): | |
|
George Yong
|
r1167 | print('%s file does not exist' % abs_file) | |
|
Miguel Valdez
|
r681 | return 0 | |
| r889 | |||
|
Miguel Valdez
|
r197 | self.projectElement = None | |
| self.procUnitConfObjDict = {} | |||
| r889 | |||
|
Miguel Valdez
|
r735 | try: | |
| self.projectElement = ElementTree().parse(abs_file) | |||
| except: | |||
|
George Yong
|
r1167 | print('Error reading %s, verify file format' % filename) | |
|
Miguel Valdez
|
r735 | return 0 | |
| r889 | |||
|
Miguel Valdez
|
r197 | self.project = self.projectElement.tag | |
| r889 | |||
|
Miguel Valdez
|
r197 | self.id = self.projectElement.get('id') | |
| self.name = self.projectElement.get('name') | |||
|
José Chávez
|
r1082 | self.description = self.projectElement.get('description') | |
| readUnitElementList = self.projectElement.iter( | |||
| ReadUnitConf().getElementName()) | |||
| r889 | |||
|
Miguel Valdez
|
r197 | for readUnitElement in readUnitElementList: | |
| readUnitConfObj = ReadUnitConf() | |||
| readUnitConfObj.readXml(readUnitElement) | |||
| r889 | |||
|
Miguel Valdez
|
r589 | if readUnitConfObj.parentId == None: | |
| readUnitConfObj.parentId = self.id | |||
| r889 | |||
|
Miguel Valdez
|
r197 | self.procUnitConfObjDict[readUnitConfObj.getId()] = readUnitConfObj | |
| r889 | |||
|
José Chávez
|
r1082 | procUnitElementList = self.projectElement.iter( | |
| ProcUnitConf().getElementName()) | |||
| r889 | |||
|
Miguel Valdez
|
r197 | for procUnitElement in procUnitElementList: | |
| procUnitConfObj = ProcUnitConf() | |||
| procUnitConfObj.readXml(procUnitElement) | |||
| r889 | |||
|
Miguel Valdez
|
r589 | if procUnitConfObj.parentId == None: | |
| procUnitConfObj.parentId = self.id | |||
| r889 | |||
|
Miguel Valdez
|
r197 | self.procUnitConfObjDict[procUnitConfObj.getId()] = procUnitConfObj | |
| r889 | |||
|
Miguel Valdez
|
r708 | self.filename = abs_file | |
| r889 | |||
|
Miguel Valdez
|
r681 | return 1 | |
| r889 | |||
|
Miguel Valdez
|
r197 | def printattr(self): | |
| r889 | |||
|
George Yong
|
r1167 | print('Project[%s]: name = %s, description = %s' % (self.id, | |
|
José Chávez
|
r1082 | self.name, | |
|
George Yong
|
r1167 | self.description)) | |
|
José Chávez
|
r1082 | ||
|
George Yong
|
r1167 | for procUnitConfObj in list(self.procUnitConfObjDict.values()): | |
|
Miguel Valdez
|
r197 | procUnitConfObj.printattr() | |
| r889 | |||
|
Miguel Valdez
|
r197 | def createObjects(self): | |
| r889 | |||
|
George Yong
|
r1167 | for procUnitConfObj in list(self.procUnitConfObjDict.values()): | |
|
George Yong
|
r1171 | print("Creating process:", procUnitConfObj.name) | |
| procUnitConfObj.createObjects(self.procUnitConfObjDict) | |||
| r889 | |||
|
George Yong
|
r1171 | print('All processes were created') | |
| r889 | |||
| r1130 | def __handleError(self, procUnitConfObj, modes=None, stdout=True): | ||
| r889 | |||
|
Miguel Valdez
|
r687 | import socket | |
| r889 | |||
| r1129 | if modes is None: | ||
| modes = self.alarm | |||
|
Juan C. Espinoza
|
r1163 | ||
| if not self.alarm: | |||
| modes = [] | |||
| r1129 | |||
|
Miguel Valdez
|
r687 | err = traceback.format_exception(sys.exc_info()[0], | |
|
José Chávez
|
r1082 | sys.exc_info()[1], | |
| sys.exc_info()[2]) | |||
|
George Yong
|
r1171 | ||
| r1128 | log.error('{}'.format(err[-1]), procUnitConfObj.name) | ||
|
Miguel Valdez
|
r735 | ||
|
Juan C. Espinoza
|
r1052 | message = ''.join(err) | |
| r889 | |||
| r1130 | if stdout: | ||
| sys.stderr.write(message) | |||
| r889 | |||
|
José Chávez
|
r1082 | subject = 'SChain v%s: Error running %s\n' % ( | |
| schainpy.__version__, procUnitConfObj.name) | |||
| r889 | |||
|
José Chávez
|
r1082 | subtitle = '%s: %s\n' % ( | |
| procUnitConfObj.getElementName(), procUnitConfObj.name) | |||
| subtitle += 'Hostname: %s\n' % socket.gethostbyname( | |||
| socket.gethostname()) | |||
| subtitle += 'Working directory: %s\n' % os.path.abspath('./') | |||
| subtitle += 'Configuration file: %s\n' % self.filename | |||
| subtitle += 'Time: %s\n' % str(datetime.datetime.now()) | |||
| r889 | |||
|
Miguel Valdez
|
r687 | readUnitConfObj = self.getReadUnitObj() | |
| if readUnitConfObj: | |||
|
Juan C. Espinoza
|
r1052 | subtitle += '\nInput parameters:\n' | |
|
José Chávez
|
r1082 | subtitle += '[Data path = %s]\n' % readUnitConfObj.path | |
| subtitle += '[Data type = %s]\n' % readUnitConfObj.datatype | |||
| subtitle += '[Start date = %s]\n' % readUnitConfObj.startDate | |||
| subtitle += '[End date = %s]\n' % readUnitConfObj.endDate | |||
| subtitle += '[Start time = %s]\n' % readUnitConfObj.startTime | |||
| subtitle += '[End time = %s]\n' % readUnitConfObj.endTime | |||
| r889 | |||
| r1129 | a = Alarm( | ||
| modes=modes, | |||
|
Juan C. Espinoza
|
r1126 | email=self.email, | |
| message=message, | |||
| subject=subject, | |||
| subtitle=subtitle, | |||
| filename=self.filename | |||
| ) | |||
|
José Chávez
|
r1082 | ||
| r1130 | return a | ||
| r1129 | |||
|
Miguel Valdez
|
r672 | def isPaused(self): | |
| return 0 | |||
| r889 | |||
|
Miguel Valdez
|
r672 | def isStopped(self): | |
| return 0 | |||
| r889 | |||
|
Miguel Valdez
|
r672 | def runController(self): | |
|
Juan C. Espinoza
|
r1052 | ''' | |
|
Miguel Valdez
|
r672 | returns 0 when this process has been stopped, 1 otherwise | |
|
Juan C. Espinoza
|
r1052 | ''' | |
| r889 | |||
|
Miguel Valdez
|
r672 | if self.isPaused(): | |
|
George Yong
|
r1167 | print('Process suspended') | |
| r889 | |||
|
Miguel Valdez
|
r672 | while True: | |
|
Juan C. Espinoza
|
r1052 | time.sleep(0.1) | |
| r889 | |||
|
Miguel Valdez
|
r672 | if not self.isPaused(): | |
| break | |||
| r889 | |||
|
Miguel Valdez
|
r672 | if self.isStopped(): | |
| break | |||
| r889 | |||
|
George Yong
|
r1167 | print('Process reinitialized') | |
| r889 | |||
|
Miguel Valdez
|
r672 | if self.isStopped(): | |
|
George Yong
|
r1167 | print('Process stopped') | |
|
Miguel Valdez
|
r672 | return 0 | |
| r889 | |||
|
Miguel Valdez
|
r672 | return 1 | |
|
Miguel Valdez
|
r691 | ||
| def setFilename(self, filename): | |||
| r889 | |||
|
Miguel Valdez
|
r691 | self.filename = filename | |
| r889 | |||
|
George Yong
|
r1171 | def setProxyCom(self): | |
| ctx = zmq.Context() | |||
| if not os.path.exists('/tmp/socketTmp'): os.mkdir('/tmp/socketTmp') | |||
| xsub = ctx.socket(zmq.XSUB) | |||
| xsub.bind('ipc:///tmp/socketTmp/a') | |||
| xpub = ctx.socket(zmq.XPUB) | |||
| xpub.bind('ipc:///tmp/socketTmp/b') | |||
| print("Controller Ready: Processes and proxy created") | |||
| zmq.proxy(xsub, xpub) | |||
| r889 | |||
|
George Yong
|
r1171 | ||
| r889 | |||
|
Juan C. Espinoza
|
r1052 | def run(self): | |
|
José Chávez
|
r1004 | ||
| r1134 | log.success('Starting {}'.format(self.name), tag='') | ||
|
Juan C. Espinoza
|
r1112 | self.start_time = time.time() | |
|
George Yong
|
r1171 | self.createObjects() | |
| self.setProxyCom() | |||
| r889 | |||
|
George Yong
|
r1171 | # Iniciar todos los procesos .start(), monitoreo de procesos. ELiminar lo de abajo | |
| r889 | |||
|
José Chávez
|
r1082 | # Closing every process | |
|
Juan C. Espinoza
|
r1112 | log.success('{} finished (time: {}s)'.format( | |
| self.name, | |||
|
George Yong
|
r1167 | time.time()-self.start_time)) |
