使用 Python 测试磁盘性能

#!python3
import time
import math


def test_speed_write(file_path, buffer=1024, size='1G'):
    size_byte = convert_to_bytes(size)
    n = math.ceil(size_byte/buffer)
    with open(file_path, 'wb') as f:
        start = time.time()
        s = b'0'*buffer
        for i in range(n):
            f.write(s)
        end = time.time()
    time_used = end - start
    return time_used




def test_speed_read(file_path, buffer=1024):
    with open(file_path, 'rb') as f:
        start = time.time()
        while True:
            data = f.read(buffer)
            if not data:
                break
        end = time.time()
    time_used = end - start
    return time_used




def human_readable_time(seconds):
    '''将秒转换为可读的形式'''
    if seconds < 2:
        return str(round(seconds, 2))+' seconds'
    days, remainder = divmod(seconds, 86400)
    hours, remainder = divmod(remainder, 3600)
    minutes, seconds = divmod(remainder, 60)
    time_parts = []
    if days > 0:
        time_parts.append(f"{int(days)} day{'s' if days > 1 else ''}")
    if hours > 0:
        time_parts.append(f"{int(hours)} hour{'s' if hours > 1 else ''}")
    if minutes > 0:
        time_parts.append(f"{int(minutes)} minute{'s' if minutes > 1 else ''}")
    if seconds > 0:
        time_parts.append(f"{int(seconds)} second{'s' if seconds > 1 else ''}")
    return ', '.join(time_parts) if len(time_parts) > 0 else '0 seconds'




def human_readable_size(size_bytes):
    """
    Convert a number of bytes into a human-readable string


    print(convert_to_bytes('1K'))   # 1024
    print(convert_to_bytes('1M'))   # 1048576
    print(convert_to_bytes('1G'))   # 1073741824


    """
    for unit in ['', 'K', 'M', 'G', 'T', 'P', 'E', 'Z']:
        if size_bytes < 1024:
            return f"{size_bytes:.2f} {unit}B"
        size_bytes = 1024
    return f"{size_bytes:.2f} YB"




def convert_to_bytes(size_str):
    """
    Convert a string with a unit (K, M, G, etc.) to a number of bytes
    """
    units = {'K': 1024, 'M': 1024**2, 'G': 1024**3, 'T': 1024**4, 'P': 1024**5}
    size_str = size_str.upper()
    for unit, multiplier in units.items():
        if size_str.endswith(unit):
            size = float(size_str[:-1]) * multiplier
            return int(size)
    return int(size_str)




def test_disk_speed(file_name, buffer=1024, size='1G'):
    size_byte = convert_to_bytes(size)
    n = math.ceil(size_byte/buffer)


    t1 = test_speed_write(file_name, buffer, size)
    t2 = test_speed_read(file_name, buffer)
    read_speed = human_readable_size(size_byte/t2)+'/s'
    write_speed = human_readable_size(size_byte/t1)+'/s'


    # 表格数据
    data = [
        ["类型", "Read", "Write"],
        ["速度", read_speed, write_speed],
        ["耗时", human_readable_time(t2), human_readable_time(t1)]
    ]


    # 计算每列的最大宽度
    widths = [max(len(str(x)) for x in col) for col in zip(*data)]


    print("样本：%s size: %s buffer: %d byte" % (file_name, size, buffer))
    print('='*50)
    # 输出数据行
    for row in data:
        print("|".join(str(x).ljust(width) for x, width in zip(row, widths)))
    print()




if __name__ == '__main__':
    test_disk_speed(r'd:\testio.log', buffer=1024, size='2G')
    test_disk_speed(r'c:\testio1.log', buffer=512, size='500M')