python实现音乐检索器 听歌识曲--用python实现一个音乐检索器的功能

# coding=utf8
import wave

import pyaudio


class recode():
 def recode(self, CHUNK=44100, FORMAT=pyaudio.paInt16, CHANNELS=2, RATE=44100, RECORD_SECONDS=200,
    WAVE_OUTPUT_FILENAME="record.wav"):
  '''

  :param CHUNK: 缓冲区大小
  :param FORMAT: 采样大小
  :param CHANNELS:通道数
  :param RATE:采样率
  :param RECORD_SECONDS:录的时间
  :param WAVE_OUTPUT_FILENAME:输出文件路径
  :return:
  '''
  p = pyaudio.PyAudio()
  stream = p.open(format=FORMAT,
      channels=CHANNELS,
      rate=RATE,
      input=True,
      frames_per_buffer=CHUNK)
  frames = []
  for i in range(0, int(RATE / CHUNK * RECORD_SECONDS)):
   data = stream.read(CHUNK)
   frames.append(data)
  stream.stop_stream()
  stream.close()
  p.terminate()
  wf = wave.open(WAVE_OUTPUT_FILENAME, 'wb')
  wf.setnchannels(CHANNELS)
  wf.setsampwidth(p.get_sample_size(FORMAT))
  wf.setframerate(RATE)
  wf.writeframes(''.join(frames))
  wf.close()


if __name__ == '__main__':
 a = recode()
 a.recode(RECORD_SECONDS=30, WAVE_OUTPUT_FILENAME='record_pianai.wav')

# coding=utf8
import os
import re
import wave
import numpy as np
import pyaudio


class voice():
 def loaddata(self, filepath):
  '''

  :param filepath: 文件路径，为wav文件
  :return: 如果无异常则返回True，如果有异常退出并返回False
  self.wave_data内储存着多通道的音频数据，其中self.wave_data[0]代表第一通道
  具体有几通道，看self.nchannels
  '''
  if type(filepath) != str:
   print 'the type of filepath must be string'
   return False
  p1 = re.compile('\.wav')
  if p1.findall(filepath) is None:
   print 'the suffix of file must be .wav'
   return False
  try:
   f = wave.open(filepath, 'rb')
   params = f.getparams()
   self.nchannels, self.sampwidth, self.framerate, self.nframes = params[:4]
   str_data = f.readframes(self.nframes)
   self.wave_data = np.fromstring(str_data, dtype=np.short)
   self.wave_data.shape = -1, self.sampwidth
   self.wave_data = self.wave_data.T
   f.close()
   self.name = os.path.basename(filepath) # 记录下文件名
   return True
  except:
   print 'File Error!'

 def fft(self, frames=40):
  '''
  :param frames: frames是指定每秒钟分块数
  :return:
  '''
  block = []
  fft_blocks = []
  self.high_point = []
  blocks_size = self.framerate / frames # block_size为每一块的frame数量
  blocks_num = self.nframes / blocks_size # 将音频分块的数量
  for i in xrange(0, len(self.wave_data[0]) - blocks_size, blocks_size):
   block.append(self.wave_data[0][i:i + blocks_size])
   fft_blocks.append(np.abs(np.fft.fft(self.wave_data[0][i:i + blocks_size])))
   self.high_point.append((np.argmax(fft_blocks[-1][:40]),
         np.argmax(fft_blocks[-1][40:80]) + 40,
         np.argmax(fft_blocks[-1][80:120]) + 80,
         np.argmax(fft_blocks[-1][120:180]) + 120,
         # np.argmax(fft_blocks[-1][180:300]) + 180,
         )) # 提取指纹的关键步骤，没有取最后一个，但是保留了这一项，可以想想为什么去掉了？

 def play(self, filepath):
  '''
  用来做音频播放的方法
  :param filepath:文件路径 
  :return: 
  '''
  chunk = 1024
  wf = wave.open(filepath, 'rb')
  p = pyaudio.PyAudio()
  # 打开声音输出流
  stream = p.open(format=p.get_format_from_width(wf.getsampwidth()),
      channels=wf.getnchannels(),
      rate=wf.getframerate(),
      output=True)
  # 写声音输出流进行播放
  while True:
   data = wf.readframes(chunk)
   if data == "":
    break
   stream.write(data)

  stream.close()
  p.terminate()


if __name__ == '__main__':
 p = voice()
 p.loaddata('record_beiyiwang.wav')
 p.fft()

