Keras CNN Sound Classify #7

edgetpu_compiler yyy.tflite -o ./tpu

5. 次に、今回の、Coral Edge TPU での、マイク入力での判定プログラムです。
おんちゃんは、Coral Edge TPU USB を今回のために購入して、使いました。
Windows10 と、Raspberrypi 3B+(Buster) でそれぞれ試しましたが、
Keras での実行と同じように、当たります。

スピードは、
 Windows10 / USB3.0 : 60 ms 位
 RaspberryPi 3B+ / USB2.0 : 600ms 位 
  Raspi の方は、USB2.0 がネックのようですが、体感は、600ms より、結構速く感じます。
  音の判定の実用には使えそうです。

sound-predict3-tflite-tpu.py
# -*- coding:utf-8 -*- ''' Created on 2020/01/19 update on 2020/06/17 sound-predict3-tflite-tpu.py https://blog.imind.jp/entry/2019/08/11/182853 https://qiita.com/__Attsun__/items/e033d689c336315435b3 https://musicinformationretrieval.com/ipython_audio.html @author: nishi ''' import time #import tensorflow as tf import tflite_runtime.interpreter as tflite import matplotlib.pyplot as plt import librosa import librosa.display import numpy as np #import matplotlib as mpl import pyaudio #import os import platform EDGETPU_SHARED_LIB = { 'Linux': 'libedgetpu.so.1', 'Darwin': 'libedgetpu.1.dylib', 'Windows': 'edgetpu.dll' }[platform.system()] np.random.seed(0) #test_name='test4f' test_name='test4h' test_id=4 class_indx={ 0: 'dog', 1: 'rooster', 2: 'pig', 3: 'cow', 4: 'frog', 5: 'cat', 6: 'hen', 7: 'insects', 8: 'sheep', 9: 'crow', 10: 'rain', 11: 'sea_waves', 12: 'crackling_fire', 13: 'crickets', 14: 'chirping_birds', 15: 'water_drops', 16: 'wind', 17: 'pouring_water', 18: 'toilet_flush', 18: 'toilet_flush', 19: 'thunderstorm', 20: 'crying_baby', 21: 'sneezing', 22: 'clapping', 23: 'breathing', 24: 'coughing', 25: 'footsteps', 26: 'laughing', 27: 'brushing_teeth', 28: 'snoring', 29: 'drinking_sipping', 30: 'door_wood_knock', 31: 'mouse_click', 32: 'keyboard_typing', 33: 'door_wood_creaks', 34: 'can_opening', 35: 'washing_machine', 36: 'vacuum_cleaner', 37: 'clock_alarm', 38: 'clock_tick', 39: 'glass_breaking', 40: 'helicopter', 41: 'chainsaw', 42: 'siren', 43: 'car_horn', 44: 'engine', 45: 'train', 46: 'church_bells', 47: 'airplane', 48: 'fireworks', 49: 'hand_saw' } #f_list=['1-137-A-32.wav','1-13572-A-46.wav','1-50455-A-44.wav'] # load a wave data def load_wave_data(audio_dir, file_name): file_path = os.path.join(audio_dir, file_name) x, fs = librosa.load(file_path, sr=44100) return x,fs # change wave data to mel-stft def calculate_melsp_org(x, n_fft=1024, hop_length=128): stft = np.abs(librosa.stft(x, n_fft=n_fft, hop_length=hop_length))**2 log_stft = librosa.power_to_db(stft) melsp = librosa.feature.melspectrogram(S=log_stft,n_mels=128) return melsp # for load-datasep.y 改 2020.4.2 def calculate_melsp1(x, n_fft=1024, hop_length=128,n_mels=128): stft = np.abs(librosa.stft(x, n_fft=n_fft, hop_length=hop_length))**2 #log_stft = librosa.power_to_db(stft) # こちらが、オリジナル log_stft = librosa.power_to_db(stft,ref=0.0) melsp = librosa.feature.melspectrogram(S=log_stft,n_mels=n_mels) #melsp -= melsp.min() melsp /= melsp.max() return melsp # for load-dataset2.py def calculate_melsp2(x, n_fft=1024, hop_length=128,sr=44100): S = librosa.feature.melspectrogram(y=x,sr=sr, n_mels=128,n_fft=n_fft, hop_length=hop_length) db_D = librosa.amplitude_to_db(S, ref=0.0) db_D -= db_D.min() return db_D def load_melsp(dx,test=3): dx=dx[np.newaxis,:,:] #print(dx.shape) #print(dx.dtype) #x = x.astype('float32') # data type convert float63 -> float32 #x = x.astype('float32') dx_shape = dx.shape # reshape # (-1,128,1723) -> (-1,128,1723,1) dx = np.reshape(dx, (-1,dx_shape[1],dx_shape[2],1)) #print(dx.shape) # normalize if test == 2: dx_max=dx.max() dx_min=dx.min() print('>dx_max=',dx_max,', dx_min=',dx_min) if dx_min < 0.0 : dx_min *= -1.0 dx += dx_min dx_max += dx_min #dx /= dx_max dx /= 6.5 #if dx_max > 4.0: print('>dx_max=',dx_max) if test==3: dx /= 80.0 #dx /= 4.1 dx_max=dx.max() dx_min=dx.min() print('>dx_max=',dx_max) print('>dx_min=',dx_min) if test ==4: dx_max=dx.max() dx_min=dx.min() print('>dx_max=',dx_max) print('>dx_min=',dx_min) return dx def make_interpreter(model_file): model_file, *device = model_file.split('@') return tflite.Interpreter( model_path=model_file, experimental_delegates=[ tflite.load_delegate(EDGETPU_SHARED_LIB, {'device': device[0]} if device else {}) ]) classes = 50 #input_shape=(128,1379,1) input_shape=(128,1723,1) model_path="tpu/xxx2_edgetpu.tflite" if test_name=='test4h': input_shape=(256,862,1) model_path="tpu/yyy_edgetpu.tflite" dx_mx=input_shape[1] # 読込 itp = make_interpreter(model_path) itp.allocate_tensors() # 入出力のdetail確認 input_fs = itp.get_input_details() #=> [{'name': 'conv2d_6_input', 'index': 7, 'shape': array([ 1, 28, 28, 1], dtype=int32), 'dtype': <class 'numpy.float32'>, 'quantization': (0.0, 0)}] output_fs = itp.get_output_details() print('input_fs=',input_fs) #input_fs= [{'name': 'input_1', 'index': 295, 'shape': array([ 1, 128, 1723, 1]), 'dtype': <class 'numpy.float32'>, 'quantization': (0.0, 0)}] print('output_fs=',output_fs) #output_fs= [{'name': 'predictions/Softmax', 'index': 315, 'shape': array([ 1, 50]), 'dtype': <class 'numpy.float32'>, 'quantization': (0.0, 0)}] #sys.exit() #CHUNK=1024 #CHUNK=44100 #CHUNK=2**11 # 2048 RATE=44100 # Sampling Rate CHUNK=int(RATE/5) # 0.2 [sec] CHUNK predict_dt = 5*5 # predict data length -> 5 [sec] #predict_dt = 5*4 # predict data length -> 4 [sec] #dbl_off = 2 # double check offset dbl_off = 0 # double check offset #level_th=1000.0 # predict start Sound Level #level_th=2000.0 # predict start Sound Level #level_th=1500.0 # predict start Sound Level level_th=2300.0 # predict start Sound Level p=pyaudio.PyAudio() stream=p.open(format = pyaudio.paInt16, channels = 1, rate = RATE, frames_per_buffer = CHUNK, input = True, output = False) # inputとoutputを同時にTrueにする smaple_f=0 hop_length=128 n_mels=128 if test_name=='test4h': hop_length=256 n_mels=256 while stream.is_active(): input = stream.read(CHUNK) #output = stream.write(input) if smaple_f==0: #print(len(input)) #dx = np.frombuffer(input, dtype='int16').astype('float32') dx = np.frombuffer(input, dtype='int16') #print(dx.shape) #print(dx.dtype) level_max = dx.max() #print('level=',level_max) if level_max < level_th: continue print('level=',level_max) smaple_f=1 data = [] dcnt=0 data.append(input) dcnt+=1 if dcnt >= predict_dt+dbl_off: print('.',end='') data = b''.join(data) #dx = np.frombuffer(data,dtype="int16").astype('float32')/2**15 if test_name=='test4h': dx = np.frombuffer(data,dtype="int16").astype('float32')/2**15 else: #dx = np.frombuffer(data,dtype="int16").astype('float16')/2**15 dx = np.frombuffer(data,dtype="int16").astype('float32')/2**15 melsp = calculate_melsp1(dx,hop_length=hop_length,n_mels=n_mels) dt=load_melsp(melsp,test=test_id) if dbl_off >0: # (-1,128,1723) -> (-1,128,1723,1) dt=dt*255 dt=dt.astype('uint8') dx_st=dt.shape[2]-dx_mx #v = model.predict(dt[:,:,:dx_mx,:]) itp.set_tensor(input_fs[0]['index'], dt[:,:,:dx_mx,:]) itp.invoke() v=itp.get_tensor(output_fs[0]['index'])[0] #v2 = model.predict(dt[:,:,dx_st:,:]) itp.set_tensor(input_fs[0]['index'], dt[:,:,dx_st:,:]) itp.invoke() v2=itp.get_tensor(output_fs[0]['index'])[0] i=v.argmax() pred=v.max() s=class_indx[i] print('>',i,':',s,' ',pred) i=v2.argmax() pred=v2.max() s=class_indx[i] print('>',i,':',s,' ',pred) else: #input_data = x_test.astype(np.float32)[i:i+1] dt=dt*255 dt=dt.astype('uint8') itp.set_tensor(input_fs[0]['index'], dt) start = time.perf_counter() #start = time.time() itp.invoke() inference_time = time.perf_counter() - start #inference_time = time.time() - start v=itp.get_tensor(output_fs[0]['index'])[0] print('%.1fms' % (inference_time * 1000)) #print('%.4fms' % (inference_time)) #v = model.predict(dt) i=v.argmax() pred=v.max() s=class_indx[i] print('>',i,':',s,' ',pred) if False: #plt.figure(figsize=(14, 5)) #plt.figure(figsize=(11, 5)) librosa.display.waveplot(dx, sr=RATE) plt.show() if False: X = librosa.stft(dx) Xdb = librosa.amplitude_to_db(abs(X)) #plt.figure(figsize=(14, 5)) librosa.display.specshow(Xdb, sr=RATE, x_axis='time', y_axis='hz') plt.show() if False: plt.figure(figsize=(14, 5)) librosa.display.specshow(melsp, sr=RATE, x_axis="time", y_axis="mel", hop_length=128) plt.colorbar(format='%+2.0f dB') plt.title("sound-predict3") plt.show() smaple_f=0 data=None stream.stop_stream() stream.start_stream() stream.stop_stream() stream.close() p.terminate()

余談ですが、Coral Edge TPU USB は、マウザーで購入しました。
注文して、決済してから、３日位でも届きました。結構早かったし、値段も、 7,655円 と安かったです。