# -*- coding: UTF-8 -*-
# -------------------------------------------------------------------------
# Name: ragileutil
# Purpose: common configuration and api
#
# Author: rd
#
# Created: 02/07/2018
# Copyright: (c) rd 2018
# -------------------------------------------------------------------------
import sys
if sys.version_info >= (3, 0):
import subprocess as commands
else:
import commands
import os
import re
import syslog
import time
import binascii
import tty
import termios
import threading
import click
import mmap
from ragileconfig import (
rg_eeprom,
FRULISTS,
MAC_DEFAULT_PARAM,
MAC_AVS_PARAM,
FANS_DEF,
FAN_PROTECT,
E2_LOC,
E2_PROTECT,
RAGILE_SERVICE_TAG,
RAGILE_DIAG_VERSION,
STARTMODULE,
RAGILE_CARDID,
RAGILE_PRODUCTNAME,
RAGILE_PART_NUMBER,
RAGILE_LABEL_REVISION,
RAGILE_MAC_SIZE,
RAGILE_MANUF_NAME,
RAGILE_MANUF_COUNTRY,
RAGILE_VENDOR_NAME,
MAILBOX_DIR,
)
try:
from eepromutil.fru import ipmifru
except Exception or SystemExit:
pass
import logging.handlers
import shutil
import gzip
import glob
__all__ = [
"MENUID",
"MENUPARENT",
"MENUVALUE",
"CHILDID",
"MENUITEMNAME",
"MENUITEMDEAL",
"GOBACK",
"GOQUIT",
"file_name",
"CRITICAL",
"FATAL",
"ERROR",
"WARNING",
"WARN",
"INFO",
"DEBUG",
"NOTSET",
"levelNames",
"TLV_INFO_ID_STRING",
"TLV_INFO_VERSION",
"TLV_INFO_LENGTH",
"TLV_INFO_LENGTH_VALUE",
"TLV_CODE_PRODUCT_NAME",
"TLV_CODE_PART_NUMBER",
"TLV_CODE_SERIAL_NUMBER",
"TLV_CODE_MAC_BASE",
"TLV_CODE_MANUF_DATE",
"TLV_CODE_DEVICE_VERSION",
"TLV_CODE_LABEL_REVISION",
"TLV_CODE_PLATFORM_NAME",
"TLV_CODE_ONIE_VERSION",
"TLV_CODE_MAC_SIZE",
"TLV_CODE_MANUF_NAME",
"TLV_CODE_MANUF_COUNTRY",
"TLV_CODE_VENDOR_NAME",
"TLV_CODE_DIAG_VERSION",
"TLV_CODE_SERVICE_TAG",
"TLV_CODE_VENDOR_EXT",
"TLV_CODE_CRC_32",
"_TLV_DISPLAY_VENDOR_EXT",
"TLV_CODE_RJ_CARID",
"_TLV_INFO_HDR_LEN",
"SYSLOG_IDENTIFIER",
"log_info",
"log_debug",
"log_warning",
"log_error",
"CompressedRotatingFileHandler",
"SETMACException",
"checkinput",
"checkinputproduct",
"getInputSetmac",
"fan_tlv",
"AVSUTIL",
"I2CUTIL",
"BMC",
"getSdkReg",
"getfilevalue",
"get_sysfs_value",
"write_sysfs_value",
"RJPRINTERR",
"strtoint",
"inttostr",
"str_to_hex",
"hex_to_str",
"str_to_bin",
"bin_to_str",
"get_mac_temp",
"get_mac_temp_sysfs",
"restartDockerService",
"wait_dhcp",
"wait_sdk",
"wait_docker",
"getTLV_BODY",
"_crc32",
"printvalue",
"generate_value",
"getsyseeprombyId",
"fac_init_cardidcheck",
"isValidMac",
"util_setmac",
"getInputCheck",
"getrawch",
"upper_input",
"changeTypeValue",
"astrcmp",
"generate_ext",
"rgi2cget",
"rgi2cset",
"rgpcird",
"rgpciwr",
"rgsysset",
"rgi2cget_word",
"rgi2cset_word",
"fan_setmac",
"checkfansninput",
"checkfanhwinput",
"util_show_fanse2",
"get_fane2_sysfs",
"util_show_fane2",
"getPid",
"fac_fans_setmac_tlv",
"fac_fan_setmac_fru",
"fac_fans_setmac",
"fac_fan_setmac",
"writeToEEprom",
"get_local_eth0_mac",
"getonieversion",
"createbmcMac",
"fac_board_setmac",
"ipmi_set_mac",
"getInputValue",
"bmc_setmac",
"closeProtocol",
"checkSdkMem",
"getch",
"get_raw_input",
"getsysvalue",
"get_pmc_register",
"decoder",
"decode_eeprom",
"get_sys_eeprom",
"getCardId",
"getsysmeminfo",
"getsysmeminfo_detail",
"getDmiSysByType",
"gethwsys",
"getsysbios",
"searchDirByName",
"getUsbLocation",
"getusbinfo",
"get_cpu_info",
"get_version_config_info",
"io_rd",
"io_wr",
]
MENUID = "menuid"
MENUPARENT = "parentid"
MENUVALUE = "value"
CHILDID = "childid"
MENUITEMNAME = "name"
MENUITEMDEAL = "deal"
GOBACK = "goBack"
GOQUIT = "quit"
file_name = "/etc/init.d/opennsl-modules-3.16.0-5-amd64"
##########################################################################
# ERROR LOG LEVEL
##########################################################################
CRITICAL = 50
FATAL = CRITICAL
ERROR = 40
WARNING = 30
WARN = WARNING
INFO = 20
DEBUG = 10
NOTSET = 0
levelNames = {
CRITICAL: "CRITICAL",
ERROR: "ERROR",
WARNING: "WARNING",
INFO: "INFO",
DEBUG: "DEBUG",
NOTSET: "NOTSET",
"CRITICAL": CRITICAL,
"ERROR": ERROR,
"WARN": WARNING,
"WARNING": WARNING,
"INFO": INFO,
"DEBUG": DEBUG,
"NOTSET": NOTSET,
}
TLV_INFO_ID_STRING = "TlvInfo\x00"
TLV_INFO_VERSION = 0x01
TLV_INFO_LENGTH = 0x00
TLV_INFO_LENGTH_VALUE = 0xBA
##########################################################################
# eeprom info
##########################################################################
TLV_CODE_PRODUCT_NAME = 0x21
TLV_CODE_PART_NUMBER = 0x22
TLV_CODE_SERIAL_NUMBER = 0x23
TLV_CODE_MAC_BASE = 0x24
TLV_CODE_MANUF_DATE = 0x25
TLV_CODE_DEVICE_VERSION = 0x26
TLV_CODE_LABEL_REVISION = 0x27
TLV_CODE_PLATFORM_NAME = 0x28
TLV_CODE_ONIE_VERSION = 0x29
TLV_CODE_MAC_SIZE = 0x2A
TLV_CODE_MANUF_NAME = 0x2B
TLV_CODE_MANUF_COUNTRY = 0x2C
TLV_CODE_VENDOR_NAME = 0x2D
TLV_CODE_DIAG_VERSION = 0x2E
TLV_CODE_SERVICE_TAG = 0x2F
TLV_CODE_VENDOR_EXT = 0xFD
TLV_CODE_CRC_32 = 0xFE
_TLV_DISPLAY_VENDOR_EXT = 1
TLV_CODE_RJ_CARID = 0x01
_TLV_INFO_HDR_LEN = 11
SYSLOG_IDENTIFIER = "UTILTOOL"
# ========================== Syslog wrappers ==========================
def log_info(msg, also_print_to_console=False):
syslog.openlog(SYSLOG_IDENTIFIER)
syslog.syslog(syslog.LOG_INFO, msg)
syslog.closelog()
if also_print_to_console:
click.echo(msg)
def log_debug(msg, also_print_to_console=False):
try:
syslog.openlog(SYSLOG_IDENTIFIER)
syslog.syslog(syslog.LOG_DEBUG, msg)
syslog.closelog()
if also_print_to_console:
click.echo(msg)
except Exception as e:
pass
def log_warning(msg, also_print_to_console=False):
syslog.openlog(SYSLOG_IDENTIFIER)
syslog.syslog(syslog.LOG_WARNING, msg)
syslog.closelog()
if also_print_to_console:
click.echo(msg)
def log_error(msg, also_print_to_console=False):
syslog.openlog(SYSLOG_IDENTIFIER)
syslog.syslog(syslog.LOG_ERR, msg)
syslog.closelog()
if also_print_to_console:
click.echo(msg)
class CompressedRotatingFileHandler(logging.handlers.RotatingFileHandler):
def doRollover(self):
"""
Do a rollover, as described in __init__().
"""
if self.stream:
self.stream.close()
self.stream = None
if self.backupCount > 0:
for i in range(self.backupCount - 1, 0, -1):
sfn = "%s.%d.gz" % (self.baseFilename, i)
dfn = "%s.%d.gz" % (self.baseFilename, i + 1)
if os.path.exists(sfn):
if os.path.exists(dfn):
os.remove(dfn)
os.rename(sfn, dfn)
dfn = self.baseFilename + ".1.gz"
if os.path.exists(dfn):
os.remove(dfn)
# These two lines below are the only new lines. I commented out the os.rename(self.baseFilename, dfn) and
# replaced it with these two lines.
with open(self.baseFilename, "rb") as f_in, gzip.open(dfn, "wb") as f_out:
shutil.copyfileobj(f_in, f_out)
self.mode = "w"
self.stream = self._open()
class SETMACException(Exception):
def __init__(self, param="ERROR", errno="-1"):
err = "Setmac fail[%s]: %s" % (errno, param)
Exception.__init__(self, err)
self.param = param
self.errno = errno
def checkinput(b):
if b.isdigit() == False:
raise Exception("Ivalid Number")
if int(b) > 0xFF or int(b) < 0:
raise Exception("Out of area")
def checkinputproduct(b):
if b.isalnum() == False:
raise Exception("Invalid string")
def getInputSetmac(val):
bia = val.boardInfoArea
pia = val.productInfoArea
if bia != None:
a = raw_input("[Board Card]Product Serial Number:")
if len(a) != 13:
raise Exception("Invalid Serial Number length")
checkinputproduct(a)
bia.boardSerialNumber = a
b = raw_input("[Board Card]Product Version:(from 1-255)")
checkinput(b)
b = "%0x" % int(b)
bia.boardextra1 = b.upper()
if pia != None:
a = raw_input("[Product Area]Product Serial Number:")
if len(a) != 13:
raise Exception("Invalid Serial Number")
checkinputproduct(a)
pia.productSerialNumber = a
b = raw_input("[Product Area]Product Version:(from 1-255)")
checkinput(b)
b = "%0x" % int(b)
pia.productVersion = b.upper()
return val
class fan_tlv(object):
VERSION = 0x01 # E2PROM Version，start from 0x01
FLAG = 0x7E # New E2PROM version flag is 0x7E
HW_VER = 0x01 # compose by master version and fixed version
TYPE = 0xF1 # hw type defination
TLV_LEN = 00 # data length (16bit)
_FAN_TLV_HDR_LEN = 6
_FAN_TLV_CRC_LEN = 2
_FAN_TLV_TYPE_NAME = 0x02
_FAN_TLV_TYPE_SN = 0x03
_FAN_TLV_TYPE_HW_INFO = 0x05
_FAN_TLV_TYPE_DEV_TYPE = 0x06
_fandecodetime = 0
@property
def dstatus(self):
return self._dstatus
@property
def typename(self):
return self._typename
@property
def typesn(self):
return self._typesn
@property
def typehwinfo(self):
return self._typehwinfo
@property
def typedevtype(self):
return self._typedevtype
@property
def fanbus(self):
return self._fanbus
@property
def fanloc(self):
return self._fanloc
@property
def fandecodetime(self):
return self._fandecodetime
def __init__(self):
self._typename = ""
self._typesn = ""
self._typehwinfo = ""
self._typedevtype = ""
self._dstatus = 0
def strtoarr(self, str):
s = []
if str is not None:
for index in range(len(str)):
s.append(str[index])
return s
def generate_fan_value(self):
bin_buffer = [chr(0xFF)] * 256
bin_buffer[0] = chr(self.VERSION)
bin_buffer[1] = chr(self.FLAG)
bin_buffer[2] = chr(self.HW_VER)
bin_buffer[3] = chr(self.TYPE)
temp_t = "%08x" % self.typedevtype # handle devtype first
typedevtype_t = hex_to_str(temp_t)
total_len = (
len(self.typename)
+ len(self.typesn)
+ len(self.typehwinfo)
+ len(typedevtype_t)
+ 8
)
bin_buffer[4] = chr(total_len >> 8)
bin_buffer[5] = chr(total_len & 0x00FF)
index_start = 6
bin_buffer[index_start] = chr(self._FAN_TLV_TYPE_NAME)
bin_buffer[index_start + 1] = chr(len(self.typename))
bin_buffer[
index_start + 2 : index_start + 2 + len(self.typename)
] = self.strtoarr(self.typename)
index_start = index_start + 2 + len(self.typename)
bin_buffer[index_start] = chr(self._FAN_TLV_TYPE_SN)
bin_buffer[index_start + 1] = chr(len(self.typesn))
bin_buffer[
index_start + 2 : index_start + 2 + len(self.typesn)
] = self.strtoarr(self.typesn)
index_start = index_start + 2 + len(self.typesn)
bin_buffer[index_start] = chr(self._FAN_TLV_TYPE_HW_INFO)
bin_buffer[index_start + 1] = chr(len(self.typehwinfo))
bin_buffer[
index_start + 2 : index_start + 2 + len(self.typehwinfo)
] = self.strtoarr(self.typehwinfo)
index_start = index_start + 2 + len(self.typehwinfo)
bin_buffer[index_start] = chr(self._FAN_TLV_TYPE_DEV_TYPE)
bin_buffer[index_start + 1] = chr(len(typedevtype_t))
bin_buffer[
index_start + 2 : index_start + 2 + len(typedevtype_t)
] = self.strtoarr(typedevtype_t)
index_start = index_start + 2 + len(typedevtype_t)
crcs = fan_tlv.fancrc("".join(bin_buffer[0:index_start])) # check 2bytes
bin_buffer[index_start] = chr(crcs >> 8)
bin_buffer[index_start + 1] = chr(crcs & 0x00FF)
return bin_buffer
def decode(self, e2):
ret = []
self.VERSION = ord(e2[0])
self.FLAG = ord(e2[1])
self.HW_VER = ord(e2[2])
self.TYPE = ord(e2[3])
self.TLV_LEN = (ord(e2[4]) << 8) | ord(e2[5])
tlv_index = self._FAN_TLV_HDR_LEN
tlv_end = self._FAN_TLV_HDR_LEN + self.TLV_LEN
# check checksum
if len(e2) < self._FAN_TLV_HDR_LEN + self.TLV_LEN + 2:
self._dstatus = -2
return ret
sumcrc = fan_tlv.fancrc(e2[0 : self._FAN_TLV_HDR_LEN + self.TLV_LEN])
readcrc = ord(e2[self._FAN_TLV_HDR_LEN + self.TLV_LEN]) << 8 | ord(
e2[self._FAN_TLV_HDR_LEN + self.TLV_LEN + 1]
)
if sumcrc != readcrc:
self._dstatus = -1
return ret
else:
self._dstatus = 0
while (tlv_index + 2) < len(e2) and tlv_index < tlv_end:
s = self.decoder(e2[tlv_index : tlv_index + 2 + ord(e2[tlv_index + 1])])
tlv_index += ord(e2[tlv_index + 1]) + 2
ret.append(s)
return ret
@staticmethod
def fancrc(t):
sum = 0
for index in range(len(t)):
sum += ord(t[index])
return sum
def decoder(self, t):
try:
name = ""
value = ""
if ord(t[0]) == self._FAN_TLV_TYPE_NAME:
name = "Product Name"
value = str(t[2 : 2 + ord(t[1])])
self._typename = value
elif ord(t[0]) == self._FAN_TLV_TYPE_SN:
name = "serial Number"
value = str(t[2 : 2 + ord(t[1])])
self._typesn = value
elif ord(t[0]) == self._FAN_TLV_TYPE_HW_INFO:
name = "hardware info"
value = str(t[2 : 2 + ord(t[1])])
self._typehwinfo = value
elif ord(t[0]) == self._FAN_TLV_TYPE_DEV_TYPE:
name = "dev type"
value = str(t[2 : 2 + ord(t[1])])
value = str_to_hex(value)
self._typedevtype = value
value = "0x08%x" % value
except Exception as e:
print(e)
return {"name": name, "code": ord(t[0]), "value": value}
def __str__(self):
formatstr = (
"VERSION : 0x%02x \n"
" FLAG : 0x%02x \n"
" HW_VER : 0x%02x \n"
" TYPE : 0x%02x \n"
"typename : %s \n"
"typesn : %s \n"
"typehwinfo : %s \n"
)
return formatstr % (
self.VERSION,
self.FLAG,
self.HW_VER,
self.TYPE,
self.typename,
self.typesn,
self.typehwinfo,
)
class AVSUTIL:
@staticmethod
def mac_avs_chip(bus, devno, loc, open, close, loop, protectaddr, level, loopaddr):
# disable protection
rgi2cset(bus, devno, protectaddr, open)
rgi2cset(bus, devno, loopaddr, loop)
rgi2cset_word(bus, devno, loc, level)
ret, value = rgi2cget_word(bus, devno, loc)
if strtoint(value) == level:
ret = 0
# enable protection
rgi2cset(bus, devno, protectaddr, close)
if ret == 0:
return True
return False
@staticmethod
def macPressure_adj(macavs, avs_param, mac_def_param):
# check whether it within range
max_adj = max(avs_param.keys())
min_adj = min(avs_param.keys())
type = mac_def_param["type"]
level = 0
if type == 0:
if macavs not in range(min_adj, max_adj + 1):
return False
else:
level = macavs
else:
if macavs not in range(min_adj, max_adj + 1):
level = mac_def_param["default"]
else:
level = macavs
ret = AVSUTIL.mac_avs_chip(
mac_def_param["bus"],
mac_def_param["devno"],
mac_def_param["addr"],
mac_def_param["open"],
mac_def_param["close"],
mac_def_param["loop"],
mac_def_param["protectaddr"],
avs_param[level],
mac_def_param["loopaddr"],
)
return ret
@staticmethod
def mac_adj():
macavs = 0
name = MAC_DEFAULT_PARAM["sdkreg"]
ret, status = getSdkReg(name)
if ret == False:
return False
status = strtoint(status)
# shift operation
if MAC_DEFAULT_PARAM["sdktype"] != 0:
status = (status >> MAC_DEFAULT_PARAM["macregloc"]) & MAC_DEFAULT_PARAM[
"mask"
]
macavs = status
ret = AVSUTIL.macPressure_adj(macavs, MAC_AVS_PARAM, MAC_DEFAULT_PARAM)
return ret
class I2CUTIL:
@staticmethod
def getvaluefromdevice(name):
ret = []
for item in DEVICE:
if item["name"] == name:
ret.append(item)
return ret
@staticmethod
def openFanE2Protect():
rgi2cset(
FAN_PROTECT["bus"],
FAN_PROTECT["devno"],
FAN_PROTECT["addr"],
FAN_PROTECT["open"],
)
@staticmethod
def closeFanE2Protect():
rgi2cset(
FAN_PROTECT["bus"],
FAN_PROTECT["devno"],
FAN_PROTECT["addr"],
FAN_PROTECT["close"],
)
@staticmethod
def writeToFanE2(bus, loc, rst_arr):
index = 0
for item in rst_arr:
rgi2cset(bus, loc, index, ord(item))
index += 1
@staticmethod
def writeToE2(bus, loc, rst_arr):
index = 0
for item in rst_arr:
rgi2cset(bus, loc, index, ord(item))
index += 1
@staticmethod
def getE2File(bus, loc):
return "/sys/bus/i2c/devices/%d-00%02x/eeprom" % (bus, loc)
class BMC:
_instance_lock = threading.Lock()
def __init__(self):
pass
def __new__(cls, *args, **kwargs):
if not hasattr(Singleton, "_instance"):
with Singleton._instance_lock:
if not hasattr(Singleton, "_instance"):
Singleton._instance = object.__new__(cls)
return Singleton._instance
# Internal interface
def getSdkReg(reg):
try:
cmd = "bcmcmd -t 1 'getr %s ' < /dev/null" % reg
ret, result = os_system(cmd)
result_t = result.strip().replace("\r", "").replace("\n", "")
if ret != 0 or "Error:" in result_t:
return False, result
patt = r"%s.(.*):(.*)>drivshell" % reg
rt = re.findall(patt, result_t, re.S)
test = re.findall("=(.*)", rt[0][0])[0]
except Exception as e:
return False, "getsdk register error"
return True, test
def getfilevalue(location):
try:
with open(location, "r") as fd:
value = fd.read()
return True, value.strip()
except Exception as e:
return False, "error"
def get_sysfs_value(location):
pos_t = str(location)
name = get_pmc_register(pos_t)
return name
def write_sysfs_value(reg_name, value):
fileLoc = MAILBOX_DIR + reg_name
try:
if not os.path.isfile(fileLoc):
print(fileLoc, "not found !")
return False
with open(fileLoc, "w") as fd:
fd.write(value)
except Exception as error:
log_error("Unable to open " + fileLoc + "file !")
return False
return True
def RJPRINTERR(str):
print("\033[0;31m%s\033[0m" % str)
def strtoint(str): # convert Hex string to int such as "4040"/"0x4040"/"0X4040" = 16448
value = 0
rest_v = str.replace("0X", "").replace("0x", "")
for index in range(len(rest_v)):
print(rest_v[index])
value |= int(rest_v[index], 16) << ((len(rest_v) - index - 1) * 4)
return value
def inttostr(vl, len): # convert int to string such as 0x3030 = 00
if type(vl) != int:
raise Exception(" type error")
index = 0
ret_t = ""
while index < len:
ret = 0xFF & (vl >> index * 8)
ret_t += chr(ret)
index += 1
return ret_t
def str_to_hex(rest_v):
value = 0
for index in range(len(rest_v)):
value |= ord(rest_v[index]) << ((len(rest_v) - index - 1) * 8)
return value
def hex_to_str(s):
len_t = len(s)
if len_t % 2 != 0:
return 0
ret = ""
for t in range(0, int(len_t / 2)):
ret += chr(int(s[2 * t : 2 * t + 2], 16))
return ret
def str_to_bin(s):
return " ".join([bin(ord(c)).replace("0b", "") for c in s])
def bin_to_str(s):
return "".join([chr(i) for i in [int(b, 2) for b in s.split(" ")]])
def get_mac_temp():
result = {}
# wait_docker()
# exec twice, get the second result
os_system('bcmcmd -t 1 "show temp" < /dev/null')
ret, log = os_system('bcmcmd -t 1 "show temp" < /dev/null')
if ret:
return False, result
else:
# decode obtained info
logs = log.splitlines()
for line in logs:
if "average" in line:
b = re.findall(r"\d+.\d+", line)
result["average"] = b[0]
elif "maximum" in line:
b = re.findall(r"\d+.\d+", line)
result["maximum"] = b[0]
return True, result
def get_mac_temp_sysfs(mactempconf):
try:
temp = -1000000
temp_list = []
mac_temp_loc = mactempconf.get("loc", [])
mac_temp_flag = mactempconf.get("flag", None)
if mac_temp_flag is not None: # check mac temperature vaild flag
gettype = mac_temp_flag.get("gettype")
okbit = mac_temp_flag.get("okbit")
okval = mac_temp_flag.get("okval")
if gettype == "io":
io_addr = mac_temp_flag.get("io_addr")
val = io_rd(io_addr)
if val is None:
raise Exception("get mac_flag by io failed.")
else:
bus = mac_temp_flag.get("bus")
loc = mac_temp_flag.get("loc")
offset = mac_temp_flag.get("offset")
ind, val = rgi2cget(bus, loc, offset)
if ind is not True:
raise Exception("get mac_flag by i2c failed.")
val_t = (int(val, 16) & (1 << okbit)) >> okbit
if val_t != okval:
raise Exception("mac_flag invalid, val_t:%d." % val_t)
for loc in mac_temp_loc:
temp_s = get_sysfs_value(loc)
if isinstance(temp_s, str) and temp_s.startswith("ERR"):
raise Exception("get mac temp error. loc:%s" % loc)
temp_t = int(temp_s)
if temp_t == -1000000:
raise Exception("mac temp invalid.loc:%s" % loc)
temp_list.append(temp_t)
temp_list.sort(reverse=True)
temp = temp_list[0]
except Exception as e:
return False, temp
return True, temp
def restartDockerService(force=False):
container_name = [
"database",
"snmp",
"syncd",
"swss",
"dhcp_relay",
"radv",
"teamd",
"pmon",
]
ret, status = os_system("docker ps")
if ret == 0:
for tmpname in container_name:
if tmpname not in status:
if force == True:
os_system("docker restart %s" % tmpname)
else:
os_system("systemctl restart %s" % tmpname)
def wait_dhcp(timeout):
time_cnt = 0
while True:
try:
ret, status = os_system("systemctl status dhcp_relay.service")
if (ret == 0 and "running" in status) or "SUCCESS" in status:
break
else:
sys.stdout.write(".")
sys.stdout.flush()
time_cnt = time_cnt + 1
if time_cnt > timeout:
raise Exception("wait_dhcp timeout")
time.sleep(1)
except Exception as e:
return False
return True
def wait_sdk(sdk_fpath, timeout):
time_cnt = 0
while True:
try:
if os.path.exists(sdk_fpath):
break
else:
sys.stdout.write(".")
sys.stdout.flush()
time_cnt = time_cnt + 1
if time_cnt > timeout:
raise Exception("wait_sdk timeout")
time.sleep(1)
except Exception as e:
return False
return True
def wait_docker(need_restart=False, timeout=180):
sdkcheck_params = STARTMODULE.get("sdkcheck", {})
if sdkcheck_params.get("checktype") == "file": # pass file check
sdk_fpath = sdkcheck_params.get("sdk_fpath")
return wait_sdk(sdk_fpath, timeout)
return wait_dhcp(timeout)
def getTLV_BODY(type, productname):
x = []
temp_t = ""
if type == TLV_CODE_MAC_BASE:
arr = productname.split(":")
for tt in arr:
temp_t += chr(int(tt, 16))
elif type == TLV_CODE_DEVICE_VERSION:
temp_t = chr(productname)
elif type == TLV_CODE_MAC_SIZE:
temp_t = chr(productname >> 8) + chr(productname & 0x00FF)
else:
temp_t = productname
x.append(chr(type))
x.append(chr(len(temp_t)))
for i in temp_t:
x.append(i)
return x
def _crc32(v):
return "0x%08x" % (
binascii.crc32(v) & 0xFFFFFFFF
) # get 8 bytes of crc32 %x return hex
def printvalue(b):
index = 0
for i in range(0, len(b)):
if index % 16 == 0:
print(" ")
print("%02x " % ord(b[i]))
index += 1
print("\n")
def generate_value(_t):
ret = []
for i in TLV_INFO_ID_STRING:
ret.append(i)
ret.append(chr(TLV_INFO_VERSION))
ret.append(chr(TLV_INFO_LENGTH))
ret.append(chr(TLV_INFO_LENGTH_VALUE))
total_len = 0
for key in _t:
x = getTLV_BODY(key, _t[key])
ret += x
total_len += len(x)
ret[10] = chr(total_len + 6)
ret.append(chr(0xFE))
ret.append(chr(0x04))
s = _crc32("".join(ret))
for t in range(0, 4):
ret.append(chr(int(s[2 * t + 2 : 2 * t + 4], 16)))
totallen = len(ret)
if totallen < 256:
for left_t in range(0, 256 - totallen):
ret.append(chr(0x00))
return (ret, True)
def getsyseeprombyId(id):
ret = get_sys_eeprom()
for item in ret:
if item["code"] == id:
return item
return None
def fac_init_cardidcheck():
rest = getsyseeprombyId(TLV_CODE_RJ_CARID) # check cardId same or not
if rest is None:
print("need to program write bin file")
return False
else:
rest_v = rest["value"]
value = strtoint(rest_v)
if value == RAGILE_CARDID:
log_debug("check card ID pass")
else:
log_debug("check card ID error")
return False
return True
def isValidMac(mac):
if re.match(r"^\s*([0-9a-fA-F]{2,2}:){5,5}[0-9a-fA-F]{2,2}\s*$", mac):
return True
return False
# Internet cardsetmac
def util_setmac(eth, mac):
rulefile = "/etc/udev/rules.d/70-persistent-net.rules"
if isValidMac(mac) == False:
return False, "MAC invaild"
cmd = "ethtool -e %s | grep 0x0010 | awk '{print \"0x\"$13$12$15$14}'" % eth
ret, log = os_system(cmd)
log_debug(log)
magic = ""
if ret == 0 and len(log):
magic = log
macs = mac.upper().split(":")
# chage ETH0 to value after setmac
ifconfigcmd = "ifconfig eth0 hw ether %s" % mac
log_debug(ifconfigcmd)
ret, status = os_system(ifconfigcmd)
if ret:
raise SETMACException("software set Internet card MAC error")
index = 0
for item in macs:
cmd = "ethtool -E %s magic %s offset %d value 0x%s" % (eth, magic, index, item)
log_debug(cmd)
index += 1
ret, log = os_system(cmd)
if ret != 0:
raise SETMACException("set hardware Internet card MAC error")
# get value after setting
cmd_t = "ethtool -e eth0 offset 0 length 6"
ret, log = os_system(cmd_t)
m = re.split(":", log)[-1].strip().upper()
mac_result = m.upper().split(" ")
for ind, s in enumerate(macs):
if s != mac_result[ind]:
RJPRINTERR("MAC comparison error")
if os.path.exists(rulefile):
os.remove(rulefile)
print("MGMT MAC[%s]" % mac)
return True
def getInputCheck(tips):
str = raw_input(tips)
if (
astrcmp(str, "y")
or astrcmp(str, "ye")
or astrcmp(str, "yes")
or astrcmp(str, "")
):
return True
else:
return False
def getrawch():
fd = sys.stdin.fileno()
old_settings = termios.tcgetattr(fd)
try:
tty.setraw(sys.stdin.fileno())
ch = sys.stdin.read(1)
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
return ch
def upper_input(tips):
sys.stdout.write(tips)
sys.stdout.flush()
passwd = []
while True:
ch = getrawch().upper()
if ch == "\r" or ch == "\n":
return "".join(passwd)
elif ch == "\b" or ord(ch) == 127:
if passwd:
del passwd[-1]
sys.stdout.write("\b \b")
else:
sys.stdout.write(ch)
passwd.append(ch)
def changeTypeValue(_value, type1, tips, example):
if type1 == TLV_CODE_PRODUCT_NAME:
while True:
print(
"please check (1)air from forward to backward/(2)air from backward to forward:"
)
option = raw_input()
if option == "1":
_value[type1] = example + "-F-RJ"
print(
"check Product is air from forward to backward device,Product Name:%s"
% _value[type1]
)
break
elif option == "2":
_value[type1] = example + "-R-RJ"
print(
"check Product is air from backward to forward device,Product Name:%s"
% _value[type1]
)
break
else:
print("input incorrect, check please")
return True
print("Please input[%s]such as(%s):" % (tips, example))
name = upper_input("")
if type1 == TLV_CODE_MAC_BASE:
if len(name) != 12:
raise SETMACException("MAC address length incorrect, check please")
release_mac = ""
for i in range(int(len(name) / 2)):
if i == 0:
release_mac += name[i * 2 : i * 2 + 2]
else:
release_mac += ":" + name[i * 2 : i * 2 + 2]
if isValidMac(release_mac) == True:
_value[type1] = release_mac
else:
raise SETMACException("MAC address invaild, check please")
elif type1 == TLV_CODE_DEVICE_VERSION:
if name.isdigit():
_value[type1] = int(name)
else:
raise SETMACException("Version is not number, check please")
elif type1 == TLV_CODE_MAC_SIZE:
if name.isdigit():
_value[type1] = int(name, 16)
else:
raise SETMACException("Version is not number, check please")
elif type1 == TLV_CODE_SERIAL_NUMBER:
if name.isalnum() == False:
raise SETMACException("Serial Number invaild string, check please")
elif len(name) != 13:
raise SETMACException("Serial Number length incorrect, check please")
else:
_value[type1] = name
elif type1 == TLV_CODE_VENDOR_EXT:
_value[type1] = name
else:
_value[type1] = name
return True
def astrcmp(str1, str2):
return str1.lower() == str2.lower()
def generate_ext(cardid):
s = "%08x" % cardid
ret = ""
for t in range(0, 4):
ret += chr(int(s[2 * t : 2 * t + 2], 16))
ret = chr(0x01) + chr(len(ret)) + ret
return ret
def rgi2cget(bus, devno, address):
command_line = "i2cget -f -y %d 0x%02x 0x%02x " % (bus, devno, address)
retrytime = 6
ret_t = ""
for i in range(retrytime):
ret, ret_t = os_system(command_line)
if ret == 0:
return True, ret_t
time.sleep(0.1)
return False, ret_t
def rgi2cset(bus, devno, address, byte):
command_line = "i2cset -f -y %d 0x%02x 0x%02x 0x%02x" % (bus, devno, address, byte)
retrytime = 6
ret_t = ""
for i in range(retrytime):
ret, ret_t = os_system(command_line)
if ret == 0:
return True, ret_t
return False, ret_t
def rgpcird(pcibus, slot, fn, bar, offset):
"""read pci register"""
if offset % 4 != 0:
return
filename = "/sys/bus/pci/devices/0000:%02x:%02x.%x/resource%d" % (
int(pcibus),
int(slot),
int(fn),
int(bar),
)
file = open(filename, "r+")
size = os.path.getsize(filename)
data = mmap.mmap(file.fileno(), size)
result = data[offset : offset + 4]
s = result[::-1]
val = 0
for i in range(0, len(s)):
val = val << 8 | s[i]
return "0x%08x" % val
def rgpciwr(pcibus, slot, fn, bar, offset, data):
"""write pci register"""
ret = inttostr(data, 4)
ret = str.encode(ret)
ret = ret.strip(b'\xc2')
filename = "/sys/bus/pci/devices/0000:%02x:%02x.%x/resource%d" % (
int(pcibus),
int(slot),
int(fn),
int(bar),
)
file = open(filename, "r+")
size = os.path.getsize(filename)
data = mmap.mmap(file.fileno(), size)
data[offset : offset + 4] = ret
result = data[offset : offset + 4]
s = result[::-1]
val = 0
for i in range(0, len(s)):
val = val << 8 | s[i]
data.close()
def rgsysset(location, value):
command_line = "echo 0x%02x > %s" % (value, location)
retrytime = 6
ret_t = ""
for i in range(retrytime):
ret, ret_t = os_system(command_line)
if ret == 0:
return True, ret_t
return False, ret_t
def rgi2cget_word(bus, devno, address):
command_line = "i2cget -f -y %d 0x%02x 0x%02x w" % (bus, devno, address)
retrytime = 3
ret_t = ""
for i in range(retrytime):
ret, ret_t = os_system(command_line)
if ret == 0:
return True, ret_t
return False, ret_t
def rgi2cset_word(bus, devno, address, byte):
command_line = "i2cset -f -y %d 0x%02x 0x%02x 0x%x w" % (bus, devno, address, byte)
os_system(command_line)
def fan_setmac():
rgi2cset(
FAN_PROTECT["bus"],
FAN_PROTECT["devno"],
FAN_PROTECT["addr"],
FAN_PROTECT["open"],
)
rgi2cset(
FAN_PROTECT["bus"],
FAN_PROTECT["devno"],
FAN_PROTECT["addr"],
FAN_PROTECT["close"],
)
def checkfansninput(fan_sn, fansntemp):
if fan_sn in fansntemp:
RJPRINTERR("exist same Serial Number，please input again")
return False
if len(fan_sn) != 13:
RJPRINTERR("Serial Number length incorrect，please input again")
return False
return True
# check hw version
def checkfanhwinput(hw):
if len(hw) != 4:
RJPRINTERR("hardware version length incorrect, please input again")
return False
if hw.find(".") != 1:
RJPRINTERR("hardware version incorrect, please input again")
return False
return True
def util_show_fanse2(fans):
formatstr = "%-8s %-20s %-20s %-20s %-20s"
print(formatstr % ("id", "Name", "hardware version", "Serial Number", "Time"))
print(
formatstr
% ("------", "---------------", "---------------", "---------------", "----")
)
for fan in fans:
# print fan.dstatus
if fan.dstatus < 0:
print("%-8s" % ("FAN%d" % (fans.index(fan) + 1)))
RJPRINTERR(" decode e2 error")
else:
print(
formatstr
% (
"FAN%d" % (fans.index(fan) + 1),
fan.typename.replace(chr(0x00), ""),
fan.typehwinfo.replace(chr(0x00), ""),
fan.typesn.replace(chr(0x00), ""),
fan.fandecodetime,
)
)
def get_fane2_sysfs(bus, loc):
rg_fan_e2 = "%d-%04x/fan" % (bus, loc)
eeprom = get_sysfs_value(rg_fan_e2)
return eeprom
def util_show_fane2():
ret = sorted(I2CUTIL.getvaluefromdevice("rg_fan"))
if len(ret) <= 0:
return None
fans = []
for index in range(len(ret)):
t1 = int(round(time.time() * 1000))
eeprom = get_fane2_sysfs(ret[index]["bus"], ret[index]["loc"])
t2 = int(round(time.time() * 1000))
fane2 = fan_tlv()
fane2.fandecodetime = t2 - t1
fane2.decode(eeprom)
fans.append(fane2)
util_show_fanse2(fans)
def getPid(name):
ret = []
for dirname in os.listdir("/proc"):
if dirname == "curproc":
continue
try:
with open("/proc/{}/cmdline".format(dirname), mode="rb") as fd:
content = fd.read()
except Exception:
continue
if name in content:
ret.append(dirname)
return ret
def fac_fans_setmac_tlv(ret):
if len(ret) <= 0:
return None
fans = []
fansntemp = []
for index in range(len(ret)):
item = ret[index]
log_debug(item)
eeprom = get_fane2_sysfs(item["bus"], item["loc"])
fane2 = fan_tlv()
fane2.decode(eeprom)
fane2.fanbus = item["bus"]
fane2.fanloc = item["loc"]
log_debug("decode eeprom success")
print("Fan[%d]-[%s]setmac" % ((index + 1), FANS_DEF[fane2.typedevtype]))
while True:
print("Please input[%s]:" % "Serial Number")
fan_sn = raw_input()
if checkfansninput(fan_sn, fansntemp) == False:
continue
fansntemp.append(fan_sn)
fan_sn = fan_sn + chr(0x00)
fane2.typesn = fan_sn + chr(0x00)
break
while True:
print("Please input[%s]:" % "hardware version")
hwinfo = raw_input()
if checkfanhwinput(hwinfo) == False:
continue
fan_hwinfo = hwinfo + chr(0x00)
fane2.typehwinfo = fan_hwinfo + chr(0x00)
break
log_debug(fane2.typedevtype)
fane2.typename = FANS_DEF[fane2.typedevtype] + chr(0x00)
fans.append(fane2)
print("\n")
print("\n*******************************\n")
util_show_fanse2(fans)
if getInputCheck("check input correctly or not（Yes/No):") == True:
for fan in fans:
log_debug("ouput fan")
fac_fan_setmac(fan)
else:
print("setmac quit")
return False
def fac_fan_setmac_fru(ret):
fans = FRULISTS.get("fans")
fanfrus = {}
newfrus = {}
# getmsg
try:
for fan in fans:
print("===============%s ================getmessage" % fan.get("name"))
eeprom = getsysvalue(I2CUTIL.getE2File(fan.get("bus"), fan.get("loc")))
fru = ipmifru()
fru.decodeBin(eeprom)
fanfrus[fan.get("name")] = fru
except Exception as e:
print(str(e))
return False
# setmsg
for fan in fans:
print("===============%s ================setmac" % fan.get("name"))
fruold = fanfrus.get(fan.get("name"))
newfru = getInputSetmac(fruold)
newfru.recalcute()
newfrus[fan.get("name")] = newfru
# writemsg
for fan in fans:
print("===============%s ================writeToE2" % fan.get("name"))
ret_t = newfrus.get(fan.get("name"))
I2CUTIL.openFanE2Protect()
I2CUTIL.writeToFanE2(fan.get("bus"), fan.get("loc"), ret_t.bindata)
I2CUTIL.closeFanE2Protect()
# check
try:
for fan in fans:
print("===============%s ================getmessage" % fan.get("name"))
eeprom = getsysvalue(I2CUTIL.getE2File(fan.get("bus"), fan.get("loc")))
fru = ipmifru()
fru.decodeBin(eeprom)
except Exception as e:
print(str(e))
return False
return True
def fac_fans_setmac():
ret = I2CUTIL.getvaluefromdevice("rg_fan")
if ret is not None and len(ret) > 0:
return fac_fans_setmac_tlv(ret)
fans = FRULISTS.get("fans", None)
if fans is not None and len(fans) > 0:
return fac_fan_setmac_fru(ret)
return False
def fac_fan_setmac(item):
I2CUTIL.openFanE2Protect()
I2CUTIL.writeToFanE2(item.fanbus, item.fanloc, item.generate_fan_value())
I2CUTIL.closeFanE2Protect()
def writeToEEprom(rst_arr):
dealtype = E2_PROTECT.get("gettype", None)
if dealtype is None:
rgi2cset(
E2_PROTECT["bus"],
E2_PROTECT["devno"],
E2_PROTECT["addr"],
E2_PROTECT["open"],
)
elif dealtype == "io":
io_wr(E2_PROTECT["io_addr"], E2_PROTECT["open"])
index = 0
for item in rst_arr:
rgi2cset(E2_LOC["bus"], E2_LOC["devno"], index, ord(item))
index += 1
if dealtype is None:
rgi2cset(
E2_PROTECT["bus"],
E2_PROTECT["devno"],
E2_PROTECT["addr"],
E2_PROTECT["close"],
)
elif dealtype == "io":
io_wr(E2_PROTECT["io_addr"], E2_PROTECT["close"])
# deal last drivers
os.system("rmmod at24 ")
os.system("modprobe at24 ")
os.system("rm -f /var/cache/sonic/decode-syseeprom/syseeprom_cache")
def get_local_eth0_mac():
cmd = "ifconfig eth0 |grep HWaddr"
print(os_system(cmd))
def getonieversion():
if not os.path.isfile("/host/machine.conf"):
return ""
machine_vars = {}
with open("/host/machine.conf") as machine_file:
for line in machine_file:
tokens = line.split("=")
if len(tokens) < 2:
continue
machine_vars[tokens[0]] = tokens[1].strip()
return machine_vars.get("onie_version")
def createbmcMac(cpumac, num=2):
bcmvalue = strtoint(cpumac[cpumac.rindex(":") + 1 : len(cpumac)]) + num
# bmcmac =
t = cpumac.split(":")
t[5] = "%02x" % bcmvalue
bmcmac = ":".join(t)
return bmcmac.upper()
def fac_board_setmac():
_value = {}
# default value
_value[TLV_CODE_VENDOR_EXT] = generate_ext(RAGILE_CARDID) # generate id
_value[TLV_CODE_PRODUCT_NAME] = RAGILE_PRODUCTNAME
_value[TLV_CODE_PART_NUMBER] = RAGILE_PART_NUMBER
_value[TLV_CODE_LABEL_REVISION] = RAGILE_LABEL_REVISION
_value[TLV_CODE_PLATFORM_NAME] = platform
_value[TLV_CODE_ONIE_VERSION] = getonieversion()
_value[TLV_CODE_MAC_SIZE] = RAGILE_MAC_SIZE
_value[TLV_CODE_MANUF_NAME] = RAGILE_MANUF_NAME
_value[TLV_CODE_MANUF_COUNTRY] = RAGILE_MANUF_COUNTRY
_value[TLV_CODE_VENDOR_NAME] = RAGILE_VENDOR_NAME
_value[TLV_CODE_DIAG_VERSION] = RAGILE_DIAG_VERSION
_value[TLV_CODE_SERVICE_TAG] = RAGILE_SERVICE_TAG
try:
if 0x00004052 == RAGILE_CARDID:
_value[TLV_CODE_PRODUCT_NAME] = RAGILE_PRODUCTNAME + "-RJ"
elif 0x00004051 == RAGILE_CARDID or 0x00004050 == RAGILE_CARDID:
changeTypeValue(
_value, TLV_CODE_PRODUCT_NAME, "Product name", RAGILE_PRODUCTNAME
)
changeTypeValue(
_value, TLV_CODE_SERIAL_NUMBER, "SN", "0000000000000"
) # add serial number
changeTypeValue(
_value, TLV_CODE_DEVICE_VERSION, "hardware version", "101"
) # hardware version
changeTypeValue(
_value, TLV_CODE_MAC_BASE, "MAC address", "58696cfb2108"
) # MAC address
_value[TLV_CODE_MANUF_DATE] = time.strftime(
"%m/%d/%Y %H:%M:%S", time.localtime()
) # add setmac time
rst, ret = generate_value(_value)
if (
util_setmac("eth0", _value[TLV_CODE_MAC_BASE]) == True
): # set Internet cardIP
writeToEEprom(rst) # write to e2
# set BMC MAC
if "bmcsetmac" in FACTESTMODULE and FACTESTMODULE["bmcsetmac"] == 1:
bmcmac = createbmcMac(_value[TLV_CODE_MAC_BASE])
if ipmi_set_mac(bmcmac) == True:
print("BMC MAC[%s]" % bmcmac)
else:
print("SET BMC MAC FAILED")
return False
else:
return False
except SETMACException as e:
# print(e)
RJPRINTERR("\n\n%s\n\n" % e)
return False
except ValueError as e:
return False
return True
def ipmi_set_mac(mac):
macs = mac.split(":")
cmdinit = "ipmitool raw 0x0c 0x01 0x01 0xc2 0x00"
cmdset = "ipmitool raw 0x0c 0x01 0x01 0x05"
for ind in range(len(macs)):
cmdset += " 0x%02x" % int(macs[ind], 16)
os_system(cmdinit)
ret, status = os_system(cmdset)
if ret:
RJPRINTERR("\n\n%s\n\n" % status)
return False
return True
def getInputValue(title, tips):
print("Please input[%s]such as(%s):" % (title, tips))
name = raw_input()
return name
def bmc_setmac():
tips = "BMC MAC"
print("Please input value you want to change[%s]:" % tips)
name = raw_input()
if len(name) != 12:
RJPRINTERR("\nMAC address invaild, try again\n")
return False
release_mac = ""
for i in range(int(len(name) / 2)):
if i == 0:
release_mac += name[i * 2 : i * 2 + 2]
else:
release_mac += ":" + name[i * 2 : i * 2 + 2]
if isValidMac(release_mac) == True:
if ipmi_set_mac(release_mac) == True:
return True
else:
RJPRINTERR("\nMAC address invaild, try again\n")
return False
def closeProtocol():
# disable LLDP
log_info("disable LLDP")
sys.stdout.write(".")
sys.stdout.flush()
os_system("systemctl stop lldp.service")
log_info("disable lldp service")
sys.stdout.write(".")
sys.stdout.flush()
os_system("systemctl stop bgp.service")
log_info("disable bgp service")
sys.stdout.write(".")
sys.stdout.flush()
# ret, status = os_system('bcmcmd "port ce,xe stp=disable"')
# check SDK memory must be 256M
def checkSdkMem():
ind = 0
file_data = ""
with open(file_name, "r") as f:
for line in f:
if "dmasize=16M" in line:
line = line.replace("dmasize=16M", "dmasize=256M")
ind = -1
file_data += line
if ind == 0:
return
with open(file_name, "w") as f:
f.write(file_data)
print("change SDK memory to 256, reboot required")
os_system("sync")
os_system("reboot")
##########################################################################
# receives a character setting
##########################################################################
def getch(msg):
ret = ""
fd = sys.stdin.fileno()
old_ttyinfo = termios.tcgetattr(fd)
new_ttyinfo = old_ttyinfo[:]
new_ttyinfo[3] &= ~termios.ICANON
new_ttyinfo[3] &= ~termios.ECHO
sys.stdout.write(msg)
sys.stdout.flush()
try:
termios.tcsetattr(fd, termios.TCSANOW, new_ttyinfo)
ret = os.read(fd, 1)
finally:
# print "try to setting"
termios.tcsetattr(fd, termios.TCSANOW, old_ttyinfo)
return ret
def get_raw_input():
ret = ""
fd = sys.stdin.fileno()
old_ttyinfo = termios.tcgetattr(fd)
new_ttyinfo = old_ttyinfo[:]
new_ttyinfo[3] &= ~termios.ICANON
new_ttyinfo[3] &= ~termios.ECHO
try:
termios.tcsetattr(fd, termios.TCSANOW, new_ttyinfo)
ret = raw_input("")
except Exception as e:
print(e)
finally:
termios.tcsetattr(fd, termios.TCSANOW, old_ttyinfo)
return ret
def getsysvalue(location):
retval = None
mb_reg_file = location
if not os.path.isfile(mb_reg_file):
print(mb_reg_file, "not found !")
return retval
try:
if not os.path.isfile(mb_reg_file):
print(mb_reg_file, "not found !")
return retval
with open(mb_reg_file, "r") as fd:
retval = fd.read()
except Exception as error:
log_error("Unable to open " + mb_reg_file + "file !")
retval = retval.rstrip("\r\n")
retval = retval.lstrip(" ")
# log_debug(retval)
return retval
# get file value
def get_pmc_register(reg_name):
retval = "ERR"
mb_reg_file = MAILBOX_DIR + reg_name
filepath = glob.glob(mb_reg_file)
if len(filepath) == 0:
return "%s %s notfound" % (retval, mb_reg_file)
mb_reg_file = filepath[0]
if not os.path.isfile(mb_reg_file):
return "%s %s notfound" % (retval, mb_reg_file)
try:
with open(mb_reg_file, "r") as fd:
retval = fd.read()
except Exception as error:
pass
retval = retval.rstrip("\r\n")
retval = retval.lstrip(" ")
return retval
# decode EEPROM
def decoder(s, t):
if ord(t[0]) == TLV_CODE_PRODUCT_NAME:
name = "Product Name"
value = str(t[2 : 2 + ord(t[1])])
elif ord(t[0]) == TLV_CODE_PART_NUMBER:
name = "Part Number"
value = t[2 : 2 + ord(t[1])]
elif ord(t[0]) == TLV_CODE_SERIAL_NUMBER:
name = "Serial Number"
value = t[2 : 2 + ord(t[1])]
elif ord(t[0]) == TLV_CODE_MAC_BASE:
name = "Base MAC Address"
value = ":".join([binascii.b2a_hex(T) for T in t[2:8]]).upper()
elif ord(t[0]) == TLV_CODE_MANUF_DATE:
name = "Manufacture Date"
value = t[2 : 2 + ord(t[1])]
elif ord(t[0]) == TLV_CODE_DEVICE_VERSION:
name = "Device Version"
value = str(ord(t[2]))
elif ord(t[0]) == TLV_CODE_LABEL_REVISION:
name = "Label Revision"
value = t[2 : 2 + ord(t[1])]
elif ord(t[0]) == TLV_CODE_PLATFORM_NAME:
name = "Platform Name"
value = t[2 : 2 + ord(t[1])]
elif ord(t[0]) == TLV_CODE_ONIE_VERSION:
name = "ONIE Version"
value = t[2 : 2 + ord(t[1])]
elif ord(t[0]) == TLV_CODE_MAC_SIZE:
name = "MAC Addresses"
value = str((ord(t[2]) << 8) | ord(t[3]))
elif ord(t[0]) == TLV_CODE_MANUF_NAME:
name = "Manufacturer"
value = t[2 : 2 + ord(t[1])]
elif ord(t[0]) == TLV_CODE_MANUF_COUNTRY:
name = "Manufacture Country"
value = t[2 : 2 + ord(t[1])]
elif ord(t[0]) == TLV_CODE_VENDOR_NAME:
name = "Vendor Name"
value = t[2 : 2 + ord(t[1])]
elif ord(t[0]) == TLV_CODE_DIAG_VERSION:
name = "Diag Version"
value = t[2 : 2 + ord(t[1])]
elif ord(t[0]) == TLV_CODE_SERVICE_TAG:
name = "Service Tag"
value = t[2 : 2 + ord(t[1])]
elif ord(t[0]) == TLV_CODE_VENDOR_EXT:
name = "Vendor Extension"
value = ""
if _TLV_DISPLAY_VENDOR_EXT:
value = t[2 : 2 + ord(t[1])]
elif ord(t[0]) == TLV_CODE_CRC_32 and len(t) == 6:
name = "CRC-32"
value = "0x%08X" % (
((ord(t[2]) << 24) | (ord(t[3]) << 16) | (ord(t[4]) << 8) | ord(t[5])),
)
elif ord(t[0]) == TLV_CODE_RJ_CARID:
name = "rj_cardid"
value = ""
for c in t[2 : 2 + ord(t[1])]:
value += "%02X" % (ord(c),)
else:
name = "Unknown"
value = ""
for c in t[2 : 2 + ord(t[1])]:
value += "0x%02X " % (ord(c),)
return {"name": name, "code": ord(t[0]), "value": value}
def decode_eeprom(e):
total_len = (ord(e[9]) << 8) | ord(e[10])
tlv_index = _TLV_INFO_HDR_LEN
tlv_end = _TLV_INFO_HDR_LEN + total_len
ret = []
while (tlv_index + 2) < len(e) and tlv_index < tlv_end:
rt = decoder(None, e[tlv_index : tlv_index + 2 + ord(e[tlv_index + 1])])
ret.append(rt)
if ord(e[tlv_index]) == TLV_CODE_CRC_32:
break
tlv_index += ord(e[tlv_index + 1]) + 2
for item in ret:
if item["code"] == TLV_CODE_VENDOR_EXT:
rt = decoder(None, item["value"][0 : 0 + 2 + ord(item["value"][0 + 1])])
ret.append(rt)
return ret
def get_sys_eeprom():
eeprom = get_sysfs_value(rg_eeprom)
return decode_eeprom(eeprom)
# get card ID
def getCardId():
ret = get_sys_eeprom()
for item in ret:
if item["code"] == TLV_CODE_RJ_CARID:
return item.get("value", None)
return None
# ====================================
# execute shell command
# ====================================
def os_system(cmd):
status, output = commands.getstatusoutput(cmd)
return status, output
###########################################
# get memory slot and number via DMI command
###########################################
def getsysmeminfo():
ret, log = os_system("which dmidecode ")
if ret != 0 or len(log) <= 0:
error = "cmd find dmidecode"
return False, error
cmd = log + '|grep -P -A5 "Memory\s+Device"|grep Size|grep -v Range'
# get total number first
result = []
ret1, log1 = os_system(cmd)
if ret1 == 0 and len(log1):
log1 = log1.lstrip()
arr = log1.split("\n")
# total = len(arr) # total slot number
for i in range(len(arr)):
val = re.sub("\D", "", arr[i])
if val == "":
val = arr[i].lstrip()
val = re.sub("Size:", "", val).lstrip()
# print val
result.append({"slot": i + 1, "size": val})
return True, result
return False, "error"
###########################################
# get memory slot and number via DMI command
# return various arrays
###########################################
def getsysmeminfo_detail():
ret, log = os_system("which dmidecode ")
if ret != 0 or len(log) <= 0:
error = "cmd find dmidecode"
return False, error
cmd = log + ' -t 17 | grep -A21 "Memory Device"' # 17
# get total number
ret1, log1 = os_system(cmd)
if ret1 != 0 or len(log1) <= 0:
return False, "command execution error[%s]" % cmd
result_t = log1.split("--")
mem_rets = []
for item in result_t:
its = item.replace("\t", "").strip().split("\n")
ret = {}
for it in its:
if ":" in it:
key = it.split(":")[0].lstrip()
value = it.split(":")[1].lstrip()
ret[key] = value
mem_rets.append(ret)
return True, mem_rets
###########################################
# get BIOS info via DMI command
###########################################
def getDmiSysByType(type_t):
ret, log = os_system("which dmidecode ")
if ret != 0 or len(log) <= 0:
error = "cmd find dmidecode"
return False, error
cmd = log + " -t %s" % type_t
# get total number
ret1, log1 = os_system(cmd)
if ret1 != 0 or len(log1) <= 0:
return False, "command execution error[%s]" % cmd
its = log1.replace("\t", "").strip().split("\n")
ret = {}
for it in its:
if ":" in it:
key = it.split(":")[0].lstrip()
value = it.split(":")[1].lstrip()
ret[key] = value
return True, ret
def gethwsys():
return getDmiSysByType(1)
###########################################
# get BIOS info via DMI command
def getsysbios():
return getDmiSysByType(0)
def searchDirByName(name, dir):
result = []
try:
files = os.listdir(dir)
for file in files:
if name in file:
result.append(os.path.join(dir, file))
except Exception as e:
pass
return result
def getUsbLocation():
dir = "/sys/block/"
spect = "sd"
usbpath = ""
result = searchDirByName(spect, dir)
if len(result) <= 0:
return False
for item in result:
with open(os.path.join(item, "removable"), "r") as fd:
value = fd.read()
if value.strip() == "1": # U-Disk found
usbpath = item
break
if usbpath == "": # no U-Disk found
log_debug("no usb found")
return False, usbpath
return True, usbpath
# judge USB file
def getusbinfo():
ret, path = getUsbLocation()
if ret == False:
return False, "not usb exists"
str = os.path.join(path, "size")
ret, value = getfilevalue(str)
if ret == True:
return (
True,
{
"id": os.path.basename(path),
"size": float(value) * 512 / 1024 / 1024 / 1024,
},
)
else:
return False, "Err"
def get_cpu_info():
cmd = "cat /proc/cpuinfo |grep processor -A18" # 17
ret, log1 = os_system(cmd)
if ret != 0 or len(log1) <= 0:
return False, "command execution error[%s]" % cmd
result_t = log1.split("--")
mem_rets = []
for item in result_t:
its = item.replace("\t", "").strip().split("\n")
ret = {}
for it in its:
if ":" in it:
key = it.split(":")[0].lstrip()
value = it.split(":")[1].lstrip()
ret[key] = value
mem_rets.append(ret)
return True, mem_rets
# read file
def get_version_config_info(attr_key, file_name=None):
if file_name is None:
version_conf_filename = "/root/version.json"
else:
version_conf_filename = file_name
if not os.path.isfile(version_conf_filename):
return None
with open(version_conf_filename) as rjconf_file:
for line in rjconf_file:
tokens = line.split("=")
if len(tokens) < 2:
continue
if tokens[0] == attr_key:
return tokens[1].strip()
return None
def io_rd(reg_addr, size=1):
path = "/dev/port"
ret = ""
fd = None
try:
reg_addr = int(reg_addr)
fd = os.open(path, os.O_RDWR|os.O_CREAT)
for i in range(size):
os.lseek(fd, reg_addr+i, os.SEEK_SET)
ret+="{:02x}".format(ord(os.read(fd, 1).decode('latin-1')))
return ret
except Exception as e:
print(str(e))
return None
finally:
if fd: os.close(fd)
def io_wr(reg_addr, reg_data):
u"""io write"""
fd = None
try:
regdata = 0
regaddr = 0
if type(reg_addr) == int:
regaddr = reg_addr
else:
regaddr = int(reg_addr, 16)
if type(reg_data) == int:
regdata = reg_data
else:
regdata = int(reg_data, 16)
devfile = "/dev/port"
fd = os.open(devfile, os.O_RDWR | os.O_CREAT)
os.lseek(fd, regaddr, os.SEEK_SET)
os.write(fd, regdata.to_bytes(2, 'little'))
return True
except ValueError as e:
print(e)
return False
except Exception as e:
print(e)
return False
finally:
if fd: os.close(fd)