# sfputil.py # # Platform-specific SFP transceiver interface for SONiC # # For S5296F-ON, hardware version X01 try: import struct import time from sonic_sfp.sfputilbase import SfpUtilBase from os import * from mmap import * import io from sonic_sfp.sff8436 import sff8436InterfaceId from sonic_sfp.sff8436 import sff8436Dom from sonic_sfp.sff8472 import sff8472Dom except ImportError as e: raise ImportError("%s - required module not found" % str(e)) # definitions of the offset and width for values in DOM info eeprom QSFP_DOM_REV_OFFSET = 1 QSFP_DOM_REV_WIDTH = 1 QSFP_TEMPE_OFFSET = 22 QSFP_TEMPE_WIDTH = 2 QSFP_VOLT_OFFSET = 26 QSFP_VOLT_WIDTH = 2 QSFP_CHANNL_MON_OFFSET = 34 QSFP_CHANNL_MON_WIDTH = 16 QSFP_CHANNL_MON_WITH_TX_POWER_WIDTH = 24 QSFP_MODULE_THRESHOLD_OFFSET = 128 QSFP_MODULE_THRESHOLD_WIDTH = 24 QSFP_CHANNL_THRESHOLD_OFFSET = 176 QSFP_CHANNL_THRESHOLD_WIDTH = 16 QSFP_CHANNL_MON_MASK_OFFSET = 242 QSFP_CHANNL_MON_MASK_WIDTH = 4 SFP_TEMPE_OFFSET = 96 SFP_TEMPE_WIDTH = 2 SFP_VOLT_OFFSET = 98 SFP_VOLT_WIDTH = 2 SFP_MODULE_THRESHOLD_OFFSET = 0 SFP_MODULE_THRESHOLD_WIDTH = 56 XCVR_DOM_CAPABILITY_OFFSET = 92 XCVR_DOM_CAPABILITY_WIDTH = 1 class SfpUtil(SfpUtilBase): """Platform-specific SfpUtil class""" PORT_START = 1 PORT_END = 104 PORTS_IN_BLOCK = 104 BASE_RES_PATH = "/sys/bus/pci/devices/0000:04:00.0/resource0" _port_to_eeprom_mapping = {} _global_port_pres_dict = {} @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(97, self.PORTS_IN_BLOCK + 1)) @property def port_to_eeprom_mapping(self): return self._port_to_eeprom_mapping def pci_mem_read(self, mm, offset): mm.seek(offset) read_data_stream = mm.read(4) reg_val = struct.unpack('I', read_data_stream) mem_val = str(reg_val)[1:-2] # print "reg_val read:%x"%reg_val return mem_val def pci_mem_write(self, mm, offset, data): mm.seek(offset) # print "data to write:%x"%data mm.write(struct.pack('I', data)) def pci_set_value(self, resource, val, offset): fd = open(resource, O_RDWR) mm = mmap(fd, 0) val = self.pci_mem_write(mm, offset, val) mm.close() close(fd) return val def pci_get_value(self, resource, offset): fd = open(resource, O_RDWR) mm = mmap(fd, 0) val = self.pci_mem_read(mm, offset) mm.close() close(fd) return val def init_global_port_presence(self): for port_num in range(self.port_start, (self.port_end + 1)): presence = self.get_presence(port_num) if(presence): self._global_port_pres_dict[port_num] = '1' else: self._global_port_pres_dict[port_num] = '0' def __init__(self): eeprom_path = "/sys/class/i2c-adapter/i2c-{0}/{0}-0050/eeprom" for x in range(self.port_start, self.port_end + 1): self.port_to_eeprom_mapping[x] = eeprom_path.format(x + 1) self.init_global_port_presence() SfpUtilBase.__init__(self) 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 # Port offset starts with 0x4004 port_offset = 16388 + ((port_num-1) * 16) status = self.pci_get_value(self.BASE_RES_PATH, port_offset) reg_value = int(status) # Absence of status throws error if (reg_value == ""): return False # Mask off bit for presence mask = (1 << 1) if (port_num > 96): mask = (1 << 4) # ModPrsL is active low if reg_value & mask == 0: return True return False 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 # Port offset starts with 0x4000 port_offset = 16384 + ((port_num-1) * 16) status = self.pci_get_value(self.BASE_RES_PATH, port_offset) reg_value = int(status) # Absence of status throws error if (reg_value == ""): return False # Mask off 4th bit for presence mask = (1 << 6) # LPMode is active high if reg_value & mask == 0: return False return True 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 # Port offset starts with 0x4000 port_offset = 16384 + ((port_num-1) * 16) status = self.pci_get_value(self.BASE_RES_PATH, port_offset) reg_value = int(status) # Absence of status throws error if (reg_value == ""): return False # Mask off 4th bit for presence mask = (1 << 6) # LPMode is active high; set or clear the bit accordingly if lpmode is True: reg_value = reg_value | mask else: reg_value = reg_value & ~mask # Convert our register value back to a hex string and write back self.pci_set_value(self.BASE_RES_PATH, reg_value, port_offset) return True def reset(self, port_num): # Check for invalid port_num if port_num < self.port_start or port_num > self.port_end: return False # Port offset starts with 0x4000 port_offset = 16384 + ((port_num-1) * 16) status = self.pci_get_value(self.BASE_RES_PATH, port_offset) reg_value = int(status) # Absence of status throws error if (reg_value == ""): return False # Mask off 4th bit for presence mask = (1 << 6) # ResetL is active low reg_value = reg_value & ~mask # Convert our register value back to a hex string and write back self.pci_set_value(self.BASE_RES_PATH, reg_value, port_offset) # Sleep 1 second to allow it to settle time.sleep(1) reg_value = reg_value | mask # Convert our register value back to a hex string and write back self.pci_set_value(self.BASE_RES_PATH, reg_value, port_offset) return True def get_transceiver_change_event(self, timeout=0): port_dict = {} sleep_time_ms = 500 # Poll interval, in milliseconds sleep_time = sleep_time_ms / 1000.0 elapsed_time_ms = 0 while True: for port_num in range(self.port_start, (self.port_end + 1)): presence = self.get_presence(port_num) if(presence and self._global_port_pres_dict[port_num] == '0'): self._global_port_pres_dict[port_num] = '1' port_dict[port_num] = '1' elif(not presence and self._global_port_pres_dict[port_num] == '1'): self._global_port_pres_dict[port_num] = '0' port_dict[port_num] = '0' if(len(port_dict) > 0): break if len(port_dict) > 0: break if timeout != 0: elapsed_time_ms += sleep_time_ms if elapsed_time_ms > timeout: break time.sleep(sleep_time) return True, port_dict def get_transceiver_dom_info_dict(self, port_num): transceiver_dom_info_dict = {} dom_info_dict_keys = ['temperature', 'voltage', 'rx1power', 'rx2power', 'rx3power', 'rx4power', 'tx1bias', 'tx2bias', 'tx3bias', 'tx4bias', 'tx1power', 'tx2power', 'tx3power', 'tx4power', ] transceiver_dom_info_dict = dict.fromkeys(dom_info_dict_keys, 'N/A') if 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 = 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_info_dict sfpi_obj = sff8436InterfaceId() if sfpi_obj is None: return transceiver_dom_info_dict # 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_qsfp_dom_capability(qsfp_dom_capability_raw, 0) else: return transceiver_dom_info_dict 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 transceiver_dom_info_dict 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 transceiver_dom_info_dict 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 transceiver_dom_info_dict 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 transceiver_dom_info_dict 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' 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 = 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 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 transceiver_dom_info_dict 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 transceiver_dom_info_dict dom_channel_monitor_raw = self._read_eeprom_specific_bytes(sysfsfile_eeprom, (offset + SFP_MODULE_THRESHOLD_OFFSET), SFP_MODULE_THRESHOLD_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 transceiver_dom_info_dict 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') 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'] return transceiver_dom_threshold_info_dict