Merge branch 'L2SS-1548-mib-for-pcon-device' into 'master'

Resolve L2SS-1548 "Mib for pcon device"

Closes L2SS-1548

See merge request !762

CDB/LOFAR_ConfigDb.json

@@ -177,7 +177,7 @@
                 "public"
               ],
               "SNMP_mib_dir": [
-                "devices/mibs/ACC-MIB.mib"
+                "devices/mibs/SP2-MIB.mib"
               ],
               "SNMP_timeout": [
                 "10.0"

CDB/stations/common.json

@@ -159,7 +159,7 @@
                 "public"
               ],
               "SNMP_mib_dir": [
-                "devices/mibs/ACC-MIB.mib"
+                "devices/mibs/SP2-MIB.mib"
               ],
               "SNMP_timeout": [
                 "10.0"

README.md

@@ -115,6 +115,7 @@ Next change the version in the following places:
 
 # Release Notes
 
+* 0.21.4 Replace `ACC-MIB.mib` with `SP2-MIB.mib` source file in PCON device
 * 0.21.3 Added DigitalBeam.Antenna_Usage_Mask_R to expose antennas used in beamforming
 * 0.21.2 Removed deprecated "Boot" device (use StationManager now)
 * 0.21.1 Implement multi project integration downstream pipeline

tangostationcontrol/VERSION

-0.21.3
+0.21.4

tangostationcontrol/integration_test/configuration/configDB/LOFAR_ConfigDb.json

@@ -80,7 +80,7 @@
                         "properties": {
                             "SNMP_host": ["127.0.0.1"],
                             "SNMP_community": ["public"],
-                            "SNMP_mib_dir": ["devices/mibs/ACC-MIB.mib"],
+                            "SNMP_mib_dir": ["devices/mibs/SP2-MIB.mib"],
                             "SNMP_timeout": ["10.0"],
                             "SNMP_version": ["1"]
                         }

tangostationcontrol/tangostationcontrol/clients/snmp/attribute_classes.py

+import logging
 import numpy
 from pysnmp import hlapi
-import logging
 
 logger = logging.getLogger()
 
 
 class SNMPAttribute:
+    """Define a SNMP attribute with its features"""
+
     snmp_to_numpy_dict = {
         hlapi.Integer32: numpy.int64,
         hlapi.Integer: numpy.int64,
@@ -51,17 +53,18 @@ class SNMPAttribute:
         """
 
         # get all of the values
-        errorIndication, errorStatus, errorIndex, *varBinds = self.comm.getter(
+        error_indication, errorStatus, errorIndex, *var_binds = self.comm.getter(
             self.objs
         )
 
-        if errorIndication is not None:
+        if error_indication is not None:
             raise IOError(
-                f"An error occurred while attempting to write '{self.name}'. errorIndication: {errorIndication}"
+                f"An error occurred while attempting to write '{self.name}'. \
+                    errorIndication: {error_indication}"
             )
 
         # get all the values in a list converted to the correct type
-        val_lst = self.convert(varBinds)
+        val_lst = self.convert(var_binds)
 
         # return the list of values
         return val_lst
@@ -78,24 +81,26 @@ class SNMPAttribute:
             hlapi.ObjectType(self.objID[i], value[i]) for i in range(len(self.objID))
         )
 
-        errorIndication, errorStatus, errorIndex, *varBinds = self.comm.setter(
+        error_indication, errorStatus, errorIndex, *var_binds = self.comm.setter(
             write_obj
         )
 
-        if errorIndication is not None:
+        if error_indication is not None:
             raise IOError(
-                f"An error occurred while attempting to write '{self.name}'. errorIndication: {errorIndication}"
+                f"An error occurred while attempting to write '{self.name}'. \
+                    errorIndication: {error_indication}"
             )
 
     def convert(self, var_binds):
         """
-        get all the values in a list, make sure to convert specific types that dont want to play nicely
+        Get all the values in a list, make sure to convert specific types
+        that dont want to play nicely
         """
         values = []
 
-        varBinds = var_binds[0]
+        var_binds = var_binds[0]
 
-        for var_bind in varBinds:
+        for var_bind in var_binds:
             value = self._convert_var_bind_value(var_bind[1])
             values.append(value)
 
@@ -153,7 +158,8 @@ class SNMPAttribute:
             result = str(value)
         else:
             raise TypeError(
-                f"Error: did not find a valid snmp type. Got: {type(value)}, expected one of: '{SNMPAttribute.snmp_to_numpy_dict.keys()}'"
+                f"Error: did not find a valid snmp type. Got: {type(value)}, \
+                    expected one of: '{SNMPAttribute.snmp_to_numpy_dict.keys()}'"
             )
 
         return result
@@ -192,8 +198,9 @@ class PSOC_sim:
         self.idx = idx - 1
 
     def write_function(self, value):
+        """Simulated attribute write function"""
         # we only write to sPDUOutletCtl and only as a scalar
-        if type(value) == list:
+        if isinstance(value, list):
             value = value[0]
 
         if self.name == "sPDUOutletCtl":
@@ -211,6 +218,7 @@ class PSOC_sim:
             )
 
     def read_function(self):
+        """Simulated attribute read function"""
         if self.name == "sPDUOutletCtl":
             return_val = [
                 "outletOn" if PSOC_sim.sockets_states[i + self.idx] else "outletOff"
@@ -219,26 +227,22 @@ class PSOC_sim:
 
             if self.len == 1:
                 return return_val[0]
-            else:
             return return_val
 
-        elif self.name == "sPDUMasterState":
+        if self.name == "sPDUMasterState":
             return_val = ""
             return return_val.join(
                 "On  " if i else "Off " for i in PSOC_sim.sockets_states
             )
 
-        elif self.name == "sysUpTime":
+        if self.name == "sysUpTime":
             # return arbitrary value of 60s uptime
             return 60000
 
-        elif self.name == "rPDULoadStatusLoad":
+        if self.name == "rPDULoadStatusLoad":
             # give a random value in amps
             return 2
-        else:
-            raise NameError(
-                f"pcon does not have any read attributes named: {self.name}"
-            )
+        raise NameError(f"pcon does not have any read attributes named: {self.name}")
 
 
 class PCON_sim:
@@ -247,6 +251,9 @@ class PCON_sim:
     are interested in. It is a replacement for the SNMPAttribute attribute class
     """
 
+    PCON_SCALAR_ATT = "powerSystemStatus"
+    PCON_STRING_ATT = "powerSystemCompany"
+
     def __init__(
         self,
         comm=None,
@@ -272,18 +279,13 @@ class PCON_sim:
         self.idx = idx
 
     def write_function(self, value):
+        """Simulated attribute write function"""
         raise NameError("PCOn has no write attributes")
 
     def read_function(self):
-        if self.name == "systemVoltage":
-            return 48.0
-        elif self.name == "rectifierCurrent":
-            return 12.0
-        elif self.name == "loadCurrent":
-            return 8.0
-        elif self.name == "batteryCurrent":
-            return 0.1
-        elif self.name == "battTemperature":
-            return 50
-        else:
-            raise NameError(f"pcon does not have any attributes named: {self.name}")
+        """Simulated attribute read function"""
+        match self.name:
+            case self.PCON_SCALAR_ATT:
+                return 1
+            case self.PCON_STRING_ATT:
+                return "Eltek"

tangostationcontrol/tangostationcontrol/devices/base_device_classes/snmp_device.py

@@ -35,6 +35,8 @@ __all__ = ["SNMPDevice"]
 
 @device_logging_to_python()
 class SNMPDevice(LOFARDevice):
+    """SNMP Device Server for LOFAR2.0"""
+
     # -----------------
     # Device Properties
     # -----------------
@@ -82,13 +84,13 @@ class SNMPDevice(LOFARDevice):
         for i in self.attr_list():
             try:
                 i.set_comm_client(self, self.snmp_manager)
-            except Exception as e:
+            except Exception as exc:
                 # 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
-                    )
+                    "error while setting the SNMP attribute %s read/write function. %s",
+                    i,
+                    exc,
                 )
 
         self.snmp_manager.start()
@@ -119,9 +121,11 @@ class SNMPDevice(LOFARDevice):
 
         for i in self.attr_list():
             try:
-                # for all of the attributes attempt to load the pre-compiled MIB. Skips already loaded ones
+                # 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:
+            except Exception as exc:
                 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
+                    f"Failed to load MIB file: {i.comms_annotation.get('mib')} \
+                        for attribute {i.name} in directory {self.get_mib_dir()} "
+                ) from exc

