diff --git a/src/sonic-py-common/sonic_py_common/device_info.py b/src/sonic-py-common/sonic_py_common/device_info.py index 8709941afcf..4a75720c3dc 100644 --- a/src/sonic-py-common/sonic_py_common/device_info.py +++ b/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 diff --git a/src/sonic-py-common/tests/device_info_test.py b/src/sonic-py-common/tests/device_info_test.py index 5362da49631..c717c6ba0e9 100644 --- a/src/sonic-py-common/tests/device_info_test.py +++ b/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):