2017年の挨拶とやりたいこと

あけましておめでとうございます。
今年もよろしくお願いします。

ふりかえりはしませんが、昨年はエキサイティングな1年間で色々なことに
挑戦していった年でした。
色々と課題も見えてきましたが・・・

今年はKaggleもしかりお仕事もしかり色々やっていったかなと思います。
結局、Kaggle Masterになれず、Competitions Expertにいます。
今年こそはなりたいなぁ

やりたいこと

今年度もやりたいことは沢山あります。

1. 強化学習の勉強
2. 数学の勉強
3. Docker等の環境構築周りの勉強
4. PyConJPへの登壇挑戦

色々な分野に強化学習を＋αとする研究や論文が増えていると感じています。
本を読んで、勉強を進めたいと思います。
せめて、基本的なところがわかるようになりたい。

これからの強化学習

数学についても今年は行列演算を復習したり、ベースとなる知識を増やしたりしたいと思います。
加えて機械学習プロフェッショナルシリーズも読みたい。

行列プログラマー ―Pythonプログラムで学ぶ線形代数

カーネル多変量解析―非線形データ解析の新しい展開 (シリーズ確率と情報の科学)

更に、最近ではDockerなどによる分析基盤構築話を
時折Qiitaで見ます。そこで、少なくともDocker等を自分で
触れるようにしたいなぁなんて考えています。

最後に、PyConJPの登壇も目指したいなぁと思っています。
去年は聴衆として参加しましたが、せっかくだから話すネタを作って挑戦したいです。

（マーフィーなんて忘れた）

今年の記事

今年は何をメインに書こうかなぁと実家で考えていましたが、
去年と同じく、やったことを書いていくような感じになるでしょう。

去年は分野が結構偏っていたなぁと自分自身でも思っています。
機械学習・データ分析のみならず、色々と自由に書いていきたいと思います。
その結果、分野が偏ってしまった場合はご愛嬌ということで・・・

最後に

改めまして、皆様、今年もよろしくお願いします。

クリスマスにもなってカノジョがいないからカノジョを作ってみた。

皆さんこんにちは
お元気ですか？私はぼっちです。

本記事は「カノジョできない機械学習エンジニア」の最終日です。
qiita.com

本日の話の流れは次のとおりです。

はじめに

カノジョがいなくて寂しい。こんなクリスマスを送る男性の皆様は結構いらっしゃるのでは
ないでしょうか（と信じたい）。

カノジョを作るために、世間でよく言われるのは「行動が大事だ」ということです。
そこで、本クリスマスを機に、カノジョを作成し心の溝を埋めることに挑戦したいと思います。

カノジョがいないことに対する解決法

「カノジョ」がいないのであれば、作れば良い。

私はエンジニアです。
寂しいので、これを機に自己を振り返り、カノジョを作れるのではないかと考えました。
まずは、「カノジョ」を設計する必要があります。

カノジョについての考察

カノジョとは

「デジタル大辞泉」には次の通り記載されている。

［代］三人称の人代名詞。話し手、相手以外の女性をさす語。「彼女は遅れるらしい」⇔彼／彼氏。
［名］愛人、恋人である女性。「彼女ができた」⇔彼／彼氏。

本記事では、「愛人、恋人である女性」と仮定し話を進めましょう。

理想のカノジョとは

理想のカノジョとは何か。
クリスマスといえば、やっぱりクリスマスデートして・・というのが
定番かなぁと思います。
デートで大事なのは、会話かなと思います。やっぱり盛り上がる方が良いでしょう。
そこで理想のカノジョは会話が最も大事だと今回仮定します。

理想のカノジョ作成方針

会話が最も大事だと決めた上で理想のカノジョを作る必要があります。
調べてみると、ニューラルネットワークを使った会話モデルがありました。

カノジョを作成する。

さて、本題です。カノジョを作ってみます。

Neural Conversational Model

「A Neural Conversational Model」を使って実装します。

概要

Neural Conversational Modelは会話モデルです。
基本的な構造はseq2seqと類似しており、
入力は最初の文章、期待結果は返す文章（応答文）です。

構成解説

