聽歌識曲--用python實現(xiàn)一個音樂檢索器的功能

更新時間：2016年11月15日 08:39:41 作者：唯心不易

本篇文章中主要介紹了用python實現(xiàn)一個音樂檢索器，類似于QQ音樂的搖一搖識曲，有興趣的同學可以了解一下。

聽歌識曲，顧名思義，用設(shè)備“聽”歌曲，然后它要告訴你這是首什么歌。而且十之八九它還得把這首歌給你播放出來。這樣的功能在QQ音樂等應(yīng)用上早就出現(xiàn)了。我們今天來自己動手做一個自己的聽歌識曲

我們設(shè)計的總體流程圖很簡單：

-----
錄音部分
-----

我們要想“聽”，就必須先有錄音的過程。在我們的實驗中，我們的曲庫也要用我們的錄音代碼來進行錄音，然后提取特征存進數(shù)據(jù)庫。我們用下面這樣的思路來錄音

# 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: 緩沖區(qū)大小
  :param FORMAT: 采樣大小
  :param CHANNELS:通道數(shù)
  :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')

我們錄完的歌曲是個什么形式？

如果只看一個聲道的話，他是一個一維數(shù)組，大概長成這個樣子

我們把他按照索引值為橫軸畫出來，就是我們常?？匆姷囊纛l的形式。

音頻處理部分

我們在這里要寫我們的核心代碼。關(guān)鍵的“如何識別歌曲”。想想我們?nèi)祟惾绾螀^(qū)分歌曲？是靠想上面那樣的一維數(shù)組嗎？是靠歌曲的響度嗎？都不是。

我們是通過耳朵所聽到的特有的頻率組成的序列來記憶歌曲的，所以我們想要寫聽歌識曲的話，就得在音頻的頻率序列上做文章。

復(fù)習一下什么是傅里葉變換。博主的《信號與系統(tǒng)》的課上的挺水，不過在課上雖然沒有記下來具體的變換形式，但是感性的理解還是有的。

傅里葉變換的實質(zhì)就是把時域信號變換成了頻域信號。也就是原本X,Y軸分別是我們的數(shù)組下標和數(shù)組元素，現(xiàn)在變成了頻率(這么說不準確，但在這里這樣理解沒錯)和在這個頻率上的分量大小。

怎么理解頻域這個事情呢？對于我們信號處理不是很懂的人來說，最重要的就是改變對音頻的構(gòu)成的理解。我們原來認為音頻就是如我們開始給出的波形那樣，在每一個時間有一個幅值，不同的幅值序列構(gòu)成了我們特定的聲音。而現(xiàn)在，我們認為聲音是不同的頻率信號混合而成的，他們每一個信號都自始至終存在著。并且他們按照他們的投影分量做貢獻。

讓我們看看把一首歌曲轉(zhuǎn)化到頻域是什么樣子？

我們可以觀察到這些頻率的分量并不是平均的，差異是非常大的。我們可以在一定程度上認為在圖中明顯凸起的峰值是輸出能量大的頻率信號，代表著在這個音頻中，這個信號占有很高的地位。于是我們就選擇這樣的信號來提取歌曲的特征。

但是別忘了，我們之前說的可是頻率序列，傅里葉變換一套上，我們就只能知道整首歌曲的頻率信息，那么我們就損失了時間的關(guān)系，我們說的“序列”也就無從談起。所以我們采用的比較折中的方法，將音頻按照時間分成一個個小塊，在這里我每秒分出了40個塊。

在這里留個問題：為什么要采用小塊，而不是每秒一塊這樣的大塊？

我們對每一個塊進行傅里葉變換，然后對其求模，得到一個個數(shù)組。我們在下標值為(0,40),(40,80),(80,120),(120,180)這四個區(qū)間分別取其模長最大的下標，合成一個四元組，這就是我們最核心的音頻“指紋”。

我們提取出來的“指紋”類似下面這樣

(39, 65, 110, 131), (15, 66, 108, 161), (3, 63, 118, 146), (11, 62, 82, 158), (15, 41, 95, 140), (2, 71, 106, 143), (15, 44, 80, 133), (36, 43, 80, 135), (22, 58, 80, 120), (29, 52, 89, 126), (15, 59, 89, 126), (37, 59, 89, 126), (37, 59, 89, 126), (37, 67, 119, 126)

