Support reading/writing SFP module EEPROM by page and offset

sonic_platform_base/sfp_base.py

@@ -422,9 +422,24 @@ def read_eeprom(self, offset, num_bytes):
         """
         raise NotImplementedError
 
+    def read_eeprom_by_page(self, page, offset, size, wire_addr=None):
+        """
+        Read EEPROM by page
+
+        Args:
+            page: EEPROM page number. Raise ValueError for invalid page.
+            offset: EEPROM page offset. Raise ValueError for invalid offset.
+            size: Number of bytes to be read. Raise ValueError for invalid size.
+            wire_addr: Wire address. Only valid for sff8472. Raise ValueError for invalid wire address.
+
+        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 +455,21 @@ def write_eeprom(self, offset, num_bytes, write_buffer):
         """
         raise NotImplementedError
 
+    def write_eeprom_by_page(self, page, offset, data, wire_addr=None):
+        """
+        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.
+
+        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/sff8472.py

@@ -7,6 +7,14 @@
 from ...fields import consts
 from ..xcvr_api import XcvrApi
 
+MAX_OFFSET_FOR_A0H_UPPER_PAGE = 255
+MAX_OFFSET_FOR_A0H_LOWER_PAGE = 127
+MAX_PAGE = 255
+MAX_OFFSET_FOR_A2H = 255
+PAGE_SIZE = 128
+PAGE_SIZE_FOR_A0H = 256
+
+
 class Sff8472Api(XcvrApi):
     NUM_CHANNELS = 1
 
@@ -37,7 +45,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],
@@ -298,3 +306,31 @@ def get_power_override_support(self):
 
     def is_copper(self):
         return self.xcvr_eeprom.read(consts.SFP_CABLE_TECH_FIELD) == 'Passive Cable'
+
+    def get_overall_offset(self, page, offset, size, wire_addr=None):
+        if not wire_addr:
+            raise ValueError("Invalid wire address for sff8472, must a0h or a2h")
+
+        is_active_cable = not self.is_copper()
+        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 = MAX_OFFSET_FOR_A0H_UPPER_PAGE if is_active_cable else MAX_OFFSET_FOR_A0H_LOWER_PAGE
+            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 > MAX_PAGE:
+                raise ValueError(f'Invalid page number {page} for wire address {wire_addr}, valid range: [0, 255]')
+            if offset < 0 or offset > MAX_OFFSET_FOR_A2H:
+                raise ValueError(f'Invalid offset {offset} for wire address {wire_addr}, valid range: [0, 255]')
+            if size <= 0 or size + offset - 1 > MAX_OFFSET_FOR_A2H:
+                raise ValueError(f'Invalid size {size} for wire address {wire_addr}, valid range: [1, {255 - offset + 1}]')
+            return page * PAGE_SIZE + offset + PAGE_SIZE_FOR_A0H

sonic_platform_base/sonic_xcvr/api/xcvr_api.py

@@ -5,6 +5,14 @@
     xcvrs in SONiC
 """
 from math import log10
+
+MAX_PAGE = 255
+MAX_OFFSET = 255
+MIN_OFFSET_FOR_PAGE0 = 0
+MIN_OFFSET_FOR_OTHER_PAGE = 128
+PAGE_SIZE = 128
+
+
 class XcvrApi(object):
     def __init__(self, xcvr_eeprom):
         self.xcvr_eeprom = xcvr_eeprom
@@ -637,3 +645,34 @@ def get_error_description(self):
         """
         raise NotImplementedError
 
+    def get_overall_offset(self, page, offset, size, wire_addr=None):
+        """
+        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.
+
+        Returns:
+            The overall offset
+        """
+        max_page = 0 if self.is_flat_memory() else MAX_PAGE
+        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 < MIN_OFFSET_FOR_PAGE0 or offset > MAX_OFFSET:
+                raise ValueError(f'Invalid offset {offset} for page 0, valid range: [0, 255]')
+        else:
+            if offset < MIN_OFFSET_FOR_OTHER_PAGE or offset > MAX_OFFSET:
+                raise ValueError(f'Invalid offset {offset} for page {page}, valid range: [128, 255]')
+
+        if size <= 0 or size + offset - 1 > MAX_OFFSET:
+            raise ValueError(f'Invalid size {size}, valid range: [1, {255 - offset + 1}]')
+
+        return page * PAGE_SIZE + 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):
+        api = self.get_xcvr_api()
+        overall_offset = api.get_overall_offset(page, offset, size, wire_addr) 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):
+        api = self.get_xcvr_api()
+        overall_offset = api.get_overall_offset(page, offset, len(data), wire_addr) 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

@@ -290,3 +290,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_copper = MagicMock(return_value=False)
+        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