エロ動画のシーンを音声から分析するシリーズ 2: 転移学習のチュートリアルをやってみる

前回のあらすじ

前回はこちら

学習済みの YAMNet で何かしら、音の分類はできてるっぽいことがわかった。なので転移学習というものを試したい

このシリーズ一貫した目的は「あえぎ声」と「セックス中の会話」を分離すること

まず転移学習のチュートリアルをやってみる

これを見ながらやってみる

とりあえず、写経

import os

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import scipy.signal

import tensorflow as tf
import tensorflow_hub as hub
import tensorflow_io as tfio
from scipy.io import wavfile

# YAMNet の学習済みモデルをロード
yamnet_model_handle = 'https://tfhub.dev/google/yamnet/1'
yamnet_model = hub.load(yamnet_model_handle)

# 事前にダウンロードしておいた猫の鳴き声の wav
testing_wav_file_name = 'test_data/miaow_16k.wav'

# wav ファイルを読み込む tf.function
# tf.function にする理由はパフォーマンスのため？
@tf.function
def load_wav_16k_mono(filename):
    file_contents = tf.io.read_file(filename)
    wav, sample_rate = tf.audio.decode_wav(file_contents, desired_channels=1)
    wav = tf.squeeze(wav, axis=-1)
    sample_rate = tf.cast(sample_rate, dtype=tf.int64)
    wav = tfio.audio.resample(wav, rate_in=sample_rate, rate_out=16000)
    return wav

# wav データを読み込む
testing_wav_data = load_wav_16k_mono(testing_wav_file_name)

# 学習済み YAMNet のクラス名のリストを読み込む
class_map_path = yamnet_model.class_map_path().numpy().decode('utf-8')
class_names =list(pd.read_csv(class_map_path)['display_name'])

# YAMNet の分類を実行して結果を表示
scores, embeddings, spectrogram = yamnet_model(testing_wav_data)
class_scores = tf.reduce_mean(scores, axis=0)
top_class = tf.argmax(class_scores)
inferred_class = class_names[top_class]
print(f'The main sound is: {inferred_class}')
print(f'The embeddings shape: {embeddings.shape}')

# 事前にダウンロードしておいた ESC-50 のデータセット
esc50_csv = './datasets/ESC-50-master/meta/esc50.csv'
base_data_path = './datasets/ESC-50-master/audio/'

# csv を読み込む
pd_data = pd.read_csv(esc50_csv)

# 猫と犬のクラスの id を定義
my_classes = ['dog', 'cat']
map_class_to_id = {'dog':0, 'cat':1}

# dog と cat だけ残す
filtered_pd = pd_data[pd_data.category.isin(my_classes)]

# class_name を id にして target 列にアサイン
class_id = filtered_pd['category'].apply(lambda name: map_class_to_id[name])
filtered_pd = filtered_pd.assign(target=class_id)

# full path 化してfilename カラムにアサイン
full_path = filtered_pd['filename'].apply(lambda row: os.path.join(base_data_path, row))
filtered_pd = filtered_pd.assign(filename=full_path)

# filename, target, fold の 3 つのカラムを持つ Dataset を作る
filenames = filtered_pd['filename']
targets = filtered_pd['target']
folds = filtered_pd['fold']
main_ds = tf.data.Dataset.from_tensor_slices((filenames, targets, folds))

# wav_data, target, fold の 3 つのカラムを持つ Dataset を作る
def load_wav_for_map(filename, label, fold):
    wav = load_wav_16k_mono(filename)
    return wav, label, fold
main_ds = main_ds.map(load_wav_for_map)

# wav_data 0.48 * n 秒ごとに (n, 1024) の embeddings ができるので、 (n, 1024), (n), (n) のデータを作って
# それを unbatch で (1024), (), () に分ける
# つまり、 0.96 秒のデータがあったら、 2 行分のデータになる　
def extract_embedding(wav_data, label, fold):
    scores, embeddings, spectrogram = yamnet_model(wav_data)
    num_embeddings = tf.shape(embeddings)[0]
    return (embeddings, tf.repeat(label, num_embeddings), tf.repeat(fold, num_embeddings))
main_ds = main_ds.map(extract_embedding)
main_ds = main_ds.unbatch()

# fold を使って trainingg, validation, test 用にデータセットを分割
# fold 使う必要あるのかな？ランダムに分けちゃダメ？
cached_ds = main_ds.cache()
train_ds = cached_ds.filter(lambda embedding, label, fold: fold < 4)
val_ds = cached_ds.filter(lambda embedding, label, fold: fold == 4)
test_ds = cached_ds.filter(lambda embedding, label, fold: fold == 5)

