はじめに
2020年も年末になったこともあり、ふと今年一年を振り返ってみると、コロナウィルスの影響で会社に出社せず自宅で仕事をするという昨年までは想像もしなかったような大きな変化がありました。
在宅勤務を開始するまでは、自宅で働くなんて仕事に支障出るでしょ！と思っていたのですが、開発をするだけであれば意外に問題なく働けるなと感じました。特に、やることが明確になったあとであれば、あとはやるだけなので必要なコミュニケーションを最低限取れば困らなかったです。ただ、やることを明確にするフェーズ(例えば、企画段階)だと、初めて一緒に仕事をする人だったり関係者が多い状況では、なかなか考え方が揃わず、共通認識にするまでに時間がかかったり、お互いを知ることに関する難しさを体感したので、実際にあって会話をしたりコミュニケーションをとるというのは重要だとも感じました。
とはいえ、今後も常に会社に行って働くということは考えづらかったので、自宅での働き方を改善するために試してみたことを共有します。
テレワークでの課題
- 自分の働き方を客観的に知ることが難しい
- 仕事に集中して休憩をとる機会が減っていた
- 逆にだらだら過ごしているかどうかもわかりづいらい
- 自分のステータス(仕事中、休憩中)が他の人に伝わりづらい
- PCの前にいないときにDiscordで話しかけられたとしても、あとで気づけない。
- SlackやDiscordのスターテスを自分で変更した方が良いのかもしれないけど、こまめに設定を変更するのがめんどくさい。忘れてしまう。
今回作るもの
- 自分のステータス（仕事中、休憩中、...等)をMLｘIoTで認識できる
- 自分のステータス、他の人に伝えられる
- Slackのステータスに自動で反映する
- 自分のステータスをもとに、アドバイスをくれる
- 例えば
- １時間以上働いていたら、休憩をとるようにアドバイスをくれる
- だらだらしていたら、真面目に働け！と怒られる
- 例えば
開発方針、使用したもの
- 自分のステータスを知るためにカメラ、画像認識を利用する。
- 転移学習
- AWS Rekognition Custom Label
- 自作
- 他の人に自分の状況を伝えるために、Slackのステータスを変える
- 働き方を改善するアドバイスをSlackに送る
全体構成
手順
1. M5Cameraでカメラ画像を取得する
M5Cameraは約2000円のIoTデバイスです。M5Cameraの環境構築やWebServer化のコードはこちらが参考になります。
PCをM5Cameraと同じWiFiに接続した上で、http://{M5CameraのローカルIP}/capture に対してHTTP GETすることで、画像を取得することができます。
また、Streamエンドポイントを有効にしておくと、すぐにM5Cameraが熱くなってしまい長時間起動することができないので、Streamに関するコードはコメントアウトしておくことをお勧めします。
2. 画像認識のモデルを作成する
学習データを作成する
定期的に画像を保存して、学習データを集めます。今回は、半日くらいデータを取得してみました。※ 一日だけだと服装などの環境差分で精度が下がってしまったので、最終的には三日間くらいの学習データを利用しました。
カテゴリとして、以下の４種類を用意しました。
a. 開発用PCで作業
b. 社内システム接続用PCで作業 ←事務作業をどのくらいしているか知りたかったので追加。
c. 不在(部屋にいない)
d. だらだらしている(寝ている、居眠りしている)
モデルを学習する
用意したデータセットをもとに、転移学習を行います。以下のコードを利用して、学習させました。
補足
はじめは、AWSのカスタムラベルを利用していたのですが、精度を向上させるために学習データを増やしたりバリエーションを変えて学習させるたびに費用が発生してしまうのこともあり、自分でコードを書いて学習させることにしました。認識精度は自作するよりAWSのカスタムラベルを使った方が良いのですが、今回はカテゴリが少ないこともあり、自作でも十分な精度が出たので、コストがかからない方を選びました。
<参考>AWS Rekogniton Custom Labelの料金表
from keras.models import Model
from keras.layers import Dense, GlobalAveragePooling2D,Input
from keras.applications.vgg16 import VGG16
from keras.preprocessing.image import ImageDataGenerator
import matplotlib.pyplot as plt
N_CATEGORIES = 4
IMAGE_SIZE = 224
BATCH_SIZE = 5
NUM_EPOCHS = 100
train_data_dir = 'train'
validation_data_dir = 'test'
NUM_TRAINING = 25
NUM_VALIDATION = 5
input_tensor = Input(shape=(IMAGE_SIZE, IMAGE_SIZE, 3))
base_model = VGG16(weights='imagenet', include_top=False, input_tensor=input_tensor)
x = base_model.output
x = GlobalAveragePooling2D()(x)
x = Dense(1024, activation='relu')(x)
predictions = Dense(N_CATEGORIES, activation='softmax')(x)
model = Model(inputs=base_model.input, outputs=predictions)
for layer in base_model.layers[:15]:
layer.trainable = False
from keras.optimizers import SGD
model.compile(optimizer=SGD(lr=0.001, momentum=0.5), loss='categorical_crossentropy',metrics=['accuracy'])
model.summary()
train_datagen = ImageDataGenerator(
rescale=1.0 / 255,
shear_range=0.2,
zoom_range=0.2,
horizontal_flip=True,
rotation_range=10)
test_datagen = ImageDataGenerator(
rescale=1.0 / 255,
)
train_generator = train_datagen.flow_from_directory(
train_data_dir,
target_size=(IMAGE_SIZE, IMAGE_SIZE),
batch_size=BATCH_SIZE,
class_mode='categorical',
shuffle=True
)
validation_generator = test_datagen.flow_from_directory(
validation_data_dir,
target_size=(IMAGE_SIZE, IMAGE_SIZE),
batch_size=BATCH_SIZE,
class_mode='categorical',
shuffle=True
)
history = model.fit_generator(train_generator,
steps_per_epoch=NUM_TRAINING//BATCH_SIZE,
epochs=NUM_EPOCHS,
verbose=1,
validation_data=validation_generator,
validation_steps=NUM_VALIDATION//BATCH_SIZE,
)
model.save('model.h5')
# グラフ描画
# Accuracy
plt.plot(range(1, NUM_EPOCHS+1), history.history['accuracy'], "o-")
plt.plot(range(1, NUM_EPOCHS+1), history.history['val_accuracy'], "o-")
plt.title('model accuracy')
plt.ylabel('accuracy')
plt.xlabel('epoch')
plt.legend(['train', 'test'], loc='upper left')
plt.show()
# loss
plt.plot(range(1, NUM_EPOCHS+1), history.history['loss'], "o-")
plt.plot(range(1, NUM_EPOCHS+1), history.history['val_loss'], "o-")
plt.title('model loss')
plt.ylabel('loss')
plt.xlabel('epoch')
plt.legend(['train', 'test'], loc='upper right')
plt.show()
3. 学習したモデルを利用して推論する
学習したモデルを利用して、推論を行います。M5Cameraから1分ごとにカメラ画像を取得し、推論を行います。
初めは学習データが少なくて精度が低いこともありましたが、３日間くらいの学習データを利用すると精度が向上して、かなりの確率で期待通りの推論結果になりました。
from keras.applications.vgg16 import preprocess_input
import keras.preprocessing.image as Image
import numpy as np
from keras.models import load_model
model = load_model('transfer.h5')
image_path = "推論したい画像ファイルのパス"
image = Image.load_img(image_path, target_size=(224, 224))
x = Image.img_to_array(image)
x = np.expand_dims(x, axis=0)
x = preprocess_input(x)
result = model.predict(x, 1)[0]
for i, score in enumerate(result):
print('{}: {:.2%}'.format(i, score))
4. 推論結果をもとに、Slackのステータスを変更
推論の結果を利用して、Slackのステータス変更を行います。
以下のコードを推論のたびに実行することで、Slackのステータスが変更され、他の人に自分のステータスを伝えられるようになります。
このおかげで、コンビニに行ったり、トイレにいったとしても、他の人に自分が今対応できないということを伝えられるので、他の人を無駄に待たせずにすみます。
import requests
import json
def set_user_profile(text, emoji):
data = {
"token" : "Slackのトークン",
"user": "SlackのユーザーID",
"profile":json.dumps({
"status_text":text,
"status_emoji":emoji
})
}
requests.post('https://slack.com/api/users.profile.set',params = data)
set_user_profile("休憩中です",":no_bell:")
5. アドバイスをSlackに送る
Slackのメッセージ送信APIを使って、１時間以上休憩を取らずに仕事を続けていたときに「休憩」をとるようにメッセージを送るようにしたり、仕事中に休んでいたら喝を入れてくれるようにしてみました。
1.休憩を取らず仕事を続けていたとき
2.仕事中にサボっているとき
終わりに
- MLとIoTを組み合わせることで、テレワークでの働き方を改善することが簡単にできるなーと感じました。少し昔だと、MLもIoTも利用するには少し手間がかかっていたのですが、今ではどちらも簡単に始められるのはよかったです。
- はじめは、他の人に"今は対応できないよ"というのを自動で伝えられたら良いなと思って作り始めたのですが、自分の働き方・過ごし方をデータ化することができるので、コーチングサービスにも応用できると思って、途中からはそちら目的で作っていました。学習データやカテゴリを増やして、もう少し細かな認識ができるように改善してみようと思います。
- ただ、土日もずっと起動していたら、休日もSlackにコーチからメッセージが届いて、逆にストレスだったので、休日や冬休みはサーバを落としておこうと思います。
- 今回の開発で一番難しかったのは、カメラを部屋のどこに配置するかとその固定方法でした。百均で買ったものでなんとか固定することができましたが、自分の思い通りの位置に、しっかり固定するのは案外難しかったです。何か良い方法があれば教えていただけると嬉しいです。