How to compute disk I/O usage percentage in Windows, exactly like Task Manager, for the current process?

Question

I'm trying to compute the disk I/O usage percentage for the current process in Windows, similar to what the Task Manager displays under the "Disk" column. However, I haven't been able to get an accurate match.

I used psutil.Process.io_counters() to retrieve the read/write byte counts for the current process, but it doesn't seem to align with what the Task Manager shows. Specifically, when using libraries like yara for file scanning, the io_counters read/write values don't seem to account for all the I/O operations correctly. It might be due to how YARA handles file reads in memory or how Windows tracks I/O for such operations.

What I need:

A method or tool to calculate the disk I/O usage percentage for the current process, in a way that matches the Windows Task Manager. Preferably, a Python-based approach, but I'm open to any reliable solutions or libraries. Insights into why psutil might not reflect accurate I/O stats for processes using specific libraries like YARA.

What I’ve tried:

Using psutil.Process.io_counters() to get read/write bytes, but the reported values don't match the Task Manager's disk usage. Exploring system-level counters (e.g., Performance Counters in Windows), but I couldn't figure out how to map these to the Task Manager's percentage values. Is there a way to reliably compute this disk I/O percentage for the current process? Or could someone clarify how Task Manager calculates this metric?

Answer 1

First word: as you mentioned to prevent system freezing, it is most likely that you want to monitor ram and cpu usage(now I am not one who is saying ram is performance), still for a task that works so much on a disk, you might have to consider monitoring disc.

Now for getting this data, I don't see a way psutil can do that exactly returning same the Taskmanager does, so use Windows Performance Counters win32pdh. The code is below:

import time
import win32pdh
#current_process = psutil.Process(os.getpid())
#process_name = current_process.name()
process_name = "explorer"  # Comment this

def get_count():
    try:
        # Counter path for % Idle Time

        counter_path_template = r"\Process({})\IO Data Bytes/sec"
        counter_path = counter_path_template.format(process_name)

        # Create a performance counter query
        query_handle = win32pdh.OpenQuery()
        counter = win32pdh.AddCounter(query_handle, counter_path)

        # Collect the initial data sample
        win32pdh.CollectQueryData(query_handle)
        time.sleep(1)  # Wait for the counter to update

        # Collect the second data sample for calculation
        win32pdh.CollectQueryData(query_handle)
        _, value = win32pdh.GetFormattedCounterValue(counter, win32pdh.PDH_FMT_DOUBLE)

        # Active Time (%) = 100 - Idle Time (%)
        val = value/1048576 #1024*1024
        # Clean up
        win32pdh.CloseQuery(query_handle)

        return val

    except Exception as e:
        print(f"Error: {e}")
        return None

if __name__ == "__main__":
    print("Press Ctrl+C to stop.")
    try:
        while True:
            val = get_count()
            if val is not None:
                print(f"Data Usage by {process_name}: {val} MB/s")
    except KeyboardInterrupt:
        print("\nMonitoring stopped.")

The values of this code output match that of task manager, and the values are nearly same(not a decimal precision, but precise like +-2%). Tested while copying a large folder for test which I later deleted.

Uncomment the lines 3,4 comment line 5 for your case of monitoring the python program itself

Running this didn't cause any increase in memory with respect to time not running this code, so I think this should fit your purpose.

helper script:

import win32pdh

def list_all_counters(category):
    """Lists all counters and their instances for a given category."""
    try:
        counters, instances = win32pdh.EnumObjectItems(None, None, category, win32pdh.PERF_DETAIL_WIZARD)
        print(f"\n=== {category} Counters ===")
        for counter in counters:
            print(f"Counter: {counter}")
        if instances:
            print(f"\nInstances:")
            for instance in instances:
                print(f"  {instance}")
    except Exception as e:
        print(f"Error accessing category '{category}': {e}")

def query_counters(category):
    """Queries and prints the values of all counters in a category."""
    try:
        counters, instances = win32pdh.EnumObjectItems(None, None, category, win32pdh.PERF_DETAIL_WIZARD)
        query_handle = win32pdh.OpenQuery()
        counter_handles = []

        # Add counters for all instances
        for counter in counters:
            if instances:
                for instance in instances:
                    counter_path = f"\\{category}({instance})\\{counter}"
                    try:
                        handle = win32pdh.AddCounter(query_handle, counter_path)
                        counter_handles.append((handle, counter_path))
                    except Exception as e:
                        print(f"Could not add counter {counter_path}: {e}")
            else:
                counter_path = f"\\{category}\\{counter}"
                try:
                    handle = win32pdh.AddCounter(query_handle, counter_path)
                    counter_handles.append((handle, counter_path))
                except Exception as e:
                    print(f"Could not add counter {counter_path}: {e}")

        
    except Exception as e:
        print(f"Error accessing category '{category}': {e}")
    finally:
        if 'query_handle' in locals():
            win32pdh.CloseQuery(query_handle)

if __name__ == "__main__":
    categories = ["Process", "Processor", "Memory", "PhysicalDisk", "LogicalDisk"]
    for category in categories:
        list_all_counters(category)  # List counters and instances
        query_counters(category)    # Query and print counter values

Run this script to get all process_name(instance), all counters which you can use. I assumed you need \Process({})\IO Data Bytes/sec change it if required with helper script.