Apply custom Si settings via CMIS: SONiC xcvrd platform common changes

sonic_platform_base/sonic_xcvr/api/public/cmis.py

@@ -31,6 +31,12 @@ def __init__(self, xcvr_eeprom):
         self.vdm = CmisVdmApi(xcvr_eeprom) if not self.is_flat_memory() else None
         self.cdb = CmisCdbApi(xcvr_eeprom) if not self.is_flat_memory() else None
 
+    def get_manufacturer(self):
+        '''
+        This function returns the manufacturer of the module
+        '''
+        return self.xcvr_eeprom.read(consts.VENDOR_NAME_FIELD)
+
     def get_model(self):
         '''
         This function returns the part number of the module
@@ -2069,7 +2075,7 @@ def get_application(self, lane):
 
         return (appl & 0xf)
 
-    def set_application(self, channel, appl_code):
+    def set_application(self, channel, appl_code, ec=0):
         """
         Update the selected application code to the specified lanes on the host side
 
@@ -2092,11 +2098,338 @@ def set_application(self, channel, appl_code):
                 lane_first = lane
             addr = "{}_{}_{}".format(consts.STAGED_CTRL_APSEL_FIELD, 0, lane + 1)
             data = (appl_code << 4) | (lane_first << 1)
+            #set EC bit
+            data|= ec
             self.xcvr_eeprom.write(addr, data)
 
-        # Apply DataPathInit
+    def stage_datapath_init(self, channel):
+        '''
+        This function applies DataPathInit
+        '''
         return self.xcvr_eeprom.write("%s_%d" % (consts.STAGED_CTRL_APPLY_DPINIT_FIELD, 0), channel)
 
+    def get_rx_output_amp_max_val(self):
+        '''
+        This function returns the supported RX output amp val
+        '''
+        rx_amp_max_val = self.xcvr_eeprom.read(consts.RX_OUTPUT_LEVEL_SUPPORT)
+        if rx_amp_max_val is None:
+            return None
+        return rx_amp_max_val
+
+    def get_rx_output_eq_pre_max_val(self):
+        '''
+        This function returns the supported RX output eq pre cursor val
+        '''
+        rx_pre_max_val = self.xcvr_eeprom.read(consts.RX_OUTPUT_EQ_PRE_CURSOR_MAX)
+        if rx_pre_max_val is None:
+            return None
+        return rx_pre_max_val
+
+    def get_rx_output_eq_post_max_val(self):
+        '''
+        This function returns the supported RX output eq post cursor val
+        '''
+        rx_post_max_val = self.xcvr_eeprom.read(consts.RX_OUTPUT_EQ_POST_CURSOR_MAX)
+        if rx_post_max_val is None:
+            return None
+        return rx_post_max_val
+
+    def get_tx_input_eq_max_val(self):
+        '''
+        This function returns the supported TX input eq val
+        '''
+        tx_input_max_val = self.xcvr_eeprom.read(consts.TX_INPUT_EQ_MAX)
+        if tx_input_max_val is None:
+            return None
+        return tx_input_max_val
+
+    def get_tx_cdr_supported(self):
+        '''
+        This function returns the supported TX CDR field
+        '''
+        tx_cdr_support = self.xcvr_eeprom.read(consts.TX_CDR_SUPPORT_FIELD)
+        if not tx_cdr_support or tx_cdr_support is None:
+            return False
+        return tx_cdr_support
+
+    def get_rx_cdr_supported(self):
+        '''
+        This function returns the supported RX CDR field
+        '''
+        rx_cdr_support = self.xcvr_eeprom.read(consts.RX_CDR_SUPPORT_FIELD)
+        if not rx_cdr_support or rx_cdr_support is None:
+            return False
+        return rx_cdr_support
+
+    def get_tx_input_eq_fixed_supported(self):
+        '''
+        This function returns the supported TX input eq field
+        '''
+        tx_fixed_support = self.xcvr_eeprom.read(consts.TX_INPUT_EQ_FIXED_MANUAL_CTRL_SUPPORT_FIELD)
+        if not tx_fixed_support or tx_fixed_support is None:
+            return False
+        return tx_fixed_support
+
+    def get_tx_input_adaptive_eq_supported(self):
+        '''
+        This function returns the supported TX input adaptive eq field
+        '''
+        tx_adaptive_support = self.xcvr_eeprom.read(consts.TX_INPUT_ADAPTIVE_EQ_SUPPORT_FIELD)
+        if not tx_adaptive_support or tx_adaptive_support is None:
+            return False
+        return tx_adaptive_support
+
+    def get_tx_input_recall_buf1_supported(self):
+        '''
+        This function returns the supported TX input recall buf1 field
+        '''
+        tx_recall_buf1_support = self.xcvr_eeprom.read(consts.TX_INPUT_EQ_RECALL_BUF1_SUPPORT_FIELD)
+        if not tx_recall_buf1_support or tx_recall_buf1_support is None:
+            return False
+        return tx_recall_buf1_support
+
+    def get_tx_input_recall_buf2_supported(self):
+        '''
+        This function returns the supported TX input recall buf2 field
+        '''
+        tx_recall_buf2_support = self.xcvr_eeprom.read(consts.TX_INPUT_EQ_RECALL_BUF2_SUPPORT_FIELD)
+        if not tx_recall_buf2_support or tx_recall_buf2_support is None:
+            return False
+        return tx_recall_buf2_support
+
+    def get_rx_ouput_amp_ctrl_supported(self):
+        '''
+        This function returns the supported RX output amp control field
+        '''
+        rx_amp_support = self.xcvr_eeprom.read(consts.RX_OUTPUT_AMP_CTRL_SUPPORT_FIELD)
+        if not rx_amp_support or rx_amp_support is None:
+            return False
+        return rx_amp_support
+
+    def get_rx_output_eq_pre_ctrl_supported(self):
+        '''
+        This function returns the supported RX output eq pre control field
+        '''
+        rx_pre_support = self.xcvr_eeprom.read(consts.RX_OUTPUT_EQ_PRE_CTRL_SUPPORT_FIELD)
+        if not rx_pre_support or rx_pre_support is None:
+            return False
+        return rx_pre_support
+
+    def get_rx_output_eq_post_ctrl_supported(self):
+        '''
+        This function returns the supported RX output eq post control field
+        '''
+        rx_post_support = self.xcvr_eeprom.read(consts.RX_OUTPUT_EQ_POST_CTRL_SUPPORT_FIELD)
+        if not rx_post_support or rx_post_support is None:
+            return False
+        return rx_post_support
+
+    def scs_nibble_read(self, si_param, lane):
+        si_key_lane = "{}{}".format(si_param, lane)
+        val = self.xcvr_eeprom.read(si_key_lane)
+        if val is None:
+            return None
+        return val
+
+    def scs_nibble_write(self, val, si_param, lane):
+        if (lane%2) == 0:
+            pre_val = self.scs_nibble_read(si_param, lane-1)
+            if pre_val is None:
+                return False
+            val = (val << 4) | pre_val
+        si_key_lane = "{}{}".format(si_param, lane)
+        return self.xcvr_eeprom.write(si_key_lane, val)
+
+    def stage_output_eq_pre_cursor_target_rx(self, host_lanes_mask, si_settings):
+        '''
+        This function applies RX output eq pre cursor settings
+        '''
+        for lane in range(self.NUM_CHANNELS):
+            if ((1 << lane) & host_lanes_mask) == 0:
+                continue
+            lane = lane+1
+            si_key_lane = "{}{}".format(consts.OUTPUT_EQ_PRE_CURSOR_TARGET_RX, lane)
+            val = si_settings[consts.OUTPUT_EQ_PRE_CURSOR_TARGET_RX][si_key_lane]
+            if val is None:
+                return False
+            rx_pre_max_val = self.get_rx_output_eq_pre_max_val()
+            if rx_pre_max_val is None or val > rx_pre_max_val:
+                return False
+            if not self.scs_nibble_write(val, consts.OUTPUT_EQ_PRE_CURSOR_TARGET_RX, lane):
+                return False
+        return True
+
+    def stage_output_eq_post_cursor_target_rx(self, host_lanes_mask, si_settings):
+        '''
+        This function applies RX output eq post cursor settings
+        '''
+        for lane in range(self.NUM_CHANNELS):
+            if ((1 << lane) & host_lanes_mask) == 0:
+                continue
+            lane = lane+1
+            si_key_lane = "{}{}".format(consts.OUTPUT_EQ_POST_CURSOR_TARGET_RX, lane)
+            val = si_settings[consts.OUTPUT_EQ_POST_CURSOR_TARGET_RX][si_key_lane]
+            if val is None:
+                return False
+            rx_post_max_val = self.get_rx_output_eq_post_max_val()
+            if rx_post_max_val is None or val > rx_post_max_val:
+                return False
+            if not self.scs_nibble_write(val, consts.OUTPUT_EQ_POST_CURSOR_TARGET_RX, lane):
+                return False
+        return True
+
+    def stage_output_amp_target_rx(self, host_lanes_mask, si_settings):
+        '''
+        This function applies RX output amp settings
+        '''
+        for lane in range(self.NUM_CHANNELS):
+            if ((1 << lane) & host_lanes_mask) == 0:
+                continue
+            lane = lane+1
+            si_key_lane = "{}{}".format(consts.OUTPUT_AMPLITUDE_TARGET_RX, lane)
+            val = si_settings[consts.OUTPUT_AMPLITUDE_TARGET_RX][si_key_lane]
+            if val is None:
+                return False
+            rx_amp_max_val = self.get_rx_output_amp_max_val()
+            if rx_amp_max_val is None or val > rx_amp_max_val:
+                return False
+            if not self.scs_nibble_write(val, consts.OUTPUT_AMPLITUDE_TARGET_RX, lane):
+                return False
+        return True
+
+    def stage_fixed_input_target_tx(self, host_lanes_mask, si_settings):
+        '''
+        This function applies fixed TX input si settings
+        '''
+        for lane in range(self.NUM_CHANNELS):
+            if ((1 << lane) & host_lanes_mask) == 0:
+                continue
+            lane = lane+1
+            si_key_lane = "{}{}".format(consts.FIXED_INPUT_EQ_TARGET_TX, lane)
+            val = si_settings[consts.FIXED_INPUT_EQ_TARGET_TX][si_key_lane]
+            if val is None:
+                return False
+            tx_fixed_input = self.get_tx_input_eq_max_val()
+            if tx_fixed_input is None or val > tx_fixed_input:
+                return False
+            if not self.scs_nibble_write(val, consts.FIXED_INPUT_EQ_TARGET_TX, lane):
+                return False
+        return True
+
+    def stage_adaptive_input_recall_tx(self, host_lanes_mask, si_settings):
+        '''
+        This function applies adaptive TX input recall si settings.
+        In this function, we set 2 bits at a time for each lane.
+        We mask the lane bits (ex: bit 0 and 1 for lane 1; bit 2 and 3 for lane 2 .. etc),
+        and 1 byte data is written
+        '''
+        val = 0
+        for lane in range(self.NUM_CHANNELS):
+            if ((1 << lane) & host_lanes_mask) == 0:
+                continue
+            si_key_lane = "{}{}".format(consts.ADAPTIVE_INPUT_EQ_RECALLED_TX, lane+1)
+            si_val = si_settings[consts.ADAPTIVE_INPUT_EQ_RECALLED_TX][si_key_lane]
+            lane %= (self.NUM_CHANNELS//2)
+            mask = ~(val << (lane*2))
+            l_data = si_val << (lane*2)
+            val = (val & mask) | l_data
+        return self.xcvr_eeprom.write(si_key_lane, val)
+
+    def scs_byte_read(self, si_keys):
+         si_byte_read = self.xcvr_eeprom.read(si_keys)
+         if si_byte_read is None:
+             return None
+         return si_byte_read
+
+    def scs_byte_write(self, si_key, host_lanes_mask, si_settings):
+        val = self.scs_byte_read(si_key)
+        if val is None:
+            return False
+        si_val = si_settings[si_key]
+        if si_val is None:
+            return False
+        nibble_val = si_val & host_lanes_mask
+        for lane in range(self.NUM_CHANNELS):
+            mask = (1 << lane)
+            if (mask & host_lanes_mask) == 0:
+                continue
+            if nibble_val:
+                val |= mask
+            else:
+                val &= ~mask
+        return self.xcvr_eeprom.write(si_key, val)
+
+    def stage_adaptive_input_eq_enable_tx(self, host_lanes_mask, si_settings):
+        '''
+        This function applies adaptive TX input enable si settings
+        '''
+        return self.scs_byte_write(consts.ADAPTIVE_INPUT_EQ_ENABLE_TX, host_lanes_mask, si_settings)
+
+    def stage_cdr_tx(self, host_lanes_mask, si_settings):
+        '''
+        This function applies TX CDR si settings
+        '''
+        return self.scs_byte_write(consts.CDR_ENABLE_TX, host_lanes_mask, si_settings)
+
+    def stage_cdr_rx(self, host_lanes_mask, si_settings):
+        '''
+        This function applies RX CDR si settings
+        '''
+        return self.scs_byte_write(consts.CDR_ENABLE_RX, host_lanes_mask, si_settings)
+
+    def stage_rx_si_settings(self, host_lanes_mask, si_settings):
+        for si_keys in si_settings:
+            if si_keys == consts.OUTPUT_EQ_PRE_CURSOR_TARGET_RX and self.get_rx_output_eq_pre_ctrl_supported():
+                if not self.stage_output_eq_pre_cursor_target_rx(host_lanes_mask, si_settings):
+                    return False
+            elif si_keys == consts.OUTPUT_EQ_POST_CURSOR_TARGET_RX and self.get_rx_output_eq_post_ctrl_supported():
+                if not self.stage_output_eq_post_cursor_target_rx(host_lanes_mask, si_settings):
+                    return False
+            elif si_keys == consts.OUTPUT_AMPLITUDE_TARGET_RX and self.get_rx_ouput_amp_ctrl_supported():
+                if not self.stage_output_amp_target_rx(host_lanes_mask, si_settings):
+                    return False
+            elif si_keys == consts.CDR_ENABLE_RX and self.get_rx_cdr_supported():
+                if not self.stage_cdr_rx(host_lanes_mask, si_settings):
+                    return False
+        return True
+
+    def stage_tx_si_settings(self, host_lanes_mask, si_settings):
+        for si_keys in si_settings:
+            if si_keys == consts.FIXED_INPUT_EQ_TARGET_TX and self.get_tx_input_eq_fixed_supported():
+                if not self.stage_fixed_input_target_tx(host_lanes_mask, si_settings):
+                    return False
+            elif (si_keys == consts.ADAPTIVE_INPUT_EQ_RECALLED_TX and
+                 (self.get_tx_input_recall_buf1_supported() or self.get_tx_input_recall_buf2_supported())):
+                if not self.stage_adaptive_input_recall_tx(host_lanes_mask, si_settings):
+                    return False
+            elif si_keys == consts.ADAPTIVE_INPUT_EQ_ENABLE_TX and self.get_tx_input_adaptive_eq_supported():
+                if not self.stage_adaptive_input_eq_enable_tx(host_lanes_mask, si_settings):
+                    return False
+            elif si_keys == consts.CDR_ENABLE_TX and self.get_tx_cdr_supported():
+                if not self.stage_cdr_tx(host_lanes_mask, si_settings):
+                    return False
+        return True
+
+    def stage_custom_si_settings(self, host_lanes_mask, optics_si_dict):
+        # Read and cache the existing SCS0 TX CTRL data
+        si_settings = self.xcvr_eeprom.read(consts.STAGED_CTRL0_TX_CTRL_FIELD)
+        if si_settings is None:
+            return False
+        # Replace the new values with cached SI values
+        for si_keys in optics_si_dict:
+            if si_keys in si_settings:
+                si_settings[si_keys] = optics_si_dict[si_keys]
+
+        # stage RX si settings
+        if not self.stage_rx_si_settings(host_lanes_mask, si_settings):
+            return False
+
+        #stage TX si settings
+        if not self.stage_tx_si_settings(host_lanes_mask, si_settings):
+            return False
+        return True
+
     def get_error_description(self):
         dp_state = self.get_datapath_state()
         conf_state = self.get_config_datapath_hostlane_status()