1_2.py

import math
from time import strftime, localtime
import time
import numpy as np
import pandas as pd
from numba import jit, njit
from multiprocessing import Pool, cpu_count
import ctypes
import os
import platform
import sys
import wmi
import numpy.ctypeslib as npct

# 全局变量
loop = 5
subloop = 5
winsize = 100
length = 1000000

# 创建测试数据 - 使用 int32 类型与原始代码保持一致
data = np.random.randint(1, 2000000000, size=(length), dtype=np.int32)

# 系统信息相关函数
def sysinfo():
    c = wmi.WMI()
    dictRet = {}
    for sys in c.Win32_OperatingSystem():
        dictRet['OS'] = '{} {} {}'.format(sys.Caption, sys.BuildNumber, sys.OSArchitecture)
    for processor in c.Win32_Processor():
        dictRet['CPU'] = '{} # Threads : {}'.format(processor.Name.strip(), cpu_count())
    listGPU = []
    for gpu in c.Win32_VideoController():
        listGPU.append('{} # {}'.format(gpu.Name, gpu.DriverVersion))
    if len(listGPU) == 1:
        dictRet['GPU'] = listGPU[0]
    if len(listGPU) > 1:
        for i in range(len(listGPU)):
            dictRet['GPU {}'.format(i)] = listGPU[i]
    listMemory = []
    for Memory in c.Win32_PhysicalMemory():
        listMemory.append(int(Memory.Capacity))
    output = ''
    for m in listMemory:
        output += '{} MB + '.format(int(m/(1024*1024)))
    output = output.rstrip().rstrip('+') + ' = {} MB'.format(int(sum(listMemory)/(1024*1024)))
    dictRet['MEM'] = output
    for n in c.Win32_ComputerSystem():
        dictRet['Name'] = n.name.replace(' ','_')
    return dictRet

def getSysInfo():
    name = ''
    output = ''
    dictInfo = sysinfo()
    output = '###### System Information \n'
    output += '||| \n'
    output += '|:---|:---| \n'
    for key in dictInfo:
        if key == 'Name':
            name = dictInfo[key]
            continue
        output += '|{}|{}| \n'.format(key,dictInfo[key])
    date = strftime("%Y-%m-%d %H:%M:%S", localtime()) 
    output += '|{}|{}| \n'.format('TIME', date)
    return name, output

def getTestInfo():
    global loop, subloop, winsize, length, data
    datatype = data.dtype
    output = '###### Test Information \n'
    output += '||| \n'
    output += '|:---|:---| \n'
    output += '|{}|{} * {}| \n'.format('Loop Number', loop, subloop)
    output += '|{}|{:,}| \n'.format('Data Size', length)
    output += '|{}|{}| \n'.format('Window Size', winsize)
    output += '|{}|{}| \n'.format('Data Type', str(datatype).lstrip('<class').rstrip('>').strip().strip("'"))
    return output

def saveResult(result, loop):
    output = ''
    name, sysinfo = getSysInfo()
    output += sysinfo
    output += getTestInfo()
    output += '###### Test Result \n'
    t = ''
    for i in range(loop):
        t += '|{}'.format(i+1)
    output += '|Name|Infomation'+t+'|Avg|Faster| \n'
    output += '|:---|:---:'+'|---:'*(loop+2)+'| \n'
    for key in result:
        info = '|{}|{}'.format(key, result[key][0])
        for data in result[key][1:]:
            info += '|{}'.format('%.3f'%data)
        info += '| \n'
        output += info
    if name != '':
        filename ='Result_{}.md'.format(name)
    else:
        filename = 'Result.md'
    f = open(filename, 'w', encoding='utf-8')
    f.write(output)
    f.close()
    return

# 方法检查
def printDiff(d1, d2):
    n = d1.size
    for i in range(n):
        if d1[i] != d2[i]:
            print(i, d1[i], d2[i])
            
def checkRet(dictRet):
    listKey = list(dictRet.keys())
    number = len(listKey)
    for i in range(number-1):
        for n in range(i+1, number):
            keyi = listKey[i]
            keyn = listKey[n]
            if (dictRet[keyi].size != dictRet[keyn].size):
                print('checkRet size difference {} : {} != {} : {} '.format(keyi, dictRet[keyi].size, keyn, dictRet[keyn].size))
                return False
    print('checkRet size', dictRet[listKey[0]].size)
    for i in range(number-1):
        for n in range(i+1, number):
            keyi = listKey[i]
            keyn = listKey[n]
            if ((dictRet[keyi] == dictRet[keyn]).all() != True):            
                print('checkRet result False: {} {}'.format(keyi, keyn))
                print(type(dictRet[keyi][0]), type(dictRet[keyn][0]))
                printDiff(dictRet[keyi], dictRet[keyn])
                return False
    print('checkRet result True')
    return True

