sysinfo_redfish.py

#!/usr/bin/env python3
import json
import warnings
from collections import namedtuple

import requests
import os

warnings.filterwarnings("ignore")
URL = {
    "redfish": "https://%s/redfish/v1",
    "chassis": "https://%s/redfish/v1/Chassis/1",
    "systems": "https://%s/redfish/v1/Systems/1",
    "bmc": "https://%s/redfish/v1/Managers/1",
    "bmcEthernet": "https://%s/redfish/v1/Managers/1/EthernetInterfaces/1",
    "memory": "https://%s/redfish/v1/Systems/1/Memory",
    "processors": "https://%s/redfish/v1/Systems/1/Processors",
}


class System(object):
    def __init__(self, ip, username, password):
        self.ip = ip
        self.assetTag = ""
        self.username = username
        self.password = password
        self.boardModel = ""
        self.biosVersion = ""
        self.bmcMAC = ""
        self.firmwareVersion = ""
        self.processorCount = ""
        self.processors = list()
        self.totalMemory = ""
        self.memory = list()
        self.session = requests.session()

    def get_board_model(self):

        url = URL.get("chassis", None) % self.ip
        response = self.session.get(
            url, verify=False, auth=(self.username, self.password)
        )
        if response.status_code == 200:
            data = response.json()
            self.boardModel = data.get("Model", None)
            self.assetTag = data.get("AssetTag", None)

    def get_firmware(self):
        url = URL.get("bmc", None) % self.ip
        response = self.session.get(
            url, verify=False, auth=(self.username, self.password)
        )
        if response.status_code == 200:
            data = response.json()
            self.firmwareVersion = data.get("FirmwareVersion")

    def get_bmc_mac(self):
        url = URL.get("bmcEthernet", None) % self.ip
        response = self.session.get(
            url, verify=False, auth=(self.username, self.password)
        )
        if response.status_code == 200:
            data = response.json()
            self.bmcMAC = data.get("MACAddress")

    def get_system_info(self):
        url = URL.get("systems", None) % self.ip
        response = self.session.get(
            url, verify=False, auth=(self.username, self.password)
        )
        if response.status_code == 200:
            data = response.json()
            cpu_summary = data["ProcessorSummary"]
            mem_summary = data["MemorySummary"]
            self.processorCount = cpu_summary.get("Count", None)
            self.totalMemory = mem_summary.get("TotalSystemMemoryGiB", None)
            self.biosVersion = data.get("BiosVersion", None)

    def get_processors(self):
        url = URL.get("processors", None) % self.ip
        response = self.session.get(
            url, verify=False, auth=(self.username, self.password)
        )
        data = response.json()

        if response.status_code != 200:
            raise requests.HTTPError(response.raise_for_status)
        for cpu in data["Members"]:
            response = self.session.get(
                "https://%s%s" % (self.ip, cpu["@odata.id"]),
                verify=False,
                auth=(self.username, self.password),
            )
            cpu_data = response.json()
            health = cpu_data["Status"]["Health"]
            model = cpu_data["Model"]
            socket = cpu_data["Socket"]
            cpuinfo = namedtuple("CPU", ["socket", "model", "health"])
            self.processors.append(cpuinfo(socket, model, health))

    def get_memory(self):
        url = URL.get("memory", None) % self.ip
        response = self.session.get(
            url, verify=False, auth=(self.username, self.password)
        )
        if response.status_code != 200:
            raise requests.HTTPError(response.raise_for_status)
        data = response.json()

        for slot in data["Members"]:
            response = self.session.get(
                "https://%s%s" % (self.ip, slot["@odata.id"]),
                verify=False,
                auth=(self.username, self.password),
            )
            if response.status_code != 200:
                raise requests.HTTPError(response.raise_for_status)
            memory = response.json()
            meminfo = namedtuple("MEMORY", ["slot", "capacity", "speed", "health"])
            capacity = int(memory["CapacityMiB"]) / 1024
            speed = memory["AllowedSpeedsMHz"][0]
            health = memory["Status"]["Health"]
            location = memory["DeviceLocator"]
            self.memory.append(meminfo(location, capacity, speed, health))

    def set_assetTag(self, assetTag):
        data = {"AssetTag": assetTag}
        header = {"Content-Type": "application/json", "OData-Version": "4.0"}
        url = URL.get("chassis", None) % self.ip
        response = self.session.patch(
            url,
            verify=False,
            data=json.dumps(data),
            header=header,
            auth=(self.username, self.password),
        )
        return assetTag if response.status_code == 200 else None

    def print_cpus(self):
        print(f"Processors Information: {len(self.processors)} CPUs")
        for cpu in self.processors:
            print(f"{cpu.socket}: {cpu.model} {cpu.health}")

    def print_memory(self):
        print(f"Memory Information: {len(self.memory)} DIMMS")
        for mem in self.memory:
            print(f"{mem.slot}: {mem.capacity}GiB {mem.speed}MHz {mem.health}")

    def get_system_summary(self):
        self.get_board_model()
        self.get_firmware()
        self.get_bmc_mac()
        self.get_system_info()
        self.get_processors()
        self.get_memory()

    def __str__(self):
        msg = f"""
System        : {self.assetTag}
IP            : {self.ip}
Board Model   : {self.boardModel}
BIOS Version  : {self.biosVersion}
BMC MAC       : {self.bmcMAC}
BMC Firmware  : {self.firmwareVersion}
CPU Total     : {self.processorCount}
Memory Total  : {self.totalMemory} GiB
        """
        return msg


def get_args():
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument("-i", "--ip", action="store")
    parser.add_argument("-f", "--file", action="store")
    parser.add_argument("-u", "--user", action="store")
    parser.add_argument("-p", "--password", action="store")

    return parser.parse_args()


def print_system_info(ip, username, password):
    system = System(ip, username, password)
    system.get_system_summary()
    print(system)
    system.print_cpus()
    system.print_memory()


if __name__ == "__main__":
    try:
        args = get_args()
        if args.ip and args.user and args.password:
            username = args.user
            password = args.password
            ip = args.ip
            print(f"Wait! Getting system information: {ip}")
            print_system_info(ip, username, password)

        elif args.file:
            filename = args.file
            if not os.path.exists(filename):
                print(f"Error, File '{filename}' does not exists")
                exit(1)
            with open(filename, "r") as f:
                lines = f.readlines()
                for num, line in enumerate(lines):
                    ip, username, password = line.split()
                    print(f"System {num+1} of {len(lines)}")
                    print_system_info(ip, username, password)
                print("Done!")
        else:
            print("usage: system.py [-h] (-i | -f) [-u USER] [-p PASSWORD]")
            exit(1)
    except KeyboardInterrupt:
        exit(1)