L2SS-1030: create stationconfiguration class

tangostationcontrol/tangostationcontrol/common/configuration.py

-from tango import DeviceProxy
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+from tango import DeviceProxy, Database
 
 from itertools import islice
 
+class StationConfiguration:
+
     DEVICE_PROPERTIES_QUERY = "SELECT device, property_device.name, property_device.value FROM property_device \
                             INNER JOIN device ON property_device.device = device.name \
                             WHERE class != 'DServer' \
@@ -19,33 +30,40 @@ SERVER_QUERY = "SELECT server, class, name FROM device \
                 WHERE class != 'DServer' \
                 ORDER BY server ASC"
 
-def get_db_data(db, tangodb_timeout:int = 10000):
-    """ Dump TANGO database into dictionary """
-    dbproxy = DeviceProxy(db.dev_name())            # TangoDB
-    dbproxy.set_timeout_millis(tangodb_timeout)     # Set a security timeout (default is 3000ms)
+    def __init__(self, db: Database, tangodb_timeout:int = 10000):
+        self.dbproxy = DeviceProxy(db.dev_name())            # TangoDB
+        self.dbproxy.set_timeout_millis(tangodb_timeout)     # Set a security timeout (default is 3000ms)
+
+    def get_tangodb_data(self) -> dict:
+        """ Dump a subset of TANGO database into dictionary. 
+        
+        The dictionary contains the info about all the Devices used in the 
+        present environment, including their Properties values, their Attribute Properties,
+        and the namespace of the DeviceServers which incapsulate each Device.
+        """
         # Create empty dictionaries to be populated
         devices_dict = {}
         server_dict = {}
         
         # Populate devices dictionary from query data
-    device_property_result = query_tangodb(dbproxy, DEVICE_PROPERTIES_QUERY, 3)
-    devices_dict = add_to_devices_dict(devices_dict, device_property_result)
+        device_property_result = self._query_tangodb(self.dbproxy, self.DEVICE_PROPERTIES_QUERY, 3)
+        devices_dict = self.add_to_devices_dict(devices_dict, device_property_result)
         
         # Populate devices dictionary from query data
-    attrs_property_result = query_tangodb(dbproxy, ATTRS_PROPERTIES_QUERY, 4)
-    devices_dict = add_to_attrs_dict(devices_dict, attrs_property_result)
+        attrs_property_result = self._query_tangodb(self.dbproxy, self.ATTRS_PROPERTIES_QUERY, 4)
+        devices_dict = self.add_to_attrs_dict(devices_dict, attrs_property_result)
         
         # Populate server dictionary from query data and merge it with devices dict
-    server_result = query_tangodb(dbproxy, SERVER_QUERY, 3)
-    server_dict = add_to_server_dict(server_dict, devices_dict, server_result)
+        server_result = self._query_tangodb(self.dbproxy, self.SERVER_QUERY, 3)
+        server_dict = self.add_to_server_dict(server_dict, devices_dict, server_result)
         return {"servers" : server_dict}
 
-def query_tangodb(dbproxy: DeviceProxy, sql_query: str, num_cols: int):
+    def _query_tangodb(self, dbproxy: DeviceProxy, sql_query: str, num_cols: int) -> list:
         """ Query TangoDb with a built-in function and return data as tuples """
         _, raw_result = dbproxy.command_inout("DbMySqlSelect", sql_query)
-    return query_to_tuples(raw_result, num_cols)
+        return self.query_to_tuples(raw_result, num_cols)
 
-def add_to_devices_dict(devices_dict:dict, result:list):
+    def add_to_devices_dict(self, devices_dict:dict, result:list) -> dict:
         """ Populate a devices dictionary with the following structure:
         'device_name': { 'properties' : { 'property_name': ['property_value'] } }
         """
@@ -60,7 +78,7 @@ def add_to_devices_dict(devices_dict:dict, result:list):
             value_data.append(value)
         return devices_dict
 
-def add_to_attrs_dict(devices_dict:dict, result:list):
+    def add_to_attrs_dict(self, devices_dict:dict, result:list) -> dict:
         """ Populate a device dictionary with the following structure : 
         'device_name': { 'attribute_properties' : { 'attribute_name': {'property_name' : ['property_value'] } } }
         """
@@ -77,7 +95,7 @@ def add_to_attrs_dict(devices_dict:dict, result:list):
             value_data.append(value)
         return devices_dict
 
