刚学python，来个大佬帮我把这个代码改成GPU加速运行的，..... - 精易论坛

[Python] 纯文本查看 复制代码
import cupy as cp  # 导入CuPy库代替NumPy

import scipy.signal as signal

import wave

import struct

import matplotlib.pyplot as plt



# 定义所有滤波器函数，但使用CuPy数组

def allpass(input_signal, delay, gain):

    B = cp.zeros(delay)

    B[0] = gain

    B[delay-1] = 1

    A = cp.zeros(delay)

    A[0] = 1

    A[delay-1] = gain

    output_signal = cp.zeros(input_signal.shape)

    output_signal = signal.lfilter(B, A, input_signal)  # 注意：lfilter目前不支持CuPy数组

    return output_signal



def comb(input_signal, delay, gain):

    B = cp.zeros(delay)

    B[delay-1] = 1

    A = cp.zeros(delay)

    A[0] = 1

    A[delay-1] = -gain

    output_signal = cp.zeros(input_signal.shape)

    output_signal = signal.lfilter(B, A, input_signal)  # 注意：lfilter目前不支持CuPy数组

    return output_signal



def comb_with_lp(input_signal, delay, g, g1):

    g2 = g * (1 - g1)

    B = cp.zeros(delay + 1)

    B[delay-1] = 1

    B[delay] = -g1

    A = cp.zeros(delay)

    A[0] = 1

    A[1] = -g1

    A[delay-1] = -g2

    output_signal = cp.zeros(input_signal.shape)

    output_signal = signal.lfilter(B, A, input_signal)  # 注意：lfilter目前不支持CuPy数组

    return output_signal



def delay(input_signal, delay, gain=1):

    output_signal = cp.concatenate((cp.zeros(delay), input_signal))

    output_signal = output_signal * gain

    return output_signal



def main():

    # 打开/生成测试WAV信号

    # 克罗内克δ函数

    # sample = cp.zeros((2, 88200))

    # sample[:, 0] = 1

    # WAV 文件

    sample_in = 'D:\\321\\555.wav'

    frame_rate = 44100.0

    wav_file = wave.open(sample_in, 'r')

    num_samples_sample = wav_file.getnframes()

    num_channels_sample = wav_file.getnchannels()

    sample = wav_file.readframes(num_samples_sample)

    total_samples_sample = num_samples_sample * num_channels_sample

    wav_file.close()



    # 将二进制数据转换为CuPy数组

    sample = struct.unpack('{n}h'.format(n=total_samples_sample), sample)

    sample = cp.array([sample[0::2], sample[1::2]], dtype=cp.float64)

    sample[0] /= cp.max(cp.abs(sample[0]), axis=0)

    sample[1] /= cp.max(cp.abs(sample[1]), axis=0)



    # 初始化算法变量

    stereospread = 23  # 立体分布大小

    delays_r = [605, 969, 1356, 1562, 1785, 2100, 2600, 3100, 3581, 4025]  # 左声道延迟

    delays_l = [d + stereospread for d in delays_r]  # 右声道延迟

    delays_early = [477, 821, 1161, 1511, 1721, 1951, 2310, 2542, 2751, 2952]  # 早期延迟

    gains_early = [0.87, 0.818, 0.635, 0.719, 0.267, 0.242, 0.192, 0.172, 0.151, 0.111]  # 增益

    g1_list = [0.21, 0.23, 0.25, 0.27, 0.28, 0.25, 0.21, 0.19, 0.17, 0.14]  # 延迟比例

    g = 0.9  # 干音比例

    rev_to_er_delay = 2000

    allpass_delay = 186

    allpass_g = 0.7

    output_gain = 0.075

    dry = 1

    wet = 1

    width = 1

    wet1 = wet * (width / 2 + 0.5)

    wet2 = wet * ((1 - width) / 2)

    early_reflections_r = cp.zeros(sample[0].size)

    early_reflections_l = cp.zeros(sample[1].size)

    combs_out_r = cp.zeros(sample[0].size)

    combs_out_l = cp.zeros(sample[1].size)



    # 主算法部分

    for i in range(10):

        early_reflections_r += delay(sample[0], delays_early, gains_early)[:sample[0].size]

        early_reflections_l += delay(sample[1], delays_early, gains_early)[:sample[1].size]



    for i in range(6):

        combs_out_r += comb_with_lp(sample[0], delays_r, g, g1_list)

        combs_out_l += comb_with_lp(sample[1], delays_l, g, g1_list)



    reverb_r = allpass(combs_out_r, allpass_delay, allpass_g)

    reverb_l = allpass(combs_out_l, allpass_delay, allpass_g)



    early_reflections_r = cp.concatenate((early_reflections_r, cp.zeros(rev_to_er_delay)))

    early_reflections_l = cp.concatenate((early_reflections_l, cp.zeros(rev_to_er_delay)))



    reverb_r = delay(reverb_r, rev_to_er_delay)

    reverb_l = delay(reverb_l, rev_to_er_delay)



    reverb_out_r = early_reflections_r + reverb_r

    reverb_out_l = early_reflections_l + reverb_l



    reverb_out_r = output_gain * ((reverb_out_r * wet1 + reverb_out_l * wet2) + cp.concatenate((sample[0], cp.zeros(rev_to_er_delay))) * dry)

    reverb_out_l = output_gain * ((reverb_out_l * wet1 + reverb_out_r * wet2) + cp.concatenate((sample[1], cp.zeros(rev_to_er_delay))) * dry)



    # 写入文件

    signal_integer_r = (reverb_out_r * int(cp.iinfo(cp.int16).max)).get().astype(cp.int16)  # 转换回主机内存

    signal_integer_l = (reverb_out_l * int(cp.iinfo(cp.int16).max)).get().astype(cp.int16)  # 转换回主机内存

    signal_to_render = cp.empty((signal_integer_r.size + signal_integer_l.size), dtype=cp.int16)

    signal_to_render[0::2] = signal_integer_r

    signal_to_render[1::2] = signal_integer_l

    nframes = total_samples_sample

    comptype = "NONE"

    compname = "not compressed"

    nchannels = 2

    sampwidth = 2

    wav_file_write = wave.open('D:\\321\\222.wav', 'w')

    wav_file_write.setparams((nchannels, sampwidth, int(frame_rate), nframes, comptype, compname))

    for s in range(nframes):

        wav_file_write.writeframes(struct.pack('h', signal_to_render.get()))  # 转换回主机内存

    wav_file_write.close()

    print("处理完毕")



if __name__ == "__main__":

    main()