[Mellanox] Add platform API support for CPO modules

platform/mellanox/mlnx-platform-api/sonic_platform/chassis.py

@@ -30,6 +30,7 @@
     from sonic_py_common import device_info
     from functools import reduce
     from .utils import extract_RJ45_ports_index
+    from .utils import extract_cpo_ports_index
     from . import module_host_mgmt_initializer
     from . import utils
     from .device_data import DeviceDataManager
@@ -120,6 +121,10 @@ def __init__(self):
         self._RJ45_port_inited = False
         self._RJ45_port_list = None
 
+        # Build the CPO port list from platform.json and hwsku.json
+        self._cpo_port_inited = False
+        self._cpo_port_list = None
+
         Chassis.chassis_instance = self
 
         self.module_host_mgmt_initializer = module_host_mgmt_initializer.ModuleHostMgmtInitializer()
@@ -139,6 +144,13 @@ def RJ45_port_list(self):
             self._RJ45_port_inited = True
         return self._RJ45_port_list
 
+    @property
+    def cpo_port_list(self):
+        if not self._cpo_port_inited:
+            self._cpo_port_list = extract_cpo_ports_index()
+            self._cpo_port_inited = True
+        return self._cpo_port_list
+
     ##############################################
     # PSU methods
     ##############################################
@@ -277,6 +289,8 @@ def initialize_single_sfp(self, index):
                         sfp_module = self._import_sfp_module()
                         if self.RJ45_port_list and index in self.RJ45_port_list:
                             self._sfp_list[index] = sfp_module.RJ45Port(index)
+                        elif self.cpo_port_list and index in self.cpo_port_list:
+                            self._sfp_list[index] = sfp_module.CpoPort(index)
                         else:
                             self._sfp_list[index] = sfp_module.SFP(index)
                         self.sfp_initialized_count += 1
@@ -294,6 +308,8 @@ def initialize_sfp(self):
                         for index in range(sfp_count):
                             if self.RJ45_port_list and index in self.RJ45_port_list:
                                 sfp_object = sfp_module.RJ45Port(index)
+                            elif self.cpo_port_list and index in self.cpo_port_list:
+                                sfp_object = sfp_module.CpoPort(index)
                             else:
                                 sfp_object = sfp_module.SFP(index)
                             self._sfp_list.append(sfp_object)
@@ -304,6 +320,8 @@ def initialize_sfp(self):
                             if self._sfp_list[index] is None:
                                 if self.RJ45_port_list and index in self.RJ45_port_list:
                                     self._sfp_list[index] = sfp_module.RJ45Port(index)
+                                elif self.cpo_port_list and index in self.cpo_port_list:
+                                    self._sfp_list[index] = sfp_module.CpoPort(index)
                                 else:
                                     self._sfp_list[index] = sfp_module.SFP(index)
                         self.sfp_initialized_count = len(self._sfp_list)
@@ -315,13 +333,22 @@ def get_num_sfps(self):
         Returns:
             An integer, the number of sfps available on this chassis
         """
+        num_sfps = 0
         if not self._RJ45_port_inited:
             self._RJ45_port_list = extract_RJ45_ports_index()
             self._RJ45_port_inited = True
+
+        if not self._cpo_port_inited:
+            self._cpo_port_list = extract_cpo_ports_index()
+            self._cpo_port_inited = True
+
+        num_sfps = DeviceDataManager.get_sfp_count()
         if self._RJ45_port_list is not None:
-            return DeviceDataManager.get_sfp_count() + len(self._RJ45_port_list)
-        else:
-            return DeviceDataManager.get_sfp_count()
+            num_sfps += len(self._RJ45_port_list)
+        if self._cpo_port_list is not None:
+            num_sfps += len(self._cpo_port_list)
+
+        return num_sfps
 
     def get_all_sfps(self):
         """

platform/mellanox/mlnx-platform-api/sonic_platform/sfp.py

@@ -36,7 +36,11 @@
     from .device_data import DeviceDataManager
     from sonic_platform_base.sonic_xcvr.sfp_optoe_base import SfpOptoeBase
     from sonic_platform_base.sonic_xcvr.fields import consts
