Python绘图动态可视化：实时音频流

在数据可视化中，动画是一种非常有效的方式，可以帮助我们更好地理解数据的变化和动态过程。Python 的 matplotlib.animation 模块提供了强大的功能来创建动画。本文将介绍如何使用 matplotlib.animation 创建简单的动画，并展示一个更复杂的实时音频信号绘制示例。

一、FuncAnimation介绍

FuncAnimation 是 Matplotlib 中用于创建动画的主要工具。它通过不断调用一个更新函数来动态更新图形，从而实现动画效果。FuncAnimation 可以用于各种类型的动画，包括实时数据可视化、动态图形展示等。

1.1 主要功能

• 动态更新：通过定期调用更新函数，动态改变图形的内容。这使得用户能够看到数据随时间变化的过程，增强了数据的可理解性。

• 高效绘制：使用 blit 参数可以提高绘制效率，仅更新需要变化的部分。通过只重绘变化的部分，blit 可以显著提高动画的性能，尤其是在处理复杂图形时。

• 灵活性：可以与其他 Matplotlib 功能结合使用，创建复杂的动画效果。您可以在动画中添加文本、图形、图例等元素，使得动画更加丰富和信息量更大。

1.2 参数

FuncAnimation 的构造函数接受多个参数，以下是一些常用参数：

• fig：要更新的图形对象。通常是通过 plt.subplots() 创建的图形。

• func：更新函数，每次动画帧更新时调用。该函数应接受一个参数（当前帧的编号），并返回要更新的图形对象。

• frames：动画的帧数，可以是整数、可迭代对象或生成器。指定动画的总帧数或提供帧数据。

• blit：布尔值，指示是否使用 blitting 来优化绘制。如果设置为 True，则只更新需要变化的部分，通常可以提高性能。

• interval：每帧之间的时间间隔（毫秒）。可以控制动画的播放速度。

• repeat：布尔值，指示动画是否循环播放。默认为 True，即动画结束后会重新开始。

1.3 简单示例

import numpy as np
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation

# 设置参数
x = np.linspace(0, 2 * np.pi, 100)  # x 轴数据
y = np.sin(x)  # y 轴数据

# 创建图形
fig, ax = plt.subplots()
line, = ax.plot(x, y)  # 初始化绘图线
ax.set_ylim(-1.5, 1.5)  # 设置 y 轴范围
ax.set_title('Sine Wave Animation')
ax.set_xlabel('x')
ax.set_ylabel('sin(x)')


# 定义更新函数
def update(frame):
    line.set_ydata(np.sin(x + frame / 10))  # 更新 y 数据
    return line,


# 创建动画
ani = FuncAnimation(fig, update, frames=100, blit=True)

# 保存动画为 GIF，使用 Pillow
ani.save('sine_wave_animation.gif', writer='pillow', fps=30)

# 显示动画
plt.show()

代码详细说明

1. 创建图形：使用 plt.subplots() 创建图形和坐标轴，并初始化绘图线。line, = ax.plot(x, y) 创建一条初始的正弦波线。

2. 定义更新函数：在每一帧中更新 y 数据，以实现动态效果。line.set_ydata(np.sin(x + frame / 10)) 根据当前帧的编号调整正弦波的相位。

3. 创建动画：使用 FuncAnimation 创建动画，并指定更新函数和帧数。frames=100 表示动画将有 100 帧。

二、实时音频信号绘制

接下来，我们将展示一个更复杂的示例，使用 PyAudio 录制实时音频并绘制其时域信号。以下是完整的代码：

import numpy as np
import matplotlib.pyplot as plt
import pyaudio
import wave
from matplotlib.animation import FuncAnimation
import threading

# 设置参数
FORMAT = pyaudio.paInt16  # 音频格式
CHANNELS = 1  # 单声道
RATE = 44100  # 采样率
CHUNK = 1024  # 每次读取的音频块大小
N = 30  # 显示的块数
WINDOW_SIZE = CHUNK * N  # 滑动窗口大小

# 创建 PyAudio 对象
p = pyaudio.PyAudio()

# 创建音频流
stream = p.open(format=FORMAT,
                channels=CHANNELS,
                rate=RATE,
                input=True,
                frames_per_buffer=CHUNK)

# 创建 WAV 文件
output_filename = "output.wav"
wf = wave.open(output_filename, 'wb')
wf.setnchannels(CHANNELS)
wf.setsampwidth(p.get_sample_size(FORMAT))
wf.setframerate(RATE)

# 创建一个空的图形
fig, ax = plt.subplots()
x = np.arange(0, WINDOW_SIZE)  # x 轴数据
line, = ax.plot(x, np.zeros(WINDOW_SIZE), color='blue')  # 初始化绘图线
ax.set_ylim(-32768, 32767)  # 设置 y 轴范围
ax.set_xlim(0, WINDOW_SIZE)  # 设置 x 轴范围
ax.set_title('Real-time Audio Signal')
ax.set_xlabel('Samples')
ax.set_ylabel('Amplitude')

# 初始化音频数据缓冲区
audio_buffer = np.zeros(WINDOW_SIZE, dtype=np.int16)


# 定义音频录制函数
def record_audio():
    global audio_buffer
    print("Recording...")
    try:
        while True:
            data = stream.read(CHUNK, exception_on_overflow=False)  # 读取音频数据
            audio_data = np.frombuffer(data, dtype=np.int16)  # 转换为 numpy 数组

            # 更新音频缓冲区
            audio_buffer = np.roll(audio_buffer, -CHUNK)  # 滚动缓冲区
            audio_buffer[-CHUNK:] = audio_data  # 添加新数据到缓冲区末尾

            # 写入 WAV 文件
            wf.writeframes(data)  # 将音频数据写入文件
    except Exception as e:
        print(f"Recording stopped: {e}")


# 定义更新函数
def update(frame):
    line.set_ydata(audio_buffer)  # 更新 y 数据
    return line,


# 创建音频录制线程
audio_thread = threading.Thread(target=record_audio)
audio_thread.daemon = True  # 设置为守护线程
audio_thread.start()

# 创建动画
ani = FuncAnimation(fig, update, blit=True, cache_frame_data=False)

# 显示动画
plt.show()

# 停止音频流
stream.stop_stream()
stream.close()
p.terminate()

# 关闭 WAV 文件
wf.close()
print(f"Audio saved to {output_filename}")