tangostationcontrol/tangostationcontrol/devices/pcon.py

@@ -29,46 +29,16 @@ __all__ = ["PCON", "main"]
 
 @device_logging_to_python()
 class PCON(SNMPDevice):
-    # ----------
-    # Attributes
-    # ----------
-    systemVoltage_R = AttributeWrapper(
-        comms_annotation={
-            "mib": "ACC-MIB",
-            "name": "systemVoltage",
-            "scaling_factor": 0.01,
-        },
-        datatype=numpy.double,
-    )
+    """PCON Device Server for LOFAR2.0"""
 
-    rectifierCurrent_R = AttributeWrapper(
-        comms_annotation={
-            "mib": "ACC-MIB",
-            "name": "rectifierCurrent",
-            "scaling_factor": 0.1,
-        },
-        datatype=numpy.double,
-    )
-    loadCurrent_R = AttributeWrapper(
-        comms_annotation={
-            "mib": "ACC-MIB",
-            "name": "loadCurrent",
-            "scaling_factor": 0.1,
-        },
-        datatype=numpy.double,
-    )
-    batteryCurrent_R = AttributeWrapper(
-        comms_annotation={
-            "mib": "ACC-MIB",
-            "name": "batteryCurrent",
-            "scaling_factor": 0.1,
-        },
-        datatype=numpy.double,
+    powerSystemStatus_R = AttributeWrapper(
+        comms_annotation={"mib": "SP2-MIB", "name": "powerSystemStatus"},
+        datatype=numpy.int32,
     )
 
-    battTemperature_R = AttributeWrapper(
-        comms_annotation={"mib": "ACC-MIB", "name": "battTemperature"},
-        datatype=numpy.double,
+    powerSystemCompany_R = AttributeWrapper(
+        comms_annotation={"mib": "SP2-MIB", "name": "powerSystemCompany"},
+        datatype=str,
     )
 
     # --------

tangostationcontrol/test/clients/test_snmp_client.py

@@ -3,6 +3,7 @@
 
 from os import path
 from unittest import mock
+from test import base
 
 import numpy
 from pysnmp import hlapi
@@ -18,8 +19,6 @@ from tangostationcontrol.clients.snmp.attribute_classes import (
     PSOC_sim,
 )
 
-from test import base
-
 
 class SNMPServerFixture:
     # conversion dict
@@ -38,6 +37,10 @@ class SNMPServerFixture:
         "spectrum": (4, 0),
     }
 
+    # object identifier correspondent to sysDescr object in SNMPv2-MIB
+    # value -> "1.3.6.1.2.1.1.1.0.1"
+    sysDescr_oid = hlapi.ObjectIdentity("SNMPv2-MIB", "sysDescr", 0, 1)
+
     def get_return_val(self, snmp_type: type, dims: tuple):
         """
         provides the return value for the set/get functions that an actual server would return.
@@ -54,7 +57,7 @@ class SNMPServerFixture:
 
     def _get_return_val_for_scalar(self, snmp_type: type):
         if snmp_type is hlapi.ObjectIdentity:
-            read_val = [(snmp_type("1.3.6.1.2.1.1.1.0"),)]
+            read_val = [(self.sysDescr_oid,)]
         elif snmp_type is hlapi.IpAddress:
             read_val = [
                 (
@@ -83,11 +86,11 @@ class SNMPServerFixture:
         if snmp_type is hlapi.ObjectIdentity:
             read_val = []
             for _i in range(dims[0]):
-                read_val.append((None, snmp_type(f"1.3.6.1.2.1.1.1.0.1")))
+                read_val.append((None, self.sysDescr_oid))
         elif snmp_type is hlapi.IpAddress:
             read_val = []
             for _i in range(dims[0]):
-                read_val.append((None, snmp_type(f"1.1.1.1")))
+                read_val.append((None, snmp_type("1.1.1.1")))
         elif snmp_type is hlapi.OctetString:
             read_val = []
             for _i in range(dims[0]):
@@ -106,7 +109,7 @@ class SNMPServerFixture:
 
         if dims == self.DIM_LIST["scalar"]:
             snmp_type_dict = {
-                hlapi.ObjectIdentity: "1.3.6.1.2.1.1.1.0.1",
+                hlapi.ObjectIdentity: self.sysDescr_oid,
                 hlapi.IpAddress: "1.1.1.1",
                 hlapi.OctetString: "1",
             }
@@ -116,7 +119,7 @@ class SNMPServerFixture:
                     check_val = v
         elif dims == self.DIM_LIST["spectrum"]:
             snmp_type_dict = {
-                hlapi.ObjectIdentity: ["1.3.6.1.2.1.1.1.0.1"] * dims[0],
+                hlapi.ObjectIdentity: [self.sysDescr_oid] * dims[0],
                 hlapi.IpAddress: ["1.1.1.1"] * dims[0],
                 hlapi.OctetString: ["1"] * dims[0],
             }
@@ -131,9 +134,12 @@ class SNMPServerFixture:
 
 
 class TestSNMP(base.TestCase):
+    """Test class for SNMP client"""
+
     def test_annotation_fail(self):
         """
-        unit test for the processing of annotation. Has 2 lists. 1 with things that should succeed and 1 with things that should fail.
+        unit test for the processing of annotation. Has 2 lists.
+        1 with things that should succeed and 1 with things that should fail.
         """
 
         client = SNMPClient(
@@ -207,7 +213,8 @@ class TestSNMP(base.TestCase):
     @mock.patch("tangostationcontrol.clients.snmp.snmp_client.SNMPComm.setter")
     def test_snmp_obj_set(self, m_next, m_nextCmd, m_obj_T, m_obj_ID):
         """
-        Attempts to write a value to an SNMP server, but instead intercepts it and compared whether the values is as expected.
+        Attempts to write a value to an SNMP server, but instead intercepts it
+        and compared whether the values is as expected.
         """
 
         server = SNMPServerFixture()
@@ -310,12 +317,15 @@ class TestSNMP(base.TestCase):
         )
 
 
-class testSNMPSimulators(base.TestCase):
+class TestSNMPSimulators(base.TestCase):
+    """Test class for SNMP device simulators"""
+
     def test_simulator(self):
+        """Test whether simulator read and write operations work as expected"""
         # read a simple scalar attribute
-        batt_temp = PCON_sim(name="battTemperature", idx=1, dim_x=1, dim_y=0)
+        pwr_status = PCON_sim(name=PCON_sim.PCON_SCALAR_ATT, idx=1, dim_x=1, dim_y=0)
         self.assertEqual(
-            batt_temp.read_function(), 50, "Could not read single value attribute"
+            pwr_status.read_function(), 1, "Could not read single value attribute"
         )
 
         # read socket 5 out of the socket list
@@ -356,6 +366,8 @@ class testSNMPSimulators(base.TestCase):
 
 
 class TestMibLoading(base.TestCase):
+    """Test class for MIB file load operations"""
+
     # name and directory of the pymib file
     MIB = "TEST-MIB"
 
@@ -364,8 +376,10 @@ class TestMibLoading(base.TestCase):
 
     def test_retrieve_mib_content(self):
         """
-        This file contains a 1 variable named: testNamedValue with oid "1.3.99.1.99" with named values: ("test_name", 1), ("other_name", 2)
-        In order to confirm that the mib is indeed loaded correctly this test has to get the oids, the values and the named values
+        This file contains a 1 variable named: testNamedValue with oid "1.3.99.1.99"
+        with named values: ("test_name", 1), ("other_name", 2)
+        In order to confirm that the mib is indeed loaded correctly this test has to get the oids,
+        the values and the named values
 
         """
 
@@ -375,7 +389,7 @@ class TestMibLoading(base.TestCase):
         loader.load_pymib(self.MIB)
 
         # used to view mibs client side
-        mibView = view.MibViewController(loader.mibBuilder)
+        mib_view = view.MibViewController(loader.mibBuilder)
 
         # The expected testNamedValue parameters as written in TEST-MIB.mib
         testNamedValue_oid = "1.3.99.1.99"
@@ -384,24 +398,25 @@ class TestMibLoading(base.TestCase):
         testNamedValue_value = 1
 
         # get testValue and set a value of 1
-        obj_T = hlapi.ObjectType(
+        obj_type = hlapi.ObjectType(
             ObjectIdentity(self.MIB, testNamedValue), hlapi.Integer32(1)
         )
-        obj_T.resolveWithMib(mibView)
+        obj_type.resolveWithMib(mib_view)
 
         # get the oid
-        self.assertEqual(str(obj_T[0]), testNamedValue_oid)
+        self.assertEqual(str(obj_type[0]), testNamedValue_oid)
 
         # get the name format: mib::name
-        self.assertEqual(obj_T[0].prettyPrint(), f"{self.MIB}::{testNamedValue}")
+        self.assertEqual(obj_type[0].prettyPrint(), f"{self.MIB}::{testNamedValue}")
 
         # get the namedValue
-        self.assertEqual(str(obj_T[1]), testNamedValue_named)
+        self.assertEqual(str(obj_type[1]), testNamedValue_named)
 
         # get the numerical value
-        self.assertEqual(int(obj_T[1]), testNamedValue_value)
+        self.assertEqual(int(obj_type[1]), testNamedValue_value)
 
     def test_mib_missing(self):
+        """Test whether MIB file missing exception is raised"""
         abs_dir = path.dirname(__file__) + "/" + self.REL_DIR + "/"
         loader = MIBLoader(abs_dir)
 

tangostationcontrol/tox.ini

@@ -80,12 +80,12 @@ commands =
     pep8: {envpython} -m doc8 docs/source/ --ignore D001
     pep8: {envpython} -m flake8
     black: {envpython} -m black --version
-    black: {envpython} -m black --check --diff . --extend-exclude=libhdbpp-python|SNMP_mib_loading
+    black: {envpython} -m black --check --diff . --extend-exclude=libhdbpp-python|SNMP_mib_loading|output_pymibs
     pylint: {envpython} -m pylint --version
     pylint: {envpython} -m pylint --ignore=test,integration_test --max-line-length=88 tangostationcontrol
     format: {envpython} -m autopep8 --version
     format: {envpython} -m autopep8 -v -aa --in-place --recursive tangostationcontrol/
-    format: {envpython} -m black -v . --extend-exclude=SNMP_mib_loading
+    format: {envpython} -m black -v . --extend-exclude=SNMP_mib_loading|output_pymibs
 
 [testenv:bandit];
 ; B104: hardcoded_bind_all_interfaces
@@ -117,4 +117,4 @@ commands =
 [flake8]
 filename = *.py,.stestr.conf,.txt
 ignore = B014, B019, B028, W291, W293, W391, E111, E114, E121, E122, E123, E124, E126, E127, E128, E131, E201, E201, E202, E203, E221, E222, E225, E226, E231, E241, E251, E252, E261, E262, E265, E271, E301, E302, E303, E305, E306, E401, E402, E501, E502, E701, E712, E721, E731, F403, F523, F541, F841, H301, H306, H401, H403, H404, H405, W503
-exclude = .tox,build,.egg-info,libhdbpp-python,SNMP_mib_loading
+exclude = .tox,build,.egg-info,libhdbpp-python,SNMP_mib_loading,output_pymibs