はじめに
既にお気づきでしょうが、私たち2lemetryはAmazonの最新AWS製品であるLambdaに興奮を抑えきれません。LambdaがMQTTプロトコル（訳者注：pub/subモデルに基づく軽量なメッセージプロトコル）やThingFabric（訳者注：2lementry社のIoTプラットフォーム）と共にどのように稼働するかを我が社の数名のエンジニアが検証しました。私も、IoT（モノのインターネット）においてLambda内でニューラルネットワークがどのように機能するのかを確かめるため、実際に使用してちょっとした実験を行ってみることにしました。私たちのCEOであるKyleが最近のブログ記事で書いたS3／MQTTの基礎環境をフル活用します。
使用ケース
膨大な数の異なる種類のセンサを多様に組み合わせて設置した、大規模な施設や工場を想像するのは簡単ですね。ここでは組立ラインに沿って油圧ポンプが設置された工場を想定しましょう。油圧ポンプには温度と圧力の主要なセンサが2組取り付けられています。温度と圧力がある組み合わせになると危険な状態になり、低い警報レベルに達した時点でアラームが作動します。さらに高い警報レベルに達するとデバイスが完全に停止します。ここで問題なのは、様々な要因があるために温度や圧力と警報信号の関係を具体的な方程式で表すことができないということです。こうした場合に機械学習が利用できます。
私が行ったLambdaの実験は次のようなシナリオに従いました。複数のセンサのセットが数秒毎にある種の中央ハブへ情報を送り、ハブがAmazon S3に複合ペイロードスナップショットをデプロイします。そしてAmazon S3がニューラルネットワークを持つLambdaを作動させます。警報の値が一定のレベルを超えると、LambdaがThingFabricにMQTTメッセージを送ります。ThingFabric側ではテキストメッセージを送るようにルールを設定できますし、別のMTQQメッセージを送ればデバイスを停止することもできます。
Lambdaのセットアップ
Lambdaのセットアップをよく理解するために、このシステムで発生するイベントを全て書き出してみましょう。
-
温度と圧力の値のペイロードが、特定のS3バケットにデプロイされます。
-
Lambdaが作動します。
-
ニューラルネットワークが適切な訓練用セットによって初期化されます。
-
ニューラルネットワークは逐次ペイロード値を評価し、ThingFabricの特定のトピックに送るMQTTパブリッシュコールを作動させます。
-
ThingFabricはメッセージを受け取り、それによって何か意味のあることを行います （例えば、値を遮断する、テキストメッセージやEメールを送るなど）。
高い警告レベルで何が起こるのか分かったところで、Lambdaをセットアップしていきましょう。まずAWSに新規のLambdaを作成し、ペイロードをデプロイするS3バケットへアクセスするためのロールを割り当てます。次にLambdaのリストから、”Configure event source”ボタンをクリックします。
次に出てくるダイアログで、デバイスのペイロードをデプロイする適切なS3バケットを選択し、先に進みます。
それではLambda用にいくつかお決まりのコードを書いていきましょう。
var aws = require('aws-sdk');
var s3 = new aws.S3({apiVersion: '2006-03-01'});
var brain = require("brain");
var mqtt = require('mqtt');
var config = {
mqtt: {
clientId: "<CLIENT ID>",
username: "<THING FABRIC USERNAME>",
md5Pass: "<THING FABRIC MD5 PASSWORD>",
outputTopic: "<THING FABRIC DOMAIN>/lambda"
},
alarm: {
maxTemp: 200,
maxPressure: 800,
alarmThreshold: 0.6
}
};
exports.handler = function(event, context) {
return context.done(null);
};
これでLambdaの初期設定と主なイベントハンドラの準備が完了しました。以降はMQTT接続やニューラルネットワークにおいてこの設定を利用します。
Amazon Lambdaの優れた点は、ほとんど全てのNode.jsライブラリがプロジェクト内で使えることです。その際はプロジェクトと同じディレクトリ構成で、ローカルに全てのライブラリをインストールする必要があります。なぜならAmazonではpackage.jsonを処理するような機能は提供されていないからです。今回の例では、3つのNodeのライブラリを使用します。aws-sdkとbrain、そしてmqttです。これらを./node_modulesフォルダにインストールすれば準備完了です。
S3ペイロードのパースとMQTTメッセージの送信
先に述べた通り、S3ペイロードのパースやThingFabricへのMQTTメッセージの送信方法については、Kyleが既に極めて詳細にわたって説明してくれています。したがって私はここではざっと内容をさらうにとどめます。まず、ペイロード用のデータの形式を定義しましょう。ここではデータは最小限にとどめて、特定のMQTTのトピックに紐づくデバイスのIDと、温度／圧力センサの値のペアだけを含むようにします。JSONでは次のような形になります。
{ "device_id": "foo", "values": { "t": 100, "p": 400 } },
{ "device_id": "bar", "values": { "t": 120, "p": 320 } },
{ "device_id": "foobar", "values": { "t": 90, "p": 220 } }
S3からオブジェクトをパースするために、シンプルなヘルパーパーサを作成します。
function S3Parser(event) {
var _this = this;
_this.event = event;
_this.s3 = new aws.S3({apiVersion: '2006-03-01'});
_this.parseS3Object = function(cb) {
_this.s3.getObject({
Bucket: _this.event.Records[0].s3.bucket.name,
Key: _this.event.Records[0].s3.object.key
}, function (err, data) {
return cb(JSON.parse(data.Body.toString()));
});
};
}
次にシンプルなMQTTクライアントを追加します。
function MqttClient(config, onClose, onError) {
var _this = this;
_this.username = config.username;
_this.pass = config.md5Pass;
_this.clientId = config.clientId;
_this.client = mqtt.connect(
"mqtt://" + _this.username + ":" + _this.pass + "@q.m2m.io:1883",
{ "clientId": _this.clientId }
);
_this.client.on("close", onClose);
_this.client.on("error", function(error) { onError(error) });
_this.publish = function(topicName, payload) {
return _this.client.publish(topicName, JSON.stringify(payload));
};
_this.disconnect = function() {
return _this.client.end();
};
}
これでコードの事前準備は全て完了です。次はニューラルネットワークのロジックを追加していきましょう。
ニューラルネットワークの追加
ニューラルネットワークの実装には、LambdaにインストールしたBrain.jsを使用します。結果の生成を開始するには訓練用のデータセットが必要です。今回の場合は以下のようになります。
var trainingData = [
{ input: { t: 10, p: 275 }, output: { alarm: 0 } },
{ input: { t: 14, p: 230 }, output: { alarm: 0 } },
{ input: { t: 65, p: 240 }, output: { alarm: 0 } },
{ input: { t: 89, p: 301 }, output: { alarm: 1 } },
...
];
これらはアラーム値で、いつアラームが作動するべきかを知っている人間の手で与えられたものです。私たちは小さな訓練エラーまで防ぐほどの大量のデータは必要としません。今回私は適切で安定したニューラルネットワークの出力を得るために、アラームの閾値を0.6として、たった19個の値だけを用いました。
訓練用データの準備ができたところで、Lambdaで面白い働きをしてくれるアラームオブジェクトを作成していきましょう。
function Alarm(config) {
var _this = this;
_this.maxTemp = config.maxTemp;
_this.maxPressure = config.maxPressure;
_this.alarmThreshold = config.alarmThreshold;
_this.nn = new brain.NeuralNetwork();
_this.tempToInput = function(t) {
return t / _this.maxTemp;
};
_this.pressureToInput = function(p) {
return p / _this.maxPressure;
};
_this.parseTrainingData = function(data) {
return data.map(function(point) {
return {
input: {
t: _this.tempToInput(point.input.t),
p: _this.pressureToInput(point.input.p)
},
output: { alarm: point.output.alarm }
}
});
};
_this.train = function(data) {
return _this.nn.train(_this.parseTrainingData(data));
};
_this.isTriggered = function(data) {
var guessedAlarm = _this.nn.run({
t: _this.tempToInput(data.t),
p: _this.pressureToInput(data.p)
}).alarm;
return {
triggered: (guessedAlarm > _this.alarmThreshold),
guessedAlarm: guessedAlarm
};
};
}
一目瞭然ですね。ここでの2つの主要な関数はtrainとisTriggeredです。前者は訓練用のデータセットを用い、ニューロンの重みを調節することでニューラルネットワークを訓練します。後者は温度と圧力の組み合わせをチェックし、アラームを作動させるかどうかの判断をします。TempToInputとpressureToInputは温度と圧力の値をニューラルネットワークに適したもの、すなわち0から1までのfloat値へと変換します。
全てを統合する
さて全てのパーツを組み合わせて、Amazon Lambdaの初期入力となるハンドラの関数を書いてみましょう。
exports.handler = function(event, context) {
var
parser = new S3Parser(event),
mqtt = new MqttClient(
config.mqtt,
function () {
return context.done(null);
},
function (error) {
return context.done(null, error);
}
),
alarm = new Alarm(config.alarm);
parser.parseFakeS3Object(function(payload) {
payload.readings.forEach(function(reading) {
var
dataPoint = reading.values,
outputTopic = config.mqtt.outputTopic + "/" + reading.device_id;
if (dataPoint.t > config.alarm.maxTemp || dataPoint.p > config.alarm.maxPressure) {
mqtt.publish(
outputTopic, { alarm: 1, values: dataPoint }
);
} else {
alarm.train(trainingData);
var trigger = alarm.isTriggered(dataPoint);
if (trigger.triggered) {
mqtt.publish(
outputTopic, { alarm: trigger.guessedAlarm, values: dataPoint }
);
}
}
});
mqtt.disconnect();
});
};
ハンドラが起動されS3のオブジェクトを読み込みます。温度か圧力のどちらかが、それぞれの臨界値である200及び800を超えているか、ニューラルネットワーク内の両者の組み合わせによってアラームが作動した場合には、MQTTがデバイスのトピックへのパブリッシュをします。MQTTで<YOUR DOMAIN>/#をサブスクライブしていれば、このようなメッセージが表示されます。
さてThingFabricにルールを設定しましょう。登録またはサインインをしてからプロジェクトを開き、左側にある”Rules”を選択してください。ルールウィザードで”Send SMS”を選択し、1（国番号）で始まるあなたの電話番号を入力し、次にメッセージを入力し、”Save”をクリックします。Advancedタブを開き、ルールを下記のように変更します。
<YOUR DOMAIN>/lambda/#" {"alarm": val where val > 0.7} -> sms to:"17209999999" text:"Temperature or pressure is too high!""
次のように表示されるはずです。
おめでとうございます。最初のルールが作成できました。これで、温度／圧力アラームの値が0.7を超えた時にテキストメッセージが送信されます。このルールエンジンはかなり高度なことができ、ペイロードの値をカスタムロジックでも使うことができます。ドキュメントをチェックして、もっとルールを作成し、プロジェクトを完成させてみてください。
次のステップ
さてこれで、S3のペイロードに基づいてニューラルネットワークに判断をさせるという、相当にクールなLambdaが作成できました。このアプローチは他にもいろいろなモノのインターネットのシナリオに応用できます。ですがもちろん、いくつか足りない点もあります。
まずは、ハンドラの関数が呼び出される度に訓練されるのではなく、S3のファイルからニューラルネットワークの状態をロードできたほうがいいですね。これに対する改善方法としては、ThingFabricルールエンジンのドキュメントをよく読んで、QoS2のMQTTメッセージを送ることによりデバイスを停止できるというルールを設定します。これで配信が100％保証されます。
このプロジェクトをいじってみたければ、ソースコードは全てGitHubに公開されています。ご意見やご質問をお待ちしています。