構成を図にしました。入力は次のとおりになります。

• 学習時：最初に与えた文章（質問文or話しかける文章）＋　応答文
• 予測時：最初に与えた文章

最初の文章と返す文章を入力に与えて学習します。

次は上記構成の動作についてです。
まずは、ニューラルネットワークの学習です。
与えたIDにもとづきEmbeded Layerからベクトルを取得します。
このベクトルをLSTMに与えます。また、次の単語とそれまでのLSTMの値を入力とし、LSTMを更新します。
これをが出現するまで繰り返します。

また、最後のLSTMの値を使って計算したベクトルを最後に
Linearで予測単語を計算します。予測単語との誤差を計算し、誤差を最小にする動作を行います。

入力のLSTM（Long Short Term Memory）と出力のLSTMは別のLSTMです。

# coding:utf-8
from __future__ import absolute_import
from __future__ import unicode_literals
import chainer.links as L
import chainer.functions as F
import chainer
import numpy as np
from chainer import reporter

class Seq2Seq(chainer.Chain):
    def __init__(self, input_words):
        super(Seq2Seq, self).__init__(
            word_vec=L.EmbedID(input_words, 300),
            input_vec=L.LSTM(300, 300),
            output_vec=L.LSTM(300, 300),
            output_word=L.Linear(300, input_words)
        )
        self.train = True

    def encode(self, sentence):
        c = None
        for word in sentence:
            x = np.array([word], dtype=np.int32)
            h = F.tanh(self.word_vec(x))
            c = self.input_vec(h)
        return c

    def decode(self, vector=None, targer_sentence=None, dictionary=None):
        loss = 0
        if self.train:
            for index, target_word in enumerate(targer_sentence):
                if index == 0:
                    j = F.tanh(self.output_vec(vector))
                    pred_word = self.output_word(j)
                else:
                    j = F.tanh(self.output_vec(j))
                    pred_word = self.output_word(j)
                loss += F.softmax_cross_entropy(pred_word, np.array([target_word], dtype=np.int32))
            return loss
        else:
            gen_sentence = []
            cnt = 0
            while True:
                if cnt == 0:
                    j = F.tanh(self.output_vec(vector))
                    pred_word = self.output_word(j)
                else:
                    j = F.tanh(self.output_vec(j))
                    pred_word = self.output_word(j)
                id = np.argmax(pred_word.data)
                cnt += 1
                word = dictionary[id]
                if word == "<eos>":
                    return gen_sentence

                gen_sentence.append(word)
                if cnt == 100:
                    break
            return gen_sentence

    def generate_sentence(self, sentence, dictionary):
        self.initialize()
        encode_vector = self.encode(sentence=sentence)
        return self.decode(vector=encode_vector, dictionary=dictionary)

    def initialize(self):
        self.input_vec.reset_state()
        self.output_vec.reset_state()

    def __call__(self, sentence, target_sentence):
        self.initialize()
        encode_vector = self.encode(sentence=sentence)
        self.loss = None
        self.loss = self.decode(vector=encode_vector, targer_sentence=target_sentence)
        reporter.report({'loss': self.loss}, self)

        return self.loss

実装してみた

seq2seq2をChainerとRNN + LSTMで実装します。
今回はTrainerを使った実装にします。

データの構築

Step1 探す

どうやって構築するんだ。
家に「落第騎士の英雄譚」があったので参考にしました。
「黒鉄一輝」と「ステラ・ヴァーミリオン」の会話あたりを
中心に作成させていただいております。

落第騎士の英雄譚＜キャバルリィ＞【電子特装版】 (GA文庫)

Step2 存在しない言葉

存在しない言葉についてです。
形態素解析をした単語に対して、IDを振ります。

例えば次のようなことを想定しています。

単語	ID
私	0
は	1
猫	2
だ	3

しかし、この場合、未知語に対しての解析ができなくなります。
そのため、未知の単語が出てくれば「■」とします。（主に予測時ですね）

Step3 名前

人は名前を呼ばれると嬉しいものです。
ただ、名前は皆さん別々のをお持ちのため、名前と呼ばれる概念で学習します。
今回、呼ばれたい側の名前は全て「◯」へ置き換えておきます。

