[Special edition] インターンシップ業務集大成-PAIを用いた音声認識

皆さんこんにちは。この度、インターンシップ生としてSBクラウドに配属されたnakaneです。今回は、私が業務体験期間の間に取り組んできた“PAIを用いた音声認識”についてブログを投稿させていただきます。

今回、このブログで扱うPAIについてやPAIを用いた画像認識に関する詳細な情報は以下のリンクをご参照ください。

Alibaba Cloudの機械学習サービスPAI

PAIで実現するImage Recognition

 

イントロダクション

着任一週目、クラウドに関する基礎勉強やハンズオントレーニングを受けながら、クラウド技術で一体全体何ができるのだろうかとワクワクしていたところ、なんと与えられた課題は以下の2つでした。

・研究テーマとクラウド技術を融合した構成の提案

・Alibaba Cloudの新プロダクトについて調査・紹介・デモ

クラウド初心者の私は、本当にこの課題を達成できるのかと不安でした。そこで、まずは自身の研究テーマの課題を洗い出し、それらの課題はどのようなクラウド技術を導入すれば解決出来るのか考えました。そして、研究テーマとクラウド技術を融合した構成として私が考えた構成が次の図になります。

構成としては、まず、シミュレーションで再現された心電図データや医療現場で測定された患者さんの心電図データを大量に集めて、ビッグデータとして医療クラウドにのせます。次に、クラウド上のデータを機械学習で分析にかけて特徴を抽出します。そして、抽出された特徴データを、医療現場のサポートや学問へのフィードバックに利用します。

全体の構成が完成したので、早速構成実現のために機械学習のアルゴリズムの実装に取り掛かりました。まずは、心電図のサンプルデータを・・・中々見つかりません。当たり前ですが、医療現場のデータを個人が手に入れるのは不可能に近いです。また、利用できるサンプルデータも数が少なく。とてもビッグデータとは言えません・・・。

何か他の物理現象で代替できないかと考えた結果、私は音声に着目しました。例えば、人間はスピーカから流れている音を耳で聞いて、その音が何かを認識します。一方の心電図は、観測された心電図をお医者さんが見て、どんな病気なのかを判断します。このように、音声と心電図は物理現象として非常に似ている現象であるため、代替案として音声認識に挑戦しました。

音声認識の流れを大まかに説明すると、波形データに対してデータ処理を行い、波形データが何の音声なのかを特定するアルゴリズムになります。各実装内容に対して必要なAlibaba Cloudのプロダクトは、データ管理としてOSS、ビッグデータ処理としてPAI、デモ環境の構築としてECSになります。

音声データから画像データへの変換

今回データ処理の部分に関しては、以前のtech.blogに記載されていた“PAIを用いたImage Recognition”を参考にしてDeep Learningの手法の一つであるCNN(Convolutional Neural Network)を導入しました。しかし、CNNは画像認識に向いたニューラルネットワークであるため、時系列で記録されている波形データを画像データに変換する必要があります。音声データを波形データの変換する方法は以下のサイトを参考にさせていただきました。

Reference
Pythonを使って音声データからスペクトログラムを作成する
https://own-search-and-study.xyz/2017/10/27/

<audio_to_image.py>

from pydub import AudioSegment
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import os
import glob

# Normalize(0-1)
def min_max(x, axis=None):
    min = x.min(axis=axis, keepdims=True)
    max = x.max(axis=axis, keepdims=True)
    result = (x-min)/(max-min)
    return result

# 0を最小値に変換(対数変換への対策)
def replaceZeroes(data):
  min_nonzero = np.min(data[np.nonzero(data)])
  data[data == 0] = min_nonzero
  return data


