Support reading/writing SFP module EEPROM by page and offset

sonic_platform_base/sfp_base.py

@@ -422,9 +422,25 @@ def read_eeprom(self, offset, num_bytes):
         """
         raise NotImplementedError
 
+    def read_eeprom_by_page(self, page, offset, size, wire_addr=None, flat=False):
+        """
+        Read EEPROM by page
+
+        Args:
+            page: EEPROM page number. Raise ValueError for invalid page.
+            offset: EEPROM page offset. Raise ValueError for invalid offset.
+            size: Read size. Raise ValueError for invalid size.
+            wire_addr: Wire address. Only valid for sff8472. Raise ValueError for invalid wire address.
+            flat: Read mode.
+
+        Returns:
+            A string contains the hex format EEPROM data.
+        """
+        raise NotImplementedError
+
     def write_eeprom(self, offset, num_bytes, write_buffer):
         """
-        write eeprom specfic bytes beginning from a random offset with size as num_bytes
+        write eeprom specific bytes beginning from a random offset with size as num_bytes
         and write_buffer as the required bytes
 
         Args:
@@ -440,6 +456,22 @@ def write_eeprom(self, offset, num_bytes, write_buffer):
         """
         raise NotImplementedError
 
+    def write_eeprom_by_page(self, page, offset, data, wire_addr=None, flat=False):
+        """
+        Write EEPROM by page
+
+        Args:
+            page: EEPROM page number. Raise ValueError for invalid page.
+            offset: EEPROM page offset. Raise ValueError for invalid offset.
+            data: bytearray EEPROM data.
+            wire_addr: Wire address. Only valid for sff8472. Raise ValueError for invalid wire address.
+            flat: Write mode.
+
+        Returns:
+            True if write successfully else False
+        """
+        raise NotImplementedError
+
     def get_error_description(self):
         """
         Retrives the error descriptions of the SFP module

sonic_platform_base/sonic_xcvr/api/public/cmis.py

@@ -763,10 +763,10 @@ def get_media_lane_count(self, appl=1):
         '''
         if self.is_flat_memory():
             return 0
-        
+
         if (appl <= 0):
             return 0
-        
+
         appl_advt = self.get_application_advertisement()
         return appl_advt[appl]['media_lane_count'] if len(appl_advt) >= appl else 0
 
@@ -795,10 +795,10 @@ def get_media_lane_assignment_option(self, appl=1):
         '''
         if self.is_flat_memory():
             return 'N/A'
-        
+
         if (appl <= 0):
             return 0
-        
+
         appl_advt = self.get_application_advertisement()
         return appl_advt[appl]['media_lane_assignment_options'] if len(appl_advt) >= appl else 0
 

sonic_platform_base/sonic_xcvr/api/public/sff8472.py

@@ -37,7 +37,7 @@ def get_transceiver_info(self):
             if len > 0:
                 cable_len = len
                 cable_type = type
