[CPO] Add library function to parse cpo.json

src/sonic-py-common/sonic_py_common/device_info.py

@@ -6,6 +6,7 @@
 import re
 import subprocess
 import yaml
+from typing import List, Optional
 from natsort import natsorted
 from sonic_py_common.general import getstatusoutput_noshell_pipe
 from swsscommon.swsscommon import ConfigDBConnector, SonicV2Connector
@@ -21,6 +22,10 @@
 # Port configuration file names
 PORT_CONFIG_FILE = "port_config.ini"
 PLATFORM_JSON_FILE = "platform.json"
+
+# CPO configuration file name
+CPO_FILE = "cpo.json"
+
 BMC_BUILD_CONFIG_FILE = '/etc/sonic/bmc_config.json'
 GLOBAL_BMC_DATA_FILE = '/etc/sonic/bmc.json'
 
@@ -201,6 +206,123 @@ def get_platform_json_data():
         return None
 
 
+def get_cpo_data() -> Optional[dict]:
+    """
+    Retrieve the data from the cpo.json file.
+
+    Locates the file using a two-stage lookup: a hwsku-specific file takes
+    precedence over a platform-wide file. Lane fields are normalized from
+    comma-separated strings ("41,42") into lists of ints ([41, 42]); all
+    other fields, including vendor-specific ones, are returned verbatim.
+    None is returned if the file does not exist or cannot be parsed.
+    """
+    if not get_platform():
+        return None
+
+    cpo_file = _find_cpo_file()
+    if not cpo_file:
+        return None
+
+    try:
+        with open(cpo_file, 'r') as f:
+            cpo_data = json.loads(f.read())
+    except (json.JSONDecodeError, IOError, TypeError, ValueError):
+        # Handle any file reading and JSON parsing errors
+        return None
+
+    _normalize_cpo_data(cpo_data)
+    return cpo_data
+
+
+def _find_cpo_file() -> Optional[str]:
+    """
+    Locate cpo.json, preferring the hwsku directory over the
+    platform directory.
+    Returns the path to the first cpo.json found, or None.
+    """
+    try:
+        hwsku_dir = get_path_to_hwsku_dir()
+    except (OSError, TypeError):
+        hwsku_dir = None
+
+    if hwsku_dir:
+        hwsku_file = os.path.join(hwsku_dir, CPO_FILE)
+        if os.path.isfile(hwsku_file):
+            return hwsku_file
+
+    try:
+        platform_dir = get_path_to_platform_dir()
+    except OSError:
+        platform_dir = None
+
+    if platform_dir:
+        platform_file = os.path.join(platform_dir, CPO_FILE)
+        if os.path.isfile(platform_file):
+            return platform_file
+
+    return None
+
+
+def _parse_lane_string(lane_string: str) -> List[int]:
+    """'41,42,43' -> [41, 42, 43]; tolerates spaces and a trailing comma."""
+    return [int(tok) for tok in lane_string.split(',') if tok.strip() != '']
+
+
+def _normalize_cpo_data(cpo_data: dict) -> None:
+    """
+    In-place normalization of the known lane fields from comma-separated
+    strings to lists of ints. All other fields (vendor-specific included) are
+    left untouched.
+
+    Example input:
+        {
+            "devices": {
+                "OE1": {
+                    "device_type": "optical_engine",
+                    "asic_lanes": "41,42,43,44",
+                    "i2c_path": "/sys/bus/i2c/devices/32-0050"
+                },
+                "ELS1": {
+                    "device_type": "external_laser_source",
+                    "laser_to_asic_lane_mapping": {
+                        "1": "41,42",
+                        "2": "43,44"
+                    }
+                }
+            }
+        }
+
+    After _normalize_cpo_data(...) the same dict becomes:
+        {
+            "devices": {
+                "OE1": {
+                    "device_type": "optical_engine",
+                    "asic_lanes": [41, 42, 43, 44],
+                    "i2c_path": "/sys/bus/i2c/devices/32-0050"
+                },
+                "ELS1": {
+                    "device_type": "external_laser_source",
+                    "laser_to_asic_lane_mapping": {
+                        1: [41, 42],
+                        2: [43, 44]
+                    }
+                }
+            }
+        }
+    """
+    for device in cpo_data.get('devices', {}).values():
+        device_type = device['device_type']
+        if device_type == 'optical_engine':
+            device['asic_lanes'] = _parse_lane_string(device['asic_lanes'])
+        elif device_type == 'external_laser_source':
+            device['laser_to_asic_lane_mapping'] = {
+                int(laser): _parse_lane_string(lanes)
+                for laser, lanes in device['laser_to_asic_lane_mapping'].items()
+            }
+        else:
+            raise ValueError(f'Unrecognized device_type: {device_type}')
+
+
 def get_asic_conf_file_path():
     """
     Retrieves the path to the ASIC configuration file on the device

src/sonic-py-common/tests/device_info_test.py

@@ -255,6 +255,107 @@ def test_get_platform_json_data(self, mock_get_platform, mock_get_path_to_platfo
         result = device_info.get_platform_json_data()
         assert result is None
 
+    @mock.patch("os.path.isfile")
+    @mock.patch("{}.open".format(BUILTINS))
+    @mock.patch("sonic_py_common.device_info.get_path_to_platform_dir")
+    @mock.patch("sonic_py_common.device_info.get_path_to_hwsku_dir")
+    @mock.patch("sonic_py_common.device_info.get_platform")
+    def test_get_cpo_data(self, mock_get_platform, mock_get_hwsku_dir, mock_get_platform_dir, mock_open, mock_isfile):
+        mock_get_platform.return_value = "x86_64-vendor_cpo-r0"
+        mock_get_hwsku_dir.return_value = "/usr/share/sonic/device/x86_64-vendor_cpo-r0/CPO-HWSKU"
+        mock_get_platform_dir.return_value = "/usr/share/sonic/device/x86_64-vendor_cpo-r0"
+
+        cpo_data = {
+            "devices": {
+                "OE1": {
+                    "device_type": "optical_engine",
+                    "max_banks": 2,
+                    "asic_lanes": "41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56",
+                    "i2c_path": "/sys/bus/i2c/devices/32-0050"
+                },
+                "ELS1": {
+                    "device_type": "external_laser_source",
+                    "lasers": 4,
+                    "max_banks": 1,
+                    "laser_to_asic_lane_mapping": {
+                        "1": "41,42,43,44",
+                        "2": "45,46,47,48",
+                        "3": "49,50,51,52",
+                        "4": "53,54,55,56"
+                    },
+                    # example vendor-specific field; must pass through verbatim.
+                    "elsfp_sysfs_path": "/sys/bus/i2c/devices/33-0051"
+                }
+            },
+            "interfaces": {
+                "Ethernet0": {
+                    "associated_devices": [
+                        {"device_id": "OE1", "bank": 0},
+                        {"device_id": "ELS1", "bank": 0}
+                    ]
+                },
+                "Ethernet8": {
+                    "associated_devices": [
+                        {"device_id": "OE1", "bank": 1},
+                        {"device_id": "ELS1", "bank": 0}
+                    ]
+                }
+            }
+        }
+
+        # Happy path: lane strings normalized, vendor field untouched.
+        mock_isfile.return_value = True
+        open_mocked = mock.mock_open(read_data=json.dumps(cpo_data))
+        mock_open.side_effect = open_mocked
+        result = device_info.get_cpo_data()
+        assert result["devices"]["OE1"]["asic_lanes"] == [41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56]
+        assert result["devices"]["ELS1"]["laser_to_asic_lane_mapping"] == {
+            1: [41, 42, 43, 44],
+            2: [45, 46, 47, 48],
+            3: [49, 50, 51, 52],
+            4: [53, 54, 55, 56],
+        }
+        assert result["devices"]["ELS1"]["elsfp_sysfs_path"] == "/sys/bus/i2c/devices/33-0051"
+        assert result["interfaces"]["Ethernet0"]["associated_devices"] == [
+            {"device_id": "OE1", "bank": 0},
+            {"device_id": "ELS1", "bank": 0},
+        ]
+        assert result["interfaces"]["Ethernet8"]["associated_devices"] == [
+            {"device_id": "OE1", "bank": 1},
+            {"device_id": "ELS1", "bank": 0},
+        ]
+
+        # hwsku file takes precedence over the platform file.
+        mock_open.side_effect = mock.mock_open(read_data=json.dumps(cpo_data))
+        device_info.get_cpo_data()
+        opened_path = mock_open.call_args[0][0]
+        assert opened_path == "/usr/share/sonic/device/x86_64-vendor_cpo-r0/CPO-HWSKU/cpo.json"
+
+        # Falls back to the platform file when no hwsku file exists.
+        def only_platform_file(path):
+            return path == "/usr/share/sonic/device/x86_64-vendor_cpo-r0/cpo.json"
+        mock_isfile.side_effect = only_platform_file
+        mock_open.side_effect = mock.mock_open(read_data=json.dumps(cpo_data))
+        device_info.get_cpo_data()
+        opened_path = mock_open.call_args[0][0]
+        assert opened_path == "/usr/share/sonic/device/x86_64-vendor_cpo-r0/cpo.json"
+
+        # Returns None when no file exists in either directory.
+        mock_isfile.side_effect = None
+        mock_isfile.return_value = False
+        assert device_info.get_cpo_data() is None
+
+        # Returns None when platform is not set.
+        mock_isfile.return_value = True
+        mock_open.side_effect = mock.mock_open(read_data=json.dumps(cpo_data))
+        mock_get_platform.return_value = None
+        assert device_info.get_cpo_data() is None
+
+        # Returns None when the JSON is invalid.
+        mock_get_platform.return_value = "x86_64-vendor_cpo-r0"
+        mock_open.side_effect = mock.mock_open(read_data="invalid json")
+        assert device_info.get_cpo_data() is None
+
     @mock.patch("sonic_py_common.device_info.get_platform_json_data")
     @mock.patch("sonic_py_common.device_info.get_platform")
     def test_is_smartswitch(self, mock_get_platform, mock_get_platform_json_data):