-def add_to_server_dict(server_dict:dict, devices_dict:dict, result:list):
+    def add_to_server_dict(self, server_dict:dict, devices_dict:dict, result:list) -> dict:
         """ Populate the server dictionary and merge it with the devices dictionary.
         At the end of the process, the dictionary will have the following structure : 
         'server_name' : { 'server_instance' : { 'server_class' :
@@ -99,6 +117,6 @@ def add_to_server_dict(server_dict:dict, devices_dict:dict, result:list):
             server_dict[sname][instance][sclass][device] = device_data
         return server_dict
 
-def query_to_tuples(result, num_cols):
+    def query_to_tuples(self, result: list, num_cols: int) -> list:
         """ Given a query result and its number of columns, transforms the raw result in a list of tuples """
         return list(zip(*[islice(result, i, None, num_cols) for i in range(num_cols)]))

tangostationcontrol/tangostationcontrol/devices/configuration_device.py

@@ -17,10 +17,10 @@ from tango import AttrWriteType, Database
 from tango.server import attribute
 
 # Additional import
-from tangostationcontrol.common.configuration import get_db_data
+from tangostationcontrol.common.configuration import StationConfiguration
 from tangostationcontrol.common.entrypoint import entry
 from tangostationcontrol.devices.lofar_device import lofar_device
-from tangostationcontrol.common.lofar_logging import device_logging_to_python
+from tangostationcontrol.common.lofar_logging import device_logging_to_python, log_exceptions
 
 import json
 import logging
@@ -53,10 +53,17 @@ class Configuration(lofar_device):
     
     def _dump_configdb(self):
         """ Returns the TangoDB station configuration as a JSON string """
-        db = Database() # TangoDB
-        dbdata = get_db_data(db)
+        dbdata = self.station_configuration.get_tangodb_data()
         return json.dumps(dbdata, ensure_ascii=False, indent=4, sort_keys=True)
     
+    # --------
+    # overloaded functions
+    # --------
+    @log_exceptions()
+    def configure_for_initialise(self):
+        super().configure_for_initialise()
+        self.station_configuration = StationConfiguration(db = Database())
+         
 
 # ----------
 # Run server

tangostationcontrol/tangostationcontrol/integration_test/default/devices/test_device_configuration.py

-# -*- coding: utf-8 -*-
-#
-# This file is part of the LOFAR 2.0 Station Software
-#
-#
-#
-# Distributed under the terms of the APACHE license.
-# See LICENSE.txt for more info.
-
-from .base import AbstractTestBases
-
-import json
-
-class TestDeviceConfiguration(AbstractTestBases.TestDeviceBase):
-
-    def setUp(self):
-        super().setUp("STAT/Configuration/1")
-    
-    def test_read_tangodb_properties(self):
-        """ Test whether the station control configuration is correctly retrieved as a JSON string """
-        tangodb_properties = self.proxy.tangodb_properties_RW
-        dbdata = json.loads(tangodb_properties)
-        self.assertTrue(type(dbdata), dict)
-        self.assertGreater(len(dbdata['servers']), 0)
-        # Verify if Configuration Device exists
-        self.assertTrue('configuration' in dbdata['servers']) # server-name
-        self.assertTrue('stat' in dbdata['servers']['configuration']) # server-instance
-        self.assertTrue('configuration' in dbdata['servers']['configuration']['stat']) # server-class
-        self.assertTrue('stat/configuration/1' in dbdata['servers']['configuration']['stat']['configuration']) # device

tangostationcontrol/tangostationcontrol/test/devices/test_configuration_device.py

+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+from tango import DevState
+from tango.test_context import DeviceTestContext
+
+from tangostationcontrol.devices import configuration_device
+from tangostationcontrol.test import base
+
+import json
+
+class TestConfigurationDevice(base.TestCase):
+
+    def setUp(self):
+        super(TestConfigurationDevice, self).setUp()
+
+    def test_read_tangodb_properties(self):
+        with DeviceTestContext(configuration_device.Configuration, properties={}, process=True) as proxy:
+            proxy.warm_boot()
+            self.assertEqual(DevState.ON, proxy.state())
+            """ Test whether the station control configuration is correctly retrieved as a JSON string """
+            tangodb_properties = self.proxy.tangodb_properties_RW
+            dbdata = json.loads(tangodb_properties)
+            self.assertTrue(type(dbdata), dict)
+            self.assertGreater(len(dbdata['servers']), 0)
+            # Verify if Configuration Device exists
+            self.assertTrue('configuration' in dbdata['servers']) # server-name
+            self.assertTrue('stat' in dbdata['servers']['configuration']) # server-instance
+            self.assertTrue('configuration' in dbdata['servers']['configuration']['stat']) # server-class
+            self.assertTrue('stat/configuration/1' in dbdata['servers']['configuration']['stat']['configuration']) # device