音頻處理的類有三個方法：載入數(shù)據(jù)，傅里葉變換，播放音樂。
如下：

# 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內(nèi)儲存著多通道的音頻數(shù)據(jù)，其中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是指定每秒鐘分塊數(shù)
  :return:
  '''
  block = []
  fft_blocks = []
  self.high_point = []
  blocks_size = self.framerate / frames # block_size為每一塊的frame數(shù)量
  blocks_num = self.nframes / blocks_size # 將音頻分塊的數(shù)量
  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,
         )) # 提取指紋的關(guān)鍵步驟，沒有取最后一個，但是保留了這一項，可以想想為什么去掉了？

 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()

這里面的self.high_point是未來應(yīng)用的核心數(shù)據(jù)。列表類型，里面的元素都是上面所解釋過的指紋的形式。

數(shù)據(jù)存儲和檢索部分

因為我們是事先做好了曲庫來等待檢索，所以必須要有相應(yīng)的持久化方法。我采用的是直接用mysql數(shù)據(jù)庫來存儲我們的歌曲對應(yīng)的指紋，這樣有一個好處：省寫代碼的時間

我們將指紋和歌曲存成這樣的形式：
順便一說：為什么各個歌曲前幾個的指紋都一樣？（當然，后面肯定是千差萬別的）其實是音樂開始之前的時間段中沒有什么能量較強的點，而由于我們44100的采樣率比較高，就會導(dǎo)致開頭會有很多重復(fù)，別擔心。

我們怎么來進行匹配呢？我們可以直接搜索音頻指紋相同的數(shù)量，不過這樣又損失了我們之前說的序列，我們必須要把時間序列用上。否則一首歌曲越長就越容易被匹配到，這種歌曲像野草一樣瘋狂的占據(jù)了所有搜索音頻的結(jié)果排行榜中的第一名。而且從理論上說，音頻所包含的信息就是在序列中體現(xiàn)，就像一句話是靠各個短語和詞匯按照一定順序才能表達出它自己的意思。單純的看兩個句子里的詞匯重疊數(shù)是完全不能判定兩句話是否相似的。我們采用的是下面的算法，不過我們這只是實驗性的代碼，算法設(shè)計的很簡單，效率不高。建議想要做更好的結(jié)果的同學可以使用改進的DTW算法。

我們在匹配過程中滑動指紋序列，每次比對模式串和源串的對應(yīng)子串，如果對應(yīng)位置的指紋相同，則這次的比對相似值加一，我們把滑動過程中得到的最大相似值作為這兩首歌的相似度。

舉例：

曲庫中的一首曲子的指紋序列:[fp13, fp20, fp10, fp29, fp14, fp25, fp13, fp13, fp20, fp33, fp14]

檢索音樂的指紋序列: [fp14, fp25, fp13, fp17]

比對過程：

最終的匹配相似值為3

存儲檢索部分的實現(xiàn)代碼

# 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:數(shù)據(jù)庫名
  '''
  self.host = host
  self.port = port
  self.user = user
  self.passwd = passwd
  self.db = db

 def addsong(self, path):
  '''
  添加歌曲方法，將指定路徑的歌曲提取指紋后放到數(shù)據(jù)庫
  :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')
   # 創(chuàng)建與數(shù)據(jù)庫的連接
  except:
   print 'DataBase error'
   return None
  cur = conn.cursor()
  namecount = cur.execute("select * from fingerprint.musicdata WHERE song_name = '%s'" % basename)
  # 查詢新添加的歌曲是否已經(jīng)在曲庫中了
  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__()))
  # 將新歌曲的名字和指紋存到數(shù)據(jù)庫中
  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):
  '''
  從數(shù)據(jù)庫檢索出
  :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):
  '''
  跟上個方法一樣，不過增加了將搜索出的最優(yōu)結(jié)果直接播放的功能
  :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')

總結(jié)

我們這個實驗很多地方都很粗糙，核心的算法是從shazam公司提出的算法吸取的“指紋”的思想。希望讀者可以提出寶貴建議。

本文轉(zhuǎn)載于:http://www.cnblogs.com/chuxiuhong/p/6063602.html

您可能感興趣的文章: