python WAV音频文件处理——(3)高效处理大型 WAV 文件

简介: python WAV音频文件处理——(3)高效处理大型 WAV 文件

由于 WAV 文件通常包含未压缩的数据,因此它们的体积可能很大。这可能会使它们的处理速度非常慢,甚至阻止您一次将整个文件放入内存中。

实时动画

您可以使用滑动窗口技术在播放时可视化音频的一小部分,而不是绘制整个或部分 WAV 文件的静态波形。这将通过实时更新绘图来产生有趣的示波器效果:

plot_oscilloscope.py

from argparse import ArgumentParser
from pathlib import Path

import numpy as np
import matplotlib.pyplot as plt

from waveio import WAVReader


def slide_window(window_seconds, wav):
    num_windows = round(wav.metadata.num_seconds / window_seconds)
    for i in range(num_windows):
        begin_seconds = i * window_seconds
        end_seconds = begin_seconds + window_seconds
        channels = wav.channels_sliced(begin_seconds, end_seconds)
        yield np.mean(tuple(channels), axis=0)

def animate(filename, seconds, windows):
    try:
        plt.style.use("dark_background")
    except OSError:
        pass  # Fall back to the default style

    fig, ax = plt.subplots(figsize=(16, 9))
    fig.canvas.manager.set_window_title(filename)

    plt.tight_layout()
    plt.box(False)

    for window in windows:
        plt.cla()
        ax.set_xticks([])
        ax.set_yticks([])
        ax.set_ylim(-1.0, 1.0)
        plt.plot(window)
        plt.pause(seconds)


def main():
    args = parse_args()
    with WAVReader(args.path) as wav:
        animate(
            args.path.name,
            args.seconds,
            slide_window(args.seconds, wav),
        )

def parse_args():
    parser = ArgumentParser(description="Animate WAV file waveform")
    parser.add_argument("path", type=Path, help="path to the WAV file")
    parser.add_argument(
        "-s",
        "--seconds",
        type=float,
        default=0.05,
        help="sliding window size in seconds",
    )
    return parser.parse_args()



if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("Aborted")

显示实时频谱图可视化

现在,从 plot_oscilloscope.py 中复制整个源代码并将其粘贴到名为 plot_spectrogram.py 的新脚本中,您将修改该脚本以创建 WAV 文件的新可视化效果。

由于您将计算短音频段的 FFT,因此您需要重叠相邻的音频段,以最大程度地减少边缘突然不连续性导致的频谱泄漏

# ...

def parse_args():
    parser = ArgumentParser(description="Animate WAV file spectrogram")
    parser.add_argument("path", type=Path, help="path to the WAV file")
    parser.add_argument(
        "-s",
        "--seconds",
        type=float,
        default=0.0015,
        help="sliding window size in seconds",
    )
    parser.add_argument(
        "-o",
        "--overlap",
        choices=range(100),
        default=50,
        type=int,
        help="sliding window overlap as a percentage",
    )
    return parser.parse_args()

# ...

--overlap 参数的值必须是介于 0 和 100 不含之间的整数,表示百分比。重叠越大,动画看起来越平滑。

现在,您可以修改 slide_window() 函数以接受该重叠百分比作为附加参数:

# ...

def slide_window(window_seconds, overlap_percentage, wav):
    step_seconds = window_seconds * (1 - overlap_percentage / 100)
    num_windows = round(wav.metadata.num_seconds / step_seconds)
    for i in range(num_windows):
        begin_seconds = i * step_seconds
        end_seconds = begin_seconds + window_seconds
        channels = wav.channels_sliced(begin_seconds, end_seconds)
        yield np.mean(tuple(channels), axis=0)

# ...


您没有像以前那样按窗口的整个持续时间移动窗口,而是引入了一个更小的步骤,从而导致总共有更多的窗口。另一方面,当重叠百分比为零时,您将窗口彼此相邻排列,它们之间没有任何重叠。

您现在可以将命令行中请求的overlap传递给生成器函数以及 animate() 函数:

# ...

def main():
    args = parse_args()
    with WAVReader(args.path) as wav:
        animate(
            args.path.name,
            args.seconds,
            args.overlap,
            fft(slide_window(args.seconds, args.overlap, wav), wav),
        )

# ...

使用numpy的fft计算每个窗口的频率:

# ...

def fft(windows, wav):
    sampling_period = 1 / wav.metadata.frames_per_second
    for window in windows:
        frequencies = np.fft.rfftfreq(window.size, sampling_period)
        magnitudes = np.abs(
            np.fft.rfft(
                (window - np.mean(window)) * np.blackman(window.size)
            )
        )
        yield frequencies, magnitudes

# ...

