本記事では、Amazon Location Serviceを利用した、人や動物が特定エリアに出入りしたことを検出するアプリケーションのデモを紹介します。

0. Motivation

先日は幼稚園の通園バスで痛ましい事故が起きてしまいました。オペレーションの問題と言えばそれまでかもしれませんが、あのような事故を防ぐにはテクノロジーを役立てられる可能性を探ることも必要と思います。

そこで、位置情報サービスとIoTの目線から、ボケてしまった親、まだまだ周りが見えていない我が子、あちこち行ってしまうペットのような大切な何かを、どうすれば守ることができるのか考え、少しでもヒントになりそうなデモを作ってみることにします。

1. Overview

デモは、位置情報をトラッキングする端末を身につけた人や動物が、特定のエリアを出入りした際に、スマホ（LINE Bot）に通知するというシナリオです。

位置情報のトラッキングには、スマホや以前の記事で紹介した端末などを利用できます。位置情報を取得し、通信（今回のデモでは、AWS IoT Coreに送信）する仕組みがあれば良いので、広範なデバイスが利用可能です。

位置情報を扱うサービスは多数存在します。今では、お迎えの幼稚園バスがどこにいるのか、リアルタイムでスマホで確認できる時代です。

クラウドサービスの雄である AWS にも、位置情報を扱うサービスがあるのはご存知でしょうか？それがAmazon Location Serviceです。Amazon Location Serviceは以下のような機能を提供しているようです。

今回は、調査も兼ねてAmazon Location Serviceの機能を利用してデモを実装することにしました。

2. AWS

Amazon Location Serviceを含め、クラウド側は全てAWSのマネージドサービスを利用しました。

まず、位置情報をIoT Coreで受け取り、Lambdaで適当に整形し、Location Serviceに送信します。Loaction Serviceのうち、ここでは位置情報を管理するトラッカーと、特定エリアへの出入りを検知するジオフェンスを利用します。検知結果はEventBridge経由でLambdaで処理し、整形してLINE Botに通知します。

エリア指定の準備

ジオフェンスで「特定のエリア」を指定するために、geojsonファイルを作成します。今回はgeojson.ioを利用し、エリアを定義するためのファイルを作成しました。

Amazon Location Service の設定

手順は、

1. ジオフェンスコレクションを作成する
2. geojsonファイルからジオフェンスを追加する
3. トラッカーを作成し、ジオフェンスコレクションをリンクする

です。

Lambda 関数の作成

以下は、IoT Coreから位置情報を受け取り、後段のLocation Serviceに送信するLambda関数のサンプルです。

import json
import datetime
import os
import boto3

tracker_name = "TRACKER_NAME"

def lambda_handler(event, context):
    os.environ["AWS_DATA_PATH"] = os.environ["LAMBDA_TASK_ROOT"]

    updates = [
      {
        "DeviceId": event["payload"]["deviceid"],
        "SampleTime": datetime.datetime.fromtimestamp(event["payload"]["timestamp"]).isoformat(),
        "Position": [
          event["payload"]["location"]["long"],
          event["payload"]["location"]["lat"]
        ]
      }
    ]

    client = boto3.client("location")
    response = client.batch_update_device_position(TrackerName=tracker_name, Updates=updates)

    return {
      "statusCode": 200,
      "body": json.dumps(response)
    }

また、以下はLINE Botに通知するLambda関数のサンプルです。

import os
import json
import urllib.request
import urllib.parse


def lambda_handler(event, context):
    type = event["detail"]["EventType"]

    msg = ""
    if type == "ENTER":
        msg = "高崎駅に入りました。"

    elif type == "EXIT":
        msg = "高崎駅から出ました。"

    # LINE BOT 通知
    if msg != "":
        line_broadcast(msg)

def line_broadcast(msg):
    url = "https://api.line.me/v2/bot/message/broadcast"
    access_token = os.environ["access_token"]

    headers = {
        "Authorization": "Bearer " + access_token,
        "Content-Type": "application/json"
    }
    payload = {
        "messages":[
            {
                "type":"text",
                "text": msg
            }
        ]
    }
    request = urllib.request.Request(url, json.dumps(payload).encode("utf-8"), method="POST", headers=headers)
    response = urllib.request.urlopen(request,timeout=10)

    if (status_code != 200):
        print("status-code: %d" % status_code)

        return

    return

前者はAWS IoTのルールに、後者はEventBridgeにそれぞれ紐づけることで、イベント駆動処理が可能となります。

動作テスト

AWS IoT Coreのコンソールから、位置情報をpublishすることで、動作確認できます。以下はジオフェンスで指定したエリア内の座標をメッセージペイロードに与えた場合の通知例です。

また、エリア外の座標を与えると以下のように通知されます。

3. Conclusion

この記事では、Amazon Location Serviceを利用し、特定エリアへの出入りを検知するデモを紹介しました。検知部分をLocation Serviceに任せることができるため、簡単にデモを作成することができました。

また、LoRaWANによる位置情報トラッキングデバイスとも相性が良さそうです。

今回はgeojson.ioで多角形の領域を定義し、ジオフェンスを作成しましたが、こちらのドキュメントにあるように、ある地点を中心とした半径 n メートルという円形のジオフェンスも作成できるようなので、用途によって使い分けたり、様々なニーズに応えることができると思います。