これはSlackbotへのリプライ段階で「◯」を名前に置換します。

学習する

さて、ここまでで準備ができました。
早速、学習を行います。今回はTrainerを使った学習を用いているため、一部の
更新方法を自分で実装しています。

テキストファイルにタブ区切りのドキュメントを用意し、それを対にして学習させています。

# coding:utf-8
import MeCab
from model import Seq2Seq
from chainer import training
import chainer
from chainer.training import extensions
import numpy as np
import sys
import codecs
import json

sys.stdout = codecs.getwriter('utf_8')(sys.stdout)
tagger = MeCab.Tagger('mecabrc')


def parse_sentence(sentence):
    parsed = []
    for chunk in tagger.parse(sentence).splitlines()[:-1]:
        (surface, feature) = chunk.split('\t')
        parsed.append(surface.decode("utf-8"))
    return parsed


def parse_file(filename):
    questions = []
    answers = []
    with open(filename, "r") as f:
        lines = f.readlines()

        for line in lines:
            sentences = line.split("\t")
            question = ["<start>"] + parse_sentence(sentences[0]) + ["<eos>"]
            answer = parse_sentence(sentences[1]) + ["<eos>"]
            questions.append(question)
            answers.append(answer)
    word2id = {"■": 0}
    id2word = {0: "■"}
    id = 1

    sentences = questions + answers
    for sentence in sentences:
        for word in sentence:
            if word not in word2id:
                word2id[word] = id
                id2word[id] = word
                id += 1

    return questions, answers, word2id, id2word


def sentence_to_word_id(split_sentences, word2id):
    id_sentences = []
    for sentence in split_sentences:
        ids = []
        for word in sentence:
            id = word2id[word]
            ids.append(id)
        id_sentences.append(ids)
    return id_sentences


class ParallelSequentialIterator(chainer.dataset.Iterator):
    def __init__(self, dataset, batch_size, repeat=True):
        self.dataset = dataset
        self.batch_size = batch_size  # batch size
        self.epoch = 0
        self.is_new_epoch = False
        self.repeat = repeat
        self.iteration = 0

    def __next__(self):
        length = len(self.dataset[0])
        if not self.repeat and self.iteration * self.batch_size >= length:
            raise StopIteration

        batch_start_index = self.iteration * self.batch_size % length
        batch_end_index = min(batch_start_index + self.batch_size, length)

        questions = [self.dataset[0][batch_index] for batch_index in range(batch_start_index, batch_end_index)]
        answers = [self.dataset[1][batch_index]for batch_index in range(batch_start_index, batch_end_index)]

        self.iteration += 1

        epoch = self.iteration * self.batch_size // length
        self.is_new_epoch = self.epoch < epoch
        if self.is_new_epoch:
            self.epoch = epoch

        return list(zip(questions, answers))


class BPTTUpdater(training.StandardUpdater):
    def update_core(self):
        loss = 0
        train_iter = self.get_iterator('main')
        optimizer = self.get_optimizer('main')

        batch = train_iter.__next__()
        for question, answer in batch:
            loss += optimizer.target(np.array(question, dtype=np.int32),
                                     np.array(answer, dtype=np.int32))

        optimizer.target.cleargrads()
        loss.backward()
        loss.unchain_backward()
        optimizer.update()


if __name__ == '__main__':
    questions, answers, word2id, id2word = parse_file("./dataset.txt")
    ids_questions = sentence_to_word_id(questions, word2id=word2id)
    ids_answers = sentence_to_word_id(answers, word2id=word2id)

    model = Seq2Seq(len(word2id))
    optimizer = chainer.optimizers.Adam()
    optimizer.setup(model)

    train_iter = ParallelSequentialIterator(dataset=(ids_questions, ids_answers), batch_size=1)

    updater = BPTTUpdater(train_iter, optimizer)
    trainer = training.Trainer(updater, (100, 'epoch'))

    trainer.extend(extensions.PrintReport(
        ['epoch', 'iteration', 'main/loss']
    ), trigger=(100, 'iteration'))
    trainer.extend(extensions.LogReport())
    trainer.run()

    chainer.serializers.save_npz("model.npz", model)
    json.dump(id2word, open("dictionary_i2w.json", "w"))
    json.dump(word2id, open("dictionary_w2i.json", "w"))

