2017年9月23日土曜日

加速度センサーで行動分類

人の行動のラベル付き加速度センサーのデータを準備したので、行動認識を行ってみます。最初なので、何も工夫せずに、LSTM3層+全結合1層で認識してみます。 具体的には、

model = Sequential()
model.add(LSTM(256, return_sequences=True, input_shape=(timesteps, data_dim)))
model.add(LSTM(256, return_sequences=True))
model.add(LSTM(256))
model.add(Dense(num_classes, activation='softmax'))
という構成です。ユニット数は256個としました。

実装を簡単にするため、学習時も評価時もデータからランダムに切り出してくるようにしました。そのため、同じモデルを学習できていても、精度は変動します。正しく評価するには、評価時は評価用データを全て使う必要があります。

入力データの長さを揃える必要があるため、データの長さの幅は64サンプルで揃えました。サンプリング周波数は32Hzなので、2秒分のデータで行動を推定することになります。データ長が64サンプル未満の場合はデータの先頭に0をパディングします。

切り出しは、クラスが均等に出現するようにしました。また、同一クラス内では、各ファイルが均等に出現するようにしました。データ量に関して、クラス間に偏りがあり、特定のクラスばかり学習されないようにするためです。これは、評価用データの切り出しに関しても同じです。

学習の経過を下図に示します。 バッチサイズ32、1エポックあたりのステップ数は100で、100エポック学習させました。評価のステップ数も100なので、3200個のデータで評価しています。

100エポック目の学習データでの精度は86.7%、評価データでの精度は64.4%となりました。残念ながら、あまりうまく学習できていません。評価データで90%くらいにはなるだろうと考えていましたが、そこまで簡単ではなさそうです。

コードは以下の通りです。

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
# This script trains a neural network model using accelerometer data
import keras, os, datetime
import numpy as np
from keras.models import Sequential
from keras.layers import Dense, Activation
from keras.layers import LSTM

TRAIN_EPOCH = 1

def make_model(timesteps, data_dim, num_classes):
    model = Sequential()
    model.add(LSTM(256, return_sequences=True,
                   input_shape=(timesteps, data_dim)))
    model.add(LSTM(256, return_sequences=True))
    model.add(LSTM(256))
    model.add(Dense(num_classes, activation='softmax'))
    print(model.summary())
    return model

def read_accel(filename):
    with open(filename) as f:
        data = []
        for line in f:
            for v in line.rstrip().split(" "):
                if not v.isdigit():
                    print(v, "is not digit")
            data.append([int(v) for v in line.rstrip().split(" ")])
        data = np.array(data)
        return data

# Data generator
def data_gen(filelist, batch_size, window_width):
    # Make a dict for converting a class ID to files
    num_class = 0
    with open(filelist) as f:
        id2f = dict()
        for line in f:
            id, filename = line.split(" ", maxsplit=2)
            id = int(id)
            id2f.setdefault(id, []).append(read_accel(filename.rstrip()))
            num_class = max(num_class, id)
        num_class += 1
    yield num_class # Return the number of classes at first

    # Generate data eternally
    while True:
        data = []
        labels = []
        for i in range(batch_size):
            while True:
                c = np.random.randint(0, num_class) # Select a class
                if c in id2f:
                    break
            # Select a file in the class
            f = np.random.randint(0, len(id2f[c]))
            # Select a window position
            p = np.random.randint(0, max(0, len(id2f[c][f])-window_width)+1)
            # Make a label for the selected class
            labels.append(keras.utils.to_categorical(c, num_classes=num_class))
            # Add selected data
            data.append(id2f[c][f][p:p+window_width,:])
            if(len(data[-1]) < window_width):
                padding = window_width - len(data[-1])
                data[-1] = np.append(np.zeros((padding, data[-1].shape[1]), dtype=np.int),
                                              data[-1], axis=0)
        yield np.array(data), np.array(labels).reshape((batch_size, num_class))

if __name__ == '__main__':
    if not os.path.exists("models"):
        os.mkdir("models")

    batch_size = 32
    window_width = 64
    dg = data_gen("train.list", batch_size, window_width)
    num_classes = next(dg)

    dgv = data_gen("val.list", batch_size, window_width)
    assert num_classes == next(dgv)

    # Make a model
    opt = keras.optimizers.Adamax(lr=0.002, beta_1=0.9, beta_2=0.999,
                                  epsilon=1e-08, decay=1e-4)
    model = make_model(window_width, 3, num_classes)
    model.compile(loss='categorical_crossentropy',
                  optimizer=opt,
                  metrics=['accuracy'])

    # Train the model
    hist = model.fit_generator(dg, steps_per_epoch=100, 
                               validation_data=dgv, validation_steps=100,
                               epochs=TRAIN_EPOCH).history

    # Write training history into a file
    with open("history.log", mode="a") as f:
        d = datetime.datetime.today()
        f.write("#"+d.strftime("%Y-%m-%d %H:%M:%S")+"\n")
        f.write("#Optimizer :"+str(type(opt))+str(opt.get_config())+"\n")
        f.write("#Data in the last line are calculated by the last model"
                " and not calculated in model.fit()\n")
        f.write("#epoch train-loss train-acc val-loss val-acc\n")
        for v in range(0, TRAIN_EPOCH):
            f.write("{0} {1:10.6f} {2:10.6f} {3:10.6f} {4:10.6f}\n"
                    .format(v, hist["loss"][v], hist["acc"][v],
                            hist["val_loss"][v], hist["val_acc"][v]))
        f.write("\n\n")

0 件のコメント :