Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
"""
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()
rattr = prox.read_attributes(list(wattr.keys()))
for attr in rattr:
value = attr.value
if value != wattr[attr.name]:
self.logger.warning("Attribute %s/%s value mismatch. Read %s, expect %s."%(
devicepath,
else:
self.logger.info("Attribute %s/%s is correctly set to value %s."%(
devicepath,
# -----------------------------------------------------
# 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,
status=False
failprop[k]=value
else:
self.logger.info("Property %s of device %s is correctly set to %s"%(
k,
devicepath,
# -----------------------------------------------------
# Append to faild context if any fail
if len(failattr.keys()) + len(failprop.keys())>0:
failcontext[devicepath]= {
"wattributes":failattr,
}
# ----------------------------------------------------------------------------------------
return (status, failcontext)