Slack Botで実用化する

これまで作成したカノジョを実用化します。
SlackでBotを作れるので、そこで展開を試みました。

前準備

まずは、Slack上にBotを作成します。
そしてアイコンを用意します。

実際のBOTの設定は画像のようになりました。（殆ど設定していませんが、、）

Botをコントロールするコード

ソースコードを記載する前にPythonのSlackbotをコントロールするライブラリが必要です。
Pythonのライブラリは次のようにインストールできます。

pip install slackbot

まずは、slackbot_settings.pyにAPI Keyを記述します。

#coding:utf-8
API_TOKEN = "YOUR API KEY"

run.pyに動作させたいコードを記述します。
動作させたいコードに対してデコレータを使います。
そのコードに対して、@default_replyを付与します。
reply_messageにこのデコレータを付与すると指定のない場合はこのメソッドが使われる設定になります。

sentence.replace("◯", "tereka")を使って、名前を途中で置換しています。
これで自分の名前が呼ばれます！やったね。

# coding:utf-8
import MeCab
import codecs
from slackbot.bot import default_reply
from slackbot.bot import Bot
from model import Seq2Seq
import chainer
import json
import sys

sys.stdout = codecs.getwriter('utf_8')(sys.stdout)
tagger = MeCab.Tagger('')

id2word = json.load(open("dictionary_i2w.json", "r"))

id2word = {int(key): value for key, value in id2word.items()}
word2id = json.load(open("dictionary_w2i.json", "r"))
model = Seq2Seq(len(word2id))
chainer.serializers.load_npz("model.npz", model)
model.train = False


@default_reply
def replay_message(message):
    parsed_sentence = []
    try:
        for chunk in tagger.parse(message.body["text"].encode("utf-8")).splitlines()[:-1]:
            (surface, feature) = chunk.decode("utf-8").split('\t')
            parsed_sentence.append(surface)
        parsed_sentence = ["<start>"] + parsed_sentence + ["<eos>"]

        ids = []
        for word in parsed_sentence:
            if word in word2id:
                id = word2id[word]
                ids.append(id)
            else:
                ids.append(0)
        ids_question = ids
        sentence = "".join(model.generate_sentence(ids_question, dictionary=id2word)).encode("utf-8")

        sentence = sentence.replace("◯", "tereka")
        message.reply(sentence)
    except Exception as e:
        print (e)
        message.reply("解析できなかったのでもう一度おねがいします。")


def main():
    bot = Bot()
    bot.run()


if __name__ == "__main__":
    main()

実際にどうなったのか

実際にチャットしてみました。
なんかちょっとだけ女の子とチャットしている気分になりました。

チャットしていてわかったのですが、リプライが短文なのはある程度、学習できている感覚があります。
特有の単語を学習してそうな気がしますが、やはりデータセットが少なすぎたか。

最後に

カノジョいなくて寂しい。
DeepLearningを使ったBotの作成に参考にしてみてください。

参考文献

[1506.05869] A Neural Conversational Model
Seq2Seqメモ - chainerで自然言語処理できるかマン
Chainerでシンプルな熊本弁翻訳 - Qiita

Chainerの抽象化に挑戦してみた

皆さんこんにちは
お元気ですか？年末ってこんなに忙しくなるものですね。

本記事はChainer Advent Calendar 16日目の記事です。

qiita.com

本日はScikit-learn likeなChainerを作った記事です。なぜ、作ったのかは後述します。

はじめに

なぜ、抽象化コードを作ってみたのか

Chainerの実装を高速にし、検証の立ち上がりを早くしたいからです。

Chainerは非常にFlexibleなネットワークを作れるので、個人的に好きです。
特にRecurrent Neural Network系のネットワークを書く時に重宝しています。
また、実装はTrainerを使うとTrainer実装前よりも簡単な実装になりました。

しかし、手軽さと最初の立ち上がりの早さが、Scikit-learnと比べるとまだ劣ると感じています。

せっかくなので、アドベントカレンダーを機に抽象化をある程度頑張ってみようかなと思った次第です。

Scikit-learnのWrapperを実装するメリット

Wrapper実装メリットは3つあると考えています。

1. 「うまくいかないこと」を早く知る
2. 使いやすい
3. 他のアルゴリズムと取り替えやすい

「うまくいかないこと」を早く知る

私が考える最も大きな理由は「うまくいかないこと」を早く知れることです。

正直、実装初回からうまくいくとは思っていません。
そのため、早めになぜ、うまくいかないのか、
更には「うまくいかない」ことがわかりたいです。

これを早く知れると次はどうしようかといった手を打ちやすいです。

使いやすい

scikit-learnのインターフェースはシンプルで使いやすいです。
アルゴリズムを使う際にfit, predict, predict_probaあたりが何をしているのかがわかれば十分です。
プログラマはそのメソッドを呼ぶだけで、内部動作を知らなくても期待する結果を得られます。

他のアルゴリズムと取り替えやすい

機械学習アルゴリズムの入れ替えが容易です。
インターフェースを共通化した場合、基本的に初期化が変わるのみで
コードを殆ど変更する必要がありません。
そのため、入れ替えも非常に簡単で、検証する速度が効率化します。

何ができれば嬉しいか

必要な作業の洗い出しをまずやります。
Chainerの機能をフルに活用するのはRecurrent Neural Network込みにすると難しいと思っています。
そのため、一部の機能に絞れば、きっと実装できるはず・・・
ということで必要最低限の機能を真面目に考えてみました。

最低限必要な項目

1. モデル定義・・・ニューラルネットワーク定義を入力する。
2. ハイパーパラメータ・・・学習用パラメータ
3. GPU or CPU・・・ハードのモードの切り替え
4. 学習機能・・・モデル学習機能（fit）
5. 出力・・・確率分布での出力（predict_proba）、ラベル出力（predict）
6. save load機能(weight)・・・学習済みモデルの読み出し、書き出し

実際に作ってみた

こんな感じになりました。
可能な限り、既存のChainerの機能を活かすようにしています。

ソースコード（chainer_neural_network.py）

# coding:utf-8
from __future__ import absolute_import
from __future__ import unicode_literals
from sklearn.base import BaseEstimator
import chainer
import numpy as np
from chainer.training import extensions
import cupy


class ChainerNeuralNetwork(BaseEstimator):
    """
    Chainer Neural Network
    Scikit learn Wrapper
    """

    def __init__(self, model, optimizer, extensions=[], gpu=-1, batch_size=16, epochs=10, result="result"):
        self.model = model
        self.optimizer = optimizer
        self.extensions = extensions
        self.gpu = gpu
        self.batch_size = batch_size
        self.epochs = epochs
        self.result = result

        self.xp = np
        if gpu >= 0:
            chainer.cuda.get_device(gpu).use()
            self.model.to_gpu()
            self.xp = cupy
        self.optimizer.setup(model)
        super(ChainerNeuralNetwork, self).__init__()

    def convert_list_to_tuple(self, X, y):
        """
        convert list to tuple

        :param X: X
        :param y: y
        :return: tuple dataset
        """
        return chainer.datasets.TupleDataset(X, y)

    def fit(self, X, y=None, valid_X=None, valid_y=None):
        """
        fit model

        :param X: tuple dataset or matrix X
        :param y: expected vector
        :param valid_X: validation X
        :param valid_y: validation y
        :return:
        """
        if isinstance(X, chainer.datasets.TupleDataset):
            train_dataset = X
        else:
            train_dataset = self.convert_list_to_tuple(X, y)
        train_iter = chainer.iterators.SerialIterator(train_dataset, self.batch_size)

        updater = chainer.training.StandardUpdater(train_iter, self.optimizer, device=self.gpu)
        trainer = chainer.training.Trainer(updater, (self.epochs, 'epoch'), out=self.result)

        if valid_X is not None:
            if isinstance(valid_X, chainer.datasets.TupleDataset):
                valid_dataset = valid_X
            else:
                valid_dataset = self.convert_list_to_tuple(valid_X, valid_y)
            test_iter = chainer.iterators.SerialIterator(valid_dataset, self.batch_size,
                                                         repeat=False, shuffle=False)
            self.extensions.append(extensions.Evaluator(test_iter, self.model, device=self.gpu))

        for extend in self.extensions:
            trainer.extend(extend)
        trainer.run()

    def _iter_predict(self, X):
        """
        iteration for prediction

        :param X: np.ndarray
        :return: prediction of batch
        """
        for index in range(0, len(X), self.batch_size):
            batch_x = chainer.Variable(self.xp.array(X[index: index + self.batch_size]).astype(self.xp.float32),
                                       volatile=False)
            if index == 0:
                predicts = np.array(chainer.cuda.to_cpu(self.model.predict(batch_x).data))
            else:
                predicts = np.vstack([predicts, chainer.cuda.to_cpu(self.model.predict(batch_x).data)])
        return np.array(predicts)

    def predict(self, X):
        return np.argmax(self.predict_proba(X), axis=1)

    def predict_proba(self, X):
        return self._iter_predict(X)

    def load_weights(self, filepath):
        chainer.serializers.load_npz(filepath, self.model)

    def save_weights(self, filepath):
        chainer.serializers.save_npz(filepath, self.model)