# 1. 标准 Python Numpy 实现
def calMa(d, window):
    adjust = int(window/2)
    n = d.size
    m = np.zeros(n, dtype=np.float64)
    for i in range(n-window+1):
        m[i+adjust] = np.sum(d[i:i+window], dtype=np.int64)/window
    return m

# 2. Python Numpy Convolve 实现
def calMaCov(d, window):
    if d.size <= window:
        return np.zeros(d.size, dtype=np.float64)
    adjust = int(window/2)
    head = np.zeros(adjust, dtype=np.float64)
    tail = np.zeros(window-adjust-1, dtype=np.float64)
    body = np.convolve(d, np.ones(window), 'valid')/window
    return np.concatenate([head, body, tail])

# 3. Numpy 优化实现
def calMaOpt(d, window):
    if d.size <= window:
        return np.zeros(d.size, dtype=np.float64)
    adjust = int(window/2)
    n = d.size
    m = np.zeros(n, dtype=np.float64)
    s = np.sum(d[:window], dtype=np.int64)
    m[adjust] = s/window
    for i in range(1, n-window+1):
        s = s - d[i-1] + d[i+window-1]
        m[i+adjust] = s/window
    return m

# 4. Numba JIT 实现
@njit
def calMaJIT(d, window):
    adjust = int(window/2)
    n = d.size
    m = np.zeros(n, dtype=np.float64)
    for i in range(n-window+1):
        m[i+adjust] = np.sum(d[i:i+window])/window
    return m

# 5. Numba AOT 实现 (使用 njit 装饰器)
@njit
def calMaAOT(d, window):
    adjust = int(window/2)
    n = d.size
    m = np.zeros(n, dtype=np.float64)
    for i in range(n-window+1):
        s = 0
        for j in range(window):
            s += d[i+j]
        m[i+adjust] = s/window
    return m

# 6. Cython 实现
def calMaCY(d, window):
    # 尝试导入已编译的Cython模块
    from calMaCY import calMaCY as cy_func
    return cy_func(d, window)
    

# 7. C 模块实现 (VC)
def calMaPyC(d, window):
    # 使用ctypes调用编译好的DLL
    import ctypes
    import os
    import numpy as np
    
    # 查找DLL路径
    current_dir = os.path.dirname(os.path.abspath(__file__))
    dll_path = os.path.join(current_dir, 'calMaDLL.dll')
    
    if not os.path.exists(dll_path):
        print(f"找不到DLL: {dll_path}")
        return calMa(d, window)  # 降级到标准Python实现
        
    # 加载DLL
    dll = ctypes.CDLL(dll_path)
    
    # 准备输入和输出数组
    n = d.size
    result = np.zeros(n, dtype=np.float64)
    
    # 确保输入数据类型正确
    if d.dtype != np.int32:
        d = d.astype(np.int32)
    
    # 设置函数签名
    dll.calMaDLL.argtypes = [
        ctypes.POINTER(ctypes.c_int),
        ctypes.POINTER(ctypes.c_double),
        ctypes.c_int,
        ctypes.c_int
    ]
    
    # 获取数组内存指针
    d_ptr = d.ctypes.data_as(ctypes.POINTER(ctypes.c_int))
    result_ptr = result.ctypes.data_as(ctypes.POINTER(ctypes.c_double))
    
    # 调用DLL函数
    dll.calMaDLL(d_ptr, result_ptr, n, window)
    
    return result

# 8. C 模块实现 (GCC)
def calMaPyCG(d, window):
    # 使用ctypes调用编译好的DLL
    import ctypes
    import os
    import numpy as np
    
    # 查找DLL路径
    current_dir = os.path.dirname(os.path.abspath(__file__))
    dll_path = os.path.join(current_dir, 'calMaDLL_gcc.dll')
        
    # 加载DLL
    dll = ctypes.CDLL(dll_path)
    
    # 准备输入和输出数组
    n = d.size
    result = np.zeros(n, dtype=np.float64)
    
    # 确保输入数据类型正确
    if d.dtype != np.int32:
        d = d.astype(np.int32)
    
    # 设置函数签名
    dll.calMaDLL.argtypes = [
        ctypes.POINTER(ctypes.c_int),
        ctypes.POINTER(ctypes.c_double),
        ctypes.c_int,
        ctypes.c_int
    ]
    
    # 获取数组内存指针
    d_ptr = d.ctypes.data_as(ctypes.POINTER(ctypes.c_int))
    result_ptr = result.ctypes.data_as(ctypes.POINTER(ctypes.c_double))
    
    # 调用DLL函数
    dll.calMaDLL(d_ptr, result_ptr, n, window)
    
    return result

