概要
- LoRaWANネットワークにおいて、エッジノードはアップリンク通信がほとんどだと思う
- LoRaってそういう用途に向く
- が、もちろんLoRa自体は双方向通信が可能なので、クラウド側からのダウンリンク通信も可能
- AWS IoTの場合は、簡単にそれを投げられるので、ざっくりその方法について説明
- 最終的には本ページの最後のほうで言及するLambdaのやり方を変形して実運用していくと思うが、検証用途なら全然CLIやマネジメントコンソールを使うパターンでもアリ
前提
- Aクラスだとそもそもノード側からのアップリンク通信がないとダウンリンク用のウィンドウが開かない
- つまりAクラスのノードに対してこのページで言及するダウンリンク通信を投げても、ノード側ですぐに受信できるかどうかは全くわからない、ということ (というか多分リアルタイムは無理)
- 検証用途でいうなら、クラスC運用でいつでもダウンリンクを受け入れられる状態にして試すのが手っ取り早いと思う
- そこで意図した感じの動きになるのを確認できたら、Aクラスに戻す…とか
マネジメントコンソールからライトに投げる
-
一番わかりやすくて簡単な手法
-
単なる検証程度ならこれで十分
-
AWS IoT > LOWANデバイス > LoRaWAN > デバイス > デバイストラフィック > `ダウンリンクメッセージをキューに入れる
-
確認モードをオンにしておくと、ノード側からのAckを待つようになるので、届いたかどうか？をクラウド側から確認可能になる
- Ackが返るまではリトライを繰り返すことになる
- 確実にノードに届いたかどうかを管理する必要があったり、あるいは明らかに通信環境が安定しないことがわかっているなら確認モードはオンのほうが良いと思う
- 逆に言えば、投機的に投げて問題ないような内容であるならば逆に確認モードはオフっておいたほうがいい
- そのほうがスループット自体はでるので
AWS CLIを使って投げる
- 以下のような感じで投げられる
- ぶっちゃけ一番ライトなのがこれかもしれない
aws iotwireless send-data-to-wireless-device \
--id "<WirelessDeviceId>" \
--transmit-mode 1 \
--payload-data "<Base64Payload>" \
--wireless-metadata LoRaWAN={FPort=1}
Lambdaから投げる
-
iotwireless経由でぶん投げれる - 例えば私の場合は、汎用的なLambdaを1つ用意してテストイベントペイロードをわたして任意のデータを投げれるようにしてたりする
- デカいペイロードの場合は設定にあわせて自動的にチャンクに分割して投げるとか
- 負荷テスト的にガンガン投げまくるとか
- そういうのも自由自在
- Gatewayを指定して投げることもできるのだが、そうすると簡単に冗長系を作れるLoRaWANの良さを消してしまいかねないので注意
import base64
from typing import Any, Dict
import boto3
from botocore.config import Config
IOTW = boto3.client(
"iotwireless",
config=Config(retries={"max_attempts": 3, "mode": "standard"})
)
DEFAULT_ENCODING = "hex"
DEFAULT_FPORT = 1
DEFAULT_CONFIRM = False
def _payload_to_base64(payload: str, encoding: str) -> str:
enc = (encoding or DEFAULT_ENCODING).lower()
if enc == "base64":
# validate -> normalize
raw = base64.b64decode(payload)
return base64.b64encode(raw).decode("ascii")
if enc == "hex":
raw = bytes.fromhex(payload)
return base64.b64encode(raw).decode("ascii")
if enc == "utf8":
raw = payload.encode("utf-8")
return base64.b64encode(raw).decode("ascii")
raise ValueError(f"unsupported encoding: {encoding} (use hex/base64/utf8)")
def lambda_handler(event: Dict[str, Any], context):
# Required
wireless_device_id = event["wirelessDeviceId"]
payload = event["payload"]
# Optional (defaults applied when omitted)
encoding = event.get("encoding", DEFAULT_ENCODING)
fport = int(event.get("fPort", DEFAULT_FPORT))
confirm = bool(event.get("confirm", DEFAULT_CONFIRM)) # True is expected when provided
payload_b64 = _payload_to_base64(payload, encoding)
resp = IOTW.send_data_to_wireless_device(
Id=wireless_device_id,
TransmitMode=1 if confirm else 0, # 1: Confirmed, 0: Unconfirmed
PayloadData=payload_b64,
WirelessMetadata={"LoRaWAN": {"FPort": fport}},
)
return {
"status": "queued",
"wirelessDeviceId": wireless_device_id,
"messageId": resp.get("MessageId"),
"fPort": fport,
"confirm": confirm,
"encoding": encoding,
}
備考
- MQTTトピックに直接パブリッシュするやり方もあるが今回は割愛
- 前提のところに書いたようにクラス設定に注意
- そうじゃないと、あれ？全然ダウンリンク受信できない！！ とかなりかねないので
- ADRオンの状態じゃないとデカいペイロードを投げすぎると受信できない恐れがあるので注意
- 本来は受信できるようなペイロードだったとしても、そもそも受信環境が厳しいと受信できない可能性も全然あるので注意
- 例えば5 byteなら通るのだが、50 byte や 100 byte だとダメ… みたいな
- そのため、最初は数 byte 程度の小さなペイロードで試しつつ、必要に応じて段階的に引き上げていくような形のほうが良いと思う