• 基于DTW算法的命令字识别

    DTW算法介绍

    DTW(Dynamic Time Warping)：按距离最近原则，构建两个序列之间的对应的关系，评估两个序列的相似性。

    要求：

    • 单向对应，不能回头；
    • 一一对应，不能有空；
    • 对应之后，距离最近。

     

    DTW代码实现

    1. import numpy as np
    2.  
    3. def dis_abs(x, y):
    4.     return abs(x - y)[0]
    5.  
    6. def estimate_twf(A, B, dis_func=dis_abs):
    7.     N_A = len(A)
    8.     N_B = len(B)
    9.     
    10.     D = np.zeros([N_A, N_B])
    11.     D[0, 0] = dis_func(A[0], B[0])
    12.     # 左边一列
    13.     for i in range(1, N_A):
    14.         D[i, 0] = D[i - 1, 0] + dis_func(A[i], B[0])
    15.     # 下边一行
    16.     for j in range(1, N_B):
    17.         D[0, j] = D[0, j-1] + dis_func(A[0], B[j])
    18.     # 中间部分
    19.     for i in range(1, N_A):
    20.         for j in range(1, N_B):
    21.             D[i, j] = dis_func(A[i], B[j]) + min(D[i-1, j], D[i, j-1], D[i-1][j-1])
    22.     
    23.     # 路径回溯
    24.     i = N_A - 1
    25.     j = N_B - 1
    26.     cnt = 0
    27.     d = np.zeros(max(N_A, N_B) * 3)
    28.     path = []
    29.     while True:
    30.         if i > 0 and j > 0:
    31.             path.append((i, j))
    32.             m = min(D[i-1, j], D[i, j-1], D[i-1, j-1])
    33.             if m == D[i-1, j-1]:
    34.                 d[cnt] = D[i,j] - D[i-1, j-1]
    35.                 i -= 1
    36.                 j -= 1
    37.                 cnt += 1
    38.             elif m == D[i, j-1]:
    39.                 d[cnt] = D[i,j] - D[i, j-1]
    40.                 j -= 1
    41.                 cnt += 1
    42.             elif m == D[i-1, j]:
    43.                 d[cnt] = D[i,j] - D[i-1, j]
    44.                 i -= 1
    45.                 cnt += 1
    46.         elif i == 0 and j == 0:
    47.             path.append((i, j))
    48.             d[cnt] = D[i, j]
    49.             cnt += 1
    50.             break
    51.         elif i == 0:
    52.             path.append((i, j))
    53.             d[cnt] = D[i, j] - D[i, j-1]
    54.             j -= 1
    55.             cnt += 1
    56.         elif j == 0:
    57.             path.append((i, j))
    58.             d[cnt] = D[i, j] - D[i-1, j]
    59.             i -= 1
    60.             cnt += 1
    61.     mean = np.sum(d) / cnt
    62.     return mean, path[::-1], D
    1. a = np.array([1,3,4,9,8,2,1,5,7,3])
    2. b = np.array([1,6,2,3,0,9,4,1,6,3])
    3. a = a[:, np.newaxis]
    4. b = b[:, np.newaxis]
    5. dis, path, D = estimate_twf(a, b, dis_func=dis_abs)
    6. print(dis, path, D)
    7.  
    8. >>:
    9. 1.0833333333333333
    10.  [(0, 0), (1, 1), (1, 2), (1, 3), (2, 4), (3, 5), (4, 5), (5, 6), (6, 7), (7, 8), (8, 8), (9, 9)] 
    11. [[ 0.  5.  6.  8.  9. 17. 20. 20. 25. 27.]
    12.  [ 2.  3.  4.  4.  7. 13. 14. 16. 19. 19.]
    13.  [ 5.  4.  5.  5.  8. 12. 12. 15. 17. 18.]
    14.  [13.  7. 11. 11. 14.  8. 13. 20. 18. 23.]
    15.  [20.  9. 13. 16. 19.  9. 12. 19. 20. 23.]
    16.  [21. 13.  9. 10. 12. 16. 11. 12. 16. 17.]
    17.  [21. 18. 10. 11. 11. 19. 14. 11. 16. 18.]
    18.  [25. 19. 13. 12. 16. 15. 15. 15. 12. 14.]
    19.  [31. 20. 18. 16. 19. 17. 18. 21. 13. 16.]
    20.  [33. 23. 19. 16. 19. 23. 18. 20. 16. 13.]]

    基于DTW算法的命令字识别

    utils.py:

    1. # -*- coding:UTF-8 -*-
    2. import streamlit as st
    3. import pyaudio
    4. import wave
    5. import librosa
    6. import soundfile as sf
    7. import numpy as np
    8. import os
    9. import time
    10.  
    11.  
    12. # 采用MFCC特征使用mcd距离
    13. def euclideanDistance(a, b):
    14.     diff = a - b
    15.     mcd = 10.0 / np.log(10) * np.sqrt(2.0 * np.sum(diff ** 2))
    16.     return mcd
    17.  
    18.  
    19. # DTW算法匹配距离
    20. class DTW:
    21.     def __init__(self, disFunc=euclideanDistance):
    22.         self.disFunc = disFunc
    23.  
    24.     def compute_distance(self, reference, test):
    25.         DTW_matrix = np.empty([reference.shape[0], test.shape[0]])
    26.         DTW_matrix[:] = np.inf
    27.         DTW_matrix[0, 0] = 0
    28.  
    29.         for i in range(reference.shape[0]):
    30.             for j in range(test.shape[0]):
    31.                 cost = self.disFunc(reference[i, :], test[j, :])
    32.                 r_index = i - 1
    33.                 c_index = j - 1
    34.                 if r_index < 0:
    35.                     r_index = 0
    36.                 if c_index < 0:
    37.                     c_index = 0
    38.                 DTW_matrix[i, j] = cost + min(DTW_matrix[r_index, j], DTW_matrix[i, c_index],
    39.                                               DTW_matrix[r_index, c_index])
    40.         return DTW_matrix[-1, -1] / (test.shape[0] + reference.shape[0])
    41.  
    42.  
    43. # 语音录制
    44. class wordRecorder:
    45.     def __init__(self, samplingFrequency=8000, threshold=20):
    46.         self.samplingFrequency = samplingFrequency
    47.         self.threshold = threshold
    48.  
    49.     def record(self):
    50.         p = pyaudio.PyAudio()
    51.         stream = p.open(format=pyaudio.paInt16, channels=1, rate=self.samplingFrequency, input=True, output=False,
    52.                         frames_per_buffer=1024)
    53.         frames = []
    54.         for i in range(int(self.samplingFrequency * 4 / 1024)):
    55.             data = stream.read(1024)
    56.             frames.append(data)
    57.         stream.stop_stream()
    58.         stream.close()
    59.         p.terminate()
    60.         return frames
    61.  
    62.     def record2File(self, path):
    63.         frames = self.record()
    64.         p = pyaudio.PyAudio()
    65.         with wave.open(path, 'wb') as wf:
    66.             wf.setnchannels(1)
    67.             wf.setsampwidth(p.get_sample_size(pyaudio.paInt16))
    68.             wf.setframerate(self.samplingFrequency)
    69.             wf.writeframes(b''.join(frames))
    70.             print('record finished!')
    71.  
    72.  
    73. # 提取mfcc特征
    74. def getmfcc(audio, isfile=True):
    75.     if isfile:
    76.         # 读取音频文件
    77.         y, fs = librosa.load(audio, sr=8000)
    78.     else:
    79.         # 音频数据，需要去除静音
    80.         y = np.array(audio)
    81.  
    82.     intervals = librosa.effects.split(y, top_db=20)
    83.     y = librosa.effects.remix(y, intervals)
    84.     # 预加重
    85.     y = librosa.effects.preemphasis(y)
    86.  
    87.     fs = 8000
    88.     N_fft = 256
    89.     win_length = 256
    90.     hop_length = 128
    91.     n_mels = 23
    92.     n_mfcc = 14
    93.     # mfcc提取
    94.     mfcc = librosa.feature.mfcc(y=y, sr=fs, n_mfcc=n_mfcc, n_mels=n_mels, n_fft=N_fft, win_length=win_length,
    95.                                 hop_length=hop_length)
    96.     mfcc = mfcc[1:, :]
    97.     # 添加差分量
    98.     mfcc_deta = librosa.feature.delta(mfcc)
    99.     mfcc_deta2 = librosa.feature.delta(mfcc, order=2)
    100.     # 特征拼接
    101.     mfcc_d1_d2 = np.concatenate([mfcc, mfcc_deta, mfcc_deta2], axis=0)
    102.     return mfcc_d1_d2.T
    103.  
    104.  
    105. # 指定文件夹下文件个数
    106. def check_file(name):
    107.     os.makedirs('data', exist_ok=True)
    108.     save_dir = os.path.join('data', name)
    109.     os.makedirs(save_dir, exist_ok=True)
    110.  
    111.     n_files = 0
    112.     for roots, dirs, files in os.walk(save_dir):
    113.         for file in files:
    114.             if file.endswith('.wav'):
    115.                 n_files += 1
    116.     return n_files
    117.  
    118.  
    119. @st.cache_resource  # 防止重载
    120. def model_load():
    121.     model1 = ModelHotWord(os.path.join('data', '向上'))
    122.     model2 = ModelHotWord(os.path.join('data', '向下'))
    123.     model3 = ModelHotWord(os.path.join('data', '向左'))
    124.     model4 = ModelHotWord(os.path.join('data', '向右'))
    125.     models = [model1, model2, model3, model4]
    126.     return models
    127.  
    128. class ModelHotWord(object):
    129.     def __init__(self, path):
    130.         self.mfccs = get_train_mfcc_list(path)
    131.  
    132.     def get_score(self, ref_mfcc):
    133.         return get_score(ref_mfcc, self.mfccs)
    134.  
    135.  
    136. def get_train_mfcc_list(data_path):
    137.     mfccs = []
    138.     for roots, dirs, files in os.walk(data_path):
    139.         for file in files:
    140.             if file.endswith('wav'):
    141.                 file_audio = os.path.join(data_path, file)
    142.                 mfcc = getmfcc(file_audio)
    143.                 mfccs.append(mfcc)
    144.     return mfccs
    145.  
    146.  
    147. def get_score(ref_mfcc, list_mfccs):
    148.     m_dtw = DTW()
    149.     N = len(list_mfccs)
    150.     scores = 0
    151.     for i in range(N):
    152.         dis = m_dtw.compute_distance(ref_mfcc, list_mfccs[i])
    153.         scores = scores + dis
    154.     return scores / N

    DTW.py:

    1. # -*- coding:UTF-8 -*-
    2. from utils import *
    3.  
    4.  
    5. st.title('基于DTW算法的命令字识别')
    6. tab1, tab2 = st.tabs(['音频录制', '识别演示'])
    7.  
    8. with tab1:
    9.     list_labs = ['向上', '向下', '向左', '向右']
    10.     col1, col2, col3, col4 = st.columns(4)
    11.     with col1:
    12.         name = st.selectbox('模型选择', list_labs)
    13.     with col2:
    14.         st.write('命令字录制')
    15.         flag_record = st.button(label='录音')
    16.     with col3:
    17.         st.write('命令字重录')
    18.         flag_cancel = st.button(label='撤销')
    19.     with col4:
    20.         st.write('试听')
    21.         flag_show_audios = st.button(label='试听')
    22.         
    23.     info_file_number = st.empty()
    24.     info_file_number.write('命令字---%s--已有%d个样本'%(name, check_file(name)))
    25.      
    26.     
    27. info_audios = st.empty()
    28. info_success = st.empty()
    29. if flag_record:
    30.     info_audios.info('')
    31.     info_success.success('')   
    32.     n_files = check_file(name)
    33.     info_audios.info('开始录制---第%d个命令字---%s--请在2s内完成录制.....'%(n_files + 1, name))
    34.     save_dir = os.path.join('data', name)
    35.     audio_name = os.path.join(save_dir, '%d.wav'%(n_files + 1))
    36.     wRec = wordRecorder()
    37.     wRec.record2File(audio_name)
    38.     info_success.success('录制完成，保存为' + audio_name)
    39.  
    40. if flag_cancel:
    41.     n_files = check_file(name)
    42.     save_dir = os.path.join('data', name)
    43.     file_del = os.path.join(save_dir, str(n_files)+'.wav')
    44.     os.remove(file_del)
    45.     info_file_number.write('命令字--%s--已有%d个样本'%(name, check_file(name)))
    46.     
    47. if flag_show_audios:
    48.     n_files = check_file(name)
    49.     save_dir = os.path.join('data', name)
    50.     if n_files > 0:
    51.         for i in range(n_files):
    52.             audio_file = open(os.path.join(save_dir, '%d.wav'%(i+1)), 'rb')
    53.             audio_bytes = audio_file.read()
    54.             st.audio(audio_bytes, format='audio/')
    55.  
    56.  
    57. with tab2:
    58.     th = 125
    59.     st.write('识别演示')
    60.     if 'run' not in st.session_state:
    61.         st.session_state['run'] = False
    62.     def start_listening():
    63.         st.session_state['run'] = True
    64.     def stop_listening():
    65.         st.session_state['run'] = False
    66.  
    67.     col1, col2 = st.columns(2)
    68.     with col1:
    69.         st.button('开始检测', on_click=start_listening)
    70.     with col2:
    71.         st.button('停止检测', on_click=stop_listening)
    72.  
    73.     det_word = st.empty()
    74.     def init_up():
    75.         det_word.write('向上')
    76.     def init_down():
    77.         det_word.write('向下')
    78.     def init_left():
    79.         det_word.write('向左')
    80.     def init_right():
    81.         det_word.write('向右')
    82.     callbacks = [init_up, init_down, init_left, init_right]
    83.  
    84.     # 加载预测模型，提取好的一些mfcc特征
    85.     models = model_load()
    86.     dic_labs = {'0': '向上', '1': '向下', '2': '向左', '3': '向右', '-1': ''}
    87.  
    88.     while st.session_state['run']:  # 循环进行检测
    89.         wRec = wordRecorder()
    90.         wRec.record2File('data/test.wav')
    91.         ref_mfcc = getmfcc('data/test.wav', True)
    92.         # 在每个模型上进行打分，扎到最小分数作为检测结果
    93.         scores = [model.get_score(ref_mfcc) for model in models]
    94.         i_word = np.argmin(scores)
    95.         score = np.min(scores)
    96.         print(i_word, score)
    97.         if score < th:
    98.             i_det_word = i_word
    99.             callback = callbacks[i_det_word]
    100.             if callback is not None:
    101.                 callback()
    102.             print('---------det word---------', dic_labs[str(i_det_word)])
    103.         else:
    104.             continue

    python命令行运行streamlit run DTW.py即会出现web网页ui，结果如下图所示：

    参考DTW关键字检测-代码实现_哔哩哔哩_bilibili

  • 相关阅读:
    最小体力消耗路径 -- dijkstra算法应用
    AIGC（生成式AI）试用 6 -- 从简单到复杂
    刷题笔记26——图论二分图判定
    Maven常用命令、坐标、依赖管理、依赖范围
    【Spring】Spring的JdbcTemplate
    使用kettle进行正则表达式组件日志分析
    Linux友人帐之网络编程基础FTP服务器
    LeetCode 450.删除二叉搜索树中的节点和669.修建二叉搜索树思路对比 及heap-use-after-free问题解决
    第十四章 配置国家语言支持 (NLS)
    记录扩充linux服务器centos-root目录过程
  • 原文地址：https://blog.csdn.net/qq_24946843/article/details/133357792