# fold はもういらないので削除
remove_fold_column = lambda embedding, label, fold: (embedding, label)
train_ds = train_ds.map(remove_fold_column)
val_ds = val_ds.map(remove_fold_column)
test_ds = test_ds.map(remove_fold_column)

# シャッフルしてバッチサイズを決める
# prefetch すると training 中にデータの読み込みをやってくれる
train_ds = train_ds.cache().shuffle(1000).batch(32).prefetch(tf.data.AUTOTUNE)
val_ds = val_ds.cache().batch(32).prefetch(tf.data.AUTOTUNE)
test_ds = test_ds.cache().batch(32).prefetch(tf.data.AUTOTUNE)

# 転移学習用のレイヤーを用意
my_model = tf.keras.Sequential([
    tf.keras.layers.Input(shape=(1024), dtype=tf.float32, name='input_embedding'),
    tf.keras.layers.Dense(512, activation='relu'),
    tf.keras.layers.Dense(len(my_classes))
], name='my_model')
my_model.compile(loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True), optimizer="adam", metrics=['accuracy'])

# 3 epoc 改善がなければ終了する
callback = tf.keras.callbacks.EarlyStopping(monitor='loss', patience=3, restore_best_weights=True)

# トレーニング
history = my_model.fit(train_ds, epochs=20, validation_data=val_ds, callbacks=callback)

# 評価
loss, accuracy = my_model.evaluate(test_ds)

print("Loss: ", loss)
print("Accuracy: ", accuracy)

# 新しいモデルでテスト wav データを評価
scores, embeddings, spectrogram = yamnet_model(testing_wav_data)
result = my_model(embeddings).numpy()

inferred_class = my_classes[result.mean(axis=0).argmax()]
print(f'The main sound is: {inferred_class}')

コード読む

tf.function って何？

@tf.function
def load_wav_16k_mono(filename):
    file_contents = tf.io.read_file(filename)
    wav, sample_rate = tf.audio.decode_wav(file_contents, desired_channels=1)
    wav = tf.squeeze(wav, axis=-1)
    sample_rate = tf.cast(sample_rate, dtype=tf.int64)
    wav = tfio.audio.resample(wav, rate_in=sample_rate, rate_out=16000)
    return wav 

tf.function は書いた関数を Tensorflow のグラフの中に組み込んだり、自動微分したりすることができるようになるということらしい

v = tf.Variable(10.0)

@tf.function
def f(a):
    return a * a 

with tf.GradientTape() as tape:
    result = f(v)

print(tape.gradient(result, v)) 

このようにすると確かに自動微分されて勾配が表示された

ただ、今回のコードにおいて必要なのか？という気はした。謎。

Dataset.map について

main_ds = main_ds.map(f)

みたいな書き方してる箇所があるが、 f の関数は即座に dataset の全要素に適用されるわけではないようだ。以下のような動きをする

• main_ds.map(f) した時に一度だけ f が呼び出され、計算がグラフ化される
• main_ds がイテレートされた時に、グラフ化された計算の評価がされる

なので、関数はグラフ化のために一度だけ呼ばれるだけなので、関数に print とかを入れて debug とかはできないし、引数に対する演算はグラフ化できるタイプの演算しか含むことができない

Dataset.cache について

cached_ds = main_ds.cache()

みたいに Dataset.cache() っていうメソッドを呼び出すと、その時点で最後までイテレートして計算されてその結果が cache されるみたい

計算のグラフが変化したら、ちゃんとキャッシュ破棄して再計算するようにできているんだろうか。多分できてるんだろうね

YAMNet が返してくる embeddings について

embeddings の理解については「ニューラルネットワークに入れるために作られた、もしくは、ニューラルネットワークから出力されたベクトル」くらいの認識しかない。あってるかどうかはわからない

で、以下のように YAMNet は分類スコア以外に、 embeddings を返してくる

cores, embeddings, spectrogram = yamnet_model(wav_data)

これはなんの embeddings なんだろう？と疑問に思ってドキュメントを読んだ

これを読むと

the embedding vector is the average-pooled output that feeds into the final classifier layer

とのことなのので、分類スコアをまとめる直前のレイヤーに出力されるベクトルってことですね。

なるほど〜

次は

前回の学習済み YAMNet を使って、ざっくり分類した後、手動で「あえぎ声」と「会話」のデータセットを作る