try: import time from sonic_sfp.sfputilbase import * import sys import os import io import string from ctypes import create_string_buffer from sonic_py_common.logger import Logger except ImportError as e: raise ImportError(str(e) + "- required module not found") SYSLOG_IDENTIFIER = "sfputil" logger = Logger(SYSLOG_IDENTIFIER) qfx5210_qsfp_cable_length_tup = ('Length(km)', 'Length OM3(2m)', 'Length OM2(m)', 'Length OM1(m)', 'Length Cable Assembly(m)') qfx5210_sfp_cable_length_tup = ('LengthSMFkm-UnitsOfKm', 'LengthSMF(UnitsOf100m)', 'Length50um(UnitsOf10m)', 'Length62.5um(UnitsOfm)', 'LengthCable(UnitsOfm)', 'LengthOM3(UnitsOf10m)') qfx5210_sfp_compliance_code_tup = ('10GEthernetComplianceCode', 'InfinibandComplianceCode', 'ESCONComplianceCodes', 'SONETComplianceCodes', 'EthernetComplianceCodes', 'FibreChannelLinkLength', 'FibreChannelTechnology', 'SFP+CableTechnology', 'FibreChannelTransmissionMedia', 'FibreChannelSpeed') qfx5210_qsfp_compliance_code_tup = ('10/40G Ethernet Compliance Code', 'SONET Compliance codes', 'SAS/SATA compliance codes', 'Gigabit Ethernet Compliant codes', 'Fibre Channel link length/Transmitter Technology', 'Fibre Channel transmission media', 'Fibre Channel Speed') class SfpUtil(SfpUtilBase): """Platform specific SfpUtill class""" _port_start = 0 _port_end = 63 ports_in_block = 64 cmd = '/var/run/sfppresence' _port_to_eeprom_mapping = {} port_to_i2c_mapping = { 60: 25, 61: 26, 62: 27, 63: 28, 54: 29, 55: 30, 52: 31, 53: 32, 8: 33, 9: 34, 10: 35, 11: 36, 0: 37, 1: 38, 2: 39, 3: 40, 5: 41, 4: 42, 7: 43, 6: 44, 12: 45, 13: 46, 14: 47, 15: 48, 16: 49, 17: 50, 18: 51, 19: 52, 24: 53, 25: 54, 26: 55, 27: 56, 28: 57, 29: 58, 30: 59, 31: 60, 20: 61, 21: 62, 22: 63, 23: 64, 40: 65, 41: 66, 42: 67, 43: 68, 32: 69, 33: 70, 34: 71, 35: 72, 44: 73, 45: 74, 46: 75, 47: 76, 36: 77, 37: 78, 38: 79, 39: 80, 56: 81, 57: 82, 58: 83, 59: 84, 48: 85, 49: 86, 50: 87, 51: 88, } # sys.path.append('/usr/local/bin') _qsfp_ports = list(range(0, ports_in_block + 1)) def __init__(self): eeprom_path = '/sys/bus/i2c/devices/{0}-0050/eeprom' for x in range(self._port_start, self._port_end + 1): port_eeprom_path = eeprom_path.format(self.port_to_i2c_mapping[x]) self._port_to_eeprom_mapping[x] = port_eeprom_path SfpUtilBase.__init__(self) def reset(self, port_num): # Check for invalid port_num if port_num < self._port_start or port_num > self._port_end: return False path = "/sys/bus/i2c/devices/19-0060/module_reset_{0}" port_ps = path.format(port_num+1) try: reg_file = open(port_ps, 'w') except IOError as e: print("Error: unable to open file: %s" % str(e)) return False # HW will clear reset after set. reg_file.seek(0) reg_file.write('1') reg_file.close() return True def get_presence(self, port_num): # Check for invalid port_num if port_num < self._port_start or port_num > self._port_end: return False path = "/sys/bus/i2c/devices/19-0060/module_present_{0}" port_ps = path.format(port_num + 1) try: reg_file = open(port_ps) except IOError as e: print("Error: unable to open file: %s" % str(e)) return False reg_value = reg_file.readline().rstrip() if reg_value == '1': return True return False @property def port_start(self): return self._port_start @property def port_end(self): return self._port_end @property def qsfp_ports(self): return list(range(0, self.ports_in_block + 1)) @property def port_to_eeprom_mapping(self): return self._port_to_eeprom_mapping # Writing to a file from a list def write_to_file(self, file_name, from_list): try: fp1 = open(file_name, 'w') except IOError as e: print("Error: unable to open file: %s" % str(e)) return False for i in from_list: fp1.write(i) fp1.write('\n') fp1.close() return True # Reading from a file to a list def read_from_file(self, file_name): try: fp = open(file_name, 'r') except IOError as e: print("Error: unable to open file: %s" % str(e)) return False to_list = fp.readlines() to_list = [x.rstrip() for x in to_list] fp.close() return to_list def sfp_detect(self): x = 0 ret_dict = {} defl_dict = {} current_sfp_values = [0] * 64 previous_sfp_values = [0] * 64 path = "/sys/bus/i2c/devices/19-0060/module_present_{0}" if not os.path.isfile(self.cmd): pass else: if (self.read_from_file(self.cmd) == False): return False, defl_dict else: previous_sfp_values = self.read_from_file(self.cmd) # Read the current values from sysfs for x in range(self._port_start, self._port_end + 1): try: new_path = path.format(x + 1) reg_file = open(new_path, 'r') except IOError as e: print("Error: unable to open file: %s" % str(e)) return False, defl_dict sfp_present = reg_file.readline().rstrip() reg_file.close() current_sfp_values[x] = sfp_present if (current_sfp_values[x] != previous_sfp_values[x]): ret_dict.update({x: current_sfp_values[x]}) if(self.write_to_file(self.cmd, current_sfp_values) == True): return True, ret_dict else: return False, defl_dict def get_transceiver_change_event(self, timeout=2000): time.sleep(3) return self.sfp_detect() def get_low_power_mode(self, port_num): # Check for invalid port_num if port_num < self._port_start or port_num > self._port_end: return False try: eeprom = None if not self.get_presence(port_num): return False eeprom = open(self.port_to_eeprom_mapping[port_num], "rb") eeprom.seek(93) lpmode = ord(eeprom.read(1)) if ((lpmode & 0x3) == 0x3): return True # Low Power Mode if "Power override" bit is 1 and "Power set" bit is 1 else: # High Power Mode if one of the following conditions is matched: # 1. "Power override" bit is 0 # 2. "Power override" bit is 1 and "Power set" bit is 0 return False except IOError as e: print("Error: unable to open file: %s" % str(e)) return False finally: if eeprom is not None: eeprom.close() time.sleep(0.01) def set_low_power_mode(self, port_num, lpmode): # Check for invalid port_num if port_num < self._port_start or port_num > self._port_end: return False try: eeprom = None if not self.get_presence(port_num): return False # Port is not present, unable to set the eeprom # Fill in write buffer regval = 0x3 if lpmode else 0x1 # 0x3:Low Power Mode, 0x1:High Power Mode buffer = create_string_buffer(1) buffer[0] = chr(regval) # Write to eeprom eeprom = open(self.port_to_eeprom_mapping[port_num], "r+b") eeprom.seek(93) eeprom.write(buffer[0]) return True except IOError as e: print("Error: unable to open file: %s" % str(e)) return False finally: if eeprom is not None: eeprom.close() time.sleep(0.01) # Read out SFP type, vendor name, PN, REV, SN from eeprom. def get_transceiver_info_dict(self, port_num): transceiver_info_dict = {} compliance_code_dict = {} # ToDo: OSFP tranceiver info parsing not fully supported. # in inf8628.py lack of some memory map definition # will be implemented when the inf8628 memory map ready if port_num in self.osfp_ports: offset = 0 vendor_rev_width = XCVR_HW_REV_WIDTH_OSFP sfpi_obj = inf8628InterfaceId() if sfpi_obj is None: print("Error: sfp_object open failed") return None file_path = self._get_port_eeprom_path(port_num, self.IDENTITY_EEPROM_ADDR) if not self._sfp_eeprom_present(file_path, 0): print("Error, file not exist %s" % file_path) return None try: sysfsfile_eeprom = open(file_path, mode="rb", buffering=0) except IOError: print("Error: reading sysfs file %s" % file_path) return None sfp_type_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + OSFP_TYPE_OFFSET), XCVR_TYPE_WIDTH) if sfp_type_raw is not None: sfp_type_data = sfpi_obj.parse_sfp_type(sfp_type_raw, 0) else: return None sfp_vendor_name_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + OSFP_VENDOR_NAME_OFFSET), XCVR_VENDOR_NAME_WIDTH) if sfp_vendor_name_raw is not None: sfp_vendor_name_data = sfpi_obj.parse_vendor_name(sfp_vendor_name_raw, 0) else: return None sfp_vendor_pn_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + OSFP_VENDOR_PN_OFFSET), XCVR_VENDOR_PN_WIDTH) if sfp_vendor_pn_raw is not None: sfp_vendor_pn_data = sfpi_obj.parse_vendor_pn(sfp_vendor_pn_raw, 0) else: return None sfp_vendor_rev_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + OSFP_HW_REV_OFFSET), vendor_rev_width) if sfp_vendor_rev_raw is not None: sfp_vendor_rev_data = sfpi_obj.parse_vendor_rev(sfp_vendor_rev_raw, 0) else: return None sfp_vendor_sn_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + OSFP_VENDOR_SN_OFFSET), XCVR_VENDOR_SN_WIDTH) if sfp_vendor_sn_raw is not None: sfp_vendor_sn_data = sfpi_obj.parse_vendor_sn(sfp_vendor_sn_raw, 0) else: return None try: sysfsfile_eeprom.close() except IOError: print("Error: closing sysfs file %s" % file_path) return None transceiver_info_dict['type'] = sfp_type_data['data']['type']['value'] transceiver_info_dict['type_abbrv_name'] = sfp_type_data['data']['type_abbrv_name']['value'] transceiver_info_dict['manufacturer'] = sfp_vendor_name_data['data']['Vendor Name']['value'] transceiver_info_dict['model'] = sfp_vendor_pn_data['data']['Vendor PN']['value'] transceiver_info_dict['hardware_rev'] = sfp_vendor_rev_data['data']['Vendor Rev']['value'] transceiver_info_dict['serial'] = sfp_vendor_sn_data['data']['Vendor SN']['value'] # Below part is added to avoid fail the xcvrd, shall be implemented later transceiver_info_dict['vendor_oui'] = 'N/A' transceiver_info_dict['vendor_date'] = 'N/A' transceiver_info_dict['connector'] = 'N/A' transceiver_info_dict['encoding'] = 'N/A' transceiver_info_dict['ext_identifier'] = 'N/A' transceiver_info_dict['ext_rateselect_compliance'] = 'N/A' transceiver_info_dict['cable_type'] = 'N/A' transceiver_info_dict['cable_length'] = 'N/A' transceiver_info_dict['specification_compliance'] = 'N/A' transceiver_info_dict['nominal_bit_rate'] = 'N/A' else: if port_num in self.qsfp_ports: offset = 128 vendor_rev_width = XCVR_HW_REV_WIDTH_QSFP cable_length_width = XCVR_CABLE_LENGTH_WIDTH_QSFP interface_info_bulk_width = XCVR_INTFACE_BULK_WIDTH_QSFP sfp_type = 'QSFP' sfpi_obj = sff8436InterfaceId() if sfpi_obj is None: print("Error: sfp_object open failed") return None else: offset = 0 vendor_rev_width = XCVR_HW_REV_WIDTH_SFP cable_length_width = XCVR_CABLE_LENGTH_WIDTH_SFP interface_info_bulk_width = XCVR_INTFACE_BULK_WIDTH_SFP sfp_type = 'SFP' sfpi_obj = sff8472InterfaceId() if sfpi_obj is None: print("Error: sfp_object open failed") return None file_path = self._get_port_eeprom_path(port_num, self.IDENTITY_EEPROM_ADDR) if not self._sfp_eeprom_present(file_path, 0): print("Error, file not exist %s" % file_path) return None try: sysfsfile_eeprom = open(file_path, mode="rb", buffering=0) except IOError: print("Error: reading sysfs file %s" % file_path) return None sfp_interface_bulk_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + XCVR_INTFACE_BULK_OFFSET), interface_info_bulk_width) if sfp_interface_bulk_raw is not None: sfp_interface_bulk_data = sfpi_obj.parse_sfp_info_bulk(sfp_interface_bulk_raw, 0) else: return None sfp_vendor_name_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + XCVR_VENDOR_NAME_OFFSET), XCVR_VENDOR_NAME_WIDTH) if sfp_vendor_name_raw is not None: sfp_vendor_name_data = sfpi_obj.parse_vendor_name(sfp_vendor_name_raw, 0) else: return None sfp_vendor_pn_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + XCVR_VENDOR_PN_OFFSET), XCVR_VENDOR_PN_WIDTH) if sfp_vendor_pn_raw is not None: sfp_vendor_pn_data = sfpi_obj.parse_vendor_pn(sfp_vendor_pn_raw, 0) else: return None sfp_vendor_rev_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + XCVR_HW_REV_OFFSET), vendor_rev_width) if sfp_vendor_rev_raw is not None: sfp_vendor_rev_data = sfpi_obj.parse_vendor_rev(sfp_vendor_rev_raw, 0) else: return None sfp_vendor_sn_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + XCVR_VENDOR_SN_OFFSET), XCVR_VENDOR_SN_WIDTH) if sfp_vendor_sn_raw is not None: sfp_vendor_sn_data = sfpi_obj.parse_vendor_sn(sfp_vendor_sn_raw, 0) else: return None sfp_vendor_oui_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + XCVR_VENDOR_OUI_OFFSET), XCVR_VENDOR_OUI_WIDTH) if sfp_vendor_oui_raw is not None: sfp_vendor_oui_data = sfpi_obj.parse_vendor_oui(sfp_vendor_oui_raw, 0) else: return None sfp_vendor_date_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + XCVR_VENDOR_DATE_OFFSET), XCVR_VENDOR_DATE_WIDTH) if sfp_vendor_date_raw is not None: sfp_vendor_date_data = sfpi_obj.parse_vendor_date(sfp_vendor_date_raw, 0) else: return None try: sysfsfile_eeprom.close() except IOError: print("Error: closing sysfs file %s" % file_path) return None transceiver_info_dict['type'] = sfp_interface_bulk_data['data']['type']['value'] transceiver_info_dict['type_abbrv_name'] = sfp_interface_bulk_data['data']['type_abbrv_name']['value'] transceiver_info_dict['manufacturer'] = sfp_vendor_name_data['data']['Vendor Name']['value'] transceiver_info_dict['model'] = sfp_vendor_pn_data['data']['Vendor PN']['value'] transceiver_info_dict['hardware_rev'] = sfp_vendor_rev_data['data']['Vendor Rev']['value'] transceiver_info_dict['serial'] = sfp_vendor_sn_data['data']['Vendor SN']['value'] transceiver_info_dict['vendor_oui'] = sfp_vendor_oui_data['data']['Vendor OUI']['value'] transceiver_info_dict['vendor_date'] = sfp_vendor_date_data[ 'data']['VendorDataCode(YYYY-MM-DD Lot)']['value'] transceiver_info_dict['connector'] = sfp_interface_bulk_data['data']['Connector']['value'] transceiver_info_dict['encoding'] = sfp_interface_bulk_data['data']['EncodingCodes']['value'] transceiver_info_dict['ext_identifier'] = sfp_interface_bulk_data['data']['Extended Identifier']['value'] transceiver_info_dict['ext_rateselect_compliance'] = sfp_interface_bulk_data['data']['RateIdentifier']['value'] if sfp_type == 'QSFP': for key in qfx5210_qsfp_cable_length_tup: if key in sfp_interface_bulk_data['data']: transceiver_info_dict['cable_type'] = key transceiver_info_dict['cable_length'] = str(sfp_interface_bulk_data['data'][key]['value']) break else: transceiver_info_dict['cable_type'] = key transceiver_info_dict['cable_length'] = 'N/A' for key in qfx5210_qsfp_compliance_code_tup: if key in sfp_interface_bulk_data['data']['Specification compliance']['value']: compliance_code_dict[key] = sfp_interface_bulk_data['data']['Specification compliance']['value'][key]['value'] transceiver_info_dict['specification_compliance'] = str(compliance_code_dict) if 'Nominal Bit Rate(100Mbs)' in sfp_interface_bulk_data['data']: transceiver_info_dict['nominal_bit_rate'] = str( sfp_interface_bulk_data['data']['Nominal Bit Rate(100Mbs)']['value']) else: transceiver_info_dict['nominal_bit_rate'] = 'N/A' else: for key in qfx5210_sfp_cable_length_tup: if key in sfp_interface_bulk_data['data']: transceiver_info_dict['cable_type'] = key transceiver_info_dict['cable_length'] = str(sfp_interface_bulk_data['data'][key]['value']) else: transceiver_info_dict['cable_type'] = key transceiver_info_dict['cable_length'] = 'N/A' for key in qfx5210_sfp_compliance_code_tup: if key in sfp_interface_bulk_data['data']['Specification compliance']['value']: compliance_code_dict[key] = sfp_interface_bulk_data['data']['Specification compliance']['value'][key]['value'] transceiver_info_dict['specification_compliance'] = str(compliance_code_dict) if 'NominalSignallingRate(UnitsOf100Mbd)' in sfp_interface_bulk_data['data']: transceiver_info_dict['nominal_bit_rate'] = str( sfp_interface_bulk_data['data']['NominalSignallingRate(UnitsOf100Mbd)']['value']) else: transceiver_info_dict['nominal_bit_rate'] = 'N/A' #transceiver_info_dict['nominal_bit_rate'] = str(sfp_interface_bulk_data['data']['NominalSignallingRate(UnitsOf100Mbd)']['value']) return transceiver_info_dict def get_transceiver_dom_info_dict(self, port_num): transceiver_dom_info_dict = {} if port_num in self.osfp_ports: # Below part is added to avoid fail xcvrd, shall be implemented later transceiver_dom_info_dict['temperature'] = 'N/A' transceiver_dom_info_dict['voltage'] = 'N/A' transceiver_dom_info_dict['rx1power'] = 'N/A' transceiver_dom_info_dict['rx2power'] = 'N/A' transceiver_dom_info_dict['rx3power'] = 'N/A' transceiver_dom_info_dict['rx4power'] = 'N/A' transceiver_dom_info_dict['tx1bias'] = 'N/A' transceiver_dom_info_dict['tx2bias'] = 'N/A' transceiver_dom_info_dict['tx3bias'] = 'N/A' transceiver_dom_info_dict['tx4bias'] = 'N/A' transceiver_dom_info_dict['tx1power'] = 'N/A' transceiver_dom_info_dict['tx2power'] = 'N/A' transceiver_dom_info_dict['tx3power'] = 'N/A' transceiver_dom_info_dict['tx4power'] = 'N/A' elif port_num in self.qsfp_ports: offset = 0 offset_xcvr = 128 file_path = self._get_port_eeprom_path(port_num, self.IDENTITY_EEPROM_ADDR) if not self._sfp_eeprom_present(file_path, 0): return None try: sysfsfile_eeprom = open(file_path, mode="rb", buffering=0) except IOError: print("Error: reading sysfs file %s" % file_path) return None sfpd_obj = sff8436Dom() if sfpd_obj is None: return None sfpi_obj = sff8436InterfaceId() if sfpi_obj is None: return None # QSFP capability byte parse, through this byte can know whether it support tx_power or not. # TODO: in the future when decided to migrate to support SFF-8636 instead of SFF-8436, # need to add more code for determining the capability and version compliance # in SFF-8636 dom capability definitions evolving with the versions. qsfp_dom_capability_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset_xcvr + XCVR_DOM_CAPABILITY_OFFSET), XCVR_DOM_CAPABILITY_WIDTH) if qsfp_dom_capability_raw is not None: qspf_dom_capability_data = sfpi_obj.parse_dom_capability(qsfp_dom_capability_raw, 0) else: return None dom_temperature_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + QSFP_TEMPE_OFFSET), QSFP_TEMPE_WIDTH) if dom_temperature_raw is not None: dom_temperature_data = sfpd_obj.parse_temperature(dom_temperature_raw, 0) else: return None dom_voltage_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + QSFP_VOLT_OFFSET), QSFP_VOLT_WIDTH) if dom_voltage_raw is not None: dom_voltage_data = sfpd_obj.parse_voltage(dom_voltage_raw, 0) else: return None qsfp_dom_rev_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + QSFP_DOM_REV_OFFSET), QSFP_DOM_REV_WIDTH) if qsfp_dom_rev_raw is not None: qsfp_dom_rev_data = sfpd_obj.parse_sfp_dom_rev(qsfp_dom_rev_raw, 0) else: return None transceiver_dom_info_dict['temperature'] = dom_temperature_data['data']['Temperature']['value'] transceiver_dom_info_dict['voltage'] = dom_voltage_data['data']['Vcc']['value'] # The tx_power monitoring is only available on QSFP which compliant with SFF-8636 # and claimed that it support tx_power with one indicator bit. dom_channel_monitor_data = {} qsfp_dom_rev = qsfp_dom_rev_data['data']['dom_rev']['value'] qsfp_tx_power_support = qspf_dom_capability_data['data']['Tx_power_support']['value'] if (qsfp_dom_rev[0:8] != 'SFF-8636' or (qsfp_dom_rev[0:8] == 'SFF-8636' and qsfp_tx_power_support != 'on')): dom_channel_monitor_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + QSFP_CHANNL_MON_OFFSET), QSFP_CHANNL_MON_WIDTH) if dom_channel_monitor_raw is not None: dom_channel_monitor_data = sfpd_obj.parse_channel_monitor_params(dom_channel_monitor_raw, 0) else: return None transceiver_dom_info_dict['tx1power'] = 'N/A' transceiver_dom_info_dict['tx2power'] = 'N/A' transceiver_dom_info_dict['tx3power'] = 'N/A' transceiver_dom_info_dict['tx4power'] = 'N/A' else: dom_channel_monitor_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + QSFP_CHANNL_MON_OFFSET), QSFP_CHANNL_MON_WITH_TX_POWER_WIDTH) if dom_channel_monitor_raw is not None: dom_channel_monitor_data = sfpd_obj.parse_channel_monitor_params_with_tx_power( dom_channel_monitor_raw, 0) else: return None transceiver_dom_info_dict['tx1power'] = dom_channel_monitor_data['data']['TX1Power']['value'] transceiver_dom_info_dict['tx2power'] = dom_channel_monitor_data['data']['TX2Power']['value'] transceiver_dom_info_dict['tx3power'] = dom_channel_monitor_data['data']['TX3Power']['value'] transceiver_dom_info_dict['tx4power'] = dom_channel_monitor_data['data']['TX4Power']['value'] try: sysfsfile_eeprom.close() except IOError: print("Error: closing sysfs file %s" % file_path) return None transceiver_dom_info_dict['temperature'] = dom_temperature_data['data']['Temperature']['value'] transceiver_dom_info_dict['voltage'] = dom_voltage_data['data']['Vcc']['value'] transceiver_dom_info_dict['rx1power'] = dom_channel_monitor_data['data']['RX1Power']['value'] transceiver_dom_info_dict['rx2power'] = dom_channel_monitor_data['data']['RX2Power']['value'] transceiver_dom_info_dict['rx3power'] = dom_channel_monitor_data['data']['RX3Power']['value'] transceiver_dom_info_dict['rx4power'] = dom_channel_monitor_data['data']['RX4Power']['value'] transceiver_dom_info_dict['tx1bias'] = dom_channel_monitor_data['data']['TX1Bias']['value'] transceiver_dom_info_dict['tx2bias'] = dom_channel_monitor_data['data']['TX2Bias']['value'] transceiver_dom_info_dict['tx3bias'] = dom_channel_monitor_data['data']['TX3Bias']['value'] transceiver_dom_info_dict['tx4bias'] = dom_channel_monitor_data['data']['TX4Bias']['value'] else: offset = 256 file_path = self._get_port_eeprom_path(port_num, self.DOM_EEPROM_ADDR) if not self._sfp_eeprom_present(file_path, 0): return None try: sysfsfile_eeprom = open(file_path, mode="rb", buffering=0) except IOError: print("Error: reading sysfs file %s" % file_path) return None sfpd_obj = sff8472Dom() if sfpd_obj is None: return None dom_temperature_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + SFP_TEMPE_OFFSET), SFP_TEMPE_WIDTH) if dom_temperature_raw is not None: dom_temperature_data = sfpd_obj.parse_temperature(dom_temperature_raw, 0) else: return None dom_voltage_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + SFP_VOLT_OFFSET), SFP_VOLT_WIDTH) if dom_voltage_raw is not None: dom_voltage_data = sfpd_obj.parse_voltage(dom_voltage_raw, 0) else: return None dom_channel_monitor_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + SFP_CHANNL_MON_OFFSET), SFP_CHANNL_MON_WIDTH) if dom_channel_monitor_raw is not None: dom_channel_monitor_data = sfpd_obj.parse_channel_monitor_params(dom_channel_monitor_raw, 0) else: return None try: sysfsfile_eeprom.close() except IOError: print("Error: closing sysfs file %s" % file_path) return None transceiver_dom_info_dict['temperature'] = dom_temperature_data['data']['Temperature']['value'] transceiver_dom_info_dict['voltage'] = dom_voltage_data['data']['Vcc']['value'] transceiver_dom_info_dict['rx1power'] = dom_channel_monitor_data['data']['RXPower']['value'] transceiver_dom_info_dict['rx2power'] = 'N/A' transceiver_dom_info_dict['rx3power'] = 'N/A' transceiver_dom_info_dict['rx4power'] = 'N/A' transceiver_dom_info_dict['tx1bias'] = dom_channel_monitor_data['data']['TXBias']['value'] transceiver_dom_info_dict['tx2bias'] = 'N/A' transceiver_dom_info_dict['tx3bias'] = 'N/A' transceiver_dom_info_dict['tx4bias'] = 'N/A' transceiver_dom_info_dict['tx1power'] = dom_channel_monitor_data['data']['TXPower']['value'] transceiver_dom_info_dict['tx2power'] = 'N/A' transceiver_dom_info_dict['tx3power'] = 'N/A' transceiver_dom_info_dict['tx4power'] = 'N/A' return transceiver_dom_info_dict def get_transceiver_dom_threshold_info_dict(self, port_num): transceiver_dom_threshold_info_dict = {} dom_info_dict_keys = ['temphighalarm', 'temphighwarning', 'templowalarm', 'templowwarning', 'vcchighalarm', 'vcchighwarning', 'vcclowalarm', 'vcclowwarning', 'rxpowerhighalarm', 'rxpowerhighwarning', 'rxpowerlowalarm', 'rxpowerlowwarning', 'txpowerhighalarm', 'txpowerhighwarning', 'txpowerlowalarm', 'txpowerlowwarning', 'txbiashighalarm', 'txbiashighwarning', 'txbiaslowalarm', 'txbiaslowwarning' ] transceiver_dom_threshold_info_dict = dict.fromkeys(dom_info_dict_keys, 'N/A') logger.log_debug("QFX5210: get_transceiver_dom_threshold_info_dict Start") if port_num in self.qsfp_ports: file_path = self._get_port_eeprom_path(port_num, self.IDENTITY_EEPROM_ADDR) if not self._sfp_eeprom_present(file_path, 0): return None try: sysfsfile_eeprom = io.open(file_path, mode="rb", buffering=0) except IOError: print("Error: reading sysfs file %s" % file_path) return None sfpd_obj = sff8436Dom() if sfpd_obj is None: return transceiver_dom_threshold_info_dict # Dom Threshold data starts from offset 384 # Revert offset back to 0 once data is retrieved offset = 384 dom_module_threshold_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + QSFP_MODULE_THRESHOLD_OFFSET), QSFP_MODULE_THRESHOLD_WIDTH) if dom_module_threshold_raw is not None: dom_module_threshold_data = sfpd_obj.parse_module_threshold_values(dom_module_threshold_raw, 0) else: return transceiver_dom_threshold_info_dict dom_channel_threshold_raw = self._read_eeprom_specific_bytes( sysfsfile_eeprom, (offset + QSFP_CHANNL_THRESHOLD_OFFSET), QSFP_CHANNL_THRESHOLD_WIDTH) if dom_channel_threshold_raw is not None: dom_channel_threshold_data = sfpd_obj.parse_channel_threshold_values(dom_channel_threshold_raw, 0) else: return transceiver_dom_threshold_info_dict try: sysfsfile_eeprom.close() except IOError: print("Error: closing sysfs file %s" % file_path) return None # Threshold Data transceiver_dom_threshold_info_dict['temphighalarm'] = dom_module_threshold_data['data']['TempHighAlarm']['value'] transceiver_dom_threshold_info_dict['temphighwarning'] = dom_module_threshold_data['data']['TempHighWarning']['value'] transceiver_dom_threshold_info_dict['templowalarm'] = dom_module_threshold_data['data']['TempLowAlarm']['value'] transceiver_dom_threshold_info_dict['templowwarning'] = dom_module_threshold_data['data']['TempLowWarning']['value'] transceiver_dom_threshold_info_dict['vcchighalarm'] = dom_module_threshold_data['data']['VccHighAlarm']['value'] transceiver_dom_threshold_info_dict['vcchighwarning'] = dom_module_threshold_data['data']['VccHighWarning']['value'] transceiver_dom_threshold_info_dict['vcclowalarm'] = dom_module_threshold_data['data']['VccLowAlarm']['value'] transceiver_dom_threshold_info_dict['vcclowwarning'] = dom_module_threshold_data['data']['VccLowWarning']['value'] transceiver_dom_threshold_info_dict['rxpowerhighalarm'] = dom_channel_threshold_data['data']['RxPowerHighAlarm']['value'] transceiver_dom_threshold_info_dict['rxpowerhighwarning'] = dom_channel_threshold_data['data']['RxPowerHighWarning']['value'] transceiver_dom_threshold_info_dict['rxpowerlowalarm'] = dom_channel_threshold_data['data']['RxPowerLowAlarm']['value'] transceiver_dom_threshold_info_dict['rxpowerlowwarning'] = dom_channel_threshold_data['data']['RxPowerLowWarning']['value'] transceiver_dom_threshold_info_dict['txbiashighalarm'] = dom_channel_threshold_data['data']['TxBiasHighAlarm']['value'] transceiver_dom_threshold_info_dict['txbiashighwarning'] = dom_channel_threshold_data['data']['TxBiasHighWarning']['value'] transceiver_dom_threshold_info_dict['txbiaslowalarm'] = dom_channel_threshold_data['data']['TxBiasLowAlarm']['value'] transceiver_dom_threshold_info_dict['txbiaslowwarning'] = dom_channel_threshold_data['data']['TxBiasLowWarning']['value'] else: offset = 256 file_path = self._get_port_eeprom_path(port_num, self.DOM_EEPROM_ADDR) if not self._sfp_eeprom_present(file_path, 0): return None try: sysfsfile_eeprom = io.open(file_path, "rb", 0) except IOError: print("Error: reading sysfs file %s" % file_path) return None sfpd_obj = sff8472Dom(None, 1) if sfpd_obj is None: return transceiver_dom_threshold_info_dict dom_module_threshold_raw = self._read_eeprom_specific_bytes(sysfsfile_eeprom, (offset + SFP_MODULE_THRESHOLD_OFFSET), SFP_MODULE_THRESHOLD_WIDTH) if dom_module_threshold_raw is not None: dom_module_threshold_data = sfpd_obj.parse_alarm_warning_threshold(dom_module_threshold_raw, 0) else: return transceiver_dom_threshold_info_dict try: sysfsfile_eeprom.close() except IOError: print("Error: closing sysfs file %s" % file_path) return None # Threshold Data transceiver_dom_threshold_info_dict['temphighalarm'] = dom_module_threshold_data['data']['TempHighAlarm']['value'] transceiver_dom_threshold_info_dict['templowalarm'] = dom_module_threshold_data['data']['TempLowAlarm']['value'] transceiver_dom_threshold_info_dict['temphighwarning'] = dom_module_threshold_data['data']['TempHighWarning']['value'] transceiver_dom_threshold_info_dict['templowwarning'] = dom_module_threshold_data['data']['TempLowWarning']['value'] transceiver_dom_threshold_info_dict['vcchighalarm'] = dom_module_threshold_data['data']['VoltageHighAlarm']['value'] transceiver_dom_threshold_info_dict['vcclowalarm'] = dom_module_threshold_data['data']['VoltageLowAlarm']['value'] transceiver_dom_threshold_info_dict['vcchighwarning'] = dom_module_threshold_data['data']['VoltageHighWarning']['value'] transceiver_dom_threshold_info_dict['vcclowwarning'] = dom_module_threshold_data['data']['VoltageLowWarning']['value'] transceiver_dom_threshold_info_dict['txbiashighalarm'] = dom_module_threshold_data['data']['BiasHighAlarm']['value'] transceiver_dom_threshold_info_dict['txbiaslowalarm'] = dom_module_threshold_data['data']['BiasLowAlarm']['value'] transceiver_dom_threshold_info_dict['txbiashighwarning'] = dom_module_threshold_data['data']['BiasHighWarning']['value'] transceiver_dom_threshold_info_dict['txbiaslowwarning'] = dom_module_threshold_data['data']['BiasLowWarning']['value'] transceiver_dom_threshold_info_dict['txpowerhighalarm'] = dom_module_threshold_data['data']['TXPowerHighAlarm']['value'] transceiver_dom_threshold_info_dict['txpowerlowalarm'] = dom_module_threshold_data['data']['TXPowerLowAlarm']['value'] transceiver_dom_threshold_info_dict['txpowerhighwarning'] = dom_module_threshold_data['data']['TXPowerHighWarning']['value'] transceiver_dom_threshold_info_dict['txpowerlowwarning'] = dom_module_threshold_data['data']['TXPowerLowWarning']['value'] transceiver_dom_threshold_info_dict['rxpowerhighalarm'] = dom_module_threshold_data['data']['RXPowerHighAlarm']['value'] transceiver_dom_threshold_info_dict['rxpowerlowalarm'] = dom_module_threshold_data['data']['RXPowerLowAlarm']['value'] transceiver_dom_threshold_info_dict['rxpowerhighwarning'] = dom_module_threshold_data['data']['RXPowerHighWarning']['value'] transceiver_dom_threshold_info_dict['rxpowerlowwarning'] = dom_module_threshold_data['data']['RXPowerLowWarning']['value'] logger.log_debug("QFX5210: get_transceiver_dom_threshold_info_dict End") return transceiver_dom_threshold_info_dict