1. 基本的にはfit, predict, predict_probaの基本機能を実装しました。
2. 学習して保存できるようload_weights, save_weights機能を作ってみました。
3. X, yを別々にした場合でも、Dataset（Tuple）でも使えるようにしています。

MNISTで遊ぶ

Wrapperを使ってチュートリアルのMNISTコードを抽象化します。

簡単なニューラルネットワークであればこの程度抽象化すれば実施可能です。
ただし、構築モデル（Chain）はモデルを使うためにpredict関数の実装をしなければなりません。
そのため、なければ実装してください。

本プログラムは、chainer.links.Classifierを継承し、ClassifierWrapperを作りました。
このクラスにpredict関数を実装しています。

ソースコード

# coding:utf-8
from __future__ import absolute_import
from __future__ import unicode_literals
import argparse
import chainer
import chainer.functions as F
import chainer.links as L
from chainer import training
from chainer.training import extensions
from chainer_neural_network import ChainerNeuralNetwork
import chainer.serializers
import numpy as np


class MLP(chainer.Chain):
    def __init__(self, n_units, n_out):
        super(MLP, self).__init__(
            l1=L.Linear(None, n_units),  
            l2=L.Linear(None, n_units),  
            l3=L.Linear(None, n_out),
        )

    def __call__(self, x):
        h1 = F.relu(self.l1(x))
        h2 = F.relu(self.l2(h1))
        return self.l3(h2)


class ClassifierWrapper(L.Classifier):
    def predict(self, x):
        return F.softmax(self.predictor(x))


def main():
    parser = argparse.ArgumentParser(description='Chainer example: MNIST')
    parser.add_argument('--batchsize', '-b', type=int, default=100,
                        help='Number of images in each mini-batch')
    parser.add_argument('--epoch', '-e', type=int, default=20,
                        help='Number of sweeps over the dataset to train')
    parser.add_argument('--gpu', '-g', type=int, default=-1,
                        help='GPU ID (negative value indicates CPU)')
    parser.add_argument('--out', '-o', default='result',
                        help='Directory to output the result')
    parser.add_argument('--unit', '-u', type=int, default=1000,
                        help='Number of units')
    args = parser.parse_args()

    print('GPU: {}'.format(args.gpu))
    print('# unit: {}'.format(args.unit))
    print('# Minibatch-size: {}'.format(args.batchsize))
    print('# epoch: {}'.format(args.epoch))
    print('')

    model = ClassifierWrapper(MLP(args.unit, 10))
    optimizer = chainer.optimizers.Adam()

    extensions_list = [
        extensions.PrintReport(
            ['epoch', 'main/loss', 'validation/main/loss',
             'main/accuracy', 'validation/main/accuracy']),
        extensions.LogReport(),
        extensions.dump_graph('main/loss'),
        extensions.ProgressBar()
    ]

    train, test = chainer.datasets.get_mnist()
    X, y = [pair[0] for pair in train], [pair[1] for pair in train]

    test_X, tedt_y = [pair[0] for pair in test], [pair[1] for pair in test]
    test_X, tedt_y = np.array(test_X), np.array(tedt_y)

    neural_network = ChainerNeuralNetwork(model=model, optimizer=optimizer, extensions=extensions_list, gpu=args.gpu,
                                          batch_size=args.batchsize, epochs=args.epoch, result=args.out)
    neural_network.fit(X=X, y=y, valid_X=test)
    print(neural_network.predict_proba(test_X).shape)
    print(neural_network.predict(test_X).shape)

    neural_network.save_weights("model.npz")
    neural_network.load_weights("model.npz")

