Python树莓派播放MP3并实时获取振幅教程

pip install pydub pyaudio

sudo apt-get update
sudo apt-get install ffmpeg libav-tools

from pydub import AudioSegment
import io
import wave

# 加载MP3文件
mp3_file_path = "kimi_no_shiranai.mp3"
audio_segment = AudioSegment.from_mp3(mp3_file_path)

# 将AudioSegment导出为WAV格式的BytesIO对象
wav_buffer = io.BytesIO()
audio_segment.export(wav_buffer, format="wav")
wav_buffer.seek(0) # 将缓冲区指针重置到开头

# 现在可以使用wave模块打开这个内存中的WAV流
wf = wave.open(wav_buffer, 'rb')

import struct
import numpy as np

def calculate_rms_amplitude(data, sample_width):
    """
    从原始音频数据中计算RMS振幅。
    data: 字节串形式的音频数据。
    sample_width: 每个样本的字节数 (e.g., 2 for 16-bit audio)。
    """
    if not data:
        return 0

    # 根据样本宽度解析数据
    # 'h' for short (2 bytes), 'i' for int (4 bytes)
    fmt = f'{len(data) // sample_width}{"h" if sample_width == 2 else "i"}'

    # 解包字节数据为整数数组
    try:
        samples = struct.unpack(fmt, data)
        # 计算RMS
        rms = np.sqrt(np.mean(np.array(samples, dtype=np.int64)**2))
        return rms
    except struct.error:
        # 数据可能不完整，返回0或处理错误
        return 0

import pyaudio
import wave
import io
from pydub import AudioSegment
import struct
import numpy as np
import time # For sleep

# 假设的Amplitude类，用于演示振幅计算和显示
# 在实际应用中，您需要实现此类的from_data方法，
# 或使用上面提供的calculate_rms_amplitude函数。
class Amplitude:
    def __init__(self, value=0):
        self.value = value

    @staticmethod
    def from_data(data, sample_width=2): # 默认16位音频
        # 实际的振幅计算逻辑，这里使用RMS作为示例
        return Amplitude(calculate_rms_amplitude(data, sample_width))

    def __gt__(self, other):
        return self.value > other.value

    def display(self, scale=100, mark=None):
        # 简单的文本振幅显示
        normalized_amp = min(int(self.value / 32767 * scale), scale) # 假设16位最大值32767
        bar = '#' * normalized_amp
        mark_str = ""
        if mark and mark.value > 0:
            normalized_mark = min(int(mark.value / 32767 * scale), scale)
            if normalized_mark > normalized_amp:
                bar = bar + '-' * (normalized_mark - normalized_amp)
            mark_str = f" Max: {mark.value:.2f}"
        print(f"[{bar.ljust(scale)}] Current: {self.value:.2f}{mark_str}\r", end="")

# RMS振幅计算函数
def calculate_rms_amplitude(data, sample_width):
    if not data:
        return 0

    fmt = f'{len(data) // sample_width}{"h" if sample_width == 2 else "i"}'
    try:
        samples = struct.unpack(fmt, data)
        rms = np.sqrt(np.mean(np.array(samples, dtype=np.int64)**2))
        return rms
    except struct.error:
        return 0

def main():
    mp3_file_path = "sound.mp3" # 替换为您的MP3文件路径
    chunk = 1024 # 每次读取的音频帧数

    audio = pyaudio.PyAudio()
    stream = None # 初始化stream为None

    try:
        # 1. MP3文件转换为WAV字节流
        print(f"Converting {mp3_file_path} to WAV in memory...")
        audio_segment = AudioSegment.from_mp3(mp3_file_path)
        wav_buffer = io.BytesIO()
        audio_segment.export(wav_buffer, format="wav")
        wav_buffer.seek(0)

        # 2. 打开内存中的WAV流
        wf = wave.open(wav_buffer, 'rb')

        # 3. 初始化PyAudio输出流
        stream = audio.open(format=audio.get_format_from_width(wf.getsampwidth()),
                            channels=wf.getnchannels(),
                            rate=wf.getframerate(),
                            output=True)

        print("Playing audio and monitoring amplitude...")
        data = wf.readframes(chunk)
        maximal_amplitude = Amplitude(0) # 记录最大振幅

        while data:
            # 写入流以播放声音
            stream.write(data)

            # 获取当前数据块的振幅
            # 注意：wf.getsampwidth() 返回的是每个样本的字节数
            current_amplitude = Amplitude.from_data(data, wf.getsampwidth())

            # 更新最大振幅
            if current_amplitude > maximal_amplitude:
                maximal_amplitude = current_amplitude

            # 显示振幅（可选）
            current_amplitude.display(scale=50, mark=maximal_amplitude)

            # 读取下一个数据块
            data = wf.readframes(chunk)

        print("\nAudio playback finished.")

    except FileNotFoundError:
        print(f"Error: MP3 file '{mp3_file_path}' not found.")
    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        # 确保关闭音频流和终止PyAudio
        if stream:
            stream.stop_stream()
            stream.close()
        audio.terminate()
        if 'wf' in locals() and wf:
            wf.close() # 关闭wave文件对象
        if 'wav_buffer' in locals() and wav_buffer:
            wav_buffer.close() # 关闭BytesIO对象

if __name__ == "__main__":
    main()