最后,您必须更新动画代码以在每个滑动窗口位置绘制频率条形图:

def animate(filename, seconds, overlap_percentage, windows):
    try:
        plt.style.use("dark_background")
    except OSError:
        pass  # Fall back to the default style

    fig, ax = plt.subplots(figsize=(16, 9))
    fig.canvas.manager.set_window_title(filename)

    plt.tight_layout()
    plt.box(False)

    bar_gap = 0.25
    for frequencies, magnitudes in windows:
        bar_width = (frequencies[-1] / frequencies.size) * (1 - bar_gap)
        plt.cla()
        ax.set_xticks([])
        ax.set_yticks([])
        ax.set_xlim(-bar_width / 2, frequencies[-1] - bar_width / 2)
        ax.set_ylim(0, np.max(magnitudes))
        ax.bar(frequencies, magnitudes, width=bar_width)
        plt.pause(seconds * (1 - overlap_percentage / 100))

运行以下命令以启动频谱图的动画:

python plot_spectrogram.py file.wav --seconds 0.001 --overlap 95

将 Internet 广播电台录制为 WAV 文件

到目前为止,您一直在使用 waveio 包来方便地读取和解码 WAV 文件,这使您可以专注于更高级别的任务。现在是时候添加拼图中缺失的部分并实现WAVReader 的对应物了。您将创建一个能够将音频数据块写入 WAV 文件的惰性写入器。

对于此任务,您将执行一个动手示例——将 Internet 广播电台流式传输到本地 WAV 文件。

为了简化连接到在线流的过程,您将使用一个微小的帮助程序类来实时获取音频帧。

注:这个需要网站di.fm,可能需要注册,比较麻烦。 建议只看writer相关的部分。

stream.py

import av

from waveio.encoding import PCMEncoding
from waveio.metadata import WAVMetadata

class RadioStream:
    def __init__(self, stream_url):
        self.container = av.open(stream_url)
        self.metadata = get_metadata(self.container)

    def __enter__(self):
        return self

    def __exit__(self, *args, **kwargs):
        self.container.close()

    def __iter__(self):
        for chunk in self.container.decode():
            yield chunk.to_ndarray()

def get_metadata(container):
    (audio_stream,) = container.streams.audio
    num_channels = audio_stream.channels
    bytes_per_sample = audio_stream.format.bytes // num_channels
    return WAVMetadata(
        encoding=PCMEncoding(bytes_per_sample),
        frames_per_second=audio_stream.rate,
        num_channels=num_channels,
    )

现在,在 waveio 包中创建writer 模块,并使用以下代码实现将音频帧增量写入新 WAV 文件的功能:

waveio/writer.py

import wave

class WAVWriter:
    def __init__(self, metadata, path):
        self.metadata = metadata
        self._wav_file = wave.open(str(path), mode="wb")
        self._wav_file.setframerate(metadata.frames_per_second)
        self._wav_file.setnchannels(metadata.num_channels)
        self._wav_file.setsampwidth(metadata.encoding)

    def __enter__(self):
        return self

    def __exit__(self, *args, **kwargs):
        self._wav_file.close()

    def append_channels(self, channels):
        self.append_amplitudes(channels.T.reshape(-1))

    def append_amplitudes(self, amplitudes):
        frames = self.metadata.encoding.encode(amplitudes)
        self._wav_file.writeframes(frames)

该WAVWriter 类采用输出 WAV 文件的 WAVMetadata 实例和路径。然后,它打开文件以二进制模式写入,并使用元数据设置适当的标头值。请注意,在此阶段,音频帧数仍然是未知的,因此无需指定它,而是让 wave 模块稍后在文件关闭时更新它。


就像reader一样,writer对象遵循上下文管理器协议。当您使用 with 关键字输入新上下文时,新 WAVWriter 实例将返回自身。相反,退出上下文将确保即使发生错误,WAV 文件也能正确关闭。


创建 的 WAVWriter 实例后,您可以通过调用 .append_channels() 二维 NumPy 通道数组作为参数来将数据块添加到 WAV 文件中。该方法将通道重塑为振幅值的平面数组,并使用元数据中指定的格式对其进行编码。

在继续操作之前,请将WAVReader添加到waveio __init__.py

from waveio.reader import WAVReader
from waveio.writer import WAVWriter

__all__ = ["WAVReader", "WAVWriter"]

最后,您可以连接这些点:

from argparse import ArgumentParser

from stream import RadioStream
from waveio import WAVWriter

def main():
    args = parse_args()
    with RadioStream(args.stream_url) as radio_stream:
        with WAVWriter(radio_stream.metadata, args.output) as writer:
            for channels_chunk in radio_stream:
                writer.append_channels(channels_chunk)