# coding=utf-8

import os

import MySQLdb

import my_audio


class memory():
 def __init__(self, host, port, user, passwd, db):
  '''
  初始化存储类
  :param host:主机位置
  :param port:端口
  :param user:用户名
  :param passwd:密码
  :param db:数据库名
  '''
  self.host = host
  self.port = port
  self.user = user
  self.passwd = passwd
  self.db = db

 def addsong(self, path):
  '''
  添加歌曲方法，将指定路径的歌曲提取指纹后放到数据库
  :param path:路径
  :return:
  '''
  if type(path) != str:
   print 'path need string'
   return None
  basename = os.path.basename(path)
  try:
   conn = MySQLdb.connect(host=self.host, port=self.port, user=self.user, passwd=self.passwd, db=self.db,
         charset='utf8')
   # 创建与数据库的连接
  except:
   print 'DataBase error'
   return None
  cur = conn.cursor()
  namecount = cur.execute("select * from fingerprint.musicdata WHERE song_name = '%s'" % basename)
  # 查询新添加的歌曲是否已经在曲库中了
  if namecount > 0:
   print 'the song has been record!'
   return None
  v = my_audio.voice()
  v.loaddata(path)
  v.fft()
  cur.execute("insert into fingerprint.musicdata VALUES('%s','%s')" % (basename, v.high_point.__str__()))
  # 将新歌曲的名字和指纹存到数据库中
  conn.commit()
  cur.close()
  conn.close()

 def fp_compare(self, search_fp, match_fp):
  '''
  指纹比对方法。
  :param search_fp: 查询指纹
  :param match_fp: 库中指纹
  :return:最大相似值 float
  '''
  if len(search_fp) > len(match_fp):
   return 0
  max_similar = 0
  search_fp_len = len(search_fp)
  match_fp_len = len(match_fp)
  for i in range(match_fp_len - search_fp_len):
   temp = 0
   for j in range(search_fp_len):
    if match_fp[i + j] == search_fp[j]:
     temp += 1
   if temp > max_similar:
    max_similar = temp
  return max_similar

 def search(self, path):
  '''
  从数据库检索出
  :param path: 需要检索的音频的路径
  :return:返回列表，元素是二元组，第一项是匹配的相似值，第二项是歌曲名
  '''
  v = my_audio.voice()
  v.loaddata(path)
  v.fft()
  try:
   conn = MySQLdb.connect(host=self.host, port=self.port, user=self.user, passwd=self.passwd, db=self.db,
         charset='utf8')
  except:
   print 'DataBase error'
   return None
  cur = conn.cursor()
  cur.execute("SELECT * FROM fingerprint.musicdata")
  result = cur.fetchall()
  compare_res = []
  for i in result:
   compare_res.append((self.fp_compare(v.high_point[:-1], eval(i[1])), i[0]))
  compare_res.sort(reverse=True)
  cur.close()
  conn.close()
  print compare_res
  return compare_res

 def search_and_play(self, path):
  '''
  跟上个方法一样，不过增加了将搜索出的最优结果直接播放的功能
  :param path: 带检索歌曲路径
  :return:
  '''
  v = my_audio.voice()
  v.loaddata(path)
  v.fft()
  # print v.high_point
  try:
   conn = MySQLdb.connect(host=self.host, port=self.port, user=self.user, passwd=self.passwd, db=self.db,
         charset='utf8')
  except:
   print 'DataBase error'
   return None
  cur = conn.cursor()
  cur.execute("SELECT * FROM fingerprint.musicdata")
  result = cur.fetchall()
  compare_res = []
  for i in result:
   compare_res.append((self.fp_compare(v.high_point[:-1], eval(i[1])), i[0]))
  compare_res.sort(reverse=True)
  cur.close()
  conn.close()
  print compare_res
  v.play(compare_res[0][1])
  return compare_res


if __name__ == '__main__':
 sss = memory('localhost', 3306, 'root', 'root', 'fingerprint')
 sss.addsong('taiyangzhaochangshengqi.wav')
 sss.addsong('beiyiwangdeshiguang.wav')
 sss.addsong('xiaozezhenger.wav')
 sss.addsong('nverqing.wav')
 sss.addsong('the_mess.wav')
 sss.addsong('windmill.wav')
 sss.addsong('end_of_world.wav')
 sss.addsong('pianai.wav')

 sss.search_and_play('record_beiyiwang.wav')