Add support for custom serdes attributes

sonic-xcvrd/tests/test_xcvrd.py

@@ -1501,6 +1501,28 @@ 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'},
+            },
+        ),
+        (
+            {'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)
 
@@ -315,6 +320,78 @@ def get_speed_lane_count_and_subport(port, cfg_port_tbl):
     return port_speed, lane_count, subport_num
 
 
+def hex_str_to_int32(hex_str):
+    """
+    Parse hex_str (e.g. '0xFFEEAABB') as a signed 32-bit integer.
+    If the hex_str is negative, convert it to a signed integer.
+
+    Args:
+        hex_str: hex string to be converted to signed integer
+    Returns:
+        signed integer value
+    """
+    u = int(hex_str, 16)            # unsigned value
+    # if sign bit set, subtract 2**32 to get negative
+    return u - (1 << 32) if (u & (1 << 31)) else u
+
+
+def str_to_int(int_str):
+    """
+    Convert a string representation of an integer to a signed integer.
+
+    Args:
+        int_str: string representation of an integer
+    Returns:
+        signed integer value
+    """
+    return hex_str_to_int32(int_str) if int_str.startswith('0x') else int(int_str)
+
+
+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
+    """
+    # Do deepcopy to avoid modifying the original media_dict
+    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 = [str_to_int(lane_value_str)
+                      for lane_value_str in get_serdes_si_setting_val_str(value, lane_count, subport_num).split(',')]
+        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 +443,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