#!/usr/bin/python3 # -*- coding: UTF-8 -*- """ * onboard temperature sensors * FAN trays * PSU """ import os import glob from lxml import etree as ET MAILBOX_DIR = "/sys/bus/i2c/devices/" PORTS_DIR = "/sys/class/net/" CONFIG_NAME = "dev.xml" def getPMCreg(location): retval = "ERR" if not os.path.isfile(location): return "%s %s notfound" % (retval, location) try: with open(location, "r") as fd: retval = fd.read() except Exception: pass # logging.error("Unable to open ", location, "file !") retval = retval.rstrip("\r\n") retval = retval.lstrip(" ") return retval # Get a mailbox register def get_pmc_register(reg_name): retval = "ERR" if reg_name[0:4] == "/rif" or reg_name[0:4] == "/ma1" or reg_name[0:4] == "/eth": mb_reg_file = PORTS_DIR + reg_name else: 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] # use first found patch 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: pass # logging.error("Unable to open ", mb_reg_file, "file !") retval = retval.rstrip("\r\n") retval = retval.lstrip(" ") return retval class checktype: def __init__(self, test1): self.test1 = test1 @staticmethod def check(name, location, bit, value, tips, err1): psu_status = int(get_pmc_register(location), 16) val = (psu_status & (1 << bit)) >> bit if val != value: err1["errmsg"] = tips err1["code"] = -1 return -1 else: err1["errmsg"] = "none" err1["code"] = 0 return 0 @staticmethod def getValue(location, bit, type): value_t = get_pmc_register(location) if value_t.startswith("ERR"): return value_t if type == 1: return float(value_t) / 1000 elif type == 2: return float(value_t) / 100 elif type == 3: psu_status = int(value_t, 16) return (psu_status & (1 << bit)) >> bit elif type == 4: return int(value_t, 10) else: return value_t # temperature @staticmethod def getTemp(self, name, location, ret_t): ret2 = self.getValue(location + "temp1_input", " ", 1) ret3 = self.getValue(location + "temp1_max", " ", 1) ret4 = self.getValue(location + "temp1_max_hyst", " ", 1) ret_t["temp1_input"] = ret2 ret_t["temp1_max"] = ret3 ret_t["temp1_max_hyst"] = ret4 @staticmethod def getLM75(name, location, result): c1 = checktype r1 = {} c1.getTemp(c1, name, location, r1) result[name] = r1 class status: def __init__(self, productname): self.productname = productname @staticmethod def getETroot(filename): tree = ET.parse(filename) root = tree.getroot() return root @staticmethod def getDecodValue(collection, decode): decodes = collection.find("decode") testdecode = decodes.find(decode) test = {} for neighbor in testdecode.iter("code"): test[neighbor.attrib["key"]] = neighbor.attrib["value"] return test @staticmethod def getfileValue(location): return checktype.getValue(location, " ", " ") @staticmethod def getETValue(a, filename, tagname): root = status.getETroot(filename) for neighbor in root.iter(tagname): prob_t = {} prob_t = neighbor.attrib prob_t["errcode"] = 0 prob_t["errmsg"] = "" for pros in neighbor.iter("property"): ret = dict(neighbor.attrib.items() + pros.attrib.items()) if "type" not in ret.keys(): val = "0" else: val = ret["type"] if "bit" not in ret.keys(): bit = "0" else: bit = ret["bit"] s = checktype.getValue(ret["location"], int(bit), int(val)) if isinstance(s, str) and s.startswith("ERR"): prob_t["errcode"] = -1 prob_t["errmsg"] = s if "default" in ret.keys(): rt = status.getDecodValue(root, ret["decode"]) prob_t["errmsg"] = rt[str(s)] if str(s) != ret["default"]: prob_t["errcode"] = -1 break else: if "decode" in ret.keys(): rt = status.getDecodValue(root, ret["decode"]) if ( ret["decode"] == "psutype" and s.replace("\x00", "").rstrip() not in rt.keys() ): # PSU type detect prob_t["errcode"] = -1 prob_t["errmsg"] = "%s" % ("The power type does not match, please check whether the power is correct!") else: s = rt[str(s).replace("\x00", "").rstrip()] name = ret["name"] prob_t[name] = str(s) a.append(prob_t) @staticmethod def getCPUValue(a, filename, tagname): root = status.getETroot(filename) for neighbor in root.iter(tagname): location = neighbor.attrib["location"] L = [] for dirpath, dirnames, filenames in os.walk(location): for file in filenames: if file.endswith("input"): L.append(os.path.join(dirpath, file)) L = sorted(L, reverse=False) for i in range(len(L)): prob_t = {} prob_t["name"] = getPMCreg("%s/temp%d_label" % (location, i + 1)) prob_t["temp"] = ( float(getPMCreg("%s/temp%d_input" % (location, i + 1))) / 1000 ) prob_t["alarm"] = ( float(getPMCreg("%s/temp%d_crit_alarm" % (location, i + 1))) / 1000 ) prob_t["crit"] = ( float(getPMCreg("%s/temp%d_crit" % (location, i + 1))) / 1000 ) prob_t["max"] = float(getPMCreg("%s/temp%d_max" % (location, i + 1))) / 1000 a.append(prob_t) @staticmethod def getFileName(): return os.path.dirname(os.path.realpath(__file__)) + "/" + CONFIG_NAME @staticmethod def getFan(ret): _filename = status.getFileName() _tagname = "fan" status.getvalue(ret, _filename, _tagname) @staticmethod def checkFan(ret): _filename = status.getFileName() # _filename = "/usr/local/bin/" + status.getFileName() _tagname = "fan" status.getETValue(ret, _filename, _tagname) @staticmethod def getTemp(ret): _filename = status.getFileName() # _filename = "/usr/local/bin/" + status.getFileName() _tagname = "temp" status.getETValue(ret, _filename, _tagname) @staticmethod def getPsu(ret): _filename = status.getFileName() # _filename = "/usr/local/bin/" + status.getFileName() _tagname = "psu" status.getETValue(ret, _filename, _tagname) @staticmethod def getcputemp(ret): _filename = status.getFileName() _tagname = "cpus" status.getCPUValue(ret, _filename, _tagname) @staticmethod def getMgmtRx(ret): _filename = status.getFileName() # _filename = "/usr/local/bin/" + status.getFileName() _tagname = "mgmt_rx" status.getETValue(ret, _filename, _tagname)