工作と競馬2

電子工作、プログラミング、木工といった工作の記録記事、競馬に関する考察記事を掲載するブログ

AWS IoT Device SDK v2でpublishするときのackCallbackとpacket idについて

技術メモ#AWS 技術メモ#Python

概要

AWS IoT Device SDK v2でのpublishで、ackCallbackとpacket idについて旧版との違いがわかったのでメモした。


背景と目的

AWS IoT Device SDK v2を使う必要が出た。APIの仕様が旧バージョンから変わっているので、ちょっと動かして確認する。


詳細

0. 参考URL


1. インストール

pip install aws-iot-sdk

Windows PCでは、これだけでよかったが、他のあるLinux環境では、参考URLに書いてある通りwheelが取得できず、まず以下をやってからpipでのインストールが必要だった。

sudo apt-get update
sudo apt-get install cmake
sudo apt-get install libssl-dev


2. 動かして気づいたことをメモ

とりあえず、pubsubのサンプルを動かしてみた。

2.1 旧版のMQTTのClientオブジェクトに相当するもの

  • 旧: AWSIoTPythonSDK.MQTTLib.AWSIoTMQTTClient
  • v2: awscrt.mqtt.Connection

だいたい同じようなメソッドが並んでいる。

2.2 publishメソッドは非同期

  • 旧版: publish=同期、publishAsync=非同期
  • v2: publish=非同期、publishAsync=ない

awscrt.mqtt.Connectionのpublishメソッドには、旧版のpublishAsyncの引数ackCallbackに相当するものがない。旧版のackCallbackはどうやって実装すればいいのか？

2.3 ackCallbackに相当するもの

こちらにある通り、publishメソッドの戻り値タプルの1つ目concurrent.futures.Futureオブジェクトなので、concurrent.futures.Futureオブジェクトのadd_done_callbackメソッドで、ackCallbackを割り当てればよい。

def ackCallback(f):
    # PUBACK受信後の処理を記述
    :

# publish
future, packetId = mqtt_connection.publish(...)

# 戻り値futureにackCallbackを割り当て
future.add_done_callback(ackCallback)
:

2.4 packetIdは旧版と同じではない？

旧版では、publishAsyncの戻り値packetIdは、ackCallbackの引数midと対応していた。publishごとにpacketIdがインクリメントされるので、どのpublishのACKかということが、midで判断できた。

v2では、APIドキュメントを見る限りpublish戻り値タプル2個目packet_idに見えるので、サンプルコードに以下を追記して動かしてみたが、publishごとにpacket_idが変化してくれない。ずっと1のまま。

def ackCallback(f):
    # f.result()にpacket_idが入っている
    print(f.result())

future, packetId = mqtt_connection.publish(...)
print(packetId) # publishのpacket_id
future.add_done_callback(ackCallback)

旧版でできたのにv2でできないのは、おかしい気がする。何か考え方間違えているのか？ 自分でpublish時に管理用のカウンタみたいなものを用意するしかないのか？


まとめと今後の課題

旧版とv2の違いがわかった。packet idがpublishごとにインクリメントされないのがいまいち解せない。


ランキング参加中
DIY

Raspberry Pi Picoを初起動

技術メモ

概要

Raspberry Pi Picoを動かした。


背景と目的

最近、Raspberry Pi Picoと名乗るマイコンボードが出た。基板サイズはArduino Nanoくらい。RAMやクロック周波数を見ると性能は全然こちらのほうが高そう。そして550円とかなり安いので、とりあえず試したくなって買ってしまった。（ESP32など同価格帯でもっと高性能なおもちゃは手元にあるのだが・・・） というわけで、ひとまず動かしてみる。


0.実施環境

  • Windows 10 PCを使用
  • MicroPythonを使用

1.調達

スイッチサイエンスで売ると書いてあったので入荷通知を待っていたのだが、通知があった数時間後に売り切れになっていた。これは入手に手間取りそうだと思っていたが、その数日後、秋月のネットショップを何気なく見たら、普通に売っていた。というわけで、秋月で購入。


2.マニュアル類を読む


3.実物を触る

3.1 MicroPython本体を書き込む

Raspberry Pi Picoには、標準で用意されたMicroPythonインタプリタ以外に、こちらのような独自拡張したMicroPythonを作って入れることもできるが、それは必要になったらおいおいやるとして、今回はとりあえず標準のMicroPythonを使用する。

https://www.raspberrypi.org/documentation/pico/getting-started/

まず、上記から、.uf2というRaspberry Pi Pico用MicroPython本体をダウンロードする。

次に、マニュアルに従い、BOOTSELボタンを押しながら、USBケーブルを接続すると、RPI-RP2というストレージが表示された。uf2ファイルをドラッグアンドドロップすると、ストレージが消える。これでMicroPython本体は書き込めたらしい。

Tera Termでシリアルポートを開くと以下のようになった。とりあえず、動きそう。

3.2 LEDを点灯させる