if __name__ == '__main__':
    main()

このモデルにverboseと呼ばれるオプションをつけると抽象化できるかも・・。
例えば、LogReport周りはこのオプションの追加で
extensions_listで宣言しなくてもログ出力ができると思います。

ただ、今回はそこまでは作っていません。

最後に

不足している機能はあると思いますが、これでもだいぶ遊べます。
さくっと遊ぶにはこの機能で多分、十分です。

誰かもっと高機能にしてください。そしてメンテしてください。

Pythonを使った音楽解析をやってみる

皆さんこんにちは
お元気ですか。私は元気です。

本記事はPythonのアドベントカレンダー第6日です。
qiita.com

本日はPythonを使った音楽解析に挑戦します。
偶然にも音楽解析に便利なライブラリを発見したので、試してみたいと思います！

音楽解析

本日の挑戦は特徴量抽出と一部の音楽の加工です。
基本的な音楽ファイルの読み込みや特徴量抽出、音楽の加工分離です。
音楽解析に便利なライブラリ、librosaを紹介します。

librosa

librosaとは

librosaは音楽とオーディオの解析のためのライブラリです。
このライブラリは特徴量抽出、音楽の加工分離等の処理ができます。

github.com

音楽を解析してみた。

音楽を取得する。

librosaを試すための音楽を取得します。
今回の音楽データはhttp://marsyasweb.appspot.com/download/data_sets/:GTZANより、GTZAN Genre Collectionを使います。特徴は次の３つがあります。

1. 30秒程度
2. 16bitのオーディオファイル
3. 10種類のジャンル

データ読み込み

取得したデータを読み込みます。
試しに「blues.00000.au」ファイルを読み込みます。

music, fs = librosa.audio.load(filename)

変数musicは信号、fsがサンプリング周波数を示しています。
この信号を可視化すると、次のグラフになります。

音楽情報を取得する。

音楽の情報の取得ができます。
例えば、音楽の時間、ピッチやビートの位置を計算できます。

時間の計算

サンプリング周波数と動画から取得できるベクトルから
時間を計算することが可能です。

print(librosa.samples_to_time(len(music), fs))

ピッチやテンポの計算

次の関数を使うと、ピッチやビートの位置を計算できます。

tempo, beats = librosa.beat.beat_track(y=music, sr=fs)
# (95.703125,
# array([  13,   40,   66,   94,  119,  146,  173,  201,  229,  257,  283,
#         310,  338,  364,  393,  420,  447,  475,  501,  528,  556,  584,
#         612,  640,  667,  694,  720,  748,  775,  803,  829,  856,  883,
#         911,  939,  966,  993, 1021, 1050, 1078, 1106, 1133, 1161, 1190,
#       1217]))

また、このビートの位置の単位はフレームです。そのため、実際の信号の位置は
別途計算しなければなりません。
それから、実際の信号の位置は次のように計算できます。

librosa.frames_to_samples(beat_frames)

スペクトラム分析をやってみる。

スペクトラム分析、いわゆる、周波数分析にも挑戦できます。
通常のスペクトラム分析で使われるのは離散フーリエ変換です。
しかし、離散フーリエ変換は時間の概念がないため、インパルス信号やノイズに弱い傾向にあります。