# 音声ファイル(WAV)を画像ファイルに変換
path = 'root_path'
kinds = ['dirname']
for (i,x) in enumerate(kinds):
    dirpath = os.path.join(path,kinds[i])
    filelist = os.listdir(dirpath)
    for (j,y) in enumerate(filelist):
        filename = os.path.join(path,kinds[i],filelist[j])
        os.chdir(filename)
        wavlist = glob.glob('*.wav')
        for (k,z) in enumerate(wavlist):
            sound = AudioSegment.from_wav(wavlist[k])
            samples = []
            sample = []
            spec = []
            freq = []
            samples = np.array(sound.get_array_of_samples())
            sample = samples[::sound.channels]
            #窓幅
            w = 1000
            #刻み
            s = 500
            #振幅スペクトル格納用
            ampList = []
            #窓関数(矩形窓)をずらしながら窓幅分のデータをフーリエ変換する
            for m in range(int((sample.shape[0]- w) / s)):
                data = sample[m*s:m*s+w]
                spec = np.fft.fft(data)
                spec = spec[:int(spec.shape[0]/2)]
                spec[0] = spec[0] / 2
                ampList.append(np.abs(spec))
            #周波数は共通なので1回だけ計算(縦軸表示に使う) 
            freq = np.fft.fftfreq(data.shape[0], 1.0/sound.frame_rate)
            freq = freq[:int(freq.shape[0]/2)]
            #時間も共通なので1回だけ計算(横軸表示に使う)
            time = np.arange(0, m+1, 1) * s / sound.frame_rate

            ampList = np.array(ampList)
            ampList = replaceZeroes(ampList)
            ampList = np.log10(ampList)
            ampList = min_max(ampList)
            df_amp = pd.DataFrame(data=ampList, index=time, columns=freq)

            #seabornのheatmapを使う
            plt.figure(figsize=(1, 1))
            sns.heatmap(data=df_amp.iloc[:,:].T, 
            xticklabels=False, 
            yticklabels=False, 
            cmap=plt.cm.gist_rainbow_r,
            cbar=False
            )
            plt.axis('off')
            output = wavlist[k].split('.',1)[0]
            plt.savefig(output+'.png') 
            plt.close()

 

データセット

データセットについて説明します。音声データセットは以下のサイトからダウンロードし、10class分(0-9)のデータを画像データに変換し、その後pickle形式でtraining用、test用、validation用のbatchファイルに保存しています。

Reference
Deep Learning Speech Recognition – MATLAB
http://download.tensorflow.org/data/speech_commands_v0.01.tar.gz

<image_to_pickle.py>

import numpy as np
import os
import glob
import pickle
import cv2
import linecache
import random
from PIL import Image

path = 'root_path'
kinds = ['dirname']

train_data = []
train_label = []
test_data = []
test_label = []
varidation_data = []
varidation_label = []

for (i,x) in enumerate(kinds):    
    dirpath = os.path.join(path,kinds[i])
    print(dirpath)
    os.chdir(dirpath)
    filelist = os.listdir(dirpath)
    print(filelist)
    with open('data_batch_1','wb') as f1:
        with open('test_batch','wb') as f2:
            with open('varidation_batch','wb') as f3:
                for (j,y) in enumerate(filelist):
                    print(j,y)
                    filename = os.path.join(path,kinds[i],filelist[j])
                    os.chdir(filename)
                    pnglist = glob.glob('*.png')
                    for (k,z) in enumerate(pnglist):
                        pngpath = os.path.join(path,kinds[i],filelist[j],pnglist[k])
                        im = cv2.imread(pngpath)
                        im = cv2.resize(im,(32,32))
                        R = np.reshape(np.array(im[:,:,0]),(1,1024))
                        G = np.reshape(np.array(im[:,:,1]),(1,1024))
                        B = np.reshape(np.array(im[:,:,2]),(1,1024))
                        data = np.array(np.concatenate([R, G, B], axis=1))

                        flag = random.randint(0,99)
                        if flag <= 69:
                            if len(train_data) == 0:
                                train_data = data
                                a = [j]
                                train_label = a
                            else:
                                train_data = np.append(train_data,data,axis=0)
                                a = [j]
                                train_label = train_label + a
                        
                        elif flag > 69 and flag <= 89:
                            if len(test_data) == 0:
                                test_data = data
                                a = [j]
                                test_label = a
                            else:
                                test_data = np.append(test_data,data,axis=0)
                                a = [j]
                                test_label = test_label + a
                        else:
                            if len(varidation_data) == 0:
                                varidation_data = data
                                a = [j]
                                varidation_label = a
                            else:
                                varidation_data = np.append(varidation_data,data,axis=0)
                                varidation_label = varidation_label + a
                image1 = {
                    'label': train_label,
                    'pixels': np.array(train_data),
                    }                       
                
                image2 = {
                    'label': test_label,
                    'pixels': np.array(test_data),
                    }

                image3 = {
                    'label': varidation_label,
                    'pixels': np.array(varidation_data),
                    }
                pickle.dump(image1, f1, protocol = 2)
                pickle.dump(image2, f2, protocol = 2)
                pickle.dump(image3, f3, protocol = 2)
                print(train_data.shape)
                print(len(train_label))
                print(test_data.shape)
                print(len(test_label))
                print(varidation_data.shape)
                print(len(varidation_label))

PAIについて

PAIについては以前の記事でも記載されているので、ここでは簡潔にまとめると、PAIはGUIで簡単に実現できる機械学習プラットフォームになります。事実、機械学習に関する事前知識が全くない私でも簡単にdrag-and-dropのみで構成を組めました。

