diff --git a/devices/test/devices/monitoring_performance_test.py b/devices/test/devices/monitoring_performance_test.py new file mode 100644 index 0000000000000000000000000000000000000000..0464cf14284cd8459ad53bc116e92648b510385e --- /dev/null +++ b/devices/test/devices/monitoring_performance_test.py @@ -0,0 +1,81 @@ +# -*- coding: utf-8 -*- +# +# This file is part of the LOFAR2.0 project +# +# +# +# Distributed under the terms of the APACHE license. +# See LICENSE.txt for more info. + +# TODO(Corne): Remove sys.path.append hack once packaging is in place! +import os, sys +currentdir = os.path.dirname(os.path.realpath(__file__)) +parentdir = os.path.dirname(currentdir) +parentdir = os.path.dirname(parentdir) +sys.path.append(parentdir) + +import logging +import numpy +from tango import DevState +from tango.server import run, Device, attribute +from numpy import random +from time import time + +__all__ = ["Monitoring_Performance_Device", "main"] + +logger = logging.getLogger() +logger.setLevel(logging.INFO) + +def read(size): + logger.info("read") + return numpy.array([random.random(), ] * size) + +class Monitoring_Performance_Device(Device): + small_array_r = attribute( + dtype = (numpy.double,), + max_dim_x = 10000, + polling_period = 1000, + period = 1000, + rel_change = 0.1, + archive_period = 1000, + archive_rel_change = 0.1, + max_value = 1.0, + min_value = 0.0, + fget = read(10000), + ) + + big_array_r = attribute( + dtype = (numpy.double,), + max_dim_x = 1000000, + period = 1000, + rel_change = 0.1, + archive_period = 1000, + archive_rel_change = 0.1, + max_value = 1.0, + min_value = 0.0, + fget = read(1000000), + ) + + def init_device(self): + Device.init_device(self) + self.set_state(DevState.OFF) + + self.small_array_r.set_data_ready_event(True) + self.set_change_event("small_array_r", True, True) + self.set_archive_event("small_array_r", True, True) + + self.big_array_r.set_data_ready_event(True) + self.set_change_event("big_array_r", True, True) + self.set_archive_event("big_array_r", True, True) + + self.set_state(DevState.ON) + + def delete_device(self): + self.set_state(DevState.OFF) + + +def main(args = None, **kwargs): + return run((Monitoring_Performance_Device, ), args = args, **kwargs) + +if __name__ == '__main__': + main()