def parse_args():
    parser = ArgumentParser(description="Record an Internet radio stream")
    parser.add_argument("stream_url", help="URL address of the stream")
    parser.add_argument(
        "-o",
        "--output",
        metavar="path",
        required=True,
        type=str,
        help="path to the output WAV file",
    )
    return parser.parse_args()

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("Aborted")

下面是一个示例命令,演示如何录制 Classic EuroDance 频道:

RADIO_URL=http://prem2.di.fm:80/classiceurodance?your-secret-token
python record_stream.py "$RADIO_URL" --output ripped.wav

扩大 WAV 文件的立体声场

在本节中,您将同时从一个 WAV 文件中读取大量音频帧,并以惰性的方式将其修改后的版本写入另一个文件。为此,您需要通过添加以下方法来增强WAVReader

# ...

class WAVReader:
    DEFAULT_MAX_FRAMES = 1024

    # ...

    @reshape("columns")
    def channels_lazy(self, max_frames=DEFAULT_MAX_FRAMES):
        self._wav_file.rewind()
        while True:
            chunk = self._read(max_frames)
            if chunk.size == 0:
                break
            yield chunk

# ...

与此类中的大多数其他方法和属性一样, .channels_lazy() 装饰用于 @reshape 以更方便的方式排列解码的振幅。不幸的是,此装饰器作用于 NumPy 数组,而您的新方法返回一个生成器对象。若要使它们兼容,必须通过处理两种情况来更新装饰器的定义:

import inspect
import wave
from functools import cached_property, wraps

from waveio.encoding import PCMEncoding
from waveio.metadata import WAVMetadata

def reshape(shape):
    if shape not in ("rows", "columns"):
        raise ValueError("shape must be either 'rows' or 'columns'")

    def decorator(method):
        if inspect.isgeneratorfunction(method):
            @wraps(method)
            def wrapper(self, *args, **kwargs):
                for values in method(self, *args, **kwargs):
                    reshaped = values.reshape(-1, self.metadata.num_channels)
                    yield reshaped if shape == "rows" else reshaped.T
        else:
            @wraps(method)
            def wrapper(self, *args, **kwargs):
                values = method(self, *args, **kwargs)
                reshaped = values.reshape(-1, self.metadata.num_channels)
                return reshaped if shape == "rows" else reshaped.T
        return wrapper

    return decorator

# ...

您可以使用inspect 模块来确定装饰器是包装常规方法还是生成器方法。两个包装器执行相同的操作,但生成器包装器在每次迭代中生成重新调整的值,而常规方法包装器返回它们。

最后,您可以添加属性stereo,告诉您 WAV 文件是否为立体声文件:

class WAVReader:
    # ...

    @cached_property
    def stereo(self):
        return 2 == self.metadata.num_channels

通过这些更改,您可以分块读取 WAV 文件并开始应用各种声音效果。例如,您可以扩大或缩小音频文件的立体声场,以增强或减少空间感。

其中一种技术涉及将包含左右声道的传统立体声信号转换为中声道和侧声道。

中声道 (M) 包含两侧共有的单声道分量,而侧声道 (S) 捕获左声道 (L) 和右声道 (R) 之间的差异。您可以使用以下公式在两种表示形式之间进行转换:

当你分离出侧声道S后,你可以增强S再与M重新组合出左右声道。

创建一个名为 stereo_booster.py 的脚本,该脚本使用可选的强度参数将输入和输出 WAV 文件的路径作为参数:

from argparse import ArgumentParser

def main():
    args = parse_args()

def parse_args():
    parser = ArgumentParser(description="Widen the stereo field")
    parser.add_argument(
        "-i",
        "--input",
        dest="input_path",
        required=True,
        type=str,
        help="path to the input WAV file",
    )
    parser.add_argument(
        "-o",
        "--output",
        dest="output_path",
        required=True,
        type=str,
        help="path to the output WAV file",
    )
    parser.add_argument(
        "-s",
        "--strength",
        type=float,
        default=1.0,
        help="strength (defaults to 1)",
    )
    return parser.parse_args()

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("Aborted")

接下来,实现通道转换公式:

from argparse import ArgumentParser

def main():
    args = parse_args()

# l,r --> mid, side
def convert_to_ms(left, right):
    return (left + right) / 2, (left - right) / 2

# m,s --> l,r
def convert_to_lr(mid, side):
    return mid + side, mid - side

# ...

最后,您可以打开立体声 WAV 文件进行读取,分块循环播放其通道,并应用mid-sid增强:

from argparse import ArgumentParser

from waveio import WAVReader, WAVWriter

