shadowとwillとretainを活用してAWS IoT CoreとPubSubする⑥awsiotsdkv2でpubsub

## この記事について  

[shadowとwillとretainを活用してAWS IoT CoreとPubSubする②AWSリソースの構築](https://qiita.com/Saito5656/items/62e46f009000991b2adf)で構築した環境を使ったラズパイGW broker <-> AWS IoT間でのMQTT通信、ここから３回はAWS IoT Device SDK v2 for Pythonでmessageのpubsub、shadowのupdate/get、retainとwillの設定をします。

今回はラズパイに接続された仮想のdevice_0のシグナルをAWS IoTとpubsubします。

<br>

<img width= 700 src = https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/565396/d52181b5-2b07-af25-585c-e7326b2085b1.png>

## ライブラリー

### GitHub

[GitHub: aws/aws-iot-device-sdk-python-v2](https://github.com/aws/aws-iot-device-sdk-python-v2)
[GitHub: awslabs/aws-crt-python](https://github.com/awslabs/aws-crt-python/blob/main/awscrt/mqtt.py)

### PyPi
[pypi: awsiotsdk 1.21.5](https://pypi.org/project/awsiotsdk/)

### APIリファレンス  

[Python 用 AWS IoT デバイス SDK v2](https://aws.github.io/aws-iot-device-sdk-python-v2/)

### 開発者ガイド
[AWS IoT Device SDKs](https://docs.aws.amazon.com/ja_jp/iot/latest/developerguide/iot-sdks.html
)

## callback

awsiotとawscrtで準備されているcallback関数のうち今回のpubsubでは6種類を設定します。

### on_connection_interrupted  

> Callback invoked whenever the MQTT connection is lost.
        The MQTT client will automatically attempt to reconnect.
        The function should take the following arguments return nothing

接続が失われたときに再接続すると呼び出されるcallback関数です。MQTT接続のクライアントobjectを作成する時に呼び出しておきます。

```python
def on_connection_interrupted(connection, error, **kwargs):
    print(f"Connection interrupted. error: {error}")
```

### on_connection_resumed  

> Callback invoked whenever the MQTT connection
        is automatically resumed. Function should take the following arguments and return nothing

再接続時に自動的にMQTTを再開するたびに呼び出されるcallback関数です。MQTT接続のクライアントobjectを作成する時に呼び出しておきます。

```python
def on_connection_resumed(connection, return_code, session_present, **kwargs):
    print(f"Connection resumed. return_code: {return_code} session_present: {return_code}")
```

### on_message  

> callback: Callback to invoke when message received, or None to disable.
            Function should take the following arguments and return nothing

sub時に呼び出されるcallback関数です。sub時のpayloadやstatusを読み出すのに使用します。

```python
def on_message(topic, payload, dup, qos, retain, **kwargs):
    payload_str = payload.decode('utf-8')
    payload_dict = json.loads(payload_str)
    print(f"on_message_received payload dict: {payload_dict}")
```

### on_connection_success  

> Optional callback invoked whenever the connection successfully connects.
        The function should take the following arguments and return nothing

MQTT接続が成功するたびに呼び出されるcallback関数です。MQTT接続のクライアントのインスタンスを作成する時に呼び出しておきます。Optionalと書かれていますが動作を見るために入れておきます。

```python
def on_connection_success(connection, callback_data):
    assert isinstance(callback_data, mqtt.OnConnectionSuccessData)
    print(f"Connection Successful with return code: {callback_data.return_code} session present: {callback_data.session_present}")
```

### on_connection_failure  

> Optional callback invoked whenever the connection fails to connect.
        The function should take the following arguments and return nothing

MQTT接続が失敗するたびに呼び出されるcallback関数です。MQTT接続のクライアントのインスタンスを作成する時に呼び出しておきます。Optionalと書かれていますが動作を見るために入れておきます。

```python
def on_connection_failure(connection, callback_data):
    assert isinstance(callback_data, mqtt.OnConnectionFailureData)
    print(f"Connection failed with error code: {callback_data.error}")
    print(f"Error details: {callback_data}")
```

### on_connection_closed  

> Optional callback invoked whenever the connection has been disconnected and shutdown successfully.
        The function should take the following arguments and return nothing

MQTT接続が正常に切断・終了すると呼び出されるcallback関数です。MQTT接続のクライアントobjectを作成する時に呼び出しておきます。Optionalと書かれていますが動作を見るために入れておきます。

```python
def on_connection_closed(connection, callback_data):
    print("Connection closed")
```

## customメソッド  

自分使い用にまとめ直したメソッドに再構築して使います。今回新たに作るメソッド以下です。

### MQTTクライアント作成  

一意のclient idでAWS IoTに接続するクライアントobjectを作成してます。  
ここで使うawsiot.mqtt_connection_builder.mtls_from_path()ではクライアントをインスタンス化する際に、paho-mqttと以下の点が異なります。

* endpointとportも紐づける  
* 接続に関わる5種類のcallbackを呼び出せる  

```python
def mqtt_init():
    mqtt_connection = mqtt_connection_builder.mtls_from_path(
        endpoint=endpoint,
        port=port,
        cert_filepath=cert,
        pri_key_filepath=key,
        ca_filepath=ca,
        on_connection_interrupted=on_connection_interrupted,
        on_connection_resumed=on_connection_resumed,
        client_id=clientid,
        clean_session=False,
        keep_alive_secs=30,
        on_connection_success=on_connection_success,
        on_connection_failure=on_connection_failure,
        on_connection_closed=on_connection_closed
        )
    print("Connecting to endpoint with client ID")
    return mqtt_connection
```

### MQTT接続  

mqtt_init()で作成したクライアントでAWS IoTに接続します。

```python
def mqtt_connect(mqtt_connection):
    connect_future = mqtt_connection.connect()
    connect_future.result()
```

### 切断  

client idによる接続をdisconnectメソッドで切断します。

```python
def mqtt_disconnoect(mqtt_connection):
    print("Disconnecting...")
    disconnect_future = mqtt_connection.disconnect()
    disconnect_future.result()
```

### topic発行(publish)  

仮想のホスト名：MQTTServer1のラズパイGW/brokerから、device_0のシグナル：msgを以下のメソッドでAWS IoTにpubします。

```python
def mqtt_publish(mqtt_connection, topic, msg):
    data = {}  
    print("topic:", topic)
    data['Timestamp'] = int(time.time())
    data['hostname'] = hostname
    data['device_no'] = device_no
    data['msg'] = msg
    message_json = json.dumps(data)
    mqtt_connection.publish(
        topic=topic,
        payload=message_json,
        qos=mqtt.QoS.AT_LEAST_ONCE,
    )
    print(f"published message payload dict: {data}")
    time.sleep(1)
```

### topic購読(subcrribe)  

以下のメソッドで自分が接続しているMQTTネットワーク内でtopicにpubされたメッセージをsubします。

```python
def mqtt_subscribe(mqtt_connection, topic):
    print(f"Subscribing to topic '{topic}'...")
    subscribe_future, packet_id = mqtt_connection.subscribe(
        topic=topic,
        qos=mqtt.QoS.AT_LEAST_ONCE,
        callback=on_message)
    subscribe_result = subscribe_future.result()
    print(f"Subscribed with {subscribe_result}")
    time.sleep(1)   
```

## ラズパイGW <-> AWS IoT でpubsub  

AWS IoTとport 8883で接続するための認証情報とラズパイの識別子(client id=hostname)を準備します。

topicはpahoの説明と同様のものとします。

```python: 接続情報
endpoint = 'xxxxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com'
hostname = os.uname()[1]
port = 8883
cert = f'./cert/MQTTServer1-certificate.pem.crt'
key = f'./cert/MQTTServer1-private.pem.key'
ca = f'./cert/AmazonRootCA1.pem'

device_no = 0
topic = f'MQTTServer/MQTTServer1'
```

device_0のシグナルもこれで行きますヨシッ👉

```python: デバイスシグナル
msg = "AWS IoT Device SDK v2接続ヨシッ👉"
```

ここまでで準備した認証情報とメソッドを組み合わせるとラズパイとAWS IoTが接続されたNWでpubsubが可能になります。

以下を実行すると接続後にsubscriberが待ち受けした状態となり、pubされた後に自分自身でもmessageをsubして回線を切断します。

上手くいかないときは1～３秒くらいのtime.sleepを挟むとpubsubが出来ると思います。

```python
clientid = f"{hostname}"
topic = TOPIC

#クライアント作成
mqtt_connection = mqtt_init()
#client idで接続
mqtt_connect(mqtt_connection)
#sub開始
mqtt_subscribe(mqtt_connection, topic)

try:
    #message発行
    mqtt_publish(mqtt_connection, topic, msg)

except KeyboardInterrupt:
    print("Waiting for shadow updates...")
    mqtt_disconnoect(mqtt_connection)
```

### Demo code 全体

コードを全てつなげると以下のようになります。

```python: pubsub_awssdk.py
#!/usr/bin/python
# -*- coding: utf-8 -*-
from awscrt import mqtt
from awsiot import mqtt_connection_builder, iotshadow
import os
import time
import json


endpoint = 'xxxxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com'
hostname = os.uname()[1]
port = 8883
cert = f'./cert/MQTTServer1-certificate.pem.crt'
key = f'./cert/MQTTServer1-private.pem.key'
ca = f'./cert/AmazonRootCA1.pem'

device_no = 0
topic = f'MQTTServer/MQTTServer1'
msg = "AWS IoT Device SDK v2接続ヨシッ👉"


def on_connection_interrupted(connection, error, **kwargs):
    print(f"Connection interrupted. error: {error}")

def on_connection_resumed(connection, return_code, session_present, **kwargs):
    print(f"Connection resumed. return_code: {return_code} session_present: {return_code}")

def on_message(topic, payload, dup, qos, retain, **kwargs):
    payload_str = payload.decode('utf-8')
    payload_dict = json.loads(payload_str)
    print(f"on_message payload dict: {payload_dict}")
    return

def on_connection_success(connection, callback_data):
    assert isinstance(callback_data, mqtt.OnConnectionSuccessData)
    print(f"Connection Successful with return code: {callback_data.return_code} session present: {callback_data.session_present}")

def on_connection_failure(connection, callback_data):
    assert isinstance(callback_data, mqtt.OnConnectionFailureData)
    print(f"Connection failed with error code: {callback_data.error}")
    print(f"Error details: {callback_data}")
    
def on_connection_closed(connection, callback_data):
    print("Connection closed")

def mqtt_init():
    mqtt_connection = mqtt_connection_builder.mtls_from_path(
        endpoint=endpoint,
        port=port,
        cert_filepath=cert,
        pri_key_filepath=key,
        ca_filepath=ca,
        on_connection_interrupted=on_connection_interrupted,
        on_connection_resumed=on_connection_resumed,
        client_id=clientid,
        clean_session=False,
        keep_alive_secs=30,
        on_connection_success=on_connection_success,
        on_connection_failure=on_connection_failure,
        on_connection_closed=on_connection_closed
        )
    print("Connecting to endpoint with client ID")
    return mqtt_connection

def mqtt_connect(mqtt_connection):
    connect_future = mqtt_connection.connect()
    connect_future.result()

def mqtt_disconnoect(mqtt_connection):
    print("Disconnecting...")
    disconnect_future = mqtt_connection.disconnect()
    disconnect_future.result()

def mqtt_publish(mqtt_connection, topic, msg):
    data = {}  
    print("topic:", topic)
    data['Timestamp'] = int(time.time())
    data['hostname'] = hostname
    data['device_no'] = device_no
    data['msg'] = msg
    message_json = json.dumps(data)
    mqtt_connection.publish(
        topic=topic,
        payload=message_json,
        qos=mqtt.QoS.AT_LEAST_ONCE,
    )
    print(f"published message payload dict: {data}")
    time.sleep(1)

def mqtt_subscribe(mqtt_connection, topic):
    print(f"Subscribing to topic '{topic}'...")
    subscribe_future, packet_id = mqtt_connection.subscribe(
        topic=topic,
        qos=mqtt.QoS.AT_LEAST_ONCE,
        callback=on_message)
    subscribe_result = subscribe_future.result()
    print(f"Subscribed with {subscribe_result}")
    time.sleep(1)    


if __name__ == '__main__':
    clientid = f"{hostname}"
    topic = TOPIC
    
    mqtt_connection = mqtt_init()
    mqtt_connect(mqtt_connection)
    mqtt_subscribe(mqtt_connection, topic)
    
    try:
        mqtt_publish(mqtt_connection, topic, msg)    
    except KeyboardInterrupt:
        print("Waiting for shadow updates...")
        mqtt_disconnoect(mqtt_connection)
```

## 実行結果

```sh
~ $ python pubsub_sdk.py
Connecting to endpoint with client ID
Connection Successful with return code: 0 session present: True
Subscribing to topic 'MQTTServer/MQTTServer1/0'...
Subscribed with {'packet_id': 1, 'topic': 'MQTTServer/MQTTServer1/0', 'qos': <QoS.AT_LEAST_ONCE: 1>}
topic: MQTTServer/MQTTServer1/0
published message payload dict: {'Timestamp': 1719715304, 'hostname': 'MQTTServer1', 'device_no': 0, 'msg': 'AWS IoT Device SDK v2接続ヨシッ👉'}
on_message payload dict: {'Timestamp': 1719715304, 'hostname': 'MQTTServer1', 'device_no': 0, 'msg': 'AWS IoT Device SDK v2接続ヨシッ👉'}
Disconnecting...
Connection closed
~ $ 
```

<br>

AWS IoTのテストクライアント上でもsubscribeを確認できました。

<br>

<img width= 700 src = https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/565396/33e9ce76-db48-cf6b-dc27-966b708a3fbf.png>

## 次回  

[shadowとwillとretainを活用してAWS IoT CoreとPubSubする⑦awsiotsdkv2でshadow update](https://qiita.com/Saito5656/items/b1163c875d56c20bfa4d)