PAIのコンソール画面やOSSに関する詳細な設定および以前の記事を参考にしてください。また、学習コードはImage Recognitionの際のコードを修正しています

<training.py>

# -*- coding: utf-8 -*-
from __future__ import division, print_function, absolute_import

import tensorflow as tf

from six.moves import urllib
import tarfile

import tflearn
from tflearn.data_utils import shuffle, to_categorical
from tflearn.layers.core import input_data, dropout, fully_connected
from tflearn.layers.conv import conv_2d, max_pool_2d
from tflearn.layers.estimator import regression
from tflearn.data_preprocessing import ImagePreprocessing
from tflearn.data_augmentation import ImageAugmentation

from tensorflow.python.lib.io import file_io
import os
import sys
import numpy as np
import pickle
import argparse

FLAGS = None

def load_data(dirname, one_hot=False):
    X_train = []
    Y_train = []

    
    fpath = os.path.join(dirname, 'data_batch_' + str(1))
    data, labels = load_batch(fpath)

    X_train = data
    Y_train = labels

    """
    if i == 1:
        X_train = data
        Y_train = labels
    else:
        X_train = np.concatenate([X_train, data], axis=0)
        Y_train = np.concatenate([Y_train, labels], axis=0)
    """

    fpath = os.path.join(dirname, 'test_batch')
    X_test, Y_test = load_batch(fpath)

    X_train = np.dstack((X_train[:, :1024], X_train[:, 1024:2048],
                         X_train[:, 2048:])) / 255.
    X_train = np.reshape(X_train, [-1, 32, 32, 3])
    X_test = np.dstack((X_test[:, :1024], X_test[:, 1024:2048],
                        X_test[:, 2048:])) / 255.
    X_test = np.reshape(X_test, [-1, 32, 32, 3])

    if one_hot:
        Y_train = to_categorical(Y_train, 10)
        Y_test = to_categorical(Y_test, 10)

    return (X_train, Y_train), (X_test, Y_test)

#reporthook from stackoverflow #13881092
def reporthook(blocknum, blocksize, totalsize):
    readsofar = blocknum * blocksize
    if totalsize > 0:
        percent = readsofar * 1e2 / totalsize
        s = "\r%5.1f%% %*d / %d" % (
            percent, len(str(totalsize)), readsofar, totalsize)
        sys.stderr.write(s)
        if readsofar >= totalsize: # near the end
            sys.stderr.write("\n")
    else: # total size is unknown
        sys.stderr.write("read %d\n" % (readsofar,))

def load_batch(fpath):
    object = file_io.read_file_to_string(fpath)
    #origin_bytes = bytes(object, encoding='latin1')
    # with open(fpath, 'rb') as f:
    if sys.version_info > (3, 0):
        # Python3
        d = pickle.loads(object, encoding='latin1')
    else:
        # Python2
        d = pickle.loads(object)
    data = d["pixels"]
    labels = d["label"]
    return data, labels

def main(_):
    dirname = os.path.join(FLAGS.buckets, "")
    (X, Y), (X_test, Y_test) = load_data(dirname)
    print("load data done")

    X, Y = shuffle(X, Y)
    Y = to_categorical(Y, 10)
    Y_test = to_categorical(Y_test, 10)

    # Real-time data preprocessing
    img_prep = ImagePreprocessing()
    img_prep.add_featurewise_zero_center()
    img_prep.add_featurewise_stdnorm()

    """
    # Real-time data augmentation
    img_aug = ImageAugmentation()
    img_aug.add_random_flip_leftright()
    img_aug.add_random_rotation(max_angle=25.)
    """

    # Convolutional network building
    network = input_data(shape=[None, 32, 32, 3],
                         data_preprocessing=img_prep)
    network = conv_2d(network, 32, 3, activation='relu')
    network = max_pool_2d(network, 2)
    network = conv_2d(network, 64, 3, activation='relu')
    network = conv_2d(network, 64, 3, activation='relu')
    network = max_pool_2d(network, 2)
    network = fully_connected(network, 512, activation='relu')
    network = dropout(network, 0.5)
    network = fully_connected(network, 10, activation='softmax')
    network = regression(network, optimizer='adam',
                         loss='categorical_crossentropy',
                         learning_rate=0.001)

    # Train using classifier
    model = tflearn.DNN(network, tensorboard_verbose=0)
    model.fit(X, Y, n_epoch=50, shuffle=True, validation_set=(X_test, Y_test),
              show_metric=True, batch_size=96, run_id='cifar10_cnn')
    model_path = os.path.join(FLAGS.checkpointDir, "model_50.tfl")
    print(model_path)
    model.save(model_path)

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--buckets', type=str, default='',
                        help='input data path')
    parser.add_argument('--checkpointDir', type=str, default='',
                        help='output model path')
    FLAGS, _ = parser.parse_known_args()
    tf.app.run(main=main)

