📍

グローバル版 Soracom LTE-M Button でも位置トラッキングしたい

2023/12/10に公開

IoT

SORACOM

tech

※ この記事はSORACOM Advent Calendar 2023の 10 日目の記事です

はじめに

はいさい〜！
(この記事を書いている 12/9) 今日は、SORACOM UG Okinawa #1に参加するために、沖縄に来ています(何気に初上陸！)。とても盛会で、懇親会にも参加された方が多く、第二回の開催が待ち遠しいです。

SORACOM UG okinawa #1

そして先週は Amazon Web Services の年次カンファレンスで、ラスベガスに行ってました。
ラスベガス空港
せっかく色々なところに出かけるので、位置情報をトラッキングしておきたいなと思いました。

さて、日本で発売されている SORCOM LTE-M Button で位置情報取得するやり方については、下記のレシピに沿ってやれば簡単に出来ます。

SORACOM IoT レシピ：IoTで位置情報取得システム

今回は海外でも同じように位置情報を取得したかったので、海外版であるSoracom LTE-M Buttonを使おうと思いました。
グローバルボタン

ところが、このグローバル向けボタンでは日本のボタンほど簡単に位置情報を記録する機能がありません。
この記事では、壮大なピタゴラスイッチの末にグローバル向けボタンで位置情報を記録できた方法をご紹介します。

どうやって位置情報を取得するか

SIM 一覧画面で SIM の詳細からセッション詳細を見てみると、以下の画像のように地図上に位置情報がプロットされています。

これは SORACOM Air のセッション情報から、基地局の情報を元に位置を推定しています。この機能は API としても公開されています。

基地局の位置情報を取得できる API を公開しました！(SORACOM公式ブログ)

この機能を使えば、なんとか実現できそうです。

やってみよう

SORACOM CLI を使って位置情報を取得してみましょう。

# 最新のセッションの情報を取得
$ soracom sims session-events --sim-id $SIM_ID --limit 1
[
	{
		"apn": "soracom.io",
		"cell": {
			"eci": 159869203,
			"mcc": 440,
			"mnc": 10,
			"radioType": "LTE",
			"tac": 37323
		},
		"createdTime": "2023-12-09T14:34:20.558Z",
		"dns0": "100.127.0.53",
		"dns1": "100.127.1.53",
		"event": "Deleted",
		"imei": "xxxxxxxxxxxxxxx",
		"imsi": "xxxxxxxxxxxxxxx",
		"operatorId": "OPxxxxxxxxxx",
		"primaryImsi": "xxxxxxxxxxxxxxx",
		"sessionId": "xxxxxxxxxxxxxxxxxxxxxxxxxx",
		"simId": "xxxxxxxxxxxxxxxxxxx",
		"subscription": "plan01s",
		"time": 1702132471259,
		"ueIpAddress": "10.xxx.xxx.xxx"
	}
]

この cell 以下の情報を元に位置を求めます。

$ soracom cell-locations get --mcc 440 --mnc 10 --tac 37323 --ecid 159869203
{
	"avg_strength": 0,
	"created": "2023-07-22T19:33:17Z",
	"exact": 0,
	"lat": 26.2132717,
	"lon": 127.6742535,
	"range": 168,
	"samples": 104,
	"updated": "2023-08-06T03:13:48Z"
}

実際に Google マップで開いてみましょう

https://maps.google.com/?q=26.2132717,127.6742535

泊まっているホテルのかなり近くを指しています！いい感じです

どうやって可視化するか

位置情報の可視化は、SORACOM Harvest が非常に簡単なので、できれば SORACOM Harvest のデータに位置情報を付与したいと思います。

そこでこんなことを考えました。

ボタンを押す
クリックイベント情報が Unified Endpoint に送信される
Harvest にデータが保存される(ただしこの時点では位置情報はついていない)
Funnel を経由して AWS IoT Core の特定のトピックにデータを Publish
AWS IoT Core に接続して上記トピックを Subscribe しているクライアントが、以下の処理を実施(※)
i. 受信データ中の SIM ID を参照して、SORACOM CLI を使って最新のセッションのセル情報を確認
ii. 同じく SORACOM CLI を使ってセル情報を元に位置情報を取得
iii. SORACOM Harvest に保存されている最新のデータを取得
iv. 上記データに位置情報を付加して同じタイムスタンプでデータ送信(上書き)

最後のパートでデータを上書きするためには、同じ SIM からデータを送信する必要があります。しかしボタンデバイスにそれをやらせるのは不可能です...。悩んだ末に思いついたのが、SORACOM Arc を利用することでした。

Soracom LTE-M Button の SIM に紐づけて、SORACOM Arc のバーチャル SIM を発行します。

バーチャルSIM追加

そして soratun を使って Arc に接続している状態だと、

$ curl -v -d @payload.json \
  -H content-type:application/json \
  -H x-soracom-timestamp:1702135069424 \
  harvest.soracom.io

のような感じで timestamp を指定して harvest のデータを上書きすることが出来ます。