- 
+
         xcvr_info = {
             "type": serial_id[consts.ID_FIELD],
             "type_abbrv_name": serial_id[consts.ID_ABBRV_FIELD],
@@ -295,3 +295,34 @@ def get_lpmode_support(self):
 
     def get_power_override_support(self):
         return False
+
+    def is_active_cable(self):
+        return self.xcvr_eeprom.read(consts.SFP_CABLE_TECH_FIELD) == 'Active Cable'
+
+    def get_overall_offset(self, page, offset, size, wire_addr=None, flat=False):
+        if not wire_addr:
+            raise ValueError("Invalid wire address for sff8472, must a0h or a2h")
+
+        is_active_cable = self.is_active_cable()
+        valid_wire_address = ('a0h', 'a2h') if is_active_cable else ('a0h',)
+        wire_addr = wire_addr.lower()
+        if wire_addr not in valid_wire_address:
+            raise ValueError(f"Invalid wire address {wire_addr} for sff8472, must be {' or '.join(valid_wire_address)}")
+
+        if wire_addr == 'a0h':
+            if page != 0:
+                raise ValueError(f'Invalid page number {page} for wire address {wire_addr}, only page 0 is supported')
+            max_offset = 255 if is_active_cable else 127
+            if offset < 0 or offset > max_offset:
+                raise ValueError(f'Invalid offset {offset} for wire address {wire_addr}, valid range: [0, {max_offset}]')
+            if size <= 0 or size + offset - 1 > max_offset:
+                raise ValueError(f'Invalid size {size} for wire address {wire_addr}, valid range: [1, {max_offset - offset + 1}]')
+            return offset
+        else:
+            if page < 0 or page > 255:
+                raise ValueError(f'Invalid page number {page} for wire address {wire_addr}, valid range: [0, 255]')
+            if offset < 0 or offset > 255:
+                raise ValueError(f'Invalid offset {offset} for wire address {wire_addr}, valid range: [0, 255]')
+            if size <= 0 or size + offset - 1 > 255:
+                raise ValueError(f'Invalid size {size} for wire address {wire_addr}, valid range: [1, {255 - offset + 1}]')
+            return page * 128 + offset + 256

sonic_platform_base/sonic_xcvr/api/xcvr_api.py

@@ -637,3 +637,35 @@ def get_error_description(self):
         """
         raise NotImplementedError
 
+    def get_overall_offset(self, page, offset, size, wire_addr=None, flat=False):
+        """
+        Retrieves the overall offset of the given page, offset and size
+
+        Args:
+            page: The page number
+            offset: The offset within the page
+            size: The size of the data
+            wire_addr: Wire address. Only valid for sff8472. Raise ValueError for invalid wire address.
+            flat: A boolean, True if flat mode
+
+        Returns:
+            The overall offset
+        """
+        max_page = 0 if self.is_flat_memory() else 255
+        if max_page == 0 and page != 0:
+            raise ValueError(f'Invalid page number {page}, only page 0 is supported')
+
+        if page < 0 or page > max_page:
+            raise ValueError(f'Invalid page number {page}, valid range: [0, {max_page}]')
+
+        if page == 0:
+            if offset < 0 or offset > 255:
+                raise ValueError(f'Invalid offset {offset} for page 0, valid range: [0, 255]')
+        else:
+            if offset < 128 or offset > 255:
+                raise ValueError(f'Invalid offset {offset} for page {page}, valid range: [128, 255]')
+
+        if size <= 0 or size + offset - 1 > 255:
+            raise ValueError(f'Invalid size {size}, valid range: [1, {255 - offset + 1}]')
+
+        return page * 128 + offset

sonic_platform_base/sonic_xcvr/sfp_optoe_base.py

@@ -183,6 +183,13 @@ def read_eeprom(self, offset, num_bytes):
         except (OSError, IOError):
             return None
 
+    def read_eeprom_by_page(self, page, offset, size, wire_addr=None, flat=False):
+        api = self.get_xcvr_api()
+        overall_offset = api.get_overall_offset(page, offset, size, wire_addr, flat) if api is not None else None
+        if overall_offset is None:
+            return None
+        return self.read_eeprom(overall_offset, size)
+
     def write_eeprom(self, offset, num_bytes, write_buffer):
         try:
             with open(self.get_eeprom_path(), mode='r+b', buffering=0) as f:
@@ -192,6 +199,13 @@ def write_eeprom(self, offset, num_bytes, write_buffer):
             return False
         return True
 
+    def write_eeprom_by_page(self, page, offset, data, wire_addr=None, flat=False):
+        api = self.get_xcvr_api()
+        overall_offset = api.get_overall_offset(page, offset, len(data), wire_addr, flat) if api is not None else None
+        if overall_offset is None:
+            return False
+        return self.write_eeprom(overall_offset, len(data), data)
+
     def reset(self):
         """
         Reset SFP and return all user module settings to their default state.

tests/sonic_xcvr/test_cmis.py

@@ -2371,3 +2371,32 @@ def test_get_transceiver_info_firmware_versions_negative_tests(self):
         self.api.get_module_fw_info.side_effect = {'result': TypeError}
         result = self.api.get_transceiver_info_firmware_versions()
         assert result == ["N/A", "N/A"]
+
+    def test_get_overall_offset(self):
+        self.api.is_flat_memory = MagicMock(return_value=False)
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(-1, 0, 1)
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(256, 0, 1)
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(0, -1, 1)
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(0, 256, 1)
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(1, 127, 1)
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(1, 256, 1)
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(0, 0, 0)
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(0, 0, 257)
+
+        assert self.api.get_overall_offset(0, 1, 1) == 1

tests/sonic_xcvr/test_sff8472.py

@@ -22,7 +22,7 @@ class TestSff8472(object):
 
     def test_api(self):
         """
-        Verify all api access valid fields    
+        Verify all api access valid fields
         """
         self.api.get_model()
         self.api.get_serial()
@@ -61,7 +61,7 @@ def test_temp(self):
         data = bytearray([0x80, 0x00])
         deps = {
            consts.INT_CAL_FIELD: True,
-           consts.EXT_CAL_FIELD: False, 
+           consts.EXT_CAL_FIELD: False,
            consts.T_SLOPE_FIELD: 1,
            consts.T_OFFSET_FIELD: 0,
         }
@@ -71,7 +71,7 @@ def test_temp(self):
         data = bytearray([0x0F, 0xFF])
         deps = {
            consts.INT_CAL_FIELD: False,
-           consts.EXT_CAL_FIELD: True, 
+           consts.EXT_CAL_FIELD: True,
            consts.T_SLOPE_FIELD: 2,
            consts.T_OFFSET_FIELD: 10,
         }
@@ -83,7 +83,7 @@ def test_voltage(self):
         data = bytearray([0xFF, 0xFF])
         deps = {
            consts.INT_CAL_FIELD: True,
-           consts.EXT_CAL_FIELD: False, 
+           consts.EXT_CAL_FIELD: False,
            consts.T_SLOPE_FIELD: 1,
            consts.T_OFFSET_FIELD: 0,
         }
@@ -94,7 +94,7 @@ def test_voltage(self):
         data = bytearray([0x7F, 0xFF])
         deps = {
            consts.INT_CAL_FIELD: False,
-           consts.EXT_CAL_FIELD: True, 
+           consts.EXT_CAL_FIELD: True,
            consts.V_SLOPE_FIELD: 2,
            consts.V_OFFSET_FIELD: 10,
         }
@@ -106,7 +106,7 @@ def test_tx_bias(self):
         data = bytearray([0xFF, 0xFF])
         deps = {
            consts.INT_CAL_FIELD: True,
-           consts.EXT_CAL_FIELD: False, 
+           consts.EXT_CAL_FIELD: False,
            consts.TX_I_SLOPE_FIELD: 1,
            consts.TX_I_OFFSET_FIELD: 0,
         }
@@ -117,7 +117,7 @@ def test_tx_bias(self):
         data = bytearray([0x7F, 0xFF])
         deps = {
            consts.INT_CAL_FIELD: False,
-           consts.EXT_CAL_FIELD: True, 
+           consts.EXT_CAL_FIELD: True,
            consts.TX_I_SLOPE_FIELD: 2,
            consts.TX_I_OFFSET_FIELD: 10,
         }
@@ -129,7 +129,7 @@ def test_tx_power(self):
         data = bytearray([0xFF, 0xFF])
         deps = {
            consts.INT_CAL_FIELD: True,
-           consts.EXT_CAL_FIELD: False, 
+           consts.EXT_CAL_FIELD: False,
            consts.TX_PWR_SLOPE_FIELD: 1,
            consts.TX_PWR_OFFSET_FIELD: 0,
         }
@@ -140,7 +140,7 @@ def test_tx_power(self):
         data = bytearray([0x7F, 0xFF])
         deps = {
            consts.INT_CAL_FIELD: False,
-           consts.EXT_CAL_FIELD: True, 
+           consts.EXT_CAL_FIELD: True,
            consts.TX_PWR_SLOPE_FIELD: 2,
            consts.TX_PWR_OFFSET_FIELD: 10,
         }
@@ -152,7 +152,7 @@ def test_rx_power(self):
         data = bytearray([0xFF, 0xFF])
         deps = {
            consts.INT_CAL_FIELD: True,
-           consts.EXT_CAL_FIELD: False, 
+           consts.EXT_CAL_FIELD: False,
            consts.RX_PWR_0_FIELD: 0,
            consts.RX_PWR_1_FIELD: 1,
            consts.RX_PWR_2_FIELD: 0,
@@ -165,7 +165,7 @@ def test_rx_power(self):
 
         deps = {
            consts.INT_CAL_FIELD: False,
-           consts.EXT_CAL_FIELD: True, 
+           consts.EXT_CAL_FIELD: True,
            consts.RX_PWR_0_FIELD: 10,
            consts.RX_PWR_1_FIELD: 2,
            consts.RX_PWR_2_FIELD: 0.1,
@@ -289,3 +289,44 @@ def test_get_transceiver_bulk_status(self, mock_response, expected):
         result = self.api.get_transceiver_bulk_status()
         assert result == expected
 
+    def test_get_overall_offset(self):
+        self.api.is_active_cable = MagicMock(return_value=True)
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(0, 0, 1)
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(0, 0, 1, wire_addr='invalid')
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(1, 0, 1, wire_addr='a0h')
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(0, -1, 1, wire_addr='A0h')
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(0, 256, 1, wire_addr='A0h')
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(0, 0, 0, wire_addr='A0h')
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(0, 0, 257, wire_addr='A0h')
+
+        assert self.api.get_overall_offset(0, 2, 2, wire_addr='A0h') == 2
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(-1, 0, 1, wire_addr='a2h')
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(256, 0, 1, wire_addr='a2h')
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(0, -1, 1, wire_addr='a2h')
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(0, 0, 0, wire_addr='A2h')
+
+        with pytest.raises(ValueError):
+            self.api.get_overall_offset(0, 0, 257, wire_addr='A2h')
+
+        assert self.api.get_overall_offset(0, 2, 2, wire_addr='A2h') == 258

tests/sonic_xcvr/test_sfp_optoe_base.py

@@ -0,0 +1,36 @@
+from mock import MagicMock
+
+from sonic_platform_base.sonic_xcvr import sfp_optoe_base
+from sonic_platform_base.sonic_xcvr.api.public.cmis import CmisApi
+from sonic_platform_base.sonic_xcvr.mem_maps.public.cmis import CmisMemMap
+from sonic_platform_base.sonic_xcvr.xcvr_eeprom import XcvrEeprom
+from sonic_platform_base.sonic_xcvr.codes.public.cmis import CmisCodes
+
+
+class TestSfpOptoeBase:
+    codes = CmisCodes
+    mem_map = CmisMemMap(codes)
+    reader = MagicMock(return_value=None)
+    writer = MagicMock()
+    eeprom = XcvrEeprom(reader, writer, mem_map)
+    old_read_func = eeprom.read
+    api = CmisApi(eeprom)
+
+    def test_read_eeprom_by_page(self):
+        sfp = sfp_optoe_base.SfpOptoeBase()
+        sfp.get_xcvr_api = MagicMock(return_value=TestSfpOptoeBase.api)
+        sfp.read_eeprom = MagicMock(return_value=bytearray([0]))
+        assert sfp.read_eeprom_by_page(0, 0, 1) is not None
+
+        sfp.get_xcvr_api.return_value = None
+        assert sfp.read_eeprom_by_page(0, 0, 1) is None
+
+    def test_write_eeprom(self):
+        sfp = sfp_optoe_base.SfpOptoeBase()
+        sfp.get_xcvr_api = MagicMock(return_value=TestSfpOptoeBase.api)
+        sfp.write_eeprom = MagicMock(return_value=True)
+        data = bytearray([0])
+        assert sfp.write_eeprom_by_page(0, 0, data) is True
+
+        sfp.get_xcvr_api.return_value = None
+        assert sfp.write_eeprom_by_page(0, 0, data) is False