import os.path
import sys
sys.path.append('/usr/share/sonic/platform/plugins')
import pddfparse
import json
try:
import time
from ctypes import create_string_buffer
from sonic_sfp.sfputilbase import SfpUtilBase
except ImportError as e:
raise ImportError(str(e) + "- required module not found")
class SfpUtil(SfpUtilBase):
"""Platform generic PDDF SfpUtil class"""
_port_to_eeprom_mapping = {}
_port_start = 0
_port_end = 0
_port_to_type_mapping = {}
_qsfp_ports = []
_sfp_ports = []
def __init__(self):
SfpUtilBase.__init__(self)
global pddf_obj
global plugin_data
with open(os.path.join(os.path.dirname(os.path.realpath(__file__)) + '/../pddf/pd-plugin.json')) as pd:
plugin_data = json.load(pd)
pddf_obj = pddfparse.PddfParse()
self.platform = pddf_obj.get_platform()
self._port_start = 0
self._port_end = self.get_num_ports()
for port_num in range(self._port_start, self._port_end):
device = "PORT" + "%d" % (port_num+1)
port_eeprom_path = pddf_obj.get_path(device, "eeprom")
self._port_to_eeprom_mapping[port_num] = port_eeprom_path
port_type = pddf_obj.get_device_type(device)
self._port_to_type_mapping[port_num] = port_type
self.populate_port_type(port_num)
def get_num_ports(self):
return int(self.platform['num_ports'])
def is_valid_port(self, port_num):
if port_num < self._port_start or port_num > self._port_end:
return False
else:
return True
def get_presence(self, port_num):
if port_num < self._port_start or port_num > self._port_end:
return False
device = "PORT" + "%d" % (port_num+1)
output = pddf_obj.get_attr_name_output(device, 'xcvr_present')
if not output:
return False
#mode = output['mode']
modpres = output['status'].rstrip()
if 'XCVR' in plugin_data:
if 'xcvr_present' in plugin_data['XCVR']:
ptype = self._port_to_type_mapping[port_num]
vtype = 'valmap-'+ptype
if vtype in plugin_data['XCVR']['xcvr_present']:
vmap = plugin_data['XCVR']['xcvr_present'][vtype]
if modpres in vmap:
return vmap[modpres]
else:
return False
# if plugin_data doesn't specify anything regarding Transceivers
if modpres == '1':
return True
return False
def populate_port_type(self, port):
if self._port_to_type_mapping[port] == 'QSFP' or self._port_to_type_mapping[port] == 'QSFP28':
self._qsfp_ports.append(port)
elif self._port_to_type_mapping[port] == 'SFP' or self._port_to_type_mapping[port] == 'SFP28':
self._sfp_ports.append(port)
@property
def port_start(self):
return self._port_start
@property
def port_end(self):
return (self._port_end - 1)
@property
def port_to_eeprom_mapping(self):
return self._port_to_eeprom_mapping
@property
def qsfp_ports(self):
return self._qsfp_ports
def reset(self, port_num):
if port_num < self._port_start or port_num > self._port_end:
return False
device = "PORT" + "%d" % (port_num+1)
port_ps = pddf_obj.get_path(device, "xcvr_reset")
if port_ps is None:
return False
try:
reg_file = open(port_ps, 'w')
except IOError as e:
print("Error: unable to open file: %s" % str(e))
return False
try:
reg_file.seek(0)
reg_file.write('1')
time.sleep(1)
reg_file.seek(0)
reg_file.write('0')
reg_file.close()
return True
except IOError as e:
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
if not self.get_presence(port_num):
return False
device = "PORT" + "%d" % (port_num+1)
output = pddf_obj.get_attr_name_output(device, 'xcvr_lpmode')
if not output:
if port_num not in self.qsfp_ports:
return False # Read from eeprom only for QSFP ports
try:
eeprom = None
eeprom = open(self.port_to_eeprom_mapping[port_num], "rb")
# check for valid connector type
eeprom.seek(2)
ctype = eeprom.read(1)
if ctype in ['21', '23']:
return False
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)
else:
#mode = output['mode']
status = int(output['status'].rstrip())
if status == 1:
return True
else:
return False
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
if not self.get_presence(port_num):
return False # Port is not present, unable to set the eeprom
device = "PORT" + "%d" % (port_num+1)
port_ps = pddf_obj.get_path(device, "xcvr_lpmode")
if port_ps is None:
if port_num not in self.qsfp_ports:
return False # Write to eeprom only for QSFP ports
try:
eeprom = None
eeprom = open(self.port_to_eeprom_mapping[port_num], "r+b")
# check for valid connector type
eeprom.seek(2)
ctype = eeprom.read(1)
if ctype in ['21', '23']:
return False
# 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.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)
else:
try:
f = open(port_ps, 'w')
if lpmode:
f.write('1')
else:
f.write('0')
f.close()
return True
except IOError as e:
return False
def get_transceiver_change_event(self):
"""
TODO: This function need to be implemented
when decide to support monitoring SFP(Xcvrd)
on this platform.
"""
raise NotImplementedError
def dump_sysfs(self):
return pddf_obj.cli_dump_dsysfs('xcvr')