# 9. DLL 实现 (VC)
# 直接定义 DLL 加载和调用函数
dlllib = npct.load_library("calMaDLL", ".")
dlllib.calMaDLL.argtypes = [
    npct.ndpointer(dtype=np.int32, ndim=1, flags="C_CONTIGUOUS"),
    npct.ndpointer(dtype=np.float64, ndim=1, flags="C_CONTIGUOUS"),
    ctypes.c_int, ctypes.c_int
]

def calMaDLL(d, window):
    n = d.size
    m = np.empty((n,), dtype=np.float64)
    dlllib.calMaDLL(d, m, n, window)
    return m

# 10. DLL 实现 (GCC)
dlllib_gcc = npct.load_library("calMaDLL_gcc", ".")
dlllib_gcc.calMaDLL.argtypes = [
    npct.ndpointer(dtype=np.int32, ndim=1, flags="C_CONTIGUOUS"),
    npct.ndpointer(dtype=np.float64, ndim=1, flags="C_CONTIGUOUS"),
    ctypes.c_int, ctypes.c_int
]

def calMaDLLG(d, window):
    n = d.size
    m = np.empty((n,), dtype=np.float64)
    dlllib_gcc.calMaDLL(d, m, n, window)
    return m

# 11. OpenMP 实现
omplib = npct.load_library("calMaOMP", ".")
omplib.calMaDLL.argtypes = [
    npct.ndpointer(dtype=np.int32, ndim=1, flags="C_CONTIGUOUS"),
    npct.ndpointer(dtype=np.float64, ndim=1, flags="C_CONTIGUOUS"),
    ctypes.c_int, ctypes.c_int
]

def calMaOMP(d, window):
    n = d.size
    m = np.empty((n,), dtype=np.float64)
    omplib.calMaDLL(d, m, n, window)
    return m

# 测量执行时间
def measure_time(func_name, loops=5, subloops=5):
    global data, winsize
    times = []
    
    func = eval(func_name)
    
    # 预热
    func(data[:1000], min(winsize, 10))
    
    for i in range(loops):
        total_time = 0
        for _ in range(subloops):
            start_time = time.time()
            func(data, winsize)
            end_time = time.time()
            total_time += (end_time - start_time) * 1000  # 转换为毫秒
        avg_time = total_time / subloops
        times.append(avg_time)
        print(f"{func_name} : {avg_time:.3f} ms")
    
    return times

# 主函数
def main():
    global loop, subloop, winsize, data

    print(f"初始化数据...大小:{length}, 窗口大小:{winsize}")
    
    # 测试方法和描述的字典
    dictTest = {
        'calMa': 'Standard Python Numpy',
        'calMaCov': 'Python Numpy Convolve',
        'calMaOpt': 'Numpy Optimized Ma',
        'calMaJIT': 'Numba JIT',
        'calMaAOT': 'Numba AOT',
        'calMaCY': 'Cython Module',
        'calMaPyC': 'C Module VC',
        'calMaPyCG': 'C Module GCC',
        'calMaDLL': 'ctypes DLL VC',
        'calMaDLLG': 'ctypes DLL GCC',
        'calMaOMP': 'ctypes DLL VC OpenMP',
    }
    
    # 测试性能
    result = {}
    base_time = None
    
    print(f"Running {loop}*{subloop} tests for each method...")
    
    for key in dictTest:
        print(f"测试 {key}...")
        result[key] = [dictTest[key]]
        times = measure_time(key, loop, subloop)
        for t in times:
            result[key].append(t)
        
        # 计算平均时间
        avg_time = sum(times) / len(times)
        result[key].append(avg_time)
        
        # 设置基准时间
        if key == 'calMa':  # 使用标准Python作为基准
            base_time = avg_time
    
    # 计算速度比
    for key in result:
        speedup = base_time / result[key][-1]
        result[key].append(speedup)
    
    # 保存结果
    saveResult(result, loop)
    
    print("\n测试完成，结果已保存")
    
    # 打印简短的性能总结
    print("\n性能比较:")
    for key in result:
        avg_time = result[key][-2]  # 倒数第二个是平均时间
        speedup = result[key][-1]   # 最后一个是加速比
        print(f"{key:<10} - 平均: {avg_time:.3f} ms, 加速比: {speedup:.2f}x")

if __name__ == "__main__":
    main()