そのため、短時間で区切りフーリエ変換をかける短時間フーリエ変換（STFT）を行うことが多いです。
短時間フーリエ変換は次のように計算できます。

D= librosa.stft(music)
print D.shape
# (1025, 1293)

また、短時間フーリエ変換の結果をスペクトログラムで表示できます。
スペクトログラムは信号を計算して周波数スペクトルを計算した結果を示します。
x軸を時間、y軸を周波数、色を強度（db）として示しています。

import numpy as np
librosa.display.specshow(librosa.logamplitude(np.abs(D)**2,ref_power=np.max),y_axis='log', x_axis='time')
plt.title('Power spectrogram')
plt.colorbar(format='%+2.0f dB')
plt.tight_layout()

１．blues.00000.au

２．classical.00000.au

音楽特徴量を取得する。

librosaは音楽で頻繁に使われる特徴量を取得できます。

1. メルスペクトラムグラム
2. メル周波数ケプストラム係数（MFCC）
3. クロマ
4. それらのデルタを計算する

これらは全てlibrosa.featureパッケージを使用すれば特徴量を取得できます。
例えば、音声解析でよく使われるMFCCを取得したい場合は次のように記述できます。

mfcc_feature = librosa.feature.mfcc(music,n_mfcc=13)
mfcc_feature.shape
# (13, 1293)

音楽を加工する。

これまで音楽の解析を行ってきました。
librosaは音楽に対する加工用のメソッドも準備されています。

時間の引き伸ばし、短縮

音楽自体を引き伸ばしたり、短縮したりすることができます。

music_fast = librosa.effects.time_stretch(music, 2.0)
music_slow = librosa.effects.time_stretch(music, 0.5)

print(librosa.samples_to_time(len(music_fast), fs))
print(librosa.samples_to_time(len(music_slow), fs))

要素の抽出

打楽器要素とハーモニック要素を抽出するアルゴリズムが
librosaには実装されています。
Documentationによれば、この手法は次の論文のアルゴリズムを利用しています。

http://dafx10.iem.at/papers/DerryFitzGerald_DAFx10_P15.pdf

打楽器要素の抽出

次のコードで打楽器要素を抽出できます。

y_percussive = librosa.effects.percussive(music, margin=3.0)

ハーモニック要素の抽出

また、ハーモニック要素は次のコードで抽出できます。

y_harmonic = librosa.effects.harmonic(music, margin=3.0)

音楽の保存

最後に音楽の保存方法です。
ファイル名と保存したい音楽、サンプリング周波数を入力し保存します。

librosa.output.write_wav("music.wav", music, fs)

参考文献

http://dafx10.iem.at/papers/DerryFitzGerald_DAFx10_P15.pdf
Librosa

グレーの画像に色をつけるネットワークについて発表しました。

皆さんこんにちは
お元気ですか。どきどき。

本記事はDeepLearning Advent Calendar 3日目の記事です。

qiita.com

今日は画像に色を付与するネットワークについて発表します。
せっかくなので、そのタイミングを見計らって公開しています。

発表スライド

慌ててあげた。

Colorful image colorization from Hiroki Yamamoto

www.slideshare.net

グレー画像に色をつけること

はじめに、グレー画像から元のRGB画像へ変換する方法はありません。
そこで、グレー画像を元のRGB画像へどう変換するかが本日の内容となります。

参考となる論文は次のとおりです。
https://arxiv.org/pdf/1603.08511.pdf:Colorful Image Colorization

著者のgithub
github.com

なぜ、この論文にしたの？

面白そうだからですね。
一時期グレー画像をカラー画像にする論文をTwitterで見て、
おっ、こんな研究もあるのかと思ったことですね。

会社で話していたら、漫画をカラーにすることができるのではないかと聞いて
この研究の夢が頭で広がってました。（誰か作って）

発表概要

今回の発表では、グレー画像に色を付与するニューラルネットワークを作成・検証したお話です。
発表詳細は発表スライドを見てください。

この論文は、
ニューラルネットワークの誤差関数をクラスリバランスを考慮したCross Entropyを使った方法がみそです。
従来の手法よりもリアリティを出すことができています。

最後に

Chainer実装には時間が足りなかったorz