Read CMIS data path state duration

sonic_platform_base/sonic_xcvr/api/public/cmis.py

@@ -675,6 +675,24 @@ def is_coherent_module(self):
         mintf = self.get_module_media_interface()
         return False if 'ZR' not in mintf else True
 
+    def get_datapath_init_duration(self):
+        '''
+        This function returns the duration of datapath init
+        '''
+        if self.is_flat_memory():
+            return 0
+        duration = self.xcvr_eeprom.read(consts.DP_PATH_INIT_DURATION)
+        return float(duration) if duration is not None else 0
+
+    def get_datapath_deinit_duration(self):
+        '''
+        This function returns the duration of datapath deinit
+        '''
+        if self.is_flat_memory():
+            return 0
+        duration = self.xcvr_eeprom.read(consts.DP_PATH_DEINIT_DURATION)
+        return float(duration) if duration is not None else 0
+
     def get_host_lane_count(self):
         '''
         This function returns number of host lanes for default application

sonic_platform_base/sonic_xcvr/codes/public/cmis.py

@@ -129,4 +129,21 @@ class CmisCodes(Sff8024):
         7: 'Command not compatible with operating status'
     }
 
+    DP_PATH_TIMINGS = {
+        0: '1',
+        1: '5',
+        2: '10',
+        3: '50',
+        4: '100',
+        5: '500',
+        6: '1000',
+        7: '5000',
+        8: '10000',
+        9: '60000',
+        10: '300000',
+        11: '600000',
+        12: '3000000',
+        13: '6000000',
+    }
+
     # TODO: Add other codes

sonic_platform_base/sonic_xcvr/fields/consts.py

@@ -170,6 +170,8 @@
 ACTIVE_FW_MINOR_REV = "ModuleActiveFirmwareMinorRevision"
 INACTIVE_FW_MAJOR_REV = "ModuleInactiveFirmwareMajorRevision"
 INACTIVE_FW_MINOR_REV = "ModuleInactiveFirmwareMinorRevision"
+DP_PATH_INIT_DURATION = "DPInitDuration"
+DP_PATH_DEINIT_DURATION = "DPDeinitDuration"
 
 # DOM
 TRANS_DOM_FIELD = "TransceiverDom"

sonic_platform_base/sonic_xcvr/mem_maps/public/cmis.py

@@ -110,11 +110,17 @@ def __init__(self, codes):
 
         # Should contain ONLY upper page fields
         self.ADVERTISING = RegGroupField(consts.ADVERTISING_FIELD,
+            NumberRegField(consts.INACTIVE_FW_MAJOR_REV, self.getaddr(0x1, 128), format="B", size=1),
+            NumberRegField(consts.INACTIVE_FW_MINOR_REV, self.getaddr(0x1, 129), format="B", size=1),
             NumberRegField(consts.HW_MAJOR_REV, self.getaddr(0x1, 130), size=1),
             NumberRegField(consts.HW_MINOR_REV, self.getaddr(0x1, 131), size=1),
+            CodeRegField(consts.DP_PATH_INIT_DURATION, self.getaddr(0x1, 144), self.codes.DP_PATH_TIMINGS,
+                 *(RegBitField("Bit%d" % (bit), bit) for bit in range (0, 4))
+            ),
+            CodeRegField(consts.DP_PATH_DEINIT_DURATION, self.getaddr(0x1, 144), self.codes.DP_PATH_TIMINGS,
+                 *(RegBitField("Bit%d" % (bit), bit) for bit in range (4, 8))
+            ),
             NumberRegField(consts.MEDIA_LANE_ASSIGNMENT_OPTION, self.getaddr(0x1, 176), format="B", size=1),
-            NumberRegField(consts.INACTIVE_FW_MAJOR_REV, self.getaddr(0x1, 128), format="B", size=1),
-            NumberRegField(consts.INACTIVE_FW_MINOR_REV, self.getaddr(0x1, 129), format="B", size=1),
 
             RegGroupField(consts.ACTIVE_APSEL_CODE,
                 *(NumberRegField("%s%d" % (consts.ACTIVE_APSEL_HOSTLANE, lane) , self.getaddr(0x11, offset),

tests/sonic_xcvr/test_cmis.py

@@ -618,6 +618,34 @@ def test_is_coherent_module(self, mock_response, expected):
         result = self.api.is_coherent_module()
         assert result == expected
 
+    @pytest.mark.parametrize("mock_response1, mock_response2, expected", [
+        (True, '1', 0 ),
+        (False, None, 0),
+        (False, '5', 5.0),
+        (False, '60000', 60000.0),
+    ])
+    def test_get_datapath_init_duration(self, mock_response1, mock_response2, expected):
+        self.api.is_flat_memory = MagicMock()
+        self.api.is_flat_memory.return_value = mock_response1
+        self.api.xcvr_eeprom.read = MagicMock()
+        self.api.xcvr_eeprom.read.return_value = mock_response2
+        result = self.api.get_datapath_init_duration()
+        assert result == expected
+
+    @pytest.mark.parametrize("mock_response1, mock_response2, expected", [
+        (True, '10', 0 ),
+        (False, None, 0),
+        (False, '50', 50.0),
+        (False, '6000000', 6000000.0),
+    ])
+    def test_get_datapath_deinit_duration(self, mock_response1, mock_response2, expected):
+        self.api.is_flat_memory = MagicMock()
+        self.api.is_flat_memory.return_value = mock_response1
+        self.api.xcvr_eeprom.read = MagicMock()
+        self.api.xcvr_eeprom.read.return_value = mock_response2
+        result = self.api.get_datapath_deinit_duration()
+        assert result == expected
+
     @pytest.mark.parametrize("mock_response, expected", [
         (8, 8)
     ])