L2SS-245: add some proto-code

devices/test/devices/monitoring_performance_test.py

+# -*- 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()