Python实现音频重采样与线程安全终止方案

唐可

发布时间：2026-10-12 | 1人浏览过

原创

本文解析Python中音频重采样的实现原理，通过时间刻度插值转换采样率。同时解决Python线程无法直接终止的问题，提供基于标志位的StoppableThread类实现方案，并附带完整的音频播放与重采样代码示例。

音频重采样原理与实现

在音频处理工作中，经常遇到不同采样率的声音文件需要统一处理的情况。重采样的核心在于将原始采样点转换到新的时间刻度上，并通过插值操作生成目标采样率下的音频信号。该方法经测试，在常规场景下表现稳定。

重采样代码解析

实现重采样的关键在于计算输入和输出的时间序列，并利用线性插值生成新信号。以下代码展示了具体的实现逻辑：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 17-7-21 下午2:32
# @Author : Lei.Jinggui
# @Site : http://blog.csdn.net/lccever
# @File : Resample.py
# @Software: PyCharm Community Edition
# @contact: lccever@126.com
import numpy as np
def Resample(input_signal,src_fs,tar_fs):
 '''
 :param input_signal:输入信号
 :param src_fs:输入信号采样率
 :param tar_fs:输出信号采样率
 :return:输出信号
 '''
 dtype = input_signal.dtype
 audio_len = len(input_signal)
 audio_time_max = 1.0*(audio_len-1) / src_fs
 src_time = 1.0 * np.linspace(0,audio_len,audio_len) / src_fs
 tar_time = 1.0 * np.linspace(0,np.int(audio_time_max*tar_fs),np.int(audio_time_max*tar_fs)) / tar_fs
 output_signal = np.interp(tar_time,src_time,input_signal).astype(dtype)
 return output_signal

if __name__ == '__main__':
 import wave
 import pyaudio
 def playSound(audio_data_short, framerate=16000, channels=1):
  preply = pyaudio.PyAudio()
  # 播放声音
  streamreply = preply.open(format=pyaudio.paInt16,
         channels=channels,
         rate=framerate,
         output=True)
  data = audio_data_short.tostring()
  streamreply.write(data)
  streamreply.close()
  preply.terminate()
 wave_file = 'test.wav'
 audio_file = wave.open(wave_file, 'rb')
 audio_data = audio_file.readframes(audio_file.getnframes())
 audio_data_short = np.fromstring(audio_data, np.short)
 src_fs = audio_file.getframerate()
 src_chanels = audio_file.getnchannels()
 if src_chanels > 1:
  audio_data_short = audio_data_short[::src_chanels]
 tar_fs = np.int(src_fs * 0.5)

 playSound(audio_data_short,framerate=src_fs)
 audio_data_short0 = Resample(audio_data_short,src_fs,tar_fs)
 playSound(audio_data_short0,framerate=tar_fs)

代码首先读取WAV文件，获取原始采样率和声道数。若为多声道，则提取单声道数据。接着，通过Resample函数将原始采样率降低一半（tar_fs = src_fs * 0.5），并使用pyaudio库分别播放原始音频和重采样后的音频，以验证效果。

Python线程安全终止方案

Python的threading模块没有提供stop或terminate方法。线程启动后，除非主进程退出，否则无法主动退出。主进程退出指的是主进程而非线程的父进程。因此，合理的做法是将线程函数放入threading.Thread的target中，或者将线程函数改写到一个继承类中。

StoppableThread实现

通过引入一个内部标志位__is_running，可以控制线程的循环执行。当调用terminate方法时，将标志位设为False，线程在下一次循环检查时会自然退出。以下是实现代码：

import threading
import time
import os
 
# 原本需要用来启动的无线循环的函数
def print_thread():
 pid = os.getpid()
 counts = 0
 while True:
  print(f'threading pid: {pid} ran: {counts:04d} s')
  counts += 1
  time.sleep(1)
 
# 把函数放到改写到类的run方法中，便可以通过调用类方法，实现线程的终止
class StoppableThread(threading.Thread):
 
 def __init__(self, daemon=None):
  super(StoppableThread, self).__init__(daemon=daemon)
  self.__is_running = True
  self.daemon = daemon
 
 def terminate(self):
  self.__is_running = False
 
 def run(self):
  pid = os.getpid()
  counts = 0
  while self.__is_running:
   print(f'threading running: {pid} ran: {counts:04d} s')
   counts += 1
   time.sleep(1)
 
def call_thread():
 thread = StoppableThread()
 thread.daemon = True
 thread.start()
 
 pid = os.getpid()
 counts = 0
 for i in range(5):
  print(f'0 call threading pid: {pid} ran: {counts:04d} s')
  counts += 2
  time.sleep(2)
 # 主动把线程退出
 thread.terminate()
 
if __name__ == '__main__':
 call_thread()
 print(f'==========call_thread finish===========')
 counts = 0
 for i in range(5):
  counts += 1
  time.sleep(1)
  print(f'main thread:{counts:04d} s')

在StoppableThread类中，__init__方法初始化标志位为True。run方法中的while循环检查该标志位。当需要终止线程时，调用terminate方法将标志位设为False，线程将在当前循环结束后自然退出。