In order to collect PM stats without loss of data following vdm freez…

…e and unfreeze request needs to be followed.

VDM freeze
PM stats collection
VDM unfreeze

sonic_platform_base/sonic_xcvr/api/public/c_cmis.py

@@ -147,6 +147,43 @@ def set_tx_power(self, tx_power):
         time.sleep(1)
         return status
 
+    def freeze_vdm_stats(self):
+        '''
+        This function freeze all the vdm statistics reporting registers.
+        When raised by the host, causes the module to freeze and hold all 
+        reported statistics reporting registers (minimum, maximum and 
+        average values)in Pages 24h-27h.
+
+        Returns True if the provision succeeds and False incase of failure.
+        '''
+        return self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_FREEZE)
+
+    def get_freeze_vdm_stats(self):
+        '''
+        This function reads and returns the vdm Freeze done status.
+
+        Returns True if the vdm stats freeze is successful and False if not freeze.
+        '''
+        return self.xcvr_eeprom.read(consts.VDM_FREEZE_DONE)
+
+    def unfreeze_vdm_stats(self):
+        '''
+        This function unfreeze all the vdm statistics reporting registers.
+        When freeze is ceased by the host, releases the freeze request, allowing the 
+        reported minimum, maximum and average values to update again.
+        
+        Returns True if the provision succeeds and False incase of failure.
+        '''
+        return self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_UNFREEZE)
+
+    def get_unfreeze_vdm_stats(self):
+        '''
+        This function reads and returns the vdm unfreeze status.
+
+        Returns True if the vdm stats unfreeze is successful and False if not unfreeze.
+        '''
+        return self.xcvr_eeprom.read(consts.VDM_UNFREEZE_DONE)
+
     def get_pm_all(self):
         '''
         This function returns the PMs reported in Page 34h and 35h in OIF C-CMIS document
@@ -163,14 +200,6 @@ def get_pm_all(self):
         SOPROC: unit in krad/s
         MER:    unit in dB
         '''
-        # When raised by the host, causes the module to freeze and hold all 
-        # reported statistics reporting registers (minimum, maximum and 
-        # average values)in Pages 24h-27h.
-        # When ceased by the host, releases the freeze request, allowing the 
-        # reported minimum, maximum and average values to update again.
-        self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_FREEZE)
-        time.sleep(1)
-        self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_UNFREEZE)
         PM_dict = dict()
 
         rx_bits_pm = self.xcvr_eeprom.read(consts.RX_BITS_PM)
@@ -256,7 +285,6 @@ def get_pm_all(self):
         PM_dict['rx_mer_max'] = self.xcvr_eeprom.read(consts.RX_MAX_MER_PM)
         return PM_dict
 