def main():
    args = parse_args()
    with (
        WAVReader(args.input_path) as source,
        WAVWriter(source.metadata, args.output_path) as target,
    ):
        if source.stereo:
            for channels_chunk in source.channels_lazy():
                mid, side = convert_to_ms(*channels_chunk)
                left, right = convert_to_lr(mid, side * args.strength)
                target.append_channels(left, right)
        else:
            print("Only stereo WAV files are supported")

# ...

请注意,您现在将修改后的频道附加为单独的参数,而您的广播录制脚本传递的是单个组合频道的 NumPy 数组。若要使.append_channels() 方法适用于这两种类型的调用,可以按如下方式更新 WAVWriter 类:

import wave

import numpy as np

class WAVWriter:
    # ...

    def append_channels(self, *channels):
        match channels:
            case [combined] if combined.ndim > 1:
                self.append_amplitudes(combined.T.reshape(-1))
            case _:
                self.append_amplitudes(np.dstack(channels).reshape(-1))

    def append_amplitudes(self, amplitudes):
        frames = self.metadata.encoding.encode(amplitudes)
        self._wav_file.writeframes(frames)

    # ...

# ...
python stereo_booster.py -i Bicycle-bell.wav -o boosted.wav -s 5
相关文章
|
12天前
|
安全 网络安全 数据安全/隐私保护
`certifi`是一个Python包,它提供了一个包含Mozilla证书颁发机构(CA)Bundle的PEM文件。
`certifi`是一个Python包,它提供了一个包含Mozilla证书颁发机构(CA)Bundle的PEM文件。
|
12天前
|
存储 对象存储 Python
`openpyxl`是一个用于读写Excel 2010 xlsx/xlsm/xltx/xltm文件的Python库。它不需要Microsoft Excel,也不需要.NET或COM组件。
`openpyxl`是一个用于读写Excel 2010 xlsx/xlsm/xltx/xltm文件的Python库。它不需要Microsoft Excel,也不需要.NET或COM组件。
|
12天前
|
存储 Python
`tempfile`模块在Python中用于创建临时文件和目录。
`tempfile`模块在Python中用于创建临时文件和目录。
|
15天前
|
Python
python存取yaml文件
python存取yaml文件
pyaudio:基于pyaudio利用Python编程实现播放音频mp3、wav等格式文件
pyaudio:基于pyaudio利用Python编程实现播放音频mp3、wav等格式文件
pyaudio:基于pyaudio利用Python编程实现播放音频mp3、wav等格式文件
|
Python API
python 播放 wav 文件
未使用其他库, 只是使用 pywin32 调用系统底层 API 播放 wav 文件。 # Our raison d'etre - playing sounds import pywintypes import struct import win32event import win32com.
834 0
|
17小时前
|
网络协议 开发者 Python
颠覆传统!Python Socket编程新思维,基础与进阶并重,打造卓越网络能力!
【7月更文挑战第25天】在数字时代,网络通信至关重要,Python的Socket编程简化了这一复杂领域,使初学者也能轻松上手。通过Python的`socket`模块,我们能快速搭建服务器与客户端,实现数据交换。示例代码展示了如何创建、绑定及监听Socket,以及收发消息。掌握基础后,可利用asyncio库探索异步编程,提升通信效率,处理多连接。Python的Socket编程,结合传统与现代技术,助力开发者在网络通信领域取得非凡成就。
|
3天前
|
存储 算法 搜索推荐
告别低效编程!Python算法设计与分析中,时间复杂度与空间复杂度的智慧抉择!
【7月更文挑战第22天】在编程中,时间复杂度和空间复杂度是评估算法效率的关键。时间复杂度衡量执行时间随数据量增加的趋势,空间复杂度关注算法所需的内存。在实际应用中,开发者需权衡两者,根据场景选择合适算法,如快速排序(平均O(n log n),最坏O(n^2),空间复杂度O(log n)至O(n))适合大规模数据,而归并排序(稳定O(n log n),空间复杂度O(n))在内存受限或稳定性要求高时更有利。通过优化,如改进基准选择或减少复制,可平衡这两者。理解并智慧地选择算法是提升代码效率的关键。
|
6天前
|
存储 开发者 Python
从理论到实践:Python中Trie树与Suffix Tree的完美结合,开启编程新篇章!
【7月更文挑战第19天】在编程实践中,Trie树和Suffix Tree优化了字符串处理。Trie树用于快速拼写检查,如在构建词库后,能高效判断单词是否存在。Suffix Tree则助力文本相似度检测,找寻共同子串。通过Python示例展示了Trie树插入和搜索方法,并指出Suffix Tree虽复杂但能提升性能。结合两者,实现复杂功能,展现数据结构的强大。
23 3