Python3: check correct CPU usage in Windows 11 for multicore CPU

check_cpu_load_win11_multicore_cpu.py3

import ctypes
import time

class CPUUsageMonitor:
    """
    A class to monitor CPU usage using the Windows PDH (Performance Data Helper) API via ctypes.
    """

    # PDH constants
    PDH_FMT_DOUBLE = 0x00000200
    ERROR_SUCCESS = 0x00000000

    # Load the PDH DLL
    pdh = ctypes.WinDLL('pdh.dll')

    # Define necessary ctypes types
    PDH_HQUERY = ctypes.c_void_p
    PDH_HCOUNTER = ctypes.c_void_p

    class PDH_FMT_COUNTERVALUE(ctypes.Structure):
        _fields_ = [
            ("CStatus", ctypes.c_ulong),
            ("doubleValue", ctypes.c_double)
        ]

    def __init__(self):
        """
        Initializes the CPUUsageMonitor by setting up the PDH query and counter.
        """
        # Initialize query and counter handles
        self.query_handle = self.PDH_HQUERY()
        self.counter_handle = self.PDH_HCOUNTER()
        self.counter_path = r'\Processor Information(_Total)\% Processor Utility'

        # Define PDH functions
        self.PdhOpenQuery = self.pdh.PdhOpenQueryW
        self.PdhAddCounter = self.pdh.PdhAddEnglishCounterW  # Use English counter names regardless of system locale
        self.PdhCollectQueryData = self.pdh.PdhCollectQueryData
        self.PdhGetFormattedCounterValue = self.pdh.PdhGetFormattedCounterValue
        self.PdhCloseQuery = self.pdh.PdhCloseQuery

        # Open the PDH query
        if not self.open_query():
            raise Exception("Failed to open PDH query.")

        # Add the counter to the query
        if not self.add_counter():
            self.close_query()
            raise Exception("Failed to add counter to PDH query.")

    def open_query(self):
        """
        Opens a new PDH query.

        Returns:
            bool: True if successful, False otherwise.
        """
        ret = self.PdhOpenQuery(None, None, ctypes.byref(self.query_handle))
        if ret != self.ERROR_SUCCESS:
            print(f'PdhOpenQuery failed with error code {ret}')
            return False
        return True

    def add_counter(self):
        """
        Adds a counter to the PDH query.

        Returns:
            bool: True if successful, False otherwise.
        """
        ret = self.PdhAddCounter(
            self.query_handle,
            self.counter_path,
            0,
            ctypes.byref(self.counter_handle)
        )
        if ret != self.ERROR_SUCCESS:
            print(f'PdhAddCounter failed with error code {ret}')
            return False
        return True

    def collect_data(self):
        """
        Collects the CPU usage data from the PDH query.

        Returns:
            float or None: The CPU usage percentage if successful, None otherwise.
        """
        # Collect initial data sample
        ret = self.PdhCollectQueryData(self.query_handle)
        if ret != self.ERROR_SUCCESS:
            print(f'PdhCollectQueryData failed with error code {ret}')
            return None

        # Wait for a second to get the next data sample
        time.sleep(1)

        # Collect second data sample
        ret = self.PdhCollectQueryData(self.query_handle)
        if ret != self.ERROR_SUCCESS:
            print(f'PdhCollectQueryData failed with error code {ret}')
            return None

        # Prepare a structure to receive the formatted counter value
        fmt_value = self.PDH_FMT_COUNTERVALUE()
        ret = self.PdhGetFormattedCounterValue(
            self.counter_handle,
            self.PDH_FMT_DOUBLE,
            None,
            ctypes.byref(fmt_value)
        )
        if ret != self.ERROR_SUCCESS:
            print(f'PdhGetFormattedCounterValue failed with error code {ret}')
            return None

        # Retrieve the CPU usage percentage
        cpu_usage = fmt_value.doubleValue
        return cpu_usage

    def close_query(self):
        """
        Closes the PDH query and releases resources.
        """
        ret = self.PdhCloseQuery(self.query_handle)
        if ret != self.ERROR_SUCCESS:
            print(f'PdhCloseQuery failed with error code {ret}')

def get_cpu_usage(cpu_monitor):
    """
    Handles the logic of collecting CPU usage data and returns the CPU usage percentage.

    Args:
        cpu_monitor (CPUUsageMonitor): An instance of the CPUUsageMonitor class.

    Returns:
        float or None: The CPU usage percentage if successful, None otherwise.
    """
    cpu_usage = cpu_monitor.collect_data()
    return cpu_usage

def main():
    """
    Main function that initializes the CPUUsageMonitor and prints the CPU usage.
    """
    try:
        cpu_monitor = CPUUsageMonitor()
    except Exception as e:
        print(f"Initialization error: {e}")
        return

    try:
        while True:
            # Get CPU usage
            cpu_usage = get_cpu_usage(cpu_monitor)
            if cpu_usage is not None:
                print('Total CPU Usage: {:.1f}%'.format(cpu_usage))
            else:
                print('Failed to retrieve CPU usage.')
            # Wait for 5 seconds before the next reading
            time.sleep(5)
    except KeyboardInterrupt:
        print("Stopping CPU usage monitoring.")
    finally:
        # Ensure the PDH query is closed properly
        cpu_monitor.close_query()

if __name__ == "__main__":
    main()

od 79 linijki to tutaj opakowałbym to w try except niż if else :)

od 79 linijki to tutaj opakowałbym to w try except niż if else :)

jasne refactoring wchodzi