-
     def get_transceiver_info(self):
         """
         Retrieves transceiver info of this SFP

sonic_platform_base/sonic_xcvr/fields/consts.py

@@ -413,6 +413,10 @@
 VDM_SUPPORTED = "VdmSupported"
 VDM_SUPPORTED_PAGE = "VdmSupportedPage"
 VDM_CONTROL = "VdmControl"
+VDM_STATUS = "VdmStatus"
+VDM_FREEZE_DONE = "VdmFreezeDone"
+VDM_UNFREEZE_DONE = "VdmUnfreezeDone"
+
 MEDIA_LANE_FEC_PM = "Media Lane FEC Performance Monitoring"
 MEDIA_LANE_LINK_PM = "Media Lane Link Performance Monitoring"
 RX_BITS_PM = "rxBitsPm"

sonic_platform_base/sonic_xcvr/mem_maps/public/cmis.py

@@ -449,6 +449,10 @@ def __init__(self, codes):
                 *(RegBitField("Bit%d" % (bit), bit) for bit in range (0, 2))
             ),
             NumberRegField(consts.VDM_CONTROL, self.getaddr(0x2f, 144), size=1, ro=False),
+            NumberRegField(consts.VDM_STATUS, self.getaddr(0x2f, 145),
+                RegBitField(consts.VDM_UNFREEZE_DONE, 6),
+                RegBitField(consts.VDM_FREEZE_DONE, 7),
+            ),
         )
 
         self.TRANS_CONFIG = RegGroupField(consts.TRANS_CONFIG_FIELD,

sonic_platform_base/sonic_xcvr/sfp_optoe_base.py

@@ -51,6 +51,47 @@ def get_transceiver_pm(self):
         api = self.get_xcvr_api()
         return api.get_transceiver_pm() if api is not None else None
 
+    def freeze_vdm_stats(self):
+        '''
+        This function freeze all the vdm statistics reporting registers.
+        When raised by the host, causes the module to freeze and hold all 
+        reported statistics reporting registers (minimum, maximum and 
+        average values)in Pages 24h-27h.
+
+        Returns True if the provision succeeds and False incase of failure.
+        '''
+        api = self.get_xcvr_api()
+        return api.freeze_vdm_stats() if api is not None else None
+
+    def unfreeze_vdm_stats(self):
+        '''
+        This function unfreeze all the vdm statistics reporting registers.
+        When freeze is ceased by the host, releases the freeze request, allowing the 
+        reported minimum, maximum and average values to update again.
+
+        Returns True if the provision succeeds and False incase of failure.
+        '''
+        api = self.get_xcvr_api()
+        return api.unfreeze_vdm_stats() if api is not None else None
+
+    def get_freeze_vdm_stats(self):
+        '''
+        This function reads and returns the vdm Freeze done status.
+
+        Returns True if the vdm stats freeze is successful and False if not freeze.
+        '''
+        api = self.get_xcvr_api()
+        return api.get_freeze_vdm_stats() if api is not None else None
+
+    def get_unfreeze_vdm_stats(self):
+        '''
+        This function reads and returns the vdm unfreeze status.
+
+        Returns True if the vdm stats unfreeze is successful and False if not unfreeze.
+        '''
+        api = self.get_xcvr_api()
+        return api.get_unfreeze_vdm_stats() if api is not None else None
+
     def get_rx_los(self):
         api = self.get_xcvr_api()
         if api is not None:

tests/sonic_xcvr/test_ccmis.py

@@ -624,3 +624,44 @@ def test_get_transceiver_pm(self, mock_response, expected):
         self.api.get_pm_all.return_value = mock_response
         result = self.api.get_transceiver_pm()
         assert result == expected
+
+    @pytest.mark.parametrize("mock_response, expected", [
+        (0, 0),
+        (1, 1),
+    ])
+    def test_get_freeze_vdm_stats(self, mock_response, expected):
+        self.api.xcvr_eeprom.read = MagicMock()
+        self.api.xcvr_eeprom.read.return_value = mock_response
+        result = self.api.get_freeze_vdm_stats()
+        assert result == expected
+
+    @pytest.mark.parametrize("mock_response, expected", [
+        (0, 0),
+        (1, 1),
+    ])
+    def test_get_unfreeze_vdm_stats(self, mock_response, expected):
+        self.api.xcvr_eeprom.read = MagicMock()
+        self.api.xcvr_eeprom.read.return_value = mock_response
+        result = self.api.get_unfreeze_vdm_stats()
+        assert result == expected
+
+    @pytest.mark.parametrize("mock_response, expected", [
+        (0, 0),
+        (1, 1),
+    ])
+    def test_freeze_vdm_stats(self, mock_response, expected):
+        self.api.xcvr_eeprom.write = MagicMock()
+        self.api.xcvr_eeprom.write.return_value = mock_response
+        result = self.api.freeze_vdm_stats()
+        assert result == expected
+
+    @pytest.mark.parametrize("mock_response, expected", [
+        (0, 0),
+        (1, 1),
+    ])
+    def test_unfreeze_vdm_stats(self, mock_response, expected):
+        self.api.xcvr_eeprom.write = MagicMock()
+        self.api.xcvr_eeprom.write.return_value = mock_response
+        result = self.api.unfreeze_vdm_stats()
+        assert result == expected
+

tests/sonic_xcvr/test_sfp_optoe_base.py

@@ -0,0 +1,74 @@
+from mock import MagicMock 
+from mock import patch 
+import pytest 
+from sonic_platform_base.sonic_xcvr.sfp_optoe_base import SfpOptoeBase 
+from sonic_platform_base.sonic_xcvr.api.public.c_cmis import CCmisApi 
+from sonic_platform_base.sonic_xcvr.mem_maps.public.c_cmis import CCmisMemMap 
+from sonic_platform_base.sonic_xcvr.xcvr_eeprom import XcvrEeprom 
+from sonic_platform_base.sonic_xcvr.codes.public.cmis import CmisCodes 
+
+class TestSfpOptoeBase(object): 
+
+    codes = CmisCodes 
+    mem_map = CCmisMemMap(codes) 
+    reader = MagicMock(return_value=None) 
+    writer = MagicMock() 
+    eeprom = XcvrEeprom(reader, writer, mem_map) 
+    sfp_optoe_api = SfpOptoeBase() 
+    ccmis_api = CCmisApi(eeprom) 
+
+    @pytest.mark.parametrize("mock_response1, mock_response2, expected", [ 
+        (0, ccmis_api, 0), 
+        (1, ccmis_api, 1), 
+    ]) 
+    def test_freeze_vdm_stats(self, mock_response1, mock_response2, expected): 
+        self.sfp_optoe_api.get_xcvr_api = MagicMock() 
+        self.sfp_optoe_api.get_xcvr_api.return_value = mock_response2  
+        self.ccmis_api.freeze_vdm_stats = MagicMock() 
+        self.ccmis_api.freeze_vdm_stats.return_value = mock_response1 
+
+        result = self.sfp_optoe_api.freeze_vdm_stats() 
+        assert result == expected
+
+    @pytest.mark.parametrize("mock_response1, mock_response2, expected", [
+        (0, ccmis_api, 0),
+        (1, ccmis_api, 1),
+        (None, None, None),
+    ])
+    def test_unfreeze_vdm_stats(self, mock_response1, mock_response2, expected):
+        self.sfp_optoe_api.get_xcvr_api = MagicMock()
+        self.sfp_optoe_api.get_xcvr_api.return_value = mock_response2
+        self.ccmis_api.unfreeze_vdm_stats = MagicMock()
+        self.ccmis_api.unfreeze_vdm_stats.return_value = mock_response1
+
+        result = self.sfp_optoe_api.unfreeze_vdm_stats()
+        assert result == expected
+
+    @pytest.mark.parametrize("mock_response1, mock_response2, expected", [
+        (0, ccmis_api, 0),
+        (1, ccmis_api, 1),
+        (None, None, None),
+    ])
+    def test_get_freeze_vdm_stats(self, mock_response1, mock_response2, expected):
+        self.sfp_optoe_api.get_xcvr_api = MagicMock()
+        self.sfp_optoe_api.get_xcvr_api.return_value = mock_response2
+        self.ccmis_api.get_freeze_vdm_stats = MagicMock()
+        self.ccmis_api.get_freeze_vdm_stats.return_value = mock_response1
+
+        result = self.sfp_optoe_api.get_freeze_vdm_stats()
+        assert result == expected
+
+    @pytest.mark.parametrize("mock_response1, mock_response2, expected", [
+        (0, ccmis_api, 0),
+        (1, ccmis_api, 1), 
+        (None, None, None),
+    ])
+    def test_get_unfreeze_vdm_stats(self, mock_response1, mock_response2, expected):
+        self.sfp_optoe_api.get_xcvr_api = MagicMock()
+        self.sfp_optoe_api.get_xcvr_api.return_value = mock_response2
+        self.ccmis_api.get_unfreeze_vdm_stats = MagicMock()
+        self.ccmis_api.get_unfreeze_vdm_stats.return_value = mock_response1
+
+        result = self.sfp_optoe_api.get_unfreeze_vdm_stats()
+        assert result == expected
+