diff --git a/sonic_platform_base/sonic_xcvr/api/public/c_cmis.py b/sonic_platform_base/sonic_xcvr/api/public/c_cmis.py index 7830a300b..ab3422be0 100644 --- a/sonic_platform_base/sonic_xcvr/api/public/c_cmis.py +++ b/sonic_platform_base/sonic_xcvr/api/public/c_cmis.py @@ -5,14 +5,12 @@ """ from sonic_py_common import logger from ...fields import consts -from .cmis import CmisApi +from .cmis import CmisApi, CMIS_VDM_KEY_TO_DB_PREFIX_KEY_MAP import time BYTELENGTH = 8 -VDM_FREEZE = 128 -VDM_UNFREEZE = 0 SYSLOG_IDENTIFIER = "CCmisApi" -VDM_KEY_TO_DB_KEY_PREFIX_MAP = { +C_CMIS_DELTA_VDM_KEY_TO_DB_PREFIX_KEY_MAP = { 'Modulator Bias X/I [%]' : 'biasxi', 'Modulator Bias X/Q [%]' : 'biasxq', 'Modulator Bias X_Phase [%]' : 'biasxp', @@ -51,6 +49,10 @@ class CCmisApi(CmisApi): def __init__(self, xcvr_eeprom): super(CCmisApi, self).__init__(xcvr_eeprom) + def _get_vdm_key_to_db_prefix_map(self): + combined_map = {**CMIS_VDM_KEY_TO_DB_PREFIX_KEY_MAP, **C_CMIS_DELTA_VDM_KEY_TO_DB_PREFIX_KEY_MAP} + return combined_map + def _update_dict_if_vdm_key_exists(self, dict_to_be_updated, new_key, vdm_dict_key, vdm_subtype_index, lane=1): ''' This function updates the dictionary with the VDM value if the vdm_dict_key exists. @@ -209,43 +211,6 @@ def set_tx_power(self, tx_power): time.sleep(1) return status - def freeze_vdm_stats(self): - ''' - This function freeze all the vdm statistics reporting registers. - When raised by the host, causes the module to freeze and hold all - reported statistics reporting registers (minimum, maximum and - average values)in Pages 24h-27h. - - Returns True if the provision succeeds and False incase of failure. - ''' - return self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_FREEZE) - - def get_vdm_freeze_status(self): - ''' - This function reads and returns the vdm Freeze done status. - - Returns True if the vdm stats freeze is successful and False if not freeze. - ''' - return self.xcvr_eeprom.read(consts.VDM_FREEZE_DONE) - - def unfreeze_vdm_stats(self): - ''' - This function unfreeze all the vdm statistics reporting registers. - When freeze is ceased by the host, releases the freeze request, allowing the - reported minimum, maximum and average values to update again. - - Returns True if the provision succeeds and False incase of failure. - ''' - return self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_UNFREEZE) - - def get_vdm_unfreeze_status(self): - ''' - This function reads and returns the vdm unfreeze status. - - Returns True if the vdm stats unfreeze is successful and False if not unfreeze. - ''' - return self.xcvr_eeprom.read(consts.VDM_UNFREEZE_DONE) - def get_pm_all(self): ''' This function returns the PMs reported in Page 34h and 35h in OIF C-CMIS document @@ -440,7 +405,7 @@ def get_transceiver_bulk_status(self): """ trans_dom = super(CCmisApi,self).get_transceiver_bulk_status() - for vdm_key, trans_dom_key in VDM_KEY_TO_DB_KEY_PREFIX_MAP.items(): + for vdm_key, trans_dom_key in C_CMIS_DELTA_VDM_KEY_TO_DB_PREFIX_KEY_MAP.items(): self._update_dict_if_vdm_key_exists(trans_dom, trans_dom_key, vdm_key, 0) trans_dom['laser_config_freq'] = self.get_laser_config_freq() @@ -561,7 +526,7 @@ def get_transceiver_threshold_info(self): """ trans_dom_th = super(CCmisApi,self).get_transceiver_threshold_info() - for vdm_key, trans_dom_th_key_prefix in VDM_KEY_TO_DB_KEY_PREFIX_MAP.items(): + for vdm_key, trans_dom_th_key_prefix in C_CMIS_DELTA_VDM_KEY_TO_DB_PREFIX_KEY_MAP.items(): for i in range(1, 5): trans_dom_th_key = trans_dom_th_key_prefix + VDM_SUBTYPE_IDX_MAP[i] self._update_dict_if_vdm_key_exists(trans_dom_th, trans_dom_th_key, vdm_key, i) @@ -753,7 +718,7 @@ def get_transceiver_status(self): trans_status['invalid_channel_num'] = 'InvalidChannel' in laser_tuning_summary trans_status['tuning_complete'] = 'TuningComplete' in laser_tuning_summary - for vdm_key, trans_status_key_prefix in VDM_KEY_TO_DB_KEY_PREFIX_MAP.items(): + for vdm_key, trans_status_key_prefix in C_CMIS_DELTA_VDM_KEY_TO_DB_PREFIX_KEY_MAP.items(): for i in range(5, 9): trans_status_key = trans_status_key_prefix + VDM_SUBTYPE_IDX_MAP[i] self._update_dict_if_vdm_key_exists(trans_status, trans_status_key, vdm_key, i) diff --git a/sonic_platform_base/sonic_xcvr/api/public/cmis.py b/sonic_platform_base/sonic_xcvr/api/public/cmis.py index 4a2c161a9..f716db1f1 100644 --- a/sonic_platform_base/sonic_xcvr/api/public/cmis.py +++ b/sonic_platform_base/sonic_xcvr/api/public/cmis.py @@ -5,6 +5,7 @@ Implementation of XcvrApi that corresponds to the CMIS specification. """ +from enum import Enum from ...fields import consts from ..xcvr_api import XcvrApi @@ -21,6 +22,34 @@ logger = logging.getLogger(__name__) logger.addHandler(logging.NullHandler()) +VDM_FREEZE = 128 +VDM_UNFREEZE = 0 + +class VdmSubtypeIndex(Enum): + VDM_SUBTYPE_REAL_VALUE = 0 + VDM_SUBTYPE_HALARM_THRESHOLD = 1 + VDM_SUBTYPE_LALARM_THRESHOLD = 2 + VDM_SUBTYPE_HWARN_THRESHOLD = 3 + VDM_SUBTYPE_LWARN_THRESHOLD = 4 + VDM_SUBTYPE_HALARM_FLAG = 5 + VDM_SUBTYPE_LALARM_FLAG = 6 + VDM_SUBTYPE_HWARN_FLAG = 7 + VDM_SUBTYPE_LWARN_FLAG = 8 + +THRESHOLD_TYPE_STR_MAP = { + VdmSubtypeIndex.VDM_SUBTYPE_HALARM_THRESHOLD: "halarm", + VdmSubtypeIndex.VDM_SUBTYPE_LALARM_THRESHOLD: "lalarm", + VdmSubtypeIndex.VDM_SUBTYPE_HWARN_THRESHOLD: "hwarn", + VdmSubtypeIndex.VDM_SUBTYPE_LWARN_THRESHOLD: "lwarn" +} + +FLAG_TYPE_STR_MAP = { + VdmSubtypeIndex.VDM_SUBTYPE_HALARM_FLAG: "halarm", + VdmSubtypeIndex.VDM_SUBTYPE_LALARM_FLAG: "lalarm", + VdmSubtypeIndex.VDM_SUBTYPE_HWARN_FLAG: "hwarn", + VdmSubtypeIndex.VDM_SUBTYPE_LWARN_FLAG: "lwarn" +} + CMIS_VDM_KEY_TO_DB_PREFIX_KEY_MAP = { "Laser Temperature [C]" : "laser_temperature_media", "eSNR Media Input [dB]" : "esnr_media_input", @@ -55,6 +84,71 @@ 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_vdm_key_to_db_prefix_map(self): + return CMIS_VDM_KEY_TO_DB_PREFIX_KEY_MAP + + def _update_vdm_dict(self, dict_to_update, new_key, vdm_raw_dict, vdm_observable_type, vdm_subtype_index, lane): + """ + Updates the dictionary with the VDM value if the vdm_observable_type exists. + If the key does not exist, it will update the dictionary with 'N/A'. + + Args: + dict_to_update (dict): The dictionary to be updated. + new_key (str): The key to be added in dict_to_update. + vdm_raw_dict (dict): The raw VDM dictionary to be parsed. + vdm_observable_type (str): Lookup key in the VDM dictionary. + vdm_subtype_index (VdmSubtypeIndex): The index of the VDM subtype in the VDM page. + lane (int): The lane number to be looked up in the VDM dictionary. + + Returns: + bool: True if the key exists in the VDM dictionary, False if not. + """ + try: + dict_to_update[new_key] = vdm_raw_dict[vdm_observable_type][lane][vdm_subtype_index.value] + except (KeyError, TypeError): + dict_to_update[new_key] = 'N/A' + logger.debug('key {} not present in VDM'.format(new_key)) + return False + + return True + + def freeze_vdm_stats(self): + ''' + This function freeze all the vdm statistics reporting registers. + When raised by the host, causes the module to freeze and hold all + reported statistics reporting registers (minimum, maximum and + average values)in Pages 24h-27h. + + Returns True if the provision succeeds and False incase of failure. + ''' + return self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_FREEZE) + + def get_vdm_freeze_status(self): + ''' + This function reads and returns the vdm Freeze done status. + + Returns True if the vdm stats freeze is successful and False if not freeze. + ''' + return self.xcvr_eeprom.read(consts.VDM_FREEZE_DONE) + + def unfreeze_vdm_stats(self): + ''' + This function unfreeze all the vdm statistics reporting registers. + When freeze is ceased by the host, releases the freeze request, allowing the + reported minimum, maximum and average values to update again. + + Returns True if the provision succeeds and False incase of failure. + ''' + return self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_UNFREEZE) + + def get_vdm_unfreeze_status(self): + ''' + This function reads and returns the vdm unfreeze status. + + Returns True if the vdm stats unfreeze is successful and False if not unfreeze. + ''' + return self.xcvr_eeprom.read(consts.VDM_UNFREEZE_DONE) + def get_manufacturer(self): ''' This function returns the manufacturer of the module @@ -270,6 +364,65 @@ def get_transceiver_bulk_status(self): return bulk_status + def get_transceiver_dom_flags(self): + dom_flag_dict = dict() + module_flag = self.get_module_level_flag() + + try: + case_temp_flags = module_flag['case_temp_flags'] + voltage_flags = module_flag['voltage_flags'] + dom_flag_dict.update({ + 'temphighalarm': case_temp_flags['case_temp_high_alarm_flag'], + 'templowalarm': case_temp_flags['case_temp_low_alarm_flag'], + 'temphighwarning': case_temp_flags['case_temp_high_warn_flag'], + 'templowwarning': case_temp_flags['case_temp_low_warn_flag'], + 'vcchighalarm': voltage_flags['voltage_high_alarm_flag'], + 'vcclowalarm': voltage_flags['voltage_low_alarm_flag'], + 'vcchighwarning': voltage_flags['voltage_high_warn_flag'], + 'vcclowwarning': voltage_flags['voltage_low_warn_flag'] + }) + except TypeError: + pass + + tx_power_flag_dict = self.get_tx_power_flag() + if tx_power_flag_dict: + for lane in range(1, self.NUM_CHANNELS+1): + dom_flag_dict['txpowerhighalarm%d' % lane] = tx_power_flag_dict['tx_power_high_alarm']['TxPowerHighAlarmFlag%d' % lane] + dom_flag_dict['txpowerlowalarm%d' % lane] = tx_power_flag_dict['tx_power_low_alarm']['TxPowerLowAlarmFlag%d' % lane] + dom_flag_dict['txpowerhighwarning%d' % lane] = tx_power_flag_dict['tx_power_high_warn']['TxPowerHighWarnFlag%d' % lane] + dom_flag_dict['txpowerlowwarning%d' % lane] = tx_power_flag_dict['tx_power_low_warn']['TxPowerLowWarnFlag%d' % lane] + rx_power_flag_dict = self.get_rx_power_flag() + if rx_power_flag_dict: + for lane in range(1, self.NUM_CHANNELS+1): + dom_flag_dict['rxpowerhighalarm%d' % lane] = rx_power_flag_dict['rx_power_high_alarm']['RxPowerHighAlarmFlag%d' % lane] + dom_flag_dict['rxpowerlowalarm%d' % lane] = rx_power_flag_dict['rx_power_low_alarm']['RxPowerLowAlarmFlag%d' % lane] + dom_flag_dict['rxpowerhighwarning%d' % lane] = rx_power_flag_dict['rx_power_high_warn']['RxPowerHighWarnFlag%d' % lane] + dom_flag_dict['rxpowerlowwarning%d' % lane] = rx_power_flag_dict['rx_power_low_warn']['RxPowerLowWarnFlag%d' % lane] + tx_bias_flag_dict = self.get_tx_bias_flag() + if tx_bias_flag_dict: + for lane in range(1, self.NUM_CHANNELS+1): + dom_flag_dict['txbiashighalarm%d' % lane] = tx_bias_flag_dict['tx_bias_high_alarm']['TxBiasHighAlarmFlag%d' % lane] + dom_flag_dict['txbiaslowalarm%d' % lane] = tx_bias_flag_dict['tx_bias_low_alarm']['TxBiasLowAlarmFlag%d' % lane] + dom_flag_dict['txbiashighwarning%d' % lane] = tx_bias_flag_dict['tx_bias_high_warn']['TxBiasHighWarnFlag%d' % lane] + dom_flag_dict['txbiaslowwarning%d' % lane] = tx_bias_flag_dict['tx_bias_low_warn']['TxBiasLowWarnFlag%d' % lane] + + try: + _, aux2_mon_type, aux3_mon_type = self.get_aux_mon_type() + if aux2_mon_type == 0: + dom_flag_dict['lasertemphighalarm'] = module_flag['aux2_flags']['aux2_high_alarm_flag'] + dom_flag_dict['lasertemplowalarm'] = module_flag['aux2_flags']['aux2_low_alarm_flag'] + dom_flag_dict['lasertemphighwarning'] = module_flag['aux2_flags']['aux2_high_warn_flag'] + dom_flag_dict['lasertemplowwarning'] = module_flag['aux2_flags']['aux2_low_warn_flag'] + elif aux2_mon_type == 1 and aux3_mon_type == 0: + dom_flag_dict['lasertemphighalarm'] = module_flag['aux3_flags']['aux3_high_alarm_flag'] + dom_flag_dict['lasertemplowalarm'] = module_flag['aux3_flags']['aux3_low_alarm_flag'] + dom_flag_dict['lasertemphighwarning'] = module_flag['aux3_flags']['aux3_high_warn_flag'] + dom_flag_dict['lasertemplowwarning'] = module_flag['aux3_flags']['aux3_low_warn_flag'] + except TypeError: + pass + + return dom_flag_dict + def get_transceiver_threshold_info(self): threshold_info_keys = ['temphighalarm', 'temphighwarning', 'templowalarm', 'templowwarning', @@ -681,6 +834,26 @@ def tx_disable_channel(self, channel, disable): return self.xcvr_eeprom.write(consts.TX_DISABLE_FIELD, channel_state) + def get_laser_tuning_summary(self): + ''' + This function returns laser tuning status summary on media lane + ''' + result = self.xcvr_eeprom.read(consts.LASER_TUNING_DETAIL) + laser_tuning_summary = [] + if (result >> 5) & 0x1: + laser_tuning_summary.append("TargetOutputPowerOOR") + if (result >> 4) & 0x1: + laser_tuning_summary.append("FineTuningOutOfRange") + if (result >> 3) & 0x1: + laser_tuning_summary.append("TuningNotAccepted") + if (result >> 2) & 0x1: + laser_tuning_summary.append("InvalidChannel") + if (result >> 1) & 0x1: + laser_tuning_summary.append("WavelengthUnlocked") + if (result >> 0) & 0x1: + laser_tuning_summary.append("TuningComplete") + return laser_tuning_summary + def get_power_override(self): return None @@ -2094,6 +2267,66 @@ def get_transceiver_status(self): pass return trans_status + def get_transceiver_status_flags(self): + """ + Retrieves transceiver status flags of this SFP + + Returns: + A dict which contains following keys/values : + ================================================================================ + ; Defines Transceiver Status info for a port + key = TRANSCEIVER_STATUS_FLAG|ifname ; Flag information for module on port + ; field = value + datapath_firmware_fault = BOOLEAN ; datapath (DSP) firmware fault + module_firmware_fault = BOOLEAN ; module firmware fault + module_state_changed = BOOLEAN ; module state changed + txfault{lane_num} = BOOLEAN ; tx fault flag on media lane {lane_num} + txlos_hostlane{lane_num} = BOOLEAN ; tx loss of signal flag on host lane {lane_num} + txcdrlol_hostlane{lane_num} = BOOLEAN ; tx clock and data recovery loss of lock flag on host lane {lane_num} + tx_eq_fault{lane_num} = BOOLEAN ; tx equalization fault flag on host lane {lane_num} + rxlos{lane_num} = BOOLEAN ; rx loss of signal flag on media lane {lane_num} + rxcdrlol{lane_num} = BOOLEAN ; rx clock and data recovery loss of lock flag on media lane {lane_num} + target_output_power_oor = BOOLEAN ; target output power out of range flag + fine_tuning_oor = BOOLEAN ; fine tuning out of range flag + tuning_not_accepted = BOOLEAN ; tuning not accepted flag + invalid_channel_num = BOOLEAN ; invalid channel number flag + tuning_complete = BOOLEAN ; tuning complete flag + ================================================================================ + """ + status_flags_dict = dict() + try: + dp_fw_fault, module_fw_fault, module_state_changed = self.get_module_firmware_fault_state_changed() + status_flags_dict.update({ + 'datapath_firmware_fault': dp_fw_fault, + 'module_firmware_fault': module_fw_fault, + 'module_state_changed': module_state_changed + }) + except TypeError: + pass + + fault_types = { + 'txfault': self.get_tx_fault(), + 'txlos_hostlane': self.get_tx_los(), + 'txcdrlol_hostlane': self.get_tx_cdr_lol(), + 'tx_eq_fault': self.get_tx_adaptive_eq_fail_flag(), + 'rxlos': self.get_rx_los(), + 'rxcdrlol': self.get_rx_cdr_lol() + } + + for fault_type, fault_values in fault_types.items(): + for lane in range(1, self.NUM_CHANNELS + 1): + key = f'{fault_type}{lane}' + status_flags_dict[key] = fault_values[lane - 1] if fault_values else "N/A" + + laser_tuning_summary = self.get_laser_tuning_summary() + status_flags_dict['target_output_power_oor'] = 'TargetOutputPowerOOR' in laser_tuning_summary + status_flags_dict['fine_tuning_oor'] = 'FineTuningOutOfRange' in laser_tuning_summary + status_flags_dict['tuning_not_accepted'] = 'TuningNotAccepted' in laser_tuning_summary + status_flags_dict['invalid_channel_num'] = 'InvalidChannel' in laser_tuning_summary + status_flags_dict['tuning_complete'] = 'TuningComplete' in laser_tuning_summary + + return status_flags_dict + def get_transceiver_loopback(self): """ Retrieves loopback mode for this xcvr @@ -2171,6 +2404,201 @@ def get_transceiver_loopback(self): trans_loopback['host_input_loopback_lane%d' % lane] = 'N/A' return trans_loopback + def get_transceiver_vdm_real_value(self): + """ + Retrieves VDM real value for this xcvr + + Returns: + A dict containing the following keys/values : + ======================================================================== + key = TRANSCEIVER_VDM_REAL_VALUE|ifname ; information module VDM sample on port + ; field = value + laser_temperature_media{lane_num} = FLOAT ; laser temperature value in Celsius for media input + esnr_media_input{lane_num} = FLOAT ; eSNR value in dB for media input + esnr_host_input{lane_num} = FLOAT ; eSNR value in dB for host input + pam4_level_transition_media_input{lane_num} = FLOAT ; PAM4 level transition parameter in dB for media input + pam4_level_transition_host_input{lane_num} = FLOAT ; PAM4 level transition parameter in dB for host input + prefec_ber_min_media_input{lane_num} = FLOAT ; Pre-FEC BER minimum value for media input + prefec_ber_max_media_input{lane_num} = FLOAT ; Pre-FEC BER maximum value for media input + prefec_ber_avg_media_input{lane_num} = FLOAT ; Pre-FEC BER average value for media input + prefec_ber_curr_media_input{lane_num} = FLOAT ; Pre-FEC BER current value for media input + prefec_ber_min_host_input{lane_num} = FLOAT ; Pre-FEC BER minimum value for host input + prefec_ber_max_host_input{lane_num} = FLOAT ; Pre-FEC BER maximum value for host input + prefec_ber_avg_host_input{lane_num} = FLOAT ; Pre-FEC BER average value for host input + prefec_ber_curr_host_input{lane_num} = FLOAT ; Pre-FEC BER current value for host input + errored_frames_min_media_input{lane_num} = FLOAT ; Errored frames minimum value for media input + errored_frames_max_media_input{lane_num} = FLOAT ; Errored frames maximum value for media input + errored_frames_avg_media_input{lane_num} = FLOAT ; Errored frames average value for media input + errored_frames_curr_media_input{lane_num} = FLOAT ; Errored frames current value for media input + errored_frames_min_host_input{lane_num} = FLOAT ; Errored frames minimum value for host input + errored_frames_max_host_input{lane_num} = FLOAT ; Errored frames maximum value for host input + errored_frames_avg_host_input{lane_num} = FLOAT ; Errored frames average value for host input + errored_frames_curr_host_input{lane_num} = FLOAT ; Errored frames current value for host input + + ;C-CMIS specific fields + biasxi{lane_num} = FLOAT ; modulator bias xi in percentage + biasxq{lane_num} = FLOAT ; modulator bias xq in percentage + biasxp{lane_num} = FLOAT ; modulator bias xp in percentage + biasyi{lane_num} = FLOAT ; modulator bias yi in percentage + biasyq{lane_num} = FLOAT ; modulator bias yq in percentage + biasyp{lane_num} = FLOAT ; modulator bias yq in percentage + cdshort{lane_num} = FLOAT ; chromatic dispersion, high granularity, short link in ps/nm + cdlong{lane_num} = FLOAT ; chromatic dispersion, high granularity, long link in ps/nm + dgd{lane_num} = FLOAT ; differential group delay in ps + sopmd{lane_num} = FLOAT ; second order polarization mode dispersion in ps^2 + soproc{lane_num} = FLOAT ; state of polarization rate of change in krad/s + pdl{lane_num} = FLOAT ; polarization dependent loss in db + osnr{lane_num} = FLOAT ; optical signal to noise ratio in db + esnr{lane_num} = FLOAT ; electrical signal to noise ratio in db + cfo{lane_num} = FLOAT ; carrier frequency offset in Hz + txcurrpower{lane_num} = FLOAT ; tx current output power in dbm + rxtotpower{lane_num} = FLOAT ; rx total power in dbm + rxsigpower{lane_num} = FLOAT ; rx signal power in dbm + ======================================================================== + """ + vdm_real_value_dict = dict() + vdm_raw_dict = self.get_vdm(self.vdm.VDM_REAL_VALUE) + for vdm_observable_type, db_key_name_prefix in self._get_vdm_key_to_db_prefix_map().items(): + for lane in range(1, self.NUM_CHANNELS + 1): + db_key_name = f"{db_key_name_prefix}{lane}" + self._update_vdm_dict(vdm_real_value_dict, db_key_name, vdm_raw_dict, vdm_observable_type, + VdmSubtypeIndex.VDM_SUBTYPE_REAL_VALUE, lane) + return vdm_real_value_dict + + def get_transceiver_vdm_thresholds(self): + """ + Retrieves VDM thresholds for this xcvr + + Returns: + A dict containing the following keys/values : + ======================================================================== + xxx refers to HALARM/LALARM/HWARN/LWARN threshold + ;Defines Transceiver VDM high/low alarm/warning threshold for a port + key = TRANSCEIVER_VDM_XXX_THRESHOLD|ifname ; information module VDM high/low alarm/warning threshold on port + ; field = value + laser_temperature_media_xxx{lane_num} = FLOAT ; laser temperature high/low alarm/warning value in Celsius for media input + esnr_media_input_xxx{lane_num} = FLOAT ; eSNR high/low alarm/warning value in dB for media input + esnr_host_input_xxx{lane_num} = FLOAT ; eSNR high/low alarm/warning value in dB for host input + pam4_level_transition_media_input_xxx{lane_num} = FLOAT ; PAM4 level transition high/low alarm/warning value in dB for media input + pam4_level_transition_host_input_xxx{lane_num} = FLOAT ; PAM4 level transition high/low alarm/warning value in dB for host input + prefec_ber_min_media_input_xxx{lane_num} = FLOAT ; Pre-FEC BER minimum high/low alarm/warning value for media input + prefec_ber_max_media_input_xxx{lane_num} = FLOAT ; Pre-FEC BER maximum high/low alarm/warning value for media input + prefec_ber_avg_media_input_xxx{lane_num} = FLOAT ; Pre-FEC BER average high/low alarm/warning value for media input + prefec_ber_curr_media_input_xxx{lane_num} = FLOAT ; Pre-FEC BER current high/low alarm/warning value for media input + prefec_ber_min_host_input_xxx{lane_num} = FLOAT ; Pre-FEC BER minimum high/low alarm/warning value for host input + prefec_ber_max_host_input_xxx{lane_num} = FLOAT ; Pre-FEC BER maximum high/low alarm/warning value for host input + prefec_ber_avg_host_input_xxx{lane_num} = FLOAT ; Pre-FEC BER average high/low alarm/warning value for host input + prefec_ber_curr_host_input_xxx{lane_num} = FLOAT ; Pre-FEC BER current high/low alarm/warning value for host input + errored_frames_min_media_input_xxx{lane_num} = FLOAT ; Errored frames minimum high/low alarm/warning value for media input + errored_frames_max_media_input_xxx{lane_num} = FLOAT ; Errored frames maximum high/low alarm/warning value for media input + errored_frames_avg_media_input_xxx{lane_num} = FLOAT ; Errored frames average high/low alarm/warning value for media input + errored_frames_curr_media_input_xxx{lane_num} = FLOAT ; Errored frames current high/low alarm/warning value for media input + errored_frames_min_host_input_xxx{lane_num} = FLOAT ; Errored frames minimum high/low alarm/warning value for host input + errored_frames_max_host_input_xxx{lane_num} = FLOAT ; Errored frames maximum high/low alarm/warning value for host input + errored_frames_avg_host_input_xxx{lane_num} = FLOAT ; Errored frames average high/low alarm/warning value for host input + errored_frames_curr_host_input_xxx{lane_num} = FLOAT ; Errored frames current high/low alarm/warning value for host input + + ;C-CMIS specific fields + biasxi_xxx{lane_num} = FLOAT ; modulator bias xi in percentage (high/low alarm/warning) + biasxq_xxx{lane_num} = FLOAT ; modulator bias xq in percentage (high/low alarm/warning) + biasxp_xxx{lane_num} = FLOAT ; modulator bias xp in percentage (high/low alarm/warning) + biasyi_xxx{lane_num} = FLOAT ; modulator bias yi in percentage (high/low alarm/warning) + biasyq_xxx{lane_num} = FLOAT ; modulator bias yq in percentage (high/low alarm/warning) + biasyp_xxx{lane_num} = FLOAT ; modulator bias yq in percentage (high/low alarm/warning) + cdshort_xxx{lane_num} = FLOAT ; chromatic dispersion, high granularity, short link in ps/nm (high/low alarm/warning) + cdlong_xxx{lane_num} = FLOAT ; chromatic dispersion, high granularity, long link in ps/nm (high/low alarm/warning) + dgd_xxx{lane_num} = FLOAT ; differential group delay in ps (high/low alarm/warning) + sopmd_xxx{lane_num} = FLOAT ; second order polarization mode dispersion in ps^2 (high/low alarm/warning) + soproc_xxx{lane_num} = FLOAT ; state of polarization rate of change in krad/s (high/low alarm/warning) + pdl_xxx{lane_num} = FLOAT ; polarization dependent loss in db (high/low alarm/warning) + osnr_xxx{lane_num} = FLOAT ; optical signal to noise ratio in db (high/low alarm/warning) + esnr_xxx{lane_num} = FLOAT ; electrical signal to noise ratio in db (high/low alarm/warning) + cfo_xxx{lane_num} = FLOAT ; carrier frequency offset in Hz (high/low alarm/warning) + txcurrpower_xxx{lane_num} = FLOAT ; tx current output power in dbm (high/low alarm/warning) + rxtotpower_xxx{lane_num} = FLOAT ; rx total power in dbm (high/low alarm/warning) + rxsigpower_xxx{lane_num} = FLOAT ; rx signal power in dbm (high/low alarm/warning) ======================================================================== + """ + vdm_thresholds_dict = dict() + vdm_raw_dict = self.get_vdm(self.vdm.VDM_THRESHOLD) + for vdm_observable_type, db_key_name_prefix in self._get_vdm_key_to_db_prefix_map().items(): + for lane in range(1, self.NUM_CHANNELS + 1): + for vdm_threshold_type in range(VdmSubtypeIndex.VDM_SUBTYPE_HALARM_THRESHOLD.value, VdmSubtypeIndex.VDM_SUBTYPE_LWARN_THRESHOLD.value + 1): + vdm_threshold_enum = VdmSubtypeIndex(vdm_threshold_type) + threshold_type_str = THRESHOLD_TYPE_STR_MAP.get(vdm_threshold_enum) + if threshold_type_str: + db_key_name = f"{db_key_name_prefix}_{threshold_type_str}{lane}" + self._update_vdm_dict(vdm_thresholds_dict, db_key_name, vdm_raw_dict, + vdm_observable_type, vdm_threshold_enum, lane) + + return vdm_thresholds_dict + + def get_transceiver_vdm_flags(self): + """ + Retrieves VDM flags for this xcvr + + Returns: + A dict containing the following keys/values : + ======================================================================== + xxx refers to HALARM/LALARM/HWARN/LWARN + ;Defines Transceiver VDM high/low alarm/warning flag for a port + key = TRANSCEIVER_VDM_XXX_FLAG|ifname ; information module VDM high/low alarm/warning flag on port + ; field = value + laser_temperature_media_xxx{lane_num} = FLOAT ; laser temperature high/low alarm/warning flag in Celsius for media input + esnr_media_input_xxx{lane_num} = FLOAT ; eSNR high/low alarm/warning flag in dB for media input + esnr_host_input_xxx{lane_num} = FLOAT ; eSNR high/low alarm/warning flag in dB for host input + pam4_level_transition_media_input_xxx{lane_num} = FLOAT ; PAM4 level transition high/low alarm/warning flag in dB for media input + pam4_level_transition_host_input_xxx{lane_num} = FLOAT ; PAM4 level transition high/low alarm/warning flag in dB for host input + prefec_ber_min_media_input_xxx{lane_num} = FLOAT ; Pre-FEC BER minimum high/low alarm/warning flag for media input + prefec_ber_max_media_input_xxx{lane_num} = FLOAT ; Pre-FEC BER maximum high/low alarm/warning flag for media input + prefec_ber_avg_media_input_xxx{lane_num} = FLOAT ; Pre-FEC BER average high/low alarm/warning flag for media input + prefec_ber_curr_media_input_xxx{lane_num} = FLOAT ; Pre-FEC BER current high/low alarm/warning flag for media input + prefec_ber_min_host_input_xxx{lane_num} = FLOAT ; Pre-FEC BER minimum high/low alarm/warning flag for host input + prefec_ber_max_host_input_xxx{lane_num} = FLOAT ; Pre-FEC BER maximum high/low alarm/warning flag for host input + prefec_ber_avg_host_input_xxx{lane_num} = FLOAT ; Pre-FEC BER average high/low alarm/warning flag for host input + prefec_ber_curr_host_input_xxx{lane_num} = FLOAT ; Pre-FEC BER current high/low alarm/warning flag for host input + errored_frames_min_media_input_xxx{lane_num} = FLOAT ; Errored frames minimum high/low alarm/warning flag for media input + errored_frames_max_media_input_xxx{lane_num} = FLOAT ; Errored frames maximum high/low alarm/warning flag for media input + errored_frames_avg_media_input_xxx{lane_num} = FLOAT ; Errored frames average high/low alarm/warning flag for media input + errored_frames_curr_media_input_xxx{lane_num} = FLOAT ; Errored frames current high/low alarm/warning flag for media input + errored_frames_min_host_input_xxx{lane_num} = FLOAT ; Errored frames minimum high/low alarm/warning flag for host input + errored_frames_max_host_input_xxx{lane_num} = FLOAT ; Errored frames maximum high/low alarm/warning flag for host input + errored_frames_avg_host_input_xxx{lane_num} = FLOAT ; Errored frames average high/low alarm/warning flag for host input + errored_frames_curr_host_input_xxx{lane_num} = FLOAT ; Errored frames current high/low alarm/warning flag for host input + + ;C-CMIS specific fields + biasxi_xxx{lane_num} = FLOAT ; modulator bias xi in percentage (high/low alarm/warning flag) + biasxq_xxx{lane_num} = FLOAT ; modulator bias xq in percentage (high/low alarm/warning flag) + biasxp_xxx{lane_num} = FLOAT ; modulator bias xp in percentage (high/low alarm/warning flag) + biasyi_xxx{lane_num} = FLOAT ; modulator bias yi in percentage (high/low alarm/warning flag) + biasyq_xxx{lane_num} = FLOAT ; modulator bias yq in percentage (high/low alarm/warning flag) + biasyp_xxx{lane_num} = FLOAT ; modulator bias yq in percentage (high/low alarm/warning flag) + cdshort_xxx{lane_num} = FLOAT ; chromatic dispersion, high granularity, short link in ps/nm (high/low alarm/warning flag) + cdlong_xxx{lane_num} = FLOAT ; chromatic dispersion, high granularity, long link in ps/nm (high/low alarm/warning flag) + dgd_xxx{lane_num} = FLOAT ; differential group delay in ps (high/low alarm/warning flag) + sopmd_xxx{lane_num} = FLOAT ; second order polarization mode dispersion in ps^2 (high/low alarm/warning flag) + soproc_xxx{lane_num} = FLOAT ; state of polarization rate of change in krad/s (high/low alarm/warning flag) + pdl_xxx{lane_num} = FLOAT ; polarization dependent loss in db (high/low alarm/warning flag) + osnr_xxx{lane_num} = FLOAT ; optical signal to noise ratio in db (high/low alarm/warning flag) + esnr_xxx{lane_num} = FLOAT ; electrical signal to noise ratio in db (high/low alarm/warning flag) + cfo_xxx{lane_num} = FLOAT ; carrier frequency offset in Hz (high/low alarm/warning flag) + txcurrpower_xxx{lane_num} = FLOAT ; tx current output power in dbm (high/low alarm/warning flag) + rxtotpower_xxx{lane_num} = FLOAT ; rx total power in dbm (high/low alarm/warning flag) + rxsigpower_xxx{lane_num} = FLOAT ; rx signal power in dbm (high/low alarm/warning flag) + """ + vdm_flags_dict = dict() + vdm_raw_dict = self.get_vdm(self.vdm.VDM_FLAG) + for vdm_observable_type, db_key_name_prefix in self._get_vdm_key_to_db_prefix_map().items(): + for lane in range(1, self.NUM_CHANNELS + 1): + for vdm_flag_type in range(VdmSubtypeIndex.VDM_SUBTYPE_HALARM_FLAG.value, VdmSubtypeIndex.VDM_SUBTYPE_LWARN_FLAG.value + 1): + vdm_flag_enum = VdmSubtypeIndex(vdm_flag_type) + flag_type_str = FLAG_TYPE_STR_MAP.get(vdm_flag_enum) + if flag_type_str: + db_key_name = f"{db_key_name_prefix}_{flag_type_str}{lane}" + self._update_vdm_dict(vdm_flags_dict, db_key_name, vdm_raw_dict, + vdm_observable_type, vdm_flag_enum, lane) + + return vdm_flags_dict + def set_datapath_init(self, channel): """ Put the CMIS datapath into the initialized state diff --git a/sonic_platform_base/sonic_xcvr/api/public/cmisVDM.py b/sonic_platform_base/sonic_xcvr/api/public/cmisVDM.py index f597613cc..ebaef758d 100644 --- a/sonic_platform_base/sonic_xcvr/api/public/cmisVDM.py +++ b/sonic_platform_base/sonic_xcvr/api/public/cmisVDM.py @@ -195,8 +195,11 @@ def get_vdm_allpage(self, field_option=ALL_FIELD ): vdm_low_warn_flag ] ''' - vdm_page_supported_raw = self.xcvr_eeprom.read(consts.VDM_SUPPORTED_PAGE) - if vdm_page_supported_raw is None: + vdm_pages_supported = self.xcvr_eeprom.read(consts.VDM_SUPPORTED) + if not vdm_pages_supported: + return None + vdm_groups_supported_raw = self.xcvr_eeprom.read(consts.VDM_SUPPORTED_PAGE) + if vdm_groups_supported_raw is None: return None VDM_START_PAGE = 0x20 vdm = dict() @@ -206,7 +209,7 @@ def get_vdm_allpage(self, field_option=ALL_FIELD ): else: vdm_flag_page = None - for page in range(VDM_START_PAGE, VDM_START_PAGE + vdm_page_supported_raw + 1): + for page in range(VDM_START_PAGE, VDM_START_PAGE + vdm_groups_supported_raw + 1): vdm_current_page = self.get_vdm_page(page, vdm_flag_page, field_option) vdm.update(vdm_current_page) return vdm diff --git a/sonic_platform_base/sonic_xcvr/api/xcvr_api.py b/sonic_platform_base/sonic_xcvr/api/xcvr_api.py index c49bf64f9..f28c852cf 100644 --- a/sonic_platform_base/sonic_xcvr/api/xcvr_api.py +++ b/sonic_platform_base/sonic_xcvr/api/xcvr_api.py @@ -103,6 +103,15 @@ def get_transceiver_bulk_status(self): """ raise NotImplementedError + def get_transceiver_dom_flags(self): + """ + Retrieves the DOM flags for this xcvr + + Returns: + A dict containing dom flags for this xcvr + """ + raise NotImplementedError + def get_transceiver_threshold_info(self): """ Retrieves threshold info for this xcvr @@ -242,6 +251,31 @@ def get_transceiver_status(self): """ raise NotImplementedError + def get_transceiver_status_flags(self): + """ + Retrieves status flags of this xcvr + """ + raise NotImplementedError + + def get_transceiver_vdm_real_value(self): + """ + Retrieves VDM real (sample) values for this xcvr (applicable for CMIS and C-CMIS) + Specifically, it retrieves sample data from pages 24h to 27h + """ + raise NotImplementedError + + def get_transceiver_vdm_thresholds(self): + """ + Retrieves VDM thresholds for this xcvr (applicable for CMIS and C-CMIS) + """ + raise NotImplementedError + + def get_transceiver_vdm_flags(self): + """ + Retrieves VDM flags for this xcvr (applicable for CMIS and C-CMIS) + """ + raise NotImplementedError + def get_transceiver_pm(self): """ Retrieves PM (Performance Monitoring) info for this xcvr (applicable for C-CMIS) diff --git a/sonic_platform_base/sonic_xcvr/sfp_optoe_base.py b/sonic_platform_base/sonic_xcvr/sfp_optoe_base.py index a1933d1e9..e175b1167 100644 --- a/sonic_platform_base/sonic_xcvr/sfp_optoe_base.py +++ b/sonic_platform_base/sonic_xcvr/sfp_optoe_base.py @@ -31,6 +31,10 @@ def get_transceiver_bulk_status(self): api = self.get_xcvr_api() return api.get_transceiver_bulk_status() if api is not None else None + def get_transceiver_dom_flags(self): + api = self.get_xcvr_api() + return api.get_transceiver_dom_flags() if api is not None else None + def get_transceiver_threshold_info(self): api = self.get_xcvr_api() return api.get_transceiver_threshold_info() if api is not None else None @@ -39,6 +43,10 @@ def get_transceiver_status(self): api = self.get_xcvr_api() return api.get_transceiver_status() if api is not None else None + def get_transceiver_status_flags(self): + api = self.get_xcvr_api() + return api.get_transceiver_status_flags() if api is not None else None + def get_transceiver_loopback(self): api = self.get_xcvr_api() return api.get_transceiver_loopback() if api is not None else None @@ -47,6 +55,22 @@ def is_coherent_module(self): api = self.get_xcvr_api() return api.is_coherent_module() if api is not None else None + def get_transceiver_vdm_real_value(self): + """ + Retrieves VDM real (sample) values for this xcvr (applicable for CMIS and C-CMIS) + Specifically, it retrieves sample data from pages 24h to 27h + """ + api = self.get_xcvr_api() + return api.get_transceiver_vdm_real_value() if api is not None else None + + def get_transceiver_vdm_thresholds(self): + api = self.get_xcvr_api() + return api.get_transceiver_vdm_thresholds() if api is not None else None + + def get_transceiver_vdm_flags(self): + api = self.get_xcvr_api() + return api.get_transceiver_vdm_flags() if api is not None else None + def get_transceiver_pm(self): api = self.get_xcvr_api() return api.get_transceiver_pm() if api is not None else None diff --git a/tests/sonic_xcvr/test_cmis.py b/tests/sonic_xcvr/test_cmis.py index 20a96b68b..5e94b3d9e 100644 --- a/tests/sonic_xcvr/test_cmis.py +++ b/tests/sonic_xcvr/test_cmis.py @@ -3,7 +3,7 @@ import pytest import traceback import random -from sonic_platform_base.sonic_xcvr.api.public.cmis import CmisApi +from sonic_platform_base.sonic_xcvr.api.public.cmis import CmisApi, CMIS_VDM_KEY_TO_DB_PREFIX_KEY_MAP, THRESHOLD_TYPE_STR_MAP, FLAG_TYPE_STR_MAP from sonic_platform_base.sonic_xcvr.mem_maps.public.cmis import CmisMemMap from sonic_platform_base.sonic_xcvr.xcvr_eeprom import XcvrEeprom from sonic_platform_base.sonic_xcvr.codes.public.cmis import CmisCodes @@ -574,6 +574,17 @@ def test_tx_disable_channel(self, mock_response, input_param): self.api.get_tx_disable_channel.return_value = mock_response self.api.tx_disable_channel(*input_param) + @pytest.mark.parametrize("mock_response, expected", [ + (1, ['TuningComplete']), + (62, ['TargetOutputPowerOOR', 'FineTuningOutOfRange', 'TuningNotAccepted', + 'InvalidChannel', 'WavelengthUnlocked']), + ]) + def test_get_laser_tuning_summary(self, mock_response, expected): + self.api.xcvr_eeprom.read = MagicMock() + self.api.xcvr_eeprom.read.return_value = mock_response + result = self.api.get_laser_tuning_summary() + assert result == expected + def test_get_power_override(self): self.api.get_power_override() @@ -1619,6 +1630,179 @@ def test_get_transceiver_bulk_status(self, mock_response, expected): result = self.api.get_transceiver_bulk_status() assert result == expected + @pytest.mark.parametrize( + "module_flag, tx_power_flag_dict, rx_power_flag_dict, tx_bias_flag_dict, aux_mon_types, expected_result", + [ + # Test case 1: All flags present + ( + { + 'case_temp_flags': { + 'case_temp_high_alarm_flag': True, + 'case_temp_low_alarm_flag': False, + 'case_temp_high_warn_flag': True, + 'case_temp_low_warn_flag': False + }, + 'voltage_flags': { + 'voltage_high_alarm_flag': True, + 'voltage_low_alarm_flag': False, + 'voltage_high_warn_flag': True, + 'voltage_low_warn_flag': False + }, + 'aux2_flags': { + 'aux2_high_alarm_flag': True, + 'aux2_low_alarm_flag': False, + 'aux2_high_warn_flag': True, + 'aux2_low_warn_flag': False + }, + 'aux3_flags': { + 'aux3_high_alarm_flag': True, + 'aux3_low_alarm_flag': False, + 'aux3_high_warn_flag': True, + 'aux3_low_warn_flag': False + } + }, + { + 'tx_power_high_alarm': {f'TxPowerHighAlarmFlag{i}': i % 2 == 1 for i in range(1, 9)}, + 'tx_power_low_alarm': {f'TxPowerLowAlarmFlag{i}': i % 2 == 0 for i in range(1, 9)}, + 'tx_power_high_warn': {f'TxPowerHighWarnFlag{i}': i % 2 == 1 for i in range(1, 9)}, + 'tx_power_low_warn': {f'TxPowerLowWarnFlag{i}': i % 2 == 0 for i in range(1, 9)} + }, + { + 'rx_power_high_alarm': {f'RxPowerHighAlarmFlag{i}': i % 2 == 1 for i in range(1, 9)}, + 'rx_power_low_alarm': {f'RxPowerLowAlarmFlag{i}': i % 2 == 0 for i in range(1, 9)}, + 'rx_power_high_warn': {f'RxPowerHighWarnFlag{i}': i % 2 == 1 for i in range(1, 9)}, + 'rx_power_low_warn': {f'RxPowerLowWarnFlag{i}': i % 2 == 0 for i in range(1, 9)} + }, + { + 'tx_bias_high_alarm': {f'TxBiasHighAlarmFlag{i}': i % 2 == 1 for i in range(1, 9)}, + 'tx_bias_low_alarm': {f'TxBiasLowAlarmFlag{i}': i % 2 == 0 for i in range(1, 9)}, + 'tx_bias_high_warn': {f'TxBiasHighWarnFlag{i}': i % 2 == 1 for i in range(1, 9)}, + 'tx_bias_low_warn': {f'TxBiasLowWarnFlag{i}': i % 2 == 0 for i in range(1, 9)} + }, + (0, 1, 0), + { + 'temphighalarm': True, + 'templowalarm': False, + 'temphighwarning': True, + 'templowwarning': False, + 'vcchighalarm': True, + 'vcclowalarm': False, + 'vcchighwarning': True, + 'vcclowwarning': False, + 'txpowerhighalarm1': True, + 'txpowerlowalarm1': False, + 'txpowerhighwarning1': True, + 'txpowerlowwarning1': False, + 'txpowerhighalarm2': False, + 'txpowerlowalarm2': True, + 'txpowerhighwarning2': False, + 'txpowerlowwarning2': True, + 'txpowerhighalarm3': True, + 'txpowerlowalarm3': False, + 'txpowerhighwarning3': True, + 'txpowerlowwarning3': False, + 'txpowerhighalarm4': False, + 'txpowerlowalarm4': True, + 'txpowerhighwarning4': False, + 'txpowerlowwarning4': True, + 'txpowerhighalarm5': True, + 'txpowerlowalarm5': False, + 'txpowerhighwarning5': True, + 'txpowerlowwarning5': False, + 'txpowerhighalarm6': False, + 'txpowerlowalarm6': True, + 'txpowerhighwarning6': False, + 'txpowerlowwarning6': True, + 'txpowerhighalarm7': True, + 'txpowerlowalarm7': False, + 'txpowerhighwarning7': True, + 'txpowerlowwarning7': False, + 'txpowerhighalarm8': False, + 'txpowerlowalarm8': True, + 'txpowerhighwarning8': False, + 'txpowerlowwarning8': True, + 'rxpowerhighalarm1': True, + 'rxpowerlowalarm1': False, + 'rxpowerhighwarning1': True, + 'rxpowerlowwarning1': False, + 'rxpowerhighalarm2': False, + 'rxpowerlowalarm2': True, + 'rxpowerhighwarning2': False, + 'rxpowerlowwarning2': True, + 'rxpowerhighalarm3': True, + 'rxpowerlowalarm3': False, + 'rxpowerhighwarning3': True, + 'rxpowerlowwarning3': False, + 'rxpowerhighalarm4': False, + 'rxpowerlowalarm4': True, + 'rxpowerhighwarning4': False, + 'rxpowerlowwarning4': True, + 'rxpowerhighalarm5': True, + 'rxpowerlowalarm5': False, + 'rxpowerhighwarning5': True, + 'rxpowerlowwarning5': False, + 'rxpowerhighalarm6': False, + 'rxpowerlowalarm6': True, + 'rxpowerhighwarning6': False, + 'rxpowerlowwarning6': True, + 'rxpowerhighalarm7': True, + 'rxpowerlowalarm7': False, + 'rxpowerhighwarning7': True, + 'rxpowerlowwarning7': False, + 'rxpowerhighalarm8': False, + 'rxpowerlowalarm8': True, + 'rxpowerhighwarning8': False, + 'rxpowerlowwarning8': True, + 'txbiashighalarm1': True, + 'txbiaslowalarm1': False, + 'txbiashighwarning1': True, + 'txbiaslowwarning1': False, + 'txbiashighalarm2': False, + 'txbiaslowalarm2': True, + 'txbiashighwarning2': False, + 'txbiaslowwarning2': True, + 'txbiashighalarm3': True, + 'txbiaslowalarm3': False, + 'txbiashighwarning3': True, + 'txbiaslowwarning3': False, + 'txbiashighalarm4': False, + 'txbiaslowalarm4': True, + 'txbiashighwarning4': False, + 'txbiaslowwarning4': True, + 'txbiashighalarm5': True, + 'txbiaslowalarm5': False, + 'txbiashighwarning5': True, + 'txbiaslowwarning5': False, + 'txbiashighalarm6': False, + 'txbiaslowalarm6': True, + 'txbiashighwarning6': False, + 'txbiaslowwarning6': True, + 'txbiashighalarm7': True, + 'txbiaslowalarm7': False, + 'txbiashighwarning7': True, + 'txbiaslowwarning7': False, + 'txbiashighalarm8': False, + 'txbiaslowalarm8': True, + 'txbiashighwarning8': False, + 'txbiaslowwarning8': True, + 'lasertemphighalarm': True, + 'lasertemplowalarm': False, + 'lasertemphighwarning': True, + 'lasertemplowwarning': False + } + ), + ] + ) + def test_get_transceiver_dom_flags(self, module_flag, tx_power_flag_dict, rx_power_flag_dict, tx_bias_flag_dict, aux_mon_types, expected_result): + self.api.get_module_level_flag = MagicMock(return_value=module_flag) + self.api.get_tx_power_flag = MagicMock(return_value=tx_power_flag_dict) + self.api.get_rx_power_flag = MagicMock(return_value=rx_power_flag_dict) + self.api.get_tx_bias_flag = MagicMock(return_value=tx_bias_flag_dict) + self.api.get_aux_mon_type = MagicMock(return_value=aux_mon_types) + + result = self.api.get_transceiver_dom_flags() + assert result == expected_result + @pytest.mark.parametrize("mock_response, expected",[ ( [ @@ -2159,6 +2343,92 @@ def test_get_transceiver_status(self, mock_response, expected): result = self.api.get_transceiver_status() assert result == expected + @pytest.mark.parametrize( + "module_faults, tx_fault, tx_los, tx_cdr_lol, tx_eq_fault, rx_los, rx_cdr_lol, laser_tuning_summary, expected_result", + [ + # Test case 1: All flags present for lanes 1 to 8 + ( + (True, False, True), + [True, False, True, False, True, False, True, False], + [False, True, False, True, False, True, False, True], + [True, False, True, False, True, False, True, False], + [False, True, False, True, False, True, False, True], + [True, False, True, False, True, False, True, False], + [False, True, False, True, False, True, False, True], + ['TargetOutputPowerOOR', 'FineTuningOutOfRange', 'TuningNotAccepted', 'InvalidChannel', 'TuningComplete'], + { + 'datapath_firmware_fault': True, + 'module_firmware_fault': False, + 'module_state_changed': True, + 'txfault1': True, + 'txfault2': False, + 'txfault3': True, + 'txfault4': False, + 'txfault5': True, + 'txfault6': False, + 'txfault7': True, + 'txfault8': False, + 'txlos_hostlane1': False, + 'txlos_hostlane2': True, + 'txlos_hostlane3': False, + 'txlos_hostlane4': True, + 'txlos_hostlane5': False, + 'txlos_hostlane6': True, + 'txlos_hostlane7': False, + 'txlos_hostlane8': True, + 'txcdrlol_hostlane1': True, + 'txcdrlol_hostlane2': False, + 'txcdrlol_hostlane3': True, + 'txcdrlol_hostlane4': False, + 'txcdrlol_hostlane5': True, + 'txcdrlol_hostlane6': False, + 'txcdrlol_hostlane7': True, + 'txcdrlol_hostlane8': False, + 'tx_eq_fault1': False, + 'tx_eq_fault2': True, + 'tx_eq_fault3': False, + 'tx_eq_fault4': True, + 'tx_eq_fault5': False, + 'tx_eq_fault6': True, + 'tx_eq_fault7': False, + 'tx_eq_fault8': True, + 'rxlos1': True, + 'rxlos2': False, + 'rxlos3': True, + 'rxlos4': False, + 'rxlos5': True, + 'rxlos6': False, + 'rxlos7': True, + 'rxlos8': False, + 'rxcdrlol1': False, + 'rxcdrlol2': True, + 'rxcdrlol3': False, + 'rxcdrlol4': True, + 'rxcdrlol5': False, + 'rxcdrlol6': True, + 'rxcdrlol7': False, + 'rxcdrlol8': True, + 'target_output_power_oor': True, + 'fine_tuning_oor': True, + 'tuning_not_accepted': True, + 'invalid_channel_num': True, + 'tuning_complete': True + } + ), + ] + ) + def test_get_transceiver_status_flags(self, module_faults, tx_fault, tx_los, tx_cdr_lol, tx_eq_fault, rx_los, rx_cdr_lol, laser_tuning_summary, expected_result): + self.api.get_module_firmware_fault_state_changed = MagicMock(return_value=module_faults) + self.api.get_tx_fault = MagicMock(return_value=tx_fault) + self.api.get_tx_los = MagicMock(return_value=tx_los) + self.api.get_tx_cdr_lol = MagicMock(return_value=tx_cdr_lol) + self.api.get_rx_los = MagicMock(return_value=rx_los) + self.api.get_rx_cdr_lol = MagicMock(return_value=rx_cdr_lol) + self.api.get_laser_tuning_summary = MagicMock(return_value=laser_tuning_summary) + with patch.object(self.api, 'get_tx_adaptive_eq_fail_flag', return_value=tx_eq_fault): + result = self.api.get_transceiver_status_flags() + assert result == expected_result + @pytest.mark.parametrize("mock_response, expected",[ ( [ @@ -2293,6 +2563,138 @@ def test_get_transceiver_loopback(self, mock_response, expected): result = self.api.get_transceiver_loopback() assert result == expected + def generate_vdm_real_value_expected_dict(base_dict): + default_dict = dict() + for _, db_prefix_key_map in CMIS_VDM_KEY_TO_DB_PREFIX_KEY_MAP.items(): + default_dict.update({f'{db_prefix_key_map}{i}': 'N/A' for i in range(1, 9)}) + + default_dict.update(base_dict) + return default_dict + @pytest.mark.parametrize( + "vdm_raw_dict, expected_result", + [ + # Test case 1: VDM descriptor partially advertised + ( + { + 'Laser Temperature [C]' : { + 1: [10.0, None, None, None, None, None, None, None, None], + 2: [20.0, None, None, None, None, None, None, None, None], + 3: [30.0, None, None, None, None, None, None, None, None], + 4: [40.0, None, None, None, None, None, None, None, None], + 5: [50.0, None, None, None, None, None, None, None, None], + 6: [60.0, None, None, None, None, None, None, None, None], + 7: [70.0, None, None, None, None, None, None, None, None], + 8: [80.0, None, None, None, None, None, None, None, None]}, + 'eSNR Media Input [dB]' : {1: [22.94921875, None, None, None, None, None, None, None, None]} + }, + generate_vdm_real_value_expected_dict( + { + 'laser_temperature_media1': 10.0, + 'laser_temperature_media2': 20.0, + 'laser_temperature_media3': 30.0, + 'laser_temperature_media4': 40.0, + 'laser_temperature_media5': 50.0, + 'laser_temperature_media6': 60.0, + 'laser_temperature_media7': 70.0, + 'laser_temperature_media8': 80.0, + 'esnr_media_input1' : 22.94921875, + } + ) + ), + ] + ) + def test_get_transceiver_vdm_real_value(self, vdm_raw_dict, expected_result): + self.api.vdm = MagicMock() + self.api.vdm.VDM_REAL_VALUE = MagicMock() + self.api.get_vdm = MagicMock(return_value=vdm_raw_dict) + + result = self.api.get_transceiver_vdm_real_value() + assert result == expected_result + + def generate_vdm_thrsholds_expected_dict(base_dict): + default_dict = dict() + for _, db_prefix_key_map in CMIS_VDM_KEY_TO_DB_PREFIX_KEY_MAP.items(): + for _, thrshold_type_str in THRESHOLD_TYPE_STR_MAP.items(): + default_dict.update({f'{db_prefix_key_map}_{thrshold_type_str}{i}': 'N/A' for i in range(1, 9)}) + + default_dict.update(base_dict) + return default_dict + + @pytest.mark.parametrize( + "vdm_raw_dict, expected_result", + [ + # Test case 1: VDM descriptor partially advertised + ( + { #Laser temperature media 1 is halarm + 'Laser Temperature [C]': { + 1: [None, 90.0, -5.0, 85.0, 0.0, None, None, None, None], + 2: [None, 90.0, -5.0, 85.0, 0.0, None, None, None, None], + 3: [None, 90.0, -5.0, 85.0, 0.0, None, None, None, None], + 4: [None, 90.0, -5.0, 85.0, 0.0, None, None, None, None], + 5: [None, 90.0, -5.0, 85.0, 0.0, None, None, None, None], + 6: [None, 90.0, -5.0, 85.0, 0.0, None, None, None, None], + 7: [None, 90.0, -5.0, 85.0, 0.0, None, None, None, None], + 8: [None, 90.0, -5.0, 85.0, 0.0, None, None, None, None], + }, + }, + generate_vdm_thrsholds_expected_dict( + { + **{f'laser_temperature_media_{alarm_type}{lane}': value for lane in range(1, 9) for alarm_type, value in zip(['halarm', 'hwarn', 'lwarn', 'lalarm'], [90.0, 85.0, 0.0, -5.0])} + } + ) + ), + ] + ) + def test_get_transceiver_vdm_thresholds(self, vdm_raw_dict, expected_result): + self.api.vdm = MagicMock() + self.api.vdm.VDM_THRESHOLD = MagicMock() + self.api.get_vdm = MagicMock(return_value=vdm_raw_dict) + + result = self.api.get_transceiver_vdm_thresholds() + assert result == expected_result + + def generate_vdm_flags_expected_dict(base_dict): + default_dict = dict() + for _, db_prefix_key_map in CMIS_VDM_KEY_TO_DB_PREFIX_KEY_MAP.items(): + for _, flag_type_str in FLAG_TYPE_STR_MAP.items(): + default_dict.update({f'{db_prefix_key_map}_{flag_type_str}{i}': 'N/A' for i in range(1, 9) for flag_type in ['halarm', 'hwarn', 'lwarn', 'lalarm']}) + + default_dict.update(base_dict) + return default_dict + @pytest.mark.parametrize( + "vdm_raw_dict, expected_result", + [ + # Test case 1: VDM descriptor partially advertised + ( + { #Laser temperature media 1 is halarm + 'Laser Temperature [C]': { + 1: [None, None, None, None, None, True, False, False, False], + 2: [None, None, None, None, None, False, False, False, False], + 3: [None, None, None, None, None, False, False, False, False], + 4: [None, None, None, None, None, False, False, False, False], + 5: [None, None, None, None, None, False, False, False, False], + 6: [None, None, None, None, None, False, False, False, False], + 7: [None, None, None, None, None, False, False, False, False], + 8: [None, None, None, None, None, False, False, False, False], + }, + }, + generate_vdm_flags_expected_dict( + { + **{f'laser_temperature_media_{alarm_type}{lane}': False for lane in range(1, 9) for alarm_type in ['halarm', 'hwarn', 'lwarn', 'lalarm']}, + 'laser_temperature_media_halarm1': True + } + ) + ), + ] + ) + def test_get_transceiver_vdm_flags(self, vdm_raw_dict, expected_result): + self.api.vdm = MagicMock() + self.api.vdm.VDM_FLAG = MagicMock() + self.api.get_vdm = MagicMock(return_value=vdm_raw_dict) + + result = self.api.get_transceiver_vdm_flags() + assert result == expected_result + def test_cable_len(self): cable_len_field = self.mem_map.get_field(consts.LENGTH_ASSEMBLY_FIELD) data = bytearray([0xFF]) @@ -2727,4 +3129,4 @@ def test_get_tx_adaptive_eq_fail_flag(self, mock_response, expected): self.api.xcvr_eeprom.read = MagicMock() self.api.xcvr_eeprom.read.return_value = mock_response[1] result = self.api.get_tx_adaptive_eq_fail_flag() - assert result == expected \ No newline at end of file + assert result == expected diff --git a/tests/sonic_xcvr/test_cmisVDM.py b/tests/sonic_xcvr/test_cmisVDM.py index 2ffd74509..f57bf67fe 100644 --- a/tests/sonic_xcvr/test_cmisVDM.py +++ b/tests/sonic_xcvr/test_cmisVDM.py @@ -115,7 +115,8 @@ def test_get_vdm_page_none_vdm_descriptor(self, input_param, mock_response, expe @pytest.mark.parametrize("mock_response, expected", [ ( [ # mock_response - 0, # vdm_page_supported_raw + 1, # vdm_pages_supported + 0, # vdm_groups_supported_raw ( # VDM_flag_page 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, @@ -140,15 +141,22 @@ def test_get_vdm_page_none_vdm_descriptor(self, input_param, mock_response, expe 'Pre-FEC BER Current Value Media Input': {1: [0, 0, 0, 0, 0, False, False, False, False]}, } - ) + ), + ([ + 0, + None, + None, + None + ], + None) ]) def test_get_vdm_allpage(self, mock_response, expected): self.api.xcvr_eeprom.read = MagicMock() - self.api.xcvr_eeprom.read.return_value = mock_response[0] + self.api.xcvr_eeprom.read.side_effect = (mock_response[0], mock_response[1]) self.api.xcvr_eeprom.read_raw = MagicMock() - self.api.xcvr_eeprom.read_raw.return_value = mock_response[1] + self.api.xcvr_eeprom.read_raw.return_value = mock_response[2] # input_param = [0x20, mock_response[1]] self.api.get_vdm_page = MagicMock() - self.api.get_vdm_page.side_effect = mock_response[2:] + self.api.get_vdm_page.side_effect = mock_response[3:] result = self.api.get_vdm_allpage() assert result == expected diff --git a/tests/sonic_xcvr/test_sfp_optoe_base.py b/tests/sonic_xcvr/test_sfp_optoe_base.py index 49b4faad5..2fdd0ef6d 100644 --- a/tests/sonic_xcvr/test_sfp_optoe_base.py +++ b/tests/sonic_xcvr/test_sfp_optoe_base.py @@ -21,61 +21,61 @@ class TestSfpOptoeBase(object): cmis_api = CmisApi(eeprom) @pytest.mark.parametrize("mock_response1, mock_response2, expected", [ - (0, ccmis_api, 0), - (1, ccmis_api, 1), + (0, cmis_api, 0), + (1, cmis_api, 1), (None, None, False), - (None, cmis_api, False), + (False, cmis_api, False), ]) def test_freeze_vdm_stats(self, mock_response1, mock_response2, expected): self.sfp_optoe_api.get_xcvr_api = MagicMock() self.sfp_optoe_api.get_xcvr_api.return_value = mock_response2 - self.ccmis_api.freeze_vdm_stats = MagicMock() - self.ccmis_api.freeze_vdm_stats.return_value = mock_response1 + self.cmis_api.freeze_vdm_stats = MagicMock() + self.cmis_api.freeze_vdm_stats.return_value = mock_response1 result = self.sfp_optoe_api.freeze_vdm_stats() assert result == expected @pytest.mark.parametrize("mock_response1, mock_response2, expected", [ - (0, ccmis_api, 0), - (1, ccmis_api, 1), + (0, cmis_api, 0), + (1, cmis_api, 1), (None, None, False), - (None, cmis_api, False), + (False, cmis_api, False), ]) def test_unfreeze_vdm_stats(self, mock_response1, mock_response2, expected): self.sfp_optoe_api.get_xcvr_api = MagicMock() self.sfp_optoe_api.get_xcvr_api.return_value = mock_response2 - self.ccmis_api.unfreeze_vdm_stats = MagicMock() - self.ccmis_api.unfreeze_vdm_stats.return_value = mock_response1 + self.cmis_api.unfreeze_vdm_stats = MagicMock() + self.cmis_api.unfreeze_vdm_stats.return_value = mock_response1 result = self.sfp_optoe_api.unfreeze_vdm_stats() assert result == expected @pytest.mark.parametrize("mock_response1, mock_response2, expected", [ - (0, ccmis_api, 0), - (1, ccmis_api, 1), + (0, cmis_api, 0), + (1, cmis_api, 1), (None, None, False), - (None, cmis_api, False), + (False, cmis_api, False), ]) def test_get_vdm_freeze_status(self, mock_response1, mock_response2, expected): self.sfp_optoe_api.get_xcvr_api = MagicMock() self.sfp_optoe_api.get_xcvr_api.return_value = mock_response2 - self.ccmis_api.get_vdm_freeze_status = MagicMock() - self.ccmis_api.get_vdm_freeze_status.return_value = mock_response1 + self.cmis_api.get_vdm_freeze_status = MagicMock() + self.cmis_api.get_vdm_freeze_status.return_value = mock_response1 result = self.sfp_optoe_api.get_vdm_freeze_status() assert result == expected @pytest.mark.parametrize("mock_response1, mock_response2, expected", [ - (0, ccmis_api, 0), - (1, ccmis_api, 1), + (0, cmis_api, 0), + (1, cmis_api, 1), (None, None, False), - (None, cmis_api, False), + (False, cmis_api, False), ]) def test_get_vdm_unfreeze_status(self, mock_response1, mock_response2, expected): self.sfp_optoe_api.get_xcvr_api = MagicMock() self.sfp_optoe_api.get_xcvr_api.return_value = mock_response2 - self.ccmis_api.get_vdm_unfreeze_status = MagicMock() - self.ccmis_api.get_vdm_unfreeze_status.return_value = mock_response1 + self.cmis_api.get_vdm_unfreeze_status = MagicMock() + self.cmis_api.get_vdm_unfreeze_status.return_value = mock_response1 result = self.sfp_optoe_api.get_vdm_unfreeze_status() assert result == expected