聽歌識曲--用python實現一個音樂檢索器

時間 2019-11-10

原文原文鏈接

聽歌識曲，顧名思義，用設備「聽」歌曲，而後它要告訴你這是首什麼歌。並且十之八九它還得把這首歌給你播放出來。這樣的功能在QQ音樂等應用上早就出現了。咱們今天來本身動手作一個本身的聽歌識曲
咱們設計的整體流程圖很簡單：
html

錄音部分

咱們要想「聽」，就必須先有錄音的過程。在咱們的實驗中，咱們的曲庫也要用咱們的錄音代碼來進行錄音，而後提取特徵存進數據庫。咱們用下面這樣的思路來錄音
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')

咱們錄完的歌曲是個什麼形式？
若是隻看一個聲道的話，他是一個一維數組，大概長成這個樣子
mysql

咱們把他按照索引值爲橫軸畫出來，就是咱們經常看見的音頻的形式。
算法

音頻處理部分

咱們在這裏要寫咱們的核心代碼。關鍵的「如何識別歌曲」。想一想咱們人類如何區分歌曲？是靠想上面那樣的一維數組嗎？是靠歌曲的響度嗎？都不是。
咱們是經過耳朵所聽到的特有的頻率組成的序列來記憶歌曲的，因此咱們想要寫聽歌識曲的話，就得在音頻的頻率序列上作文章。
複習一下什麼是傅里葉變換。博主的《信號與系統》的課上的挺水，不過在課上雖然沒有記下來具體的變換形式，可是感性的理解仍是有的。
傅里葉變換的實質就是把時域信號變換成了頻域信號。也就是本來X,Y軸分別是咱們的數組下標和數組元素，如今變成了頻率(這麼說不許確，但在這裏這樣理解沒錯)和在這個頻率上的份量大小。sql

上面的圖來自知乎，很是感謝Heinrich寫的文章，原文連接：點我跳轉數據庫

怎麼理解頻域這個事情呢？對於咱們信號處理不是很懂的人來講，最重要的就是改變對音頻的構成的理解。咱們原來認爲音頻就是如咱們開始給出的波形那樣，在每個時間有一個幅值，不一樣的幅值序列構成了咱們特定的聲音。而如今，咱們認爲聲音是不一樣的頻率信號混合而成的，他們每個信號都自始至終存在着。而且他們按照他們的投影份量作貢獻。數組

讓咱們看看把一首歌曲轉化到頻域是什麼樣子？
app

咱們能夠觀察到這些頻率的份量並非平均的，差別是很是大的。咱們能夠在必定程度上認爲在圖中明顯凸起的峯值是輸出能量大的頻率信號，表明着在這個音頻中，這個信號佔有很高的地位。因而咱們就選擇這樣的信號來提取歌曲的特徵。fetch

可是別忘了，咱們以前說的但是頻率序列，傅里葉變換一套上，咱們就只能知道整首歌曲的頻率信息，那麼咱們就損失了時間的關係，咱們說的「序列」也就無從談起。因此咱們採用的比較折中的方法，將音頻按照時間分紅一個個小塊，在這裏我每秒分出了40個塊。
在這裏留個問題：爲何要採用小塊，而不是每秒一塊這樣的大塊？設計

咱們對每個塊進行傅里葉變換，而後對其求模，獲得一個個數組。咱們在下標值爲(0,40),(40,80),(80,120),(120,180)這四個區間分別取其模長最大的下標，合成一個四元組，這就是咱們最核心的音頻「指紋」。

咱們提取出來的「指紋」相似下面這樣

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

音頻處理的類有三個方法：載入數據，傅里葉變換，播放音樂。
以下：

# 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:
            raise TypeError, 'the type of filepath must be string'
        p1 = re.compile('\.wav')
        if p1.findall(filepath) is None:
            raise IOError, 'the suffix of file must be .wav'
        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:
            raise IOError, '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.play('the_mess.wav')
    print p.name

這裏面的self.high_point是將來應用的核心數據。列表類型，裏面的元素都是上面所解釋過的指紋的形式。

數據存儲和檢索部分

由於咱們是事先作好了曲庫來等待檢索，因此必需要有相應的持久化方法。我採用的是直接用mysql數據庫來存儲咱們的歌曲對應的指紋，這樣有一個好處：省寫代碼的時間

咱們將指紋和歌曲存成這樣的形式：

順便一說：爲何各個歌曲前幾個的指紋都同樣？（固然，後面確定是千差萬別的）實際上是音樂開始以前的時間段中沒有什麼能量較強的點，而因爲咱們44100的採樣率比較高，就會致使開頭會有不少重複，別擔憂。
咱們怎麼來進行匹配呢？咱們能夠直接搜索音頻指紋相同的數量，不過這樣又損失了咱們以前說的序列，咱們必需要把時間序列用上。不然一首歌曲越長就越容易被匹配到，這種歌曲像野草同樣瘋狂的佔據了全部搜索音頻的結果排行榜中的第一名。並且從理論上說，音頻所包含的信息就是在序列中體現，就像一句話是靠各個短語和詞彙按照必定順序才能表達出它本身的意思。單純的看兩個句子裏的詞彙重疊數是徹底不能斷定兩句話是否類似的。咱們採用的是下面的算法，不過咱們這只是實驗性的代碼，算法設計的很簡單，效率不高。建議想要作更好的結果的同窗可使用改進的DTW算法。

咱們在匹配過程當中滑動指紋序列，每次比對模式串和源串的對應子串，若是對應位置的指紋相同，則此次的比對類似值加一，咱們把滑動過程當中獲得的最大類似值做爲這兩首歌的類似度。
舉例：
曲庫中的一首曲子的指紋序列:[fp13, fp20, fp10, fp29, fp14, fp25, fp13, fp13, fp20, fp33, fp14]
檢索音樂的指紋序列: [fp14, fp25, fp13, fp17]
比對過程：

最終的匹配類似值爲3

存儲檢索部分的實現代碼

# 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:
            raise TypeError, 'path need string'
        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: 按照類似度排序後的列表，元素類型爲tuple，二元組，歌曲名和類似匹配值
        '''
        #先計算出來咱們的音頻指紋
        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:
            raise IOError, 'DataBase error'
        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()
        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', '', '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_pianai.wav')