diff --git a/AppChecker.py b/AppChecker.py new file mode 100644 index 0000000000000000000000000000000000000000..52568a1ad836f03e6437bf21e5bf85665e1a054c --- /dev/null +++ b/AppChecker.py @@ -0,0 +1,189 @@ +""" +Generic Application checker class + +Describe a set of functions to operate to check an application +""" + +import PyTango as tango +import logging + +class AppChecker: + + def __init__(self, appname): + """ + Constructor. + + Parameters + ========== + appname: string + Name of the application. Usefull for logs. + + Returns + ======= + AppChecker object + """ + + # Memorize name + self.appname = appname + + # Get a logger with the app name + self.logger = logging.getLogger(self.appname) + + # We will use this level + log_level = logging.DEBUG + + # Create a stream handler + s_handler = logging.StreamHandler() + s_handler.setFormatter(logging.Formatter("%(levelname)s\t[%(funcName)s]\t%(message)s")) + s_handler.setLevel(level=log_level) + + # Attach the stream handler + self.logger.addHandler(s_handler) + + # Set level of logger + self.logger.setLevel(level=log_level) + + + ################################## + ## CHECK DEVICES STATUS + ################################## + def check_status(self, device_list): + """ + Check the DS state of all devices used by this equipment. + + Handles ConnectionFailed exceptions, i.e. when device is not started. + + Parameters + ========== + device_list: array of string + List of tango path to devices. Will check status of all those devices. + + Returns + ======= + tuple (status, devlist) + status: boolean + False if a problem was encountered + devlist: array of string + List of devices with problem + """ + self.logger.info("Checking state of tango devices...") + + # Function returns + status=True + devlist=[] + + # ---------------------------------------------------------------------------------------- + # For all devices in list, try to connect and check that state is running, on or standby. + for device in device_list: + prox = tango.DeviceProxy(device) + + try: + state = prox.state() + except tango.ConnectionFailed: + self.logger.warning("Device %s is probably not started, connection failed."%device) + status=False + devlist.append(device) + continue + + if not state in [tango.DevState.RUNNING, + tango.DevState.ON, + tango.DevState.STANDBY, + ]: + self.logger.warning("Device %s is in state %s"%(device, state)) + status=False + devlist.append(device) + else: + self.logger.info("Device %s is in state %s"%(device, state)) + + # ---------------------------------------------------------------------------------------- + + return (status, devlist) + + + ################################## + ## CHECK CONFIGURATION + ################################## + def check_configuration(self, context): + """ + Check the context (dict of attributes and properties on devices) + + Parameters + ========== + context: dict + Dict of device, attributes and properties with their values + + Returns + ======= + tuple (status, attrlist) + status: boolean + False if a problem was encountered + context: dict + Context that mismatch, with the actual values + + """ + + # Function returns + status=True + failcontext=dict() + + + # ---------------------------------------------------------------------------------------- + # Loop on every device of the context + for devicepath in context.keys(): + + # Get a proxy to device + prox = tango.DeviceProxy(devicepath) + + # ----------------------------------------------------- + # Check write attributes + wattr = context[devicepath]["wattributes"] + failattr = dict() + for attr in wattr.keys(): + value = prox[attr].value + if value != wattr[attr] + self.logger.warning("Attribute %s/%s value mismatch. Read %s, expect %s."%( + devicepath, + attr, + value, + wattr[attr]) + status=False + failattr[attr]=value + else: + self.logger.info("Attribute %s/%s is correctly set to value %s."%( + devicepath, + attr, + value) + + # ----------------------------------------------------- + # Check properties + props = context[devicepath]["properties"] + failprop = dict() + + rprops = prox.get_property(list(props.keys())) + for k in props.keys(): + value = list(rprops[k]) + if props[k] != value: + self.logger.warning("Property %s of device %s value mismatch.\nRead %s\nExpect %s"%( + k, + devicepath, + value, + props[k]) + status=False + failprop[k]=value + else: + self.logger.info("Property %s of device %s is correctly set to %s"%( + k, + devicepath, + value) + + # ----------------------------------------------------- + # Append to faild context if any fail + if len(failattr.keys()) + len(failprop.keys())>0: + failcontext[devicepath]= { + "properties":failprops, + "wattributes":failattr, + } + + # ---------------------------------------------------------------------------------------- + return (status, failcontext) +