Merge branch 'master' into L2SS-777-add-observations-to-prometheus-exporter

tangostationcontrol/tangostationcontrol/devices/pcon.py

+# -*- coding: utf-8 -*-
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+""" PCON Device Server for LOFAR2.0
+
+"""
+
+# Additional import
+from tangostationcontrol.common.entrypoint import entry
+from tangostationcontrol.common.lofar_logging import device_logging_to_python
+
+import logging
+from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
+from tangostationcontrol.devices.snmp_device import snmp_device
+
+import numpy
+
+from pysmi import debug
+
+debug.setLogger(debug.Debug('searcher', "compiler", "borrower", "reader"))
+
+logger = logging.getLogger()
+
+__all__ = ["PCON", "main"]
+
+
+@device_logging_to_python()
+class PCON(snmp_device):
+
+    # ----------
+    # Attributes
+    # ----------
+    systemVoltage_R = attribute_wrapper(comms_annotation={"mib": "ACC-MIB", "name": "systemVoltage", "scaling_factor": 0.01}, datatype=numpy.double)
+
+    rectifierCurrent_R = attribute_wrapper(comms_annotation={"mib": "ACC-MIB", "name": "rectifierCurrent", "scaling_factor": 0.1}, datatype=numpy.double)
+    loadCurrent_R = attribute_wrapper(comms_annotation={"mib": "ACC-MIB", "name": "loadCurrent", "scaling_factor": 0.1}, datatype=numpy.double)
+    batteryCurrent_R = attribute_wrapper(comms_annotation={"mib": "ACC-MIB", "name": "batteryCurrent", "scaling_factor": 0.1}, datatype=numpy.double)
+
+    battTemperature_R = attribute_wrapper(comms_annotation={"mib": "ACC-MIB", "name": "battTemperature"}, datatype=numpy.double)
+
+
+# ----------
+# Run server
+# ----------
+def main(**kwargs):
+    """Main function of the PCON module."""
+    return entry(PCON, **kwargs)

tangostationcontrol/tangostationcontrol/devices/psoc.py

@@ -9,18 +9,16 @@
 
 # Additional import
 from tangostationcontrol.common.entrypoint import entry
-from tangostationcontrol.devices.lofar_device import lofar_device
 from tangostationcontrol.common.lofar_logging import device_logging_to_python, log_exceptions
 
 from tango.server import device_property, command
-import os
 
 import logging
 from tangostationcontrol.clients.attribute_wrapper import attribute_wrapper
-from tangostationcontrol.clients.snmp_client import SNMP_client, mib_loader, snmp_attribute
+from tangostationcontrol.clients.snmp_client import snmp_attribute
+from tangostationcontrol.devices.snmp_device import snmp_device
 
 import numpy
-import pkg_resources
 from datetime import timedelta
 
 from pysmi import debug
@@ -33,35 +31,10 @@ __all__ = ["PSOC", "main"]
 
 
 @device_logging_to_python()
-class PSOC(lofar_device):
+class PSOC(snmp_device):
     # -----------------
     # Device Properties
     # -----------------
