579 lines
17 KiB
Python
Executable File
579 lines
17 KiB
Python
Executable File
# Copyright (c) 2019 Edgecore Networks Corporation
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# THIS CODE IS PROVIDED ON AN *AS IS* BASIS, WITHOUT WARRANTIES OR
|
|
# CONDITIONS OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING WITHOUT
|
|
# LIMITATION ANY IMPLIED WARRANTIES OR CONDITIONS OF TITLE, FITNESS
|
|
# FOR A PARTICULAR PURPOSE, MERCHANTABLITY OR NON-INFRINGEMENT.
|
|
#
|
|
# See the Apache Version 2.0 License for specific language governing
|
|
# permissions and limitations under the License.
|
|
|
|
# ------------------------------------------------------------------
|
|
# HISTORY:
|
|
# mm/dd/yyyy (A.D.)
|
|
# 5/29/2019: Jostar create for minipack
|
|
# -----------------------------------------------------------
|
|
try:
|
|
import os
|
|
import sys, getopt
|
|
import subprocess
|
|
import click
|
|
import imp
|
|
import logging
|
|
import logging.config
|
|
import logging.handlers
|
|
import types
|
|
import time # this is only being used as part of the example
|
|
import traceback
|
|
from tabulate import tabulate
|
|
import fbfpgaio
|
|
import re
|
|
import time
|
|
from select import select
|
|
#from ctypes import fbfpgaio
|
|
|
|
except ImportError as e:
|
|
raise ImportError('%s - required module not found' % str(e))
|
|
|
|
|
|
# pimutil.py
|
|
#
|
|
# Platform-specific PIM interface for SONiC
|
|
#
|
|
|
|
iob = {
|
|
"revision": 0x0,
|
|
"scratchpad": 0x4,
|
|
"interrupt_status": 0x2C,
|
|
"pim_status": 0x40,
|
|
"pim_present_intr_mask": 0x44,
|
|
}
|
|
|
|
dom_base = [
|
|
0xFFFFFFFF, # Padding
|
|
0x40000,
|
|
0x48000,
|
|
0x50000,
|
|
0x58000,
|
|
0x60000,
|
|
0x68000,
|
|
0x70000,
|
|
0x78000,
|
|
]
|
|
|
|
|
|
dom = {
|
|
"revision": 0x0,
|
|
"system_led": 0xC,
|
|
"intr_status": 0x2C,
|
|
"qsfp_present": 0x48,
|
|
"qsfp_present_intr": 0x50,
|
|
"qsfp_present_intr_mask": 0x58,
|
|
"qsfp_intr": 0x60,
|
|
"qsfp_intr_mask": 0x68,
|
|
"qsfp_reset": 0x70,
|
|
"qsfp_lp_mode": 0x78,
|
|
"device_power_bad_status": 0x90,
|
|
"port_led_color_profile": {
|
|
0: 0x300,
|
|
1: 0x300,
|
|
2: 0x304,
|
|
3: 0x304,
|
|
4: 0x308,
|
|
5: 0x308,
|
|
6: 0x30C,
|
|
7: 0x30C,
|
|
},
|
|
"port_led_control": {
|
|
1: 0x310,
|
|
2: 0x314,
|
|
3: 0x318,
|
|
4: 0x31C,
|
|
5: 0x320,
|
|
6: 0x324,
|
|
7: 0x328,
|
|
8: 0x32C,
|
|
9: 0x330,
|
|
10: 0x334,
|
|
11: 0x338,
|
|
12: 0x33C,
|
|
13: 0x340,
|
|
14: 0x344,
|
|
15: 0x348,
|
|
16: 0x34C,
|
|
},
|
|
"dom_control_config": 0x410,
|
|
"dom_global_status": 0x414,
|
|
"dom_data": 0x4000,
|
|
|
|
|
|
"mdio": {
|
|
"config": 0x0200,
|
|
"command": 0x0204,
|
|
"write": 0x0208,
|
|
"read": 0x020C,
|
|
"status": 0x0210,
|
|
"intr_mask": 0x0214,
|
|
"source_sel": 0x0218,
|
|
}, # mdio
|
|
}
|
|
|
|
mdio_read_cmd = 0x1
|
|
mdio_write_cmd = 0x0
|
|
mdio_device_type = 0x1F
|
|
|
|
|
|
#fbfpgaio=cdll.LoadLibrary('./fbfpgaio.so')
|
|
|
|
def init_resources():
|
|
fbfpgaio.hw_init()
|
|
return
|
|
|
|
def release_resources():
|
|
fbfpgaio.hw_release()
|
|
return
|
|
|
|
def fpga_io(offset, data=None):
|
|
if data is None:
|
|
return fbfpgaio.hw_io(offset)
|
|
else:
|
|
fbfpgaio.hw_io(offset, data)
|
|
return
|
|
|
|
def pim_io(pim, offset, data=None):
|
|
global dom_base
|
|
target_offset = dom_base[pim]+offset
|
|
if data is None:
|
|
retval = fpga_io(target_offset)
|
|
#print ("0x%04X" % retval) # idebug
|
|
return retval
|
|
else:
|
|
retval = fpga_io(target_offset, data)
|
|
return retval
|
|
|
|
|
|
def show_pim_present():
|
|
pim_status = fpga_io(iob["pim_status"])
|
|
|
|
header = "PIM # "
|
|
status_str = " "
|
|
for shift in range(0,8):
|
|
status = pim_status & (0x10000 << shift) #[23:16] from pim_0 to pim_7
|
|
header += " %d " % (shift+1)
|
|
if status:
|
|
status_str += (" | ")
|
|
else:
|
|
status_str += (" X ")
|
|
print(header)
|
|
print(status_str)
|
|
|
|
def show_qsfp_present_status(pim_num):
|
|
status = fpga_io(dom_base[pim_num]+dom["qsfp_present"])
|
|
interrupt = fpga_io(dom_base[pim_num]+dom["qsfp_present_intr"])
|
|
mask = fpga_io(dom_base[pim_num]+dom["qsfp_present_intr_mask"])
|
|
|
|
print()
|
|
print(" (0x48) (0x50) (0x58)")
|
|
print((" 0x%08X 0x%08X 0x%08X" %(status, interrupt, mask)))
|
|
print(" Status Interrupt Mask")
|
|
for row in range(8):
|
|
output_str = str()
|
|
status_left = bool(status & (0x1 << row*2))
|
|
status_right = bool(status & (0x2 << row*2))
|
|
interrupt_left = bool(interrupt & (0x1 << row*2))
|
|
interrupt_right = bool(interrupt & (0x2 << row*2))
|
|
mask_left = bool(mask & (0x1 << row*2))
|
|
mask_right = bool(mask & (0x2 << row*2))
|
|
print(("%2d: %d %d %d %d %d %d" % \
|
|
(row*2+1, status_left, status_right, \
|
|
interrupt_left, interrupt_right, \
|
|
mask_left, mask_right)))
|
|
print()
|
|
|
|
|
|
|
|
#pim_index start from 0 to 7
|
|
#port_index start from 0 to 127. Each 16-port is to one pim card.
|
|
class PimUtil(object):
|
|
|
|
PORT_START = 0
|
|
PORT_END = 127
|
|
|
|
def __init__(self):
|
|
self.value=1
|
|
|
|
def __del__(self):
|
|
self.value=0
|
|
|
|
def init_pim_fpga(self):
|
|
init_resources()
|
|
|
|
def release_pim_fpga(self):
|
|
release_resources()
|
|
|
|
def get_pim_by_port(self, port_num):
|
|
if port_num < self.PORT_START or port_num > self.PORT_END:
|
|
return False
|
|
pim_num=port_num/16
|
|
return True, pim_num+1
|
|
|
|
def get_onepimport_by_port(self, port_num):
|
|
if port_num < self.PORT_START or port_num > self.PORT_END:
|
|
return False
|
|
if port_num < 16:
|
|
return True, port_num
|
|
else:
|
|
return True, port_num%16
|
|
|
|
def get_pim_presence(self, pim_num):
|
|
if pim_num <0 or pim_num > 7:
|
|
return 0
|
|
pim_status = fpga_io(iob["pim_status"])
|
|
status = pim_status & (0x10000 << pim_num)
|
|
if status:
|
|
return 1 #present
|
|
else:
|
|
return 0 #not present
|
|
|
|
#return code=0:100G. return code=1:400G
|
|
def get_pim_board_id(self, pim_num):
|
|
if pim_num <0 or pim_num > 7:
|
|
return False
|
|
board_id = fpga_io(dom_base[pim_num+1]+dom["revision"])
|
|
board_id = board_id & 0x1
|
|
if board_id==0x0:
|
|
return 0
|
|
else:
|
|
return 1
|
|
|
|
|
|
def get_pim_status(self, pim_num):
|
|
if pim_num <0 or pim_num > 7:
|
|
return 0xFF
|
|
power_status =0
|
|
#device_power_bad_status
|
|
status=fpga_io(dom_base[pim_num+1]+dom["device_power_bad_status"])
|
|
|
|
for x in range(0, 5):
|
|
if status & ( (0xf) << (4*x) ) :
|
|
power_status = power_status | (0x1 << x)
|
|
|
|
if ( status & 0x1000000):
|
|
power_status=power_status | (0x1 << 5)
|
|
if ( status & 0x2000000):
|
|
power_status=power_status | (0x1 << 6)
|
|
if ( status & 0x8000000):
|
|
power_status=power_status | (0x1 << 7)
|
|
if ( status & 0x10000000):
|
|
power_status=power_status | (0x1 << 8)
|
|
if ( status & 0x40000000):
|
|
power_status=power_status | (0x1 << 9)
|
|
if ( status & 0x80000000):
|
|
power_status=power_status | (0x1 << 10)
|
|
|
|
return power_status
|
|
#path=0:MDIO path is set on TH3. path=1:MDIO path is set on FPGA.
|
|
def set_pim_mdio_source_sel(self, pim_num, path):
|
|
if pim_num <0 or pim_num > 7:
|
|
return False
|
|
status= pim_io(pim_num+1, dom["mdio"]["source_sel"])
|
|
|
|
if path==1:
|
|
status = status | 0x2
|
|
else:
|
|
status = status & 0xfffffffd
|
|
|
|
pim_io(pim_num+1, dom["mdio"]["source_sel"], status)
|
|
return True
|
|
#retrun code=0, path is TH3. retrun code=1, path is FPGA
|
|
def get_pim_mdio_source_sel(sefl, pim_num):
|
|
if pim_num <0 or pim_num > 7:
|
|
return False
|
|
path= pim_io(pim_num+1, dom["mdio"]["source_sel"])
|
|
path = path & 0x2
|
|
if path:
|
|
return 1
|
|
else:
|
|
return 0
|
|
|
|
#This api will set mdio path to MAC side.(At default, mdio path is set to FPGA side).
|
|
def pim_init(self, pim_num):
|
|
if pim_num <0 or pim_num > 7:
|
|
return False
|
|
status=self.set_pim_mdio_source_sel(pim_num, 0)
|
|
#put init phy cmd here
|
|
|
|
|
|
#return code="pim_dict[pim_num]='1' ":insert evt. return code="pim_dict[pim_num]='0' ":remove evt
|
|
def get_pim_change_event(self, timeout=0):
|
|
start_time = time.time()
|
|
pim_dict = {}
|
|
forever = False
|
|
|
|
if timeout == 0:
|
|
forever = True
|
|
elif timeout > 0:
|
|
timeout = timeout / float(1000) # Convert to secs
|
|
else:
|
|
print("get_transceiver_change_event:Invalid timeout value", timeout)
|
|
return False, {}
|
|
|
|
end_time = start_time + timeout
|
|
if start_time > end_time:
|
|
print('get_transceiver_change_event:' \
|
|
'time wrap / invalid timeout value', timeout)
|
|
|
|
return False, {} # Time wrap or possibly incorrect timeout
|
|
|
|
pim_mask_status = fpga_io(iob["pim_present_intr_mask"], 0xffff00000)
|
|
|
|
while timeout >= 0:
|
|
new_pim_status=0
|
|
pim_status = fpga_io(iob["pim_status"])
|
|
present_status= pim_status & 0xff0000
|
|
change_status=pim_status & 0xff
|
|
interrupt_status = fpga_io(iob["interrupt_status"])
|
|
|
|
for pim_num in range(0,8):
|
|
if change_status & (0x1 << pim_num) :
|
|
status = present_status & (0x10000 << pim_num)
|
|
new_pim_status = new_pim_status | (0x1 << pim_num) #prepare to W1C to clear
|
|
if status:
|
|
pim_dict[pim_num]='1'
|
|
else:
|
|
pim_dict[pim_num]='0'
|
|
|
|
if change_status:
|
|
new_pim_status = pim_status | new_pim_status #Write one to clear interrupt bit
|
|
fpga_io(iob["pim_status"], new_pim_status)
|
|
return True, pim_dict
|
|
if forever:
|
|
time.sleep(1)
|
|
else:
|
|
timeout = end_time - time.time()
|
|
if timeout >= 1:
|
|
time.sleep(1) # We poll at 1 second granularity
|
|
else:
|
|
if timeout > 0:
|
|
time.sleep(timeout)
|
|
return True, {}
|
|
print("get_evt_change_event: Should not reach here.")
|
|
return False, {}
|
|
|
|
|
|
def get_pim_max_number(self):
|
|
return 8
|
|
|
|
#pim_num start from 0 to 7
|
|
#color:0=amber, 1=blue
|
|
#contrl:off(0),on(1), flash(2)
|
|
def set_pim_led(self, pim_num, color, control):
|
|
if pim_num <0 or pim_num > 7:
|
|
return False
|
|
|
|
led_val=fpga_io(dom_base[pim_num+1]+dom["system_led"])
|
|
|
|
if color==1:
|
|
led_val = led_val | (0x8000 | 0x4000) #blue
|
|
elif color==0:
|
|
led_val = (led_val & ( ~ 0x8000)) | 0x4000 #amber
|
|
else:
|
|
print("Set RGB control to Green1")
|
|
led_val = led_val & (~ 0x4000)
|
|
led_val = led_val & (~ 0xfff)
|
|
led_val = led_val | 0x0f0 #B.G.R Birghtness, set to Green
|
|
|
|
if control==0:
|
|
led_val = led_val & ( ~ 0x3000) #Off
|
|
elif control==1:
|
|
led_val = led_val & ( ~ 0x3000) #Off
|
|
led_val = led_val | 0x1000 #On
|
|
else:
|
|
led_val = led_val | 0x3000 #Flash
|
|
|
|
fpga_io(dom_base[pim_num+1]+dom["system_led"], led_val)
|
|
|
|
def get_qsfp_presence(self, port_num):
|
|
#xlate port to get pim_num
|
|
status, pim_num=self.get_pim_by_port(port_num)
|
|
|
|
if status==0:
|
|
return False
|
|
else:
|
|
present = fpga_io(dom_base[pim_num]+dom["qsfp_present"])
|
|
status, shift = self.get_onepimport_by_port(port_num)
|
|
if status==0:
|
|
return False
|
|
else:
|
|
if bool(present & (0x1 << shift)):
|
|
return 1 #present
|
|
else:
|
|
return 0 #not present
|
|
|
|
#return code: low_power(1) or high_power(0)
|
|
def get_low_power_mode(self, port_num):
|
|
status, pim_num=self.get_pim_by_port(port_num)
|
|
|
|
if status==0:
|
|
return False
|
|
else:
|
|
lp_mode = fpga_io(dom_base[pim_num]+dom["qsfp_lp_mode"])
|
|
|
|
status, shift=self.get_onepimport_by_port(port_num)
|
|
if status==0:
|
|
return False
|
|
else:
|
|
if (lp_mode & (0x1 << shift)):
|
|
return 1 #low
|
|
else:
|
|
return 0 #high
|
|
|
|
#lpmode=1 to hold QSFP in low power mode. lpmode=0 to release QSFP from low power mode.
|
|
def set_low_power_mode(self, port_num, mode):
|
|
status, pim_num=self.get_pim_by_port(port_num)
|
|
if status==0:
|
|
return False
|
|
val = fpga_io(dom_base[pim_num]+dom["qsfp_lp_mode"])
|
|
status, shift=self.get_onepimport_by_port(port_num)
|
|
if status==0:
|
|
return False
|
|
else:
|
|
if mode==0:
|
|
new_val = val & (~(0x1 << shift))
|
|
else:
|
|
new_val=val|(0x1 << shift)
|
|
status=fpga_io(dom_base[pim_num]+dom["qsfp_lp_mode"], new_val)
|
|
return status
|
|
|
|
#port_dict[idx]=1 means get interrupt(change evt), port_dict[idx]=0 means no get interrupt
|
|
def get_qsfp_interrupt(self):
|
|
port_dict={}
|
|
#show_qsfp_present_status(1)
|
|
for pim_num in range(0, 8):
|
|
fpga_io(dom_base[pim_num+1]+dom["qsfp_present_intr_mask"], 0xffff0000)
|
|
fpga_io(dom_base[pim_num+1]+dom["qsfp_intr_mask"], 0xffff0000)
|
|
for pim_num in range(0, 8):
|
|
clear_bit=0
|
|
qsfp_present_intr_status = fpga_io(dom_base[pim_num+1]+dom["qsfp_present_intr"])
|
|
interrupt_status = qsfp_present_intr_status & 0xffff
|
|
#time.sleep(2)
|
|
if interrupt_status:
|
|
for idx in range (0,16):
|
|
port_idx=idx + (pim_num*16)
|
|
if interrupt_status & (0x1<<idx):
|
|
port_dict[port_idx]=1
|
|
clear_bit=clear_bit | (0x1<<idx) #W1C to clear
|
|
else:
|
|
port_dict[port_idx]=0
|
|
|
|
#W1C to clear
|
|
fpga_io(dom_base[pim_num+1]+dom["qsfp_present_intr"], qsfp_present_intr_status | clear_bit)
|
|
|
|
return port_dict
|
|
|
|
def reset(self, port_num):
|
|
status, pim=self.get_pim_by_port(port_num)
|
|
if status==0:
|
|
return False
|
|
|
|
val=fpga_io(dom_base[pim]+dom["qsfp_reset"])
|
|
status, shift=self.get_onepimport_by_port(port_num)
|
|
if status==0:
|
|
return False
|
|
else:
|
|
val = val & (~(0x1 << shift))
|
|
fpga_io(dom_base[pim]+dom["qsfp_reset"], val)
|
|
return True
|
|
|
|
#color:white(0), blue(1),red(2), orange(3),green(4)
|
|
def set_port_led(self, port_num, color, control):
|
|
status, pim_num=self.get_pim_by_port(port_num)
|
|
if status==0:
|
|
return False
|
|
status, port=self.get_onepimport_by_port(port_num)
|
|
port=port+1
|
|
led_val=fpga_io(dom_base[pim_num] + dom["port_led_control"][port])
|
|
if control==0:
|
|
led_val = led_val & (~ 0x3) #Off
|
|
elif control==1:
|
|
led_val = led_val & (~ 0x3) #Off
|
|
led_val = led_val | 0x1 #On
|
|
else:
|
|
led_val = led_val | 0x3 #Flash
|
|
|
|
led_val=led_val & (~ 0x1C)
|
|
if color==0:
|
|
led_val=led_val & (~ 0x1C) #white
|
|
elif color==1:
|
|
led_val=led_val | 0x8 #blue
|
|
elif color==2:
|
|
led_val=led_val | 0x10 #red
|
|
elif color==3:
|
|
led_val=led_val | 0x14 #oragne
|
|
else:
|
|
led_val=led_val | 0x1C #green
|
|
|
|
fpga_io(dom_base[pim_num] + dom["port_led_control"][port], led_val)
|
|
|
|
return True
|
|
|
|
def get_port_led(self, port_num):
|
|
status, pim_num=self.get_pim_by_port(port_num)
|
|
if status==0:
|
|
return False
|
|
status, port=self.get_onepimport_by_port(port_num)
|
|
led_val=fpga_io(dom_base[pim_num] + dom["port_led_control"][port+1])
|
|
control=led_val & 0x3
|
|
if control==0x3:
|
|
control=2
|
|
elif control==0x1:
|
|
control=1
|
|
else:
|
|
control=0
|
|
|
|
color = led_val & 0x1C
|
|
if color==0:
|
|
color=0 #white
|
|
elif color==0xC:
|
|
color=1 #blue
|
|
elif color==0x14:
|
|
color=2 #red
|
|
elif color==0x18:
|
|
color=3 #oragne
|
|
elif color==0x1C:
|
|
color=4 #green
|
|
|
|
print("color=%d, control=%d"%(color, control))
|
|
return color, control
|
|
|
|
def main(argv):
|
|
init_resources()
|
|
pim=PimUtil()
|
|
print("Test Board ID")
|
|
for x in range(0,8):
|
|
val=pim.get_pim_board_id(x)
|
|
print("pim=%d"%x)
|
|
if val==0:
|
|
print("100G board")
|
|
else:
|
|
print("400G board")
|
|
|
|
print("Test pim presence")
|
|
for x in range(0,8):
|
|
pres=pim.get_pim_presence(x)
|
|
print("pim=%d, presence=%d"%(x, pres))
|
|
|
|
print("Test pim status")
|
|
for x in range(0,8):
|
|
power_status=pim.get_pim_status(x)
|
|
print("pim=%d power_status=0x%x"%(x, power_status))
|
|
|
|
release_resources()
|
|
|
|
if __name__ == '__main__':
|
|
main(sys.argv[1:])
|