Add support for custom serdes attributes

sonic-xcvrd/tests/test_xcvrd.py

@@ -1501,6 +1501,62 @@ def _check_notify_media_setting(self, index, expected_found=False, expected_valu
         assert found == expected_found
         assert result_dict == expected_value
 
+    @pytest.mark.parametrize("media_dict, lane_count, subport_num, expected", [
+        (
+            {
+                'CUSTOM:XYZ': {'lane0': '0xa', 'lane1': '0xb', 'lane2': '0xc', 'lane3': '0xd'},
+                'CUSTOM:ABC': {'lane0': '0x1', 'lane1': '0x2', 'lane2': '0x3', 'lane3': '0x4'},
+                'main':       {'lane0': '0x11', 'lane1': '0x12', 'lane2': '0x13', 'lane3': '0x14'},
+            },
+            2, 2,
+            {
+                'custom_serdes_attrs': '{"attributes":[{"XYZ":{"value":[12,13]}},{"ABC":{"value":[3,4]}}]}',
+                'main':                {'lane0': '0x11', 'lane1': '0x12', 'lane2': '0x13', 'lane3': '0x14'},
+            },
+        ),
+        (
+            {
+                'CUSTOM:XYZ': {'lane0': '10', 'lane1': '11', 'lane2': '12', 'lane3': '13'},
+                'CUSTOM:ABC': {'lane0': '1', 'lane1': '2', 'lane2': '3', 'lane3': '4'},
+            },
+            2, 2,
+            {
+                'custom_serdes_attrs': '{"attributes":[{"XYZ":{"value":[12,13]}},{"ABC":{"value":[3,4]}}]}',
+            },
+        ),
+        (
+            {
+                'CUSTOM:XYZ': {'lane0': 10, 'lane1': 11, 'lane2': 12, 'lane3': 13},
+                'CUSTOM:ABC': {'lane0': 1, 'lane1': 2, 'lane2': 3, 'lane3': 4},
+            },
+            2, 2,
+            {
+                'custom_serdes_attrs': '{"attributes":[{"XYZ":{"value":[12,13]}},{"ABC":{"value":[3,4]}}]}',
+            },
+        ),
+        (
+            {
+                'CUSTOM:XYZ': {'lane0': 10, 'lane1': 11, 'lane2': 17592186044415, 'lane3': 'x11'},
+            },
+            2, 2,
+            {},
+        ),
+        (
+            {
+                'CUSTOM:XYZ': {'lane0': '0x10', 'lane1': '0x11', 'lane2': '0xFFFFFFFFFFF', 'lane3': '0x13x'},
+            },
+            2, 2,
+            {},
+        ),
+        (
+            {'main': {'lane0': '0x11', 'lane1': '0x12', 'lane2': '0x13', 'lane3': '0x14'}},
+            2, 2,
+            {'main': {'lane0': '0x11', 'lane1': '0x12', 'lane2': '0x13', 'lane3': '0x14'}},
+        ),
+    ])
+    def test_handle_custom_serdes_attrs(self, media_dict, lane_count, subport_num, expected):
+        assert expected == media_settings_parser.handle_custom_serdes_attrs(media_dict, lane_count, subport_num)
+
     @patch('xcvrd.xcvrd_utilities.optics_si_parser.g_optics_si_dict', optics_si_settings_dict)
     @patch('xcvrd.xcvrd._wrapper_get_presence', MagicMock(return_value=True))
     def test_fetch_optics_si_setting(self):

sonic-xcvrd/xcvrd/xcvrd_utilities/media_settings_parser.py

@@ -7,6 +7,7 @@
 import ast
 import re
 from natsort import natsorted
+from copy import deepcopy
 
 from sonic_py_common import device_info, syslogger
 from swsscommon import swsscommon
@@ -23,6 +24,10 @@
 DEFAULT_KEY = 'Default'
 # This is useful if default value is desired when no match is found for lane speed key
 LANE_SPEED_DEFAULT_KEY = LANE_SPEED_KEY_PREFIX + DEFAULT_KEY
+CUSTOM_SERDES_ATTR_PREFIX = 'CUSTOM:'
+CUSTOM_SERDES_ATTRS_TOP_LEVEL_KEY = 'attributes'
+CUSTOM_SERDES_ATTRS_KEY_IN_DB = 'custom_serdes_attrs'
+
 SYSLOG_IDENTIFIER = "xcvrd"
 helper_logger = syslogger.SysLogger(SYSLOG_IDENTIFIER, enable_runtime_config=True)
 
@@ -163,7 +168,7 @@ def is_si_per_speed_supported(media_dict):
     return LANE_SPEED_KEY_PREFIX in list(media_dict.keys())[0]
 
 
-def get_serdes_si_setting_val_str(val_dict, lane_count, subport_num=0):
+def get_serdes_si_setting_val(val_dict, lane_count, subport_num=0):
     """
     Get ASIC side SerDes SI settings for the given logical port (subport)
 
@@ -174,8 +179,8 @@ def get_serdes_si_setting_val_str(val_dict, lane_count, subport_num=0):
         subport_num: subport number (1-based), 0 for non-breakout case
 
     Returns:
-        string containing SerDes settings for the given subport, separated by comma
-        e.g. '0x1f,0x1f,0x1f,0x1f'
+        list containing SerDes settings for the given subport
+        e.g. ['0x1f', '0x1f', '0x1f', '0x1f']
     """
     start_lane_idx = (subport_num - 1) * lane_count if subport_num else 0
     if start_lane_idx + lane_count > len(val_dict):
@@ -186,7 +191,24 @@ def get_serdes_si_setting_val_str(val_dict, lane_count, subport_num=0):
         start_lane_idx = 0
     val_list = [val_dict[lane_key] for lane_key in natsorted(val_dict)]
     # If subport_num ('subport') is not specified in config_db, return values for first lane_count number of lanes
-    return ','.join(val_list[start_lane_idx:start_lane_idx + lane_count])
+    return val_list[start_lane_idx:start_lane_idx + lane_count]
+
+
+def get_serdes_si_setting_val_str(val_dict, lane_count, subport_num=0):
+    """
+    Get ASIC side SerDes SI settings string for the given logical port (subport)
+
+    Args:
+        val_dict: dictionary containing SerDes settings for all lanes of the port
+                  e.g. {'lane0': '0x1f', 'lane1': '0x1f', 'lane2': '0x1f', 'lane3': '0x1f'}
+        lane_count: number of lanes for this subport
+        subport_num: subport number (1-based), 0 for non-breakout case
+
+    Returns:
+        string containing SerDes settings for the given subport, separated by comma
+        e.g. '0x1f,0x1f,0x1f,0x1f'
+    """
+    return ','.join(get_serdes_si_setting_val(val_dict, lane_count, subport_num))
 
 
 def get_media_settings_for_speed(settings_dict, lane_speed_key):
@@ -315,6 +337,97 @@ def get_speed_lane_count_and_subport(port, cfg_port_tbl):
     return port_speed, lane_count, subport_num
 
 
+def convert_to_int32(value):
+    """
+    Convert value to a signed 32-bit integer.
+
+    Args:
+        value: hex string (starting with '0x') or decimal string or integer
+    Returns:
+        signed 32-bit integer value, or None if the value is out of range
+    """
+    # hex string:
+    if isinstance(value, str) and value.startswith('0x'):
+        try:
+            int_value = int(value, 16)  # unsigned value
+        except ValueError:
+            helper_logger.log_error("Invalid hex string value {}".format(value))
+            return None
+        if int_value < 0 or int_value > (1 << 32) - 1:
+            helper_logger.log_error("Hex string value {} out of 32 bits range".format(value))
+            return None
+        # if sign bit set, subtract 2**32 to get negative
+        # e.g. 0xFFFFFFFF -> -1, 0xFFFFFFFE -> -2, etc.
+        return int_value - (1 << 32) if (int_value & (1 << 31)) else int_value
+
+    # decimal string or integer:
+    try:
+        # int() can handle both decimal string representations and integer types.
+        int_value = int(value)
+    except ValueError:
+        helper_logger.log_error(
+            "Input value {} must be a hex string (starting with '0x'), decimal string, or integer".format(value)
+        )
+        return None
+
+    if int_value < -(1 << 31) or int_value > (1 << 31) - 1:
+        helper_logger.log_error("Integer value {} out of signed 32-bit integer range".format(int_value))
+        return None
+
+    return int_value
+
+
+def handle_custom_serdes_attrs(media_dict, lane_count, subport_num):
+    """
+    Handle custom SerDes attributes in the media_dict and convert them to JSON.
+
+    Args:
+        media_dict: dictionary containing SerDes settings for all lanes of the port
+        lane_count: number of lanes for this subport
+        subport_num: subport number (1-based), 0 for non-breakout case
+
+    Returns:
+        media_dict: updated media_dict with custom SerDes attributes converted to JSON
+    """
+    # Perform a deepcopy to avoid modifying the original media_dict, which may be a reference to the global 'g_dict'.
+    # Modifying g_dict would affect subsequent lookups and cause incorrect behavior.
+    media_dict = deepcopy(media_dict)
+    attrs_list = []
+
+    for key, value in list(media_dict.items()):
+        if not key.startswith(CUSTOM_SERDES_ATTR_PREFIX):
+            continue
+
+        custom_serdes_attr = key[len(CUSTOM_SERDES_ATTR_PREFIX):]
+        value_list = [convert_to_int32(lane_value)
+                      for lane_value in get_serdes_si_setting_val(value, lane_count, subport_num)]
+        if None in value_list:
+            helper_logger.log_error("Skipping custom serdes attr {} due to invalid integer value".format(custom_serdes_attr))
+            media_dict.pop(key)
+            continue
+
+        attr_dict = {
+            custom_serdes_attr: {
+                'value': value_list
+            }
+        }
+        attrs_list.append(attr_dict)
+
+        # Remove the key from media_dict to avoid duplication
+        media_dict.pop(key)
+
+    if not attrs_list:
+        return media_dict
+
+    # Combine all the custom serdes attributes to a single element,
+    # and put it back into the media_dict to be published in APP DB
+    media_dict[CUSTOM_SERDES_ATTRS_KEY_IN_DB] = json.dumps(
+        {CUSTOM_SERDES_ATTRS_TOP_LEVEL_KEY: attrs_list},
+        separators=(',', ':')  # remove whitespace for optimal payload
+    )
+    return media_dict
+
+
 def notify_media_setting(logical_port_name, transceiver_dict,
                          xcvr_table_helper, port_mapping):
 
@@ -366,6 +479,8 @@ def notify_media_setting(logical_port_name, transceiver_dict,
             helper_logger.log_info("Error in obtaining media setting for {}".format(logical_port_name))
             return
 
+        media_dict = handle_custom_serdes_attrs(media_dict, lane_count, subport_num)
+
         fvs = swsscommon.FieldValuePairs(len(media_dict))
 
         index = 0