-    SNMP_community = device_property(
-        dtype='DevString',
-        mandatory=True
-    )
-
-    SNMP_host = device_property(
-        dtype='DevString',
-        mandatory=True
-    )
-
-    SNMP_mib_dir = device_property(
-        dtype='DevString',
-        mandatory=True
-    )
-
-    SNMP_timeout = device_property(
-        dtype='DevDouble',
-        mandatory=True
-    )
-
-    SNMP_version = device_property(
-        dtype='DevULong',
-        mandatory=True
-    )
-
     PSOC_sockets = device_property(
         dtype=[str],
         mandatory=True
@@ -70,9 +43,9 @@ class PSOC(lofar_device):
     # ----------
     # Attributes
     # ----------
-    sockets_state_R = attribute_wrapper(comms_annotation={"mib": "PowerNet-MIB", "name": "sPSOCOutletCtl", "index": 1}, dims=(8,), datatype=str)
-    master_state_R = attribute_wrapper(comms_annotation={"mib": "PowerNet-MIB", "name": "sPSOCMasterState"}, datatype=str)
-    current_load_R = attribute_wrapper(comms_annotation={"mib": "PowerNet-MIB", "name": "rPSOCLoadStatusLoad", "index": 1}, datatype=numpy.int64)
+    sockets_state_R = attribute_wrapper(comms_annotation={"mib": "PowerNet-MIB", "name": "sPDUOutletCtl", "index": 1}, dims=(8,), datatype=str)
+    master_state_R = attribute_wrapper(comms_annotation={"mib": "PowerNet-MIB", "name": "sPDUMasterState"}, datatype=str)
+    current_load_R = attribute_wrapper(comms_annotation={"mib": "PowerNet-MIB", "name": "rPDULoadStatusLoad", "index": 1}, datatype=numpy.int64)
     uptime_R = attribute_wrapper(comms_annotation={"mib": "SNMPv2-MIB", "name": "sysUpTime"}, datatype=numpy.int64)
 
     # --------
@@ -94,26 +67,11 @@ class PSOC(lofar_device):
             self.socket_dict = {self.PSOC_sockets[f]: f + 1 for f in range(len(self.PSOC_sockets))}
             logger.debug(f"Configured PSOC with the following socket names: {self.PSOC_sockets}")
 
-        # set up the SNMP ua client
-        self.snmp_manager = SNMP_client(self.SNMP_community, self.SNMP_host, self.SNMP_timeout, self.SNMP_version, self.Fault, self)
-
-        # map an access helper class
-        for i in self.attr_list():
-            try:
-                i.set_comm_client(self, self.snmp_manager)
-            except Exception as e:
-                # use the pass function instead of setting read/write fails
-                i.set_pass_func(self)
-                logger.warning("error while setting the SNMP attribute {} read/write function. {}".format(i, e))
-
-        self.snmp_manager.start()
-
+        super().configure_for_initialise()
 
         # prepares this object for the readable_uptime command
         self.uptime_attr = snmp_attribute(self.snmp_manager.SNMP_comm, "SNMPv2-MIB", name="sysUpTime", idx=0, dtype=numpy.int64, dim_x=1, dim_y=0)
 
-        super().configure_for_initialise()
-
     def _toggle_socket(self, socket_name, on: bool):
         """
         This function is tailored to the "APS switched rack PSOC", changing the psoc will require some changes to this function
@@ -131,7 +89,7 @@ class PSOC(lofar_device):
             socket_set = "outletOff"
 
         # create the snmp_attribute for the correct socket
-        attr = snmp_attribute(self.snmp_manager.SNMP_comm, "PowerNet-MIB", name="sPSOCOutletCtl", idx=socket_nr, dtype=str, dim_x=1, dim_y=0)
+        attr = snmp_attribute(self.snmp_manager.SNMP_comm, "PowerNet-MIB", name="sPDUOutletCtl", idx=socket_nr, dtype=str, dim_x=1, dim_y=0)
 
         # write the correct value
         attr.write_function([socket_set])
@@ -155,36 +113,6 @@ class PSOC(lofar_device):
         # for whatever reason, the uptime is given in hundredts of a second
         return str(timedelta(seconds=self.uptime_attr.read_function()/100))
 
-
-    @log_exceptions()
-    def configure_for_on(self):
-        super().configure_for_on()
-
-    @log_exceptions()
-    def configure_for_off(self):
-        super().configure_for_off()
-
-    def get_mib_dir(self):
-        mib_filename_path = pkg_resources.resource_filename('tangostationcontrol', self.SNMP_mib_dir)
-        mib_path = os.path.dirname(mib_filename_path)
-
-        return mib_path
-
-    def init_device(self):
-        super().init_device()
-
-        # create the mib_loader and set the mib path
-        self.loader = mib_loader(self.get_mib_dir())
-
-        for i in self.attr_list():
-            try:
-                # for all of the attributes attempt to load the pre-compiled MIB. Skips already loaded ones
-                self.loader.load_pymib(i.comms_annotation["mib"])
-            except Exception as e:
-                raise Exception(
-                    f"Failed to load MIB file: {i.comms_annotation.get('mib')} for attribute {i.name} in directory {self.get_mib_dir()} ") from e
-
-
 # ----------
 # Run server
 # ----------

tangostationcontrol/tangostationcontrol/devices/snmp_device.py

+# -*- coding: utf-8 -*-
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+""" SNMP Device Server for LOFAR2.0
+
+"""
+
+# Additional import
+from tangostationcontrol.devices.lofar_device import lofar_device
+from tangostationcontrol.common.lofar_logging import device_logging_to_python, log_exceptions
+
+from tango.server import device_property
+import os
+
+import logging
+from tangostationcontrol.clients.snmp_client import SNMP_client, mib_loader
+
+import pkg_resources
+
+from pysmi import debug
+
+debug.setLogger(debug.Debug('searcher', "compiler", "borrower", "reader"))
+
+logger = logging.getLogger()
+
+__all__ = ["snmp_device"]
+
+
+@device_logging_to_python()
+class snmp_device(lofar_device):
+    # -----------------
+    # Device Properties
+    # -----------------
+    SNMP_community = device_property(
+        dtype='DevString',
+        mandatory=True
+    )
+
+    SNMP_host = device_property(
+        dtype='DevString',
+        mandatory=True
+    )
+
+    SNMP_mib_dir = device_property(
+        dtype='DevString',
+        mandatory=True
+    )
+
+    SNMP_timeout = device_property(
+        dtype='DevDouble',
+        mandatory=True
+    )
+
+    SNMP_version = device_property(
+        dtype='DevULong',
+        mandatory=True
+    )
+
+    @log_exceptions()
+    def configure_for_initialise(self):
+
+        # set up the SNMP client
+        self.snmp_manager = SNMP_client(self.SNMP_community, self.SNMP_host, self.SNMP_timeout, self.SNMP_version, self.Fault, self)
+
+        # map an access helper class
+        for i in self.attr_list():
+            try:
+                i.set_comm_client(self, self.snmp_manager)
+            except Exception as e:
+                # use the pass function instead of setting read/write fails
+                i.set_pass_func(self)
+                logger.warning("error while setting the SNMP attribute {} read/write function. {}".format(i, e))
+
+        self.snmp_manager.start()
+
+        super().configure_for_initialise()
+
+    @log_exceptions()
+    def configure_for_on(self):
+        super().configure_for_on()
+
+    @log_exceptions()
+    def configure_for_off(self):
+        super().configure_for_off()
+
+    def get_mib_dir(self):
+        mib_filename_path = pkg_resources.resource_filename('tangostationcontrol', self.SNMP_mib_dir)
+        mib_path = os.path.dirname(mib_filename_path)
+
+        return mib_path
+
+    def init_device(self):
+        super().init_device()
+
+        # create the mib_loader and set the mib path
+        self.loader = mib_loader(self.get_mib_dir())
+
+        for i in self.attr_list():
+            try:
+                # for all of the attributes attempt to load the pre-compiled MIB. Skips already loaded ones
+                self.loader.load_pymib(i.comms_annotation["mib"])
+            except Exception as e:
+                raise Exception(
+                    f"Failed to load MIB file: {i.comms_annotation.get('mib')} for attribute {i.name} in directory {self.get_mib_dir()} ") from e