-    from sonic_platform_base.sonic_xcvr.api.public import cmis, sff8636, sff8436
+    from sonic_platform_base.sonic_xcvr.api.public import sff8636, sff8436
+
+    from sonic_platform_base.sonic_xcvr.api.public import cmis as cmis_api
+    from sonic_platform_base.sonic_xcvr.codes.public import cmis as cmis_codes
+    from sonic_platform_base.sonic_xcvr.mem_maps.public import cmis as cmis_mem
 
 except ImportError as e:
     raise ImportError (str(e) + "- required module not found")
@@ -72,6 +76,7 @@
 ]
 
 RJ45_TYPE = "RJ45"
+CPO_TYPE = "CPO"
 
 #variables for sdk
 REGISTER_NUM = 1
@@ -1086,7 +1091,7 @@ def is_cmis_api(self, xcvr_api):
         Returns:
             bool: True if the api is of type CMIS
         """
-        return isinstance(xcvr_api, cmis.CmisApi)
+        return isinstance(xcvr_api, cmis_api.CmisApi)
 
     def is_sff_api(self, xcvr_api):
         """Check if the api type is SFF
@@ -1811,3 +1816,36 @@ def get_module_status(self):
         """
         status = super().get_module_status()
         return SFP_STATUS_REMOVED if status == SFP_STATUS_UNKNOWN else status
+
+
+class CpoPort(SFP):
+    """class derived from SFP, representing CPO ports"""
+
+    def __init__(self, sfp_index):
+        super(CpoPort, self).__init__(sfp_index)
+        self._sfp_type_str = None
+        self.sfp_type = CPO_TYPE
+
+    def get_transceiver_info(self):
+        transceiver_info_dict = super().get_transceiver_info()
+        transceiver_info_dict['type'] = self.sfp_type
+        return transceiver_info_dict
+
+    def get_xcvr_api(self):
+        if self._xcvr_api is None:
+            self._xcvr_api = self._xcvr_api_factory._create_api(cmis_codes.CmisCodes, cmis_mem.CmisMemMap, cmis_api.CmisApi)
+        return self._xcvr_api
+
+    def get_presence(self):
+        file_path = SFP_SDK_MODULE_SYSFS_ROOT_TEMPLATE.format(self.sdk_index) + SFP_SYSFS_PRESENT
+        present = utils.read_int_from_file(file_path)
+        return present == 1
+
+    def reinit(self):
+        """
+        Nothing to do for cpo. Just provide it to avoid exception
+        :return:
+        """
+        return
+
+

platform/mellanox/mlnx-platform-api/sonic_platform/utils.py

@@ -32,6 +32,7 @@
 PORT_INDEX_KEY = "index"
 PORT_TYPE_KEY = "port_type"
 RJ45_PORT_TYPE = "RJ45"
+CPO_PORT_TYPE = "CPO"
 
 logger = Logger()
 
@@ -239,19 +240,14 @@ def load_json_file(filename, log_func=logger.log_error):
         return None
 
 
-def extract_RJ45_ports_index():
-    # Cross check 'platform.json' and 'hwsku.json' to extract the RJ45 port index if exists.
-    hwsku_path = device_info.get_path_to_hwsku_dir()
-    hwsku_file = os.path.join(hwsku_path, HWSKU_JSON)
-    if not os.path.exists(hwsku_file):
-        # Platforms having no hwsku.json do not have RJ45 port
+def _extract_ports_index_by_type(port_type, num_of_asics=1):
+    platform_file = os.path.join(device_info.get_path_to_platform_dir(), device_info.PLATFORM_JSON_FILE)
+    if not os.path.exists(platform_file):
         return None
 
-    platform_file = device_info.get_path_to_port_config_file()
     platform_dict = load_json_file(platform_file)['interfaces']
-    hwsku_dict = load_json_file(hwsku_file)['interfaces']
     port_name_to_index_map_dict = {}
-    RJ45_port_index_list = []
+    port_index_list = []
 
     # Compose a interface name to index mapping from 'platform.json'
     for i, (key, value) in enumerate(platform_dict.items()):
@@ -264,12 +260,37 @@ def extract_RJ45_ports_index():
     if not bool(port_name_to_index_map_dict):
         return None
 
-    # Check if "port_type" specified as "RJ45", if yes, add the port index to the list.
+    hwsku_jsons = get_path_list_to_asic_hwsku_dir(num_of_asics)
+    hwsku_dict = {}
+    for hwsku_json in hwsku_jsons:
+        hwsku_dict.update(load_json_file(hwsku_json)['interfaces'])
+
+    # Check if "port_type" matches, if yes, add the port index to the list.
     for i, (key, value) in enumerate(hwsku_dict.items()):
-        if key in port_name_to_index_map_dict and PORT_TYPE_KEY in value and value[PORT_TYPE_KEY] == RJ45_PORT_TYPE:
-            RJ45_port_index_list.append(int(port_name_to_index_map_dict[key])-1)
+        if key in port_name_to_index_map_dict and PORT_TYPE_KEY in value and value[PORT_TYPE_KEY] == port_type:
+            port_index_list.append(int(port_name_to_index_map_dict[key]) - 1)
+
+    # Remove duplicates
+    port_index_list = list(dict.fromkeys(port_index_list))
+
+    return port_index_list if port_index_list else None
+
+
+def get_path_list_to_asic_hwsku_dir(num_of_asics):
+    platform_path = device_info.get_path_to_platform_dir()
+    hwsku = device_info.get_hwsku()
+    if num_of_asics == 1:
+        return [os.path.join(platform_path, hwsku, HWSKU_JSON)]
+    else:
+        return [os.path.join(platform_path, hwsku, str(asic_id), HWSKU_JSON) for asic_id in range(num_of_asics)]
+
+
+def extract_RJ45_ports_index(num_of_asics=1):
+    return _extract_ports_index_by_type(RJ45_PORT_TYPE, num_of_asics)
+
 
-    return RJ45_port_index_list if bool(RJ45_port_index_list) else None
+def extract_cpo_ports_index(num_of_asics=1):
+    return _extract_ports_index_by_type(CPO_PORT_TYPE, num_of_asics)
 
 
 def wait_until(predict, timeout, interval=1, *args, **kwargs):

platform/mellanox/mlnx-platform-api/tests/test_change_event.py

@@ -39,6 +39,7 @@ class TestChangeEvent:
     @mock.patch('sonic_platform.device_data.DeviceDataManager.is_module_host_management_mode', mock.MagicMock(return_value=False))
     @mock.patch('sonic_platform.device_data.DeviceDataManager.get_sfp_count', mock.MagicMock(return_value=1))
     @mock.patch('sonic_platform.chassis.extract_RJ45_ports_index', mock.MagicMock(return_value=[]))
+    @mock.patch('sonic_platform.chassis.extract_cpo_ports_index', mock.MagicMock(return_value=[]))
     @mock.patch('sonic_platform.sfp.SFP.get_module_status')
     def test_get_change_event_legacy(self, mock_status, mock_time, mock_create_poll, mock_get_fd):
         c = chassis.Chassis()
@@ -92,6 +93,7 @@ def test_get_change_event_legacy(self, mock_status, mock_time, mock_create_poll,
     @mock.patch('sonic_platform.device_data.DeviceDataManager.is_module_host_management_mode', mock.MagicMock(return_value=True))
     @mock.patch('sonic_platform.device_data.DeviceDataManager.get_sfp_count', mock.MagicMock(return_value=1))
     @mock.patch('sonic_platform.chassis.extract_RJ45_ports_index', mock.MagicMock(return_value=[]))
+    @mock.patch('sonic_platform.chassis.extract_cpo_ports_index', mock.MagicMock(return_value=[]))
     @mock.patch('sonic_platform.module_host_mgmt_initializer.ModuleHostMgmtInitializer.initialize', mock.MagicMock())
     def test_get_change_event_for_module_host_management_mode(self, mock_time, mock_create_poll, mock_get_fd, mock_ready):
         """Test steps:

