controller.py
918 lines
| 25.5 KiB
| text/x-python
|
PythonLexer
/ schainpy / controller.py
Juan C. Espinoza
|
r1329 | # Copyright (c) 2012-2020 Jicamarca Radio Observatory | ||
| # All rights reserved. | ||||
| # | ||||
| # Distributed under the terms of the BSD 3-clause license. | ||||
| """API to create signal chain projects | ||||
| The API is provide through class: Project | ||||
| """ | ||||
|
Juan C. Espinoza
|
r1287 | |||
| import re | ||||
Miguel Valdez
|
r672 | import sys | ||
Julio Valdez
|
r514 | import ast | ||
|
Miguel Valdez
|
r687 | import datetime | ||
|
Miguel Valdez
|
r672 | import traceback | ||
José Chávez
|
r931 | import time | ||
| r1342 | import multiprocessing | |||
|
Juan C. Espinoza
|
r1287 | from multiprocessing import Process, Queue | ||
|
Juan C. Espinoza
|
r1177 | from threading import Thread | ||
|
Juan C. Espinoza
|
r1287 | from xml.etree.ElementTree import ElementTree, Element, SubElement | ||
George Yong
|
r1171 | |||
| r1129 | from schainpy.admin import Alarm, SchainWarning | |||
|
George Yong
|
r1191 | from schainpy.model import * | ||
|
Juan C. Espinoza
|
r1052 | from schainpy.utils import log | ||
|
George Yong
|
r1191 | |||
| r1342 | if 'darwin' in sys.platform and sys.version_info[0] == 3 and sys.version_info[1] > 7: | |||
| multiprocessing.set_start_method('fork') | ||||
|
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 | ||
| r1279 | 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 | ||||
|
Juan C. Espinoza
|
r1177 | def wait(context): | ||
| r1279 | ||||
|
Juan C. Espinoza
|
r1177 | time.sleep(1) | ||
| c = zmq.Context() | ||||
| receiver = c.socket(zmq.SUB) | ||||
| r1279 | receiver.connect('ipc:///tmp/schain_{}_pub'.format(self.id)) | |||
|
Juan C. Espinoza
|
r1177 | receiver.setsockopt(zmq.SUBSCRIBE, self.id.encode()) | ||
| msg = receiver.recv_multipart()[1] | ||||
| context.terminate() | ||||
|
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': | ||
|
Juan C. Espinoza
|
r1193 | strList = [s.strip() for s in value.split(',')] | ||
|
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) | ||||
| r1279 | ||||
|
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
|
r1184 | print('Parameter[%s]: name = %s, value = %s, format = %s, project_id = %s' % (self.id, self.name, self.value, self.format, self.project_id)) | ||
|
José Chávez
|
r1082 | |||
|
Juan C. Espinoza
|
r1052 | class OperationConf(): | ||
| 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 | ||||
|
Juan C. Espinoza
|
r1287 | self.parameters = {} | ||
| self.object = None | ||||
| self.operations = [] | ||||
|
Miguel Valdez
|
r596 | |||
|
George Yong
|
r1171 | def getId(self): | ||
|
Juan C. Espinoza
|
r1287 | |||
|
George Yong
|
r1171 | return self.id | ||
|
Juan C. Espinoza
|
r1287 | def getNewId(self): | ||
| return int(self.id) * 10 + len(self.operations) + 1 | ||||
|
George Yong
|
r1171 | |||
|
Miguel Valdez
|
r596 | def updateId(self, new_id): | ||
| r889 | ||||
|
Miguel Valdez
|
r596 | self.id = str(new_id) | ||
| r889 | ||||
|
Miguel Valdez
|
r596 | n = 1 | ||
|
Juan C. Espinoza
|
r1287 | for conf in self.operations: | ||
| conf_id = str(int(new_id) * 10 + n) | ||||
| conf.updateId(conf_id) | ||||
|
Miguel Valdez
|
r596 | n += 1 | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def getKwargs(self): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | params = {} | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | for key, value in self.parameters.items(): | ||
| if value not in (None, '', ' '): | ||||
| params[key] = value | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | return params | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def update(self, **kwargs): | ||
|
Miguel Valdez
|
r577 | |||
|
Juan C. Espinoza
|
r1287 | for key, value in kwargs.items(): | ||
| self.addParameter(name=key, value=value) | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def addParameter(self, name, value, format=None): | ||
| ''' | ||||
| ''' | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | if isinstance(value, str) and re.search(r'(\d+/\d+/\d+)', value): | ||
| self.parameters[name] = datetime.date(*[int(x) for x in value.split('/')]) | ||||
| elif isinstance(value, str) and re.search(r'(\d+:\d+:\d+)', value): | ||||
| self.parameters[name] = datetime.time(*[int(x) for x in value.split(':')]) | ||||
| else: | ||||
| try: | ||||
| self.parameters[name] = ast.literal_eval(value) | ||||
| except: | ||||
| if isinstance(value, str) and ',' in value: | ||||
| self.parameters[name] = value.split(',') | ||||
| else: | ||||
| self.parameters[name] = value | ||||
| def getParameters(self): | ||||
| params = {} | ||||
| for key, value in self.parameters.items(): | ||||
| s = type(value).__name__ | ||||
| if s == 'date': | ||||
| params[key] = value.strftime('%Y/%m/%d') | ||||
| elif s == 'time': | ||||
| params[key] = value.strftime('%H:%M:%S') | ||||
| else: | ||||
| params[key] = str(value) | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | return params | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def makeXml(self, element): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | xml = SubElement(element, self.ELEMENTNAME) | ||
| for label in self.xml_labels: | ||||
| xml.set(label, str(getattr(self, label))) | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | for key, value in self.getParameters().items(): | ||
| xml_param = SubElement(xml, 'Parameter') | ||||
| xml_param.set('name', key) | ||||
| xml_param.set('value', value) | ||||
|
José Chávez
|
r1082 | |||
|
Juan C. Espinoza
|
r1287 | for conf in self.operations: | ||
| conf.makeXml(xml) | ||||
|
José Chávez
|
r1082 | |||
|
Juan C. Espinoza
|
r1287 | def __str__(self): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | if self.ELEMENTNAME == 'Operation': | ||
| s = ' {}[id={}]\n'.format(self.name, self.id) | ||||
| else: | ||||
| s = '{}[id={}, inputId={}]\n'.format(self.name, self.id, self.inputId) | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | for key, value in self.parameters.items(): | ||
| if self.ELEMENTNAME == 'Operation': | ||||
| s += ' {}: {}\n'.format(key, value) | ||||
| else: | ||||
| s += ' {}: {}\n'.format(key, value) | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | for conf in self.operations: | ||
| s += str(conf) | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | return s | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | class OperationConf(ConfBase): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | ELEMENTNAME = 'Operation' | ||
| xml_labels = ['id', 'name'] | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def setup(self, id, name, priority, project_id, err_queue): | ||
| r889 | ||||
|
Miguel Valdez
|
r596 | self.id = str(id) | ||
|
Juan C. Espinoza
|
r1177 | self.project_id = project_id | ||
|
Miguel Valdez
|
r197 | self.name = name | ||
|
Juan C. Espinoza
|
r1287 | self.type = 'other' | ||
| r1241 | self.err_queue = err_queue | |||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def readXml(self, element, project_id, err_queue): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | self.id = element.get('id') | ||
| self.name = element.get('name') | ||||
| self.type = 'other' | ||||
| r1279 | self.project_id = str(project_id) | |||
|
Juan C. Espinoza
|
r1287 | self.err_queue = err_queue | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | for elm in element.iter('Parameter'): | ||
| self.addParameter(elm.get('name'), elm.get('value')) | ||||
| r889 | ||||
|
George Yong
|
r1171 | def createObject(self): | ||
| r889 | ||||
|
George Yong
|
r1171 | className = eval(self.name) | ||
|
George Yong
|
r1184 | |||
|
Juan C. Espinoza
|
r1308 | if 'Plot' in self.name or 'Writer' in self.name or 'Send' in self.name or 'print' in self.name: | ||
|
Juan C. Espinoza
|
r1177 | kwargs = self.getKwargs() | ||
|
Juan C. Espinoza
|
r1287 | opObj = className(self.id, self.id, self.project_id, self.err_queue, **kwargs) | ||
|
Juan C. Espinoza
|
r1177 | opObj.start() | ||
|
Juan C. Espinoza
|
r1287 | self.type = 'external' | ||
| else: | ||||
| opObj = className() | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | self.object = opObj | ||
|
Miguel Valdez
|
r197 | return opObj | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | class ProcUnitConf(ConfBase): | ||
| r889 | ||||
|
Miguel Valdez
|
r197 | ELEMENTNAME = 'ProcUnit' | ||
|
Juan C. Espinoza
|
r1287 | xml_labels = ['id', 'inputId', 'name'] | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def setup(self, project_id, id, name, datatype, inputId, err_queue): | ||
|
George Yong
|
r1171 | ''' | ||
| ''' | ||||
| r1279 | ||||
|
José Chávez
|
r1082 | if datatype == None and name == None: | ||
|
George Yong
|
r1167 | raise ValueError('datatype or name should be defined') | ||
| r889 | ||||
|
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) | ||
|
Juan C. Espinoza
|
r1177 | self.project_id = project_id | ||
|
Miguel Valdez
|
r197 | self.name = name | ||
|
Miguel Valdez
|
r199 | self.datatype = datatype | ||
| r1241 | self.inputId = inputId | |||
| self.err_queue = err_queue | ||||
|
Juan C. Espinoza
|
r1287 | self.operations = [] | ||
| self.parameters = {} | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def removeOperation(self, id): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | i = [1 if x.id==id else 0 for x in self.operations] | ||
| self.operations.pop(i.index(1)) | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def getOperation(self, id): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | for conf in self.operations: | ||
| if conf.id == id: | ||||
| return conf | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1177 | def addOperation(self, name, optype='self'): | ||
|
George Yong
|
r1171 | ''' | ||
| ''' | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | id = self.getNewId() | ||
| conf = OperationConf() | ||||
| conf.setup(id, name=name, priority='0', project_id=self.project_id, err_queue=self.err_queue) | ||||
| self.operations.append(conf) | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | return conf | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def readXml(self, element, project_id, err_queue): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | self.id = element.get('id') | ||
| self.name = element.get('name') | ||||
| self.inputId = None if element.get('inputId') == 'None' else element.get('inputId') | ||||
| self.datatype = element.get('datatype', self.name.replace(self.ELEMENTNAME.replace('Unit', ''), '')) | ||||
|
George Yong
|
r1184 | self.project_id = str(project_id) | ||
|
Juan C. Espinoza
|
r1287 | self.err_queue = err_queue | ||
| self.operations = [] | ||||
| self.parameters = {} | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | for elm in element: | ||
| if elm.tag == 'Parameter': | ||||
| self.addParameter(elm.get('name'), elm.get('value')) | ||||
| elif elm.tag == 'Operation': | ||||
| conf = OperationConf() | ||||
| conf.readXml(elm, project_id, err_queue) | ||||
| self.operations.append(conf) | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1177 | def createObjects(self): | ||
|
George Yong
|
r1171 | ''' | ||
|
Juan C. Espinoza
|
r1177 | Instancia de unidades de procesamiento. | ||
|
George Yong
|
r1171 | ''' | ||
|
George Yong
|
r1192 | |||
|
Miguel Valdez
|
r197 | className = eval(self.name) | ||
| r1279 | #print(self.name) | |||
| r889 | kwargs = self.getKwargs() | |||
|
Juan C. Espinoza
|
r1287 | procUnitObj = className() | ||
| procUnitObj.name = self.name | ||||
|
Juan C. Espinoza
|
r1177 | log.success('creating process...', self.name) | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | for conf in self.operations: | ||
| r1279 | ||||
|
Juan C. Espinoza
|
r1287 | opObj = conf.createObject() | ||
| r1279 | ||||
| r1241 | log.success('adding operation: {}, type:{}'.format( | |||
|
Juan C. Espinoza
|
r1287 | conf.name, | ||
| conf.type), self.name) | ||||
| r1279 | ||||
|
Juan C. Espinoza
|
r1287 | procUnitObj.addOperation(conf, opObj) | ||
| r1279 | ||||
|
Juan C. Espinoza
|
r1287 | self.object = procUnitObj | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def run(self): | ||
| ''' | ||||
| ''' | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | return self.object.call(**self.getKwargs()) | ||
| r889 | ||||
|
José Chávez
|
r1082 | |||
|
Miguel Valdez
|
r197 | class ReadUnitConf(ProcUnitConf): | ||
| 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 | ||
|
Juan C. Espinoza
|
r1287 | self.operations = [] | ||
| self.parameters = {} | ||||
| r1279 | ||||
| r1241 | def setup(self, project_id, id, name, datatype, err_queue, path='', startDate='', endDate='', | |||
|
Juan C. Espinoza
|
r1177 | startTime='', endTime='', server=None, **kwargs): | ||
|
Juan C. Espinoza
|
r1052 | |||
|
José Chávez
|
r1082 | 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 | ||
|
Juan C. Espinoza
|
r1177 | self.project_id = project_id | ||
|
Miguel Valdez
|
r197 | self.name = name | ||
|
Miguel Valdez
|
r199 | self.datatype = datatype | ||
| r1279 | self.err_queue = err_queue | |||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | self.addParameter(name='path', value=path) | ||
| self.addParameter(name='startDate', value=startDate) | ||||
| self.addParameter(name='endDate', value=endDate) | ||||
| self.addParameter(name='startTime', value=startTime) | ||||
| self.addParameter(name='endTime', value=endTime) | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | for key, value in kwargs.items(): | ||
| self.addParameter(name=key, value=value) | ||||
| r889 | ||||
|
José Chávez
|
r1082 | |||
|
José Chávez
|
r1040 | class Project(Process): | ||
|
Juan C. Espinoza
|
r1329 | """API to create signal chain projects""" | ||
|
Juan C. Espinoza
|
r1052 | |||
|
Miguel Valdez
|
r224 | ELEMENTNAME = 'Project' | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1330 | def __init__(self, name=''): | ||
|
Juan C. Espinoza
|
r1052 | |||
|
José Chávez
|
r1040 | Process.__init__(self) | ||
|
Juan C. Espinoza
|
r1329 | self.id = '1' | ||
| if name: | ||||
| self.name = '{} ({})'.format(Process.__name__, name) | ||||
|
Juan C. Espinoza
|
r1177 | self.filename = None | ||
|
Miguel Valdez
|
r197 | self.description = None | ||
|
Juan C. Espinoza
|
r1126 | self.email = None | ||
|
Juan C. Espinoza
|
r1287 | self.alarm = [] | ||
| self.configurations = {} | ||||
| # self.err_queue = Queue() | ||||
| self.err_queue = None | ||||
| self.started = False | ||||
|
Juan C. Espinoza
|
r1052 | |||
|
Juan C. Espinoza
|
r1287 | def getNewId(self): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | idList = list(self.configurations.keys()) | ||
|
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
|
r596 | def updateId(self, new_id): | ||
| r889 | ||||
|
Miguel Valdez
|
r596 | self.id = str(new_id) | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | keyList = list(self.configurations.keys()) | ||
|
Miguel Valdez
|
r596 | keyList.sort() | ||
| r889 | ||||
|
Miguel Valdez
|
r596 | n = 1 | ||
|
Juan C. Espinoza
|
r1287 | new_confs = {} | ||
| r889 | ||||
|
Miguel Valdez
|
r596 | for procKey in keyList: | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | conf = self.configurations[procKey] | ||
|
José Chávez
|
r1082 | idProcUnit = str(int(self.id) * 10 + n) | ||
|
Juan C. Espinoza
|
r1287 | conf.updateId(idProcUnit) | ||
| new_confs[idProcUnit] = conf | ||||
|
Miguel Valdez
|
r596 | n += 1 | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | self.configurations = new_confs | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1177 | def setup(self, id=1, name='', description='', email=None, alarm=[]): | ||
| r889 | ||||
|
Miguel Valdez
|
r596 | self.id = str(id) | ||
| r1279 | self.description = description | |||
|
Juan C. Espinoza
|
r1126 | self.email = email | ||
| self.alarm = alarm | ||||
| r1241 | if name: | |||
| self.name = '{} ({})'.format(Process.__name__, name) | ||||
|
Miguel Valdez
|
r577 | |||
|
Juan C. Espinoza
|
r1126 | def update(self, **kwargs): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | for key, value in kwargs.items(): | ||
|
Juan C. Espinoza
|
r1126 | setattr(self, key, value) | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1052 | def clone(self): | ||
| p = Project() | ||||
|
Juan C. Espinoza
|
r1287 | p.id = self.id | ||
| p.name = self.name | ||||
| p.description = self.description | ||||
| p.configurations = self.configurations.copy() | ||||
|
Juan C. Espinoza
|
r1052 | return p | ||
|
Miguel Valdez
|
r687 | def addReadUnit(self, id=None, datatype=None, name=None, **kwargs): | ||
|
Juan C. Espinoza
|
r1052 | |||
|
George Yong
|
r1171 | ''' | ||
| ''' | ||||
|
Miguel Valdez
|
r687 | if id is None: | ||
|
Juan C. Espinoza
|
r1287 | idReadUnit = self.getNewId() | ||
|
Miguel Valdez
|
r687 | else: | ||
| idReadUnit = str(id) | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | conf = ReadUnitConf() | ||
| conf.setup(self.id, idReadUnit, name, datatype, self.err_queue, **kwargs) | ||||
| self.configurations[conf.id] = conf | ||||
| r1279 | ||||
|
Juan C. Espinoza
|
r1287 | return conf | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def addProcUnit(self, id=None, inputId='0', datatype=None, name=None): | ||
| r889 | ||||
|
George Yong
|
r1171 | ''' | ||
| ''' | ||||
|
Juan C. Espinoza
|
r1287 | if id is None: | ||
| idProcUnit = self.getNewId() | ||||
| else: | ||||
| idProcUnit = id | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | conf = ProcUnitConf() | ||
| conf.setup(self.id, idProcUnit, name, datatype, inputId, self.err_queue) | ||||
| self.configurations[conf.id] = conf | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | return conf | ||
| r889 | ||||
|
Miguel Valdez
|
r587 | def removeProcUnit(self, id): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | if id in self.configurations: | ||
| self.configurations.pop(id) | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def getReadUnit(self): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | for obj in list(self.configurations.values()): | ||
| if obj.ELEMENTNAME == 'ReadUnit': | ||||
|
Miguel Valdez
|
r577 | return obj | ||
| r889 | ||||
|
Miguel Valdez
|
r577 | return None | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def getProcUnit(self, id): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | return self.configurations[id] | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def getUnits(self): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | keys = list(self.configurations) | ||
| keys.sort() | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | for key in keys: | ||
| yield self.configurations[key] | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def updateUnit(self, id, **kwargs): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | conf = self.configurations[id].update(**kwargs) | ||
| r889 | ||||
| def makeXml(self): | ||||
|
Juan C. Espinoza
|
r1287 | xml = Element('Project') | ||
| xml.set('id', str(self.id)) | ||||
| xml.set('name', self.name) | ||||
| xml.set('description', self.description) | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | for conf in self.configurations.values(): | ||
| conf.makeXml(xml) | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | self.xml = xml | ||
| 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 | ||||
|
Juan C. Espinoza
|
r1287 | ElementTree(self.xml).write(abs_file, method='xml') | ||
| r889 | ||||
|
Miguel Valdez
|
r708 | self.filename = abs_file | ||
| r889 | ||||
|
Miguel Valdez
|
r681 | return 1 | ||
|
Miguel Valdez
|
r197 | |||
|
Juan C. Espinoza
|
r1287 | def readXml(self, filename): | ||
| r889 | ||||
|
Miguel Valdez
|
r687 | abs_file = os.path.abspath(filename) | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | self.configurations = {} | ||
| r889 | ||||
|
Miguel Valdez
|
r735 | try: | ||
|
Juan C. Espinoza
|
r1287 | self.xml = ElementTree().parse(abs_file) | ||
|
Miguel Valdez
|
r735 | except: | ||
|
Juan C. Espinoza
|
r1287 | log.error('Error reading %s, verify file format' % filename) | ||
|
Miguel Valdez
|
r735 | return 0 | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | self.id = self.xml.get('id') | ||
| self.name = self.xml.get('name') | ||||
| self.description = self.xml.get('description') | ||||
| for element in self.xml: | ||||
| if element.tag == 'ReadUnit': | ||||
| conf = ReadUnitConf() | ||||
| conf.readXml(element, self.id, self.err_queue) | ||||
| self.configurations[conf.id] = conf | ||||
| elif element.tag == 'ProcUnit': | ||||
| conf = ProcUnitConf() | ||||
| input_proc = self.configurations[element.get('inputId')] | ||||
| conf.readXml(element, self.id, self.err_queue) | ||||
| self.configurations[conf.id] = conf | ||||
| r889 | ||||
|
Miguel Valdez
|
r708 | self.filename = abs_file | ||
| r889 | ||||
|
Miguel Valdez
|
r681 | return 1 | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1177 | def __str__(self): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | text = '\nProject[id=%s, name=%s, description=%s]\n\n' % ( | ||
| self.id, | ||||
| self.name, | ||||
| self.description, | ||||
| ) | ||||
| for conf in self.configurations.values(): | ||||
| text += '{}'.format(conf) | ||||
|
José Chávez
|
r1082 | |||
|
Juan C. Espinoza
|
r1287 | return text | ||
| r889 | ||||
|
Miguel Valdez
|
r197 | def createObjects(self): | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | keys = list(self.configurations.keys()) | ||
|
George Yong
|
r1192 | keys.sort() | ||
| for key in keys: | ||||
|
Juan C. Espinoza
|
r1287 | conf = self.configurations[key] | ||
| conf.createObjects() | ||||
| if conf.inputId is not None: | ||||
| conf.object.setInput(self.configurations[conf.inputId].object) | ||||
| r889 | ||||
| r1241 | def monitor(self): | |||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | t = Thread(target=self._monitor, args=(self.err_queue, self.ctx)) | ||
| r1241 | t.start() | |||
| r1279 | ||||
|
Juan C. Espinoza
|
r1287 | def _monitor(self, queue, ctx): | ||
| r889 | ||||
| r1241 | import socket | |||
| r1279 | ||||
| r1241 | procs = 0 | |||
| err_msg = '' | ||||
| r1279 | ||||
| r1241 | while True: | |||
| msg = queue.get() | ||||
| if '#_start_#' in msg: | ||||
| procs += 1 | ||||
| elif '#_end_#' in msg: | ||||
| procs -=1 | ||||
| else: | ||||
| err_msg = msg | ||||
| r1279 | ||||
| if procs == 0 or 'Traceback' in err_msg: | ||||
| r1241 | break | |||
| time.sleep(0.1) | ||||
| r1279 | ||||
| r1241 | if '|' in err_msg: | |||
| name, err = err_msg.split('|') | ||||
| if 'SchainWarning' in err: | ||||
| log.warning(err.split('SchainWarning:')[-1].split('\n')[0].strip(), name) | ||||
| elif 'SchainError' in err: | ||||
| log.error(err.split('SchainError:')[-1].split('\n')[0].strip(), name) | ||||
| else: | ||||
| log.error(err, name) | ||||
| r1279 | else: | |||
| r1241 | name, err = self.name, err_msg | |||
| r1279 | ||||
|
Juan C. Espinoza
|
r1287 | time.sleep(1) | ||
| r1279 | ||||
| r1241 | ctx.term() | |||
|
Miguel Valdez
|
r735 | |||
|
Juan C. Espinoza
|
r1052 | message = ''.join(err) | ||
| r889 | ||||
| r1241 | if err_msg: | |||
| subject = 'SChain v%s: Error running %s\n' % ( | ||||
| schainpy.__version__, self.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()) | ||||
|
Juan C. Espinoza
|
r1287 | readUnitConfObj = self.getReadUnit() | ||
| r1241 | if readUnitConfObj: | |||
| subtitle += '\nInput parameters:\n' | ||||
|
Juan C. Espinoza
|
r1287 | subtitle += '[Data path = %s]\n' % readUnitConfObj.parameters['path'] | ||
| subtitle += '[Start date = %s]\n' % readUnitConfObj.parameters['startDate'] | ||||
| subtitle += '[End date = %s]\n' % readUnitConfObj.parameters['endDate'] | ||||
| subtitle += '[Start time = %s]\n' % readUnitConfObj.parameters['startTime'] | ||||
| subtitle += '[End time = %s]\n' % readUnitConfObj.parameters['endTime'] | ||||
| r1241 | ||||
| a = Alarm( | ||||
| r1279 | modes=self.alarm, | |||
| r1241 | email=self.email, | |||
| message=message, | ||||
| subject=subject, | ||||
| subtitle=subtitle, | ||||
| filename=self.filename | ||||
| ) | ||||
| a.start() | ||||
| r1129 | ||||
|
Miguel Valdez
|
r691 | def setFilename(self, filename): | ||
| r889 | ||||
|
Miguel Valdez
|
r691 | self.filename = filename | ||
| r889 | ||||
|
Juan C. Espinoza
|
r1287 | def runProcs(self): | ||
|
Juan C. Espinoza
|
r1177 | |||
|
Juan C. Espinoza
|
r1287 | err = False | ||
| n = len(self.configurations) | ||||
| r1279 | ||||
|
Juan C. Espinoza
|
r1287 | while not err: | ||
| for conf in self.getUnits(): | ||||
| r1370 | ok = conf.run() | |||
| r1342 | if ok == 'Error': | |||
|
Juan C. Espinoza
|
r1287 | n -= 1 | ||
| continue | ||||
| elif not ok: | ||||
| break | ||||
| if n == 0: | ||||
| err = True | ||||
| r889 | ||||
|
Juan C. Espinoza
|
r1052 | def run(self): | ||
|
José Chávez
|
r1004 | |||
|
Juan C. Espinoza
|
r1287 | log.success('\nStarting Project {} [id={}]'.format(self.name, self.id), tag='') | ||
| self.started = True | ||||
| r1279 | self.start_time = time.time() | |||
| self.createObjects() | ||||
|
Juan C. Espinoza
|
r1287 | self.runProcs() | ||
| log.success('{} Done (Time: {:4.2f}s)'.format( | ||||
|
Juan C. Espinoza
|
r1112 | self.name, | ||
| r1241 | time.time()-self.start_time), '') | |||
