diff --git a/devices/toolkit/attribute_polling_stats.py b/devices/toolkit/attribute_polling_stats.py new file mode 100644 index 0000000000000000000000000000000000000000..fcfb502320d4de084f219244289f6ddfa064e996 --- /dev/null +++ b/devices/toolkit/attribute_polling_stats.py @@ -0,0 +1,39 @@ +import numpy +import tango + +def attribute_polling_stats(dp: tango._tango.DeviceProxy = None, iterations: int = 10, polling_time: float = 1.0, quiet = False): + if dp is not None: + print('Will sample the device server\'s polling time {} times with a pause of {}s between each sampling.'.format(iterations, polling_time)) + else: + print('A DeviceProxy object is needed!') + return + import numpy + from time import sleep + polling_durations = [] + polling_delays = [] + value = numpy.double(0) + iterations_left = iterations + while iterations_left > 0: + iterations_left -= 1 + string = dp.polling_status()[0].split('\n') + polling_duration = numpy.double(string[3].split('=')[-1].strip()) / 1e3 + polling_delay = numpy.double(string[5].split('=')[-1].split(',')[0].strip()) / 1e3 + polling_durations.append(polling_duration) + polling_delays.append(polling_delay) + if not quiet: + print('Iteration #{}, {} iterations left, polling duration = {}s, polling delay = {}s.'.format(iterations - iterations_left, iterations_left, polling_duration, polling_delay)) + sleep(polling_time) + durations = numpy.array(polling_durations) + delays = numpy.array(polling_delays) + def compute_and_print(result): + min = numpy.min(result) + max = numpy.max(result) + median = numpy.median(result) + mean = numpy.mean(result) + std = numpy.std(result) + print("\tmin = {}[s]\n\tmax = {}[s]\n\tmedian = {}[s]\n\tmean = {}[s]\n\tstddev = {}[s]".format(min, max, median, mean, std)) + print("\n\titerations = {}\n\n\tPolling duration".format(iterations)) + compute_and_print(durations) + print("\n\tPolling delay") + compute_and_print(delays) + return (durations, delays)