platform/mellanox/mlnx-platform-api/tests/test_chassis.py

@@ -1,6 +1,6 @@
 #
 # SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
-# Copyright (c) 2021-2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# Copyright (c) 2021-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -39,6 +39,7 @@
 from sonic_platform.device_data import DeviceDataManager
 
 sonic_platform.chassis.extract_RJ45_ports_index = mock.MagicMock(return_value=[])
+sonic_platform.chassis.extract_cpo_ports_index = mock.MagicMock(return_value=[])
 
 class TestChassis:
     """Test class to test chassis.py. The test cases covers:
@@ -179,6 +180,13 @@ def test_sfp(self):
         assert chassis.get_num_sfps() == 6
         sonic_platform.chassis.extract_RJ45_ports_index = mock.MagicMock(return_value=[])
 
+        # Get all SFPs, with CPO ports
+        sonic_platform.chassis.extract_cpo_ports_index = mock.MagicMock(return_value=[3, 4])
+        DeviceDataManager.get_sfp_count = mock.MagicMock(return_value=3)
+        chassis = Chassis()
+        assert chassis.get_num_sfps() == 5
+        sonic_platform.chassis.extract_cpo_ports_index = mock.MagicMock(return_value=[])
+
     @mock.patch('sonic_platform.device_data.DeviceDataManager.is_module_host_management_mode', mock.MagicMock(return_value=False))
     def test_create_sfp_in_multi_thread(self):
         DeviceDataManager.get_sfp_count = mock.MagicMock(return_value=3)

platform/mellanox/mlnx-platform-api/tests/test_module_initializer.py

@@ -1,5 +1,6 @@
 #
-# Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
+# Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -59,9 +60,9 @@ def test_wait_module_ready(self, mock_is_host, mock_wait, mock_exists):
         mock_exists.return_value = False
         initializer.wait_module_ready()
         assert not initializer.initialized
-
 
     @mock.patch('sonic_platform.chassis.extract_RJ45_ports_index', mock.MagicMock(return_value=[]))
+    @mock.patch('sonic_platform.chassis.extract_cpo_ports_index', mock.MagicMock(return_value=[]))
     @mock.patch('sonic_platform.device_data.DeviceDataManager.get_sfp_count', mock.MagicMock(return_value=1))
     @mock.patch('sonic_platform.sfp.SFP.initialize_sfp_modules', mock.MagicMock())
     @mock.patch('sonic_platform.module_host_mgmt_initializer.ModuleHostMgmtInitializer.is_initialization_owner')

platform/mellanox/mlnx-platform-api/tests/test_sfp.py

@@ -29,7 +29,7 @@
 modules_path = os.path.dirname(test_path)
 sys.path.insert(0, modules_path)
 
-from sonic_platform.sfp import SFP, RJ45Port, SX_PORT_MODULE_STATUS_INITIALIZING, SX_PORT_MODULE_STATUS_PLUGGED, SX_PORT_MODULE_STATUS_UNPLUGGED, SX_PORT_MODULE_STATUS_PLUGGED_WITH_ERROR, SX_PORT_MODULE_STATUS_PLUGGED_DISABLED
+from sonic_platform.sfp import SFP, RJ45Port, CpoPort, CPO_TYPE, cmis_api, SX_PORT_MODULE_STATUS_INITIALIZING, SX_PORT_MODULE_STATUS_PLUGGED, SX_PORT_MODULE_STATUS_UNPLUGGED, SX_PORT_MODULE_STATUS_PLUGGED_WITH_ERROR, SX_PORT_MODULE_STATUS_PLUGGED_DISABLED
 from sonic_platform.chassis import Chassis
 
 
@@ -319,6 +319,18 @@ def test_rj45_basic(self):
         assert sfp.get_transceiver_threshold_info()
         sfp.reinit()
 
+    @mock.patch('sonic_platform.sfp.CpoPort.read_eeprom')
+    def test_cpo_get_xcvr_api(self, mock_read):
+        sfp = CpoPort(0)
+        api = sfp.get_xcvr_api()
+        assert isinstance(api, cmis_api.CmisApi)
+
+    @mock.patch('sonic_platform.sfp.SfpOptoeBase.get_transceiver_info', return_value={})
+    def test_cpo_get_transceiver_info(self, mock_get_info):
+        sfp = CpoPort(0)
+        info = sfp.get_transceiver_info()
+        assert info['type'] == CPO_TYPE
+
     @mock.patch('os.path.exists')
     @mock.patch('sonic_platform.utils.read_int_from_file')
     def test_get_temperature(self, mock_read, mock_exists):
@@ -507,6 +519,7 @@ def test_get_error_info_from_sdk_error_type(self, mock_read):
         assert error_desc is None
 
     @mock.patch('sonic_platform.chassis.extract_RJ45_ports_index', mock.MagicMock(return_value=[]))
+    @mock.patch('sonic_platform.chassis.extract_cpo_ports_index', mock.MagicMock(return_value=[]))
     @mock.patch('sonic_platform.device_data.DeviceDataManager.get_sfp_count', mock.MagicMock(return_value=1))
     def test_initialize_sfp_modules(self):
         c = Chassis()