<prediction.py>

# -*- coding: utf-8 -*-
from __future__ import division, print_function, absolute_import

import tensorflow as tf

from six.moves import urllib
import tarfile

import tflearn
from tflearn.data_utils import shuffle, to_categorical
from tflearn.layers.core import input_data, dropout, fully_connected
from tflearn.layers.conv import conv_2d, max_pool_2d
from tflearn.layers.estimator import regression
from tflearn.data_preprocessing import ImagePreprocessing
from tflearn.data_augmentation import ImageAugmentation

from tensorflow.python.lib.io import file_io
import os
import sys
import numpy as np
import pickle
import argparse
import scipy
FLAGS = None

def load_data(dirname, one_hot=False):
    X_varid = []
    Y_varid = []

    
    fpath = os.path.join(dirname, 'varidation_batch')
    data, labels = load_batch(fpath)

    X_varid = data
    Y_varid = labels
    X_varid = np.dstack((X_varid[:, :1024], X_varid[:, 1024:2048],
                         X_varid[:, 2048:])) / 255.
    X_varid = np.reshape(X_varid, [-1, 32, 32, 3])

    if one_hot:
        Y_varid = to_categorical(Y_varid, 10)
    return (X_varid, Y_varid)


def load_batch(fpath):
    object = file_io.read_file_to_string(fpath)
    #origin_bytes = bytes(object, encoding='latin1')
    # with open(fpath, 'rb') as f:
    if sys.version_info > (3, 0):
        # Python3
        d = pickle.loads(object, encoding='latin1')
    else:
        # Python2
        d = pickle.loads(object)
    data = d["pixels"]
    labels = d["label"]
    return data, labels


def main(_):
    dirname = os.path.join(FLAGS.buckets, "")
    (X, Y) = load_data(dirname)
    print("load data done")
    Y = to_categorical(Y, 10)

    # Convolutional network building
    network = input_data(shape=[None, 32, 32, 3])
    network = conv_2d(network, 32, 3, activation='relu')
    network = max_pool_2d(network, 2)
    network = conv_2d(network, 64, 3, activation='relu')
    network = conv_2d(network, 64, 3, activation='relu')
    network = max_pool_2d(network, 2)
    network = fully_connected(network, 512, activation='relu')
    network = dropout(network, 0.5)
    network = fully_connected(network, 10, activation='softmax')
    network = regression(network, optimizer='adam',
                         loss='categorical_crossentropy',
                         learning_rate=0.001)

    model = tflearn.DNN(network, tensorboard_verbose=0)
    model_path = os.path.join(FLAGS.checkpointDir, "model_50.tfl")
    print(model_path)
    model.load(model_path)
    
    """
    predict_pic = os.path.join(FLAGS.buckets, "prediction.jpg")
    img_obj = file_io.read_file_to_string(predict_pic)
    file_io.write_string_to_file("prediction.jpg", img_obj)

    img = scipy.ndimage.imread("prediction.jpg", mode="RGB")

    # Scale it to 32x32
    img = scipy.misc.imresize(img, (32, 32), interp="bicubic").astype(np.float32, casting='unsafe')
    print(img.shape)
    print(img)
    # Predict
    prediction = model.predict([img])
    print (prediction[0])
    num = ['eight','five','four','nine','one','seven','six','three','two','zero']
    print ("This is a %s"%(num[prediction[:].tolist().index(max(prediction[:]))]))
    """

    # Predict
    score = model.evaluate(X,Y)
    print(len(score))
    print(score)


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--buckets', type=str, default='',
                        help='input data path')
    parser.add_argument('--checkpointDir', type=str, default='',
                        help='output model path')
    FLAGS, _ = parser.parse_known_args()
    tf.app.run(main=main)

音声認識の結果

今回はIteration(一つの訓練データを何回繰り返して学習させるか)をパラメータとして学習モデルの精度を評価しました。精度評価の結果、学習モデルの最高精度はIteration=50で72%の精度でした。Iterationが50回を超えたところで、精度が下降していくのは、オーバーフィッティングという現象です。簡単に説明すると、勉強しすぎてアドリブが出来なくなってしまったみたいな状態です。また、モデル作成にかかった時間は、デモ用のECS(CPU)では約50分だったのに対して、PAI(シングルノードのGPU8カード)では約4分で10倍程度の差が出ました。

精度の改善

今回精度が72%という結果でしたが、より精度を向上させていくためには、欠損データを除外したりデータの一般化を行ったり、その他にも時系列データに適したネットワークの設計とまだまだやれることはあります。しかし、今回は限られた期間での実装であったため、十分な結果だったと思います。

デモ環境の構築

デモ環境に用意したECSはecs.n4.smallで、スペックは1CPU2GBになります。ECS上に、Jupyter Notebook環境を整え、以下のソースコードを準備します。あとは、Jupyter環境で実行するだけです。

<prediction_demo.ipynb>

#prediction number
number = 0-9


#setup
convert_list = [9,4,8,7,2,1,6,5,0,3]
num_list = ['eight','five','four','nine','one','seven','six','three','two','zero']
input = num_list[convert_list[number]]
model_name = 'model_50.tfl'
checkpointDir = 'check_point/'
predictiondir = 'predictionfile/'
wav_path = predictiondir + input + '.wav'
img_path = predictiondir + input + '.png'


from __future__ import division, print_function, absolute_import
import tensorflow as tf
from six.moves import urllib
import tarfile
import tflearn
from tflearn.data_utils import shuffle, to_categorical
from tflearn.layers.core import input_data, dropout, fully_connected
from tflearn.layers.conv import conv_2d, max_pool_2d
from tflearn.layers.estimator import regression
from tflearn.data_preprocessing import ImagePreprocessing
from tflearn.data_augmentation import ImageAugmentation
from tensorflow.python.lib.io import file_io
import os
import sys
import numpy as np
import pickle
import argparse
import scipy
import matplotlib.pyplot as plt
import matplotlib.image as mpimg
import IPython.display as ipd
import cv2
from pydub import AudioSegment

network = input_data(shape=[None, 32, 32, 3])
network = conv_2d(network, 32, 3, activation='relu')
network = max_pool_2d(network, 2)
network = conv_2d(network, 64, 3, activation='relu')
network = conv_2d(network, 64, 3, activation='relu')
network = max_pool_2d(network, 2)
network = fully_connected(network, 512, activation='relu')
network = dropout(network, 0.5)
network = fully_connected(network, 10, activation='softmax')
network = regression(network, optimizer='adam',
                     loss='categorical_crossentropy',
                     learning_rate=0.001)

model = tflearn.DNN(network, tensorboard_verbose=0)
model_path = os.path.join(checkpointDir, model_name)
model.load(model_path)


#play sound
ipd.Audio(wav_path)


# wave image
sound = AudioSegment.from_wav(wav_path)
samples = np.array(sound.get_array_of_samples())
sample = samples[::sound.channels]
plt.xlabel('Timse [sec]')
plt.ylabel('Amplitude')
plt.plot(sample[::10])


# FFT
#窓幅
w = 1000
#刻み
s = 500
#スペクトル格納用
ampList = []
#刻みずつずらしながら窓幅分のデータをフーリエ変換する
for i in range(int((sample.shape[0]- w) / s)):
    data = sample[i*s:i*s+w]
    spec = np.fft.fft(data)
    spec = spec[:int(spec.shape[0]/2)]
    spec[0] = spec[0] / 2
    ampList.append(np.abs(spec))

#周波数は共通なので1回だけ計算(縦軸表示に使う)  
freq = np.fft.fftfreq(data.shape[0], 1.0/sound.frame_rate)
freq = freq[:int(freq.shape[0]/2)]
plt.xlabel('Frequency [Hz]')
plt.ylabel('Amplitude ')
plt.plot(freq,np.abs(spec))


#play image
image = mpimg.imread(img_path)
plt.imshow(image)


# Predict
img = cv2.imread(img_path)
img = cv2.resize(img,(32,32))
prediction = model.predict([img])
num = ['eight','five','four','nine','one','seven','six','three','two','zero']
print ("This is a %s"%(num[prediction[0].tolist().index(max(prediction[0]))]))

まとめ

今回、クラウド・機械学習・Python未経験の三重苦の状態で業務がスタートし、中々思うように業務が進まず苦労しましたが、無事にブログという形で残すまで完成させることができました。業務を通しては、ユーザー側視点でクラウド技術の有難さを経験することが出来、クラウド技術を駆使したからこそ短期間で音声認識の実装にたどり着けたと思います。最後になりますが、インターンシップでお世話になったSBクラウドの社員の方々への感謝の言葉で締めくくらせていただきます。

一か月と短い期間でしたが、誠にありがとうございました。

この記事をシェアする