構成図にすると、こんな感じになります。

構成図

思ったよりも複雑な構成になってしまいました...

※ クライアントのコードは最後に

動作確認

さて、ボタンをクリックしてみましょう。
データが送信された直後は Harvest 上でデータを確認しても位置情報が付いていません。

位置情報のないデータ

しかし数秒後にリロードしてみると...

位置情報付きのデータ

無事位置情報がついています！

地図モードに変えてみると、ちゃんとプロットされています。

那覇地図

過去２週間分のデータを見てみると、東京・ホノルル(経由地)・ラスベガス・沖縄にプロットされてます！

世界地図

ラスベガスを拡大してみると、空港やあちこちの会場で押した様子がわかります。

ラスベガス

まとめ

思ったよりも大変でしたが、無事にグローバルボタンでも位置情報トラッキングができるようになりました！
グローバルボタンにも接点インターフェースがあるので、次はボタンを押さなくても定期的に送信するようにしてみようかなーと思っています。

Appendix

#!/usr/bin/env python
import json
import os
import subprocess
import sys

print("python " + sys.version)
SORACOM_CLI_PATH = "/opt/bin/soracom"
SORACOM_CLI = [SORACOM_CLI_PATH, "--coverage-type", "g", "--auth-key-id", os.getenv("AUTH_KEY_ID"), "--auth-key", os.getenv("AUTH_KEY")]

import random
from paho.mqtt import client as mqtt_client

broker = "beam.soracom.io"
port = 1883
topic = "button-locator/g"
client_id = f"python-mqtt-{random.randint(0, 1000)}"


def get_last_cellular_info(sim_id):
    cmd = SORACOM_CLI + ["sims", "session-events", "--sim-id", sim_id, "--limit", "1"]
    print(" ".join(cmd))
    res = subprocess.run(cmd, capture_output=True)
    if res.stdout != None:
        print(res.stdout.decode("utf-8"))
    if res.stderr != None:
        print(res.stderr.decode("utf-8"))
    if res.returncode != 0:
        print(f"error {res.returncode}")
        return None
    data = json.loads(res.stdout)
    if len(data) == 0:
        return None
    return data[0].get("cell")


def get_cellular_location_soracom(cell):
    cmd = SORACOM_CLI + f"cell-locations get --mcc {cell['mcc']} --mnc {cell['mnc']} --tac {cell['tac']} --ecid {cell['eci']}".split()

    print(" ".join(cmd))
    res = subprocess.run(cmd, capture_output=True)
    if res.stdout != None:
        print(res.stdout.decode("utf-8"))
    if res.stderr != None:
        print(res.stderr.decode("utf-8"))
    if res.returncode == 0:
        data = json.loads(res.stdout)
        if data.get("lat") and data.get("lon"):
            return [data.get("lon"), data.get("lat")]
    return None


def update_harvest_data(imsi, location):
    import urllib.request

    cmd = SORACOM_CLI + ["data", "get-entries", "--resource-type", "Subscriber", "--resource-id", imsi, "--limit", "1"]
    print(" ".join(cmd))
    res = subprocess.run(cmd, capture_output=True)
    if res.stdout != None:
        print(res.stdout.decode("utf-8"))
    if res.stderr != None:
        print(res.stderr.decode("utf-8"))
    if res.returncode != 0:
        print(f"error {res.returncode}")
        return None
    data = json.loads(res.stdout)
    if len(data) == 0:
        return None
    #    print(data[0])

    content = json.loads(data[0]["content"])
    content["location"] = {
        "lat": location[1],
        "lon": location[0],
    }
    print(content)
    headers = {
        "content-type": "application/json",
        "x-soracom-timestamp": data[0]["time"],
    }

    url = "http://harvest.soracom.io"
    req = urllib.request.Request(url, json.dumps(content).encode(), headers)
    with urllib.request.urlopen(req) as res:
        print(res.code)


def update_location_data(sim_id, imsi):
    if (latest_cell := get_last_cellular_info(sim_id)) == None:
        return None
    if (location := get_cellular_location_soracom(latest_cell)) == None:
        return None

    print(f"https://maps.google.com/?q={location[1]},{location[0]}")

    update_harvest_data(imsi, location)

    return location


def connect_mqtt():
    def on_connect(client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT Broker!")
        else:
            print("Failed to connect, return code %d\n", rc)

    # Set Connecting Client ID
    client = mqtt_client.Client(client_id)
    client.on_connect = on_connect
    client.connect(broker, port)
    return client


def subscribe(client: mqtt_client):
    def on_message(client, userdata, msg):
        print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")
        try:
            data = json.loads(msg.payload)
        except:
            print("Invalid JSON")
        else:
            if data.get("simId") and data.get("imsi"):
                update_location_data(data.get("simId"), data.get("imsi"))

    client.subscribe(topic)
    client.on_message = on_message


def run():
    client = connect_mqtt()
    subscribe(client)
    client.loop_forever()


if __name__ == "__main__":
    run()