SDKのドキュメントを見ながら、基板に載っているLEDをON/OFFしてみた。 Tera Termで直接入力したところ、LEDがON/OFFできた。成功。 GPIOを直接触るには、machineモジュールのPinを使う。Pinの引数modeにOUT/INなどを設定して入出力を設定する。オープンドレインやALTという周期動作をするモードもあるらしい。SDKのドキュメントには詳しく描いてあるので、そのうち試す。

from machine import Pin

led = Pin(25, mode=Pin.OUT)

# 点灯
led.on()

# 消灯
led.off()

3.3 MicroPython IDE Thonny

ターミナルに直接入力していてもしょうがないので、IDEでコードを作成してpyファイルを実行させたい。 Raspberry Pi Pico Python SDKのドキュメントにあるMicroPython IDE Thonnyを使うことにした。

https://thonny.org/

上記URLから、Windows用をダウンロードしてインストール。

マニュアルに従って、インタプリタの設定を行う。

いよいよ、コードを書く。

実行ボタンを押したところ、1秒周期で点滅した。成功。


まとめと今後の課題

Raspberry Pi Picoを動かした。次は、すぐに何かしたいわけではないが、ADCあたりを使ってみるか。


システムシャットダウンの際、Pythonアプリのデーモンを安全に終了する

技術メモ#Python

概要

Pythonアプリのデーモンをシャットダウン時に安全に終了するためのメモ。

背景と目的

Pythonアプリのデーモンを、システムシャットダウン時に安全に終了するための基本的な実装方法を確認する。


詳細

0. 環境


1. 実装

Python標準モジュールのsignalを使うことで、システムシャットダウン時に、デーモンにSIGNALが送られたのを、アプリが受信できる。

以下、具体的な実装をメモ。

1.1 Pythonアプリ

  • SIGTERMを受け取ったら、handlerという関数が呼ばれるようにする。
  • SIGTERM受信したら、フラグを立てる。
  • while ループでフラグを調べて、立っていたら抜ける。→アプリが終了

test_signal.py

# coding: utf-8

import signal
import time

x = {
    "end": False,
    "signum": None,
    "frame": None
}

def handler(signum, frame):
    x["end"] = True # 終了させる
    x["signum"] = signum
    x["frame"] = frame

print("test_signal start!")

signal.signal(signal.SIGTERM, handler)

print("test_signal loop!")

while True:
    time.sleep(1)
    if x["end"]:
        print("test_signal handler! {},{}".format(x["signum"], x["frame"]))
        break

1.2 呼び出し用シェルスクリプト

Pythonアプリを直接デーモンとして実行してもいいが、起動用シェルスクリプトをかませることが多いので、試しにシェルスクリプトから呼ぶ。

test_signal.sh

#!/bin/bash

HERE=$(cd $(dirname $0); pwd)

echo "test_signal sh start"

python3 $HERE/test_signal.py

echo "test_signal sh end"

1.3 サービスファイル

test_signal.shが呼ばれるようにしておく。

test_signal.service

[Unit]
Description=test_signal
After=multi-user.target

[Service]
ExecStart=/bin/bash /path/to/test_signal.sh
Restart=on-failure
Type=simple

[Install]
WantedBy=multi-user.target

配置してデーモン有効化、起動。

cp ./test_signal.service /etc/systemd/system
systemctl daemon-reload
systemctl enable test_signal
systemctl start test_signal


2. 動作確認

2.1 手動で止めてみる

systemctl stop test_signal

ログを見ると、以下の感じでちゃんと止まった。 shから呼ばれたPythonアプリでも、SIGTERMを受信して、Pythonアプリで処理できている。

Mar 10 07:14:27 ****** bash[2048]: start!
Mar 10 07:14:27 ****** bash[2048]: handler! 15,<frame at 0x7690fc30, file '/r
Mar 10 07:14:27 ****** bash[2048]: terminate!
Mar 10 07:14:27 ****** systemd[1]: Stopped test_signal Application.

2.2 システムシャットダウン

シャットダウンしてみた。再起動後、syslogを確認したところ、以下。ちゃんと止まった模様。 ただし、なぜかハンドラ関数が2回呼ばれる(SIGTERMを2回受信？)ようなので、念のため、ハンドラ関数内で2回実行されると困るものはやらないようにするほうがいいかもしれない。今回のように、メインループを抜けるためのフラグを立てるだけみたいな単純なものであれば問題ないだろう。

Mar 10 07:40:07 ********** systemd[1]: Started test_signal Application.
Mar 10 07:40:07 ********** bash[2550]: test_signal sh start
Mar 10 07:40:13 ********** systemd[1]: Stopping test_signal Application...
Mar 10 07:40:14 ********** bash[2550]: test_signal start!
Mar 10 07:40:14 ********** bash[2550]: test_signal loop!
Mar 10 07:40:14 ********** bash[2550]: test_signal handler! 15,<frame at 0x769bbc30, file '/root/test/test_signal.py', line 28, code <module>>
Mar 10 07:40:14 ********** bash[2550]: test_signal end!
Mar 10 07:40:14 ********** systemd[1]: Stopped test_signal Application.


まとめと今後の課題

とりあえず、安全にデーモンを終了させられそう。