Comment puis-je publier des messages MQTT sur AWS IoT Core depuis mon appareil lorsque j'utilise Python ?

Brève description

Vérifiez que l’objet AWS IoT est correctement configuré et que ses certificats sont correctement associés. Pour tester votre configuration, vous pouvez utiliser le client AWS IoT MQTT et l'exemple de code Python fourni dans cet article.

Résolution

Configuration d’un répertoire pour tester la publication MQTT

1.    Créez un répertoire de travail dans votre environnement de développement. Par exemple : iot-test-publish.

2.    Créez un sous-répertoire pour les certificats dans votre nouveau répertoire de travail. Par exemple : certificats.

3.    Dans la ligne de commande, remplacez le répertoire par votre nouveau répertoire de travail.

Installation de pip et du kit SDK AWS IoT pour Python

1.    Si ce n'est pas déjà fait, installez le package pip pour Python 3. Pour plus d'informations, consultez la page Installation sur le site Web de Python Packaging Authority (PyPA).

2.    Installez le kit SDK AWS IoT pour Python v2 en exécutant la commande suivante depuis la ligne de commande :

pip install awsiotsdk

-ou-

Installez le kit SDK AWS IoT Device pour Python (la version précédente du SDK) en exécutant la commande suivante :

pip install AWSIoTPythonSDK

Pour plus d'informations, consultez Kit SDK AWS IoT pour Python v2 ou Kit SDK AWS IoT Device pour Python sur GitHub.

Remarque : ces kits SDK sont recommandés pour la connexion à AWS IoT Core, mais ils ne sont pas obligatoires. Vous pouvez également vous connecter à l'aide de n'importe quel client MQTT tiers conforme.

Création d'une politique AWS IoT Core

1.    Ouvrez la console AWS IoT Core.

2.    Dans le volet de navigation gauche, choisissez Sécurité.

3.    Sous Sécurité, sélectionnez Stratégies.

4.    Si vous disposez de stratégies AWS IoT Core existantes, choisissez Créer pour créer une stratégie.
-ou-
Si vous n'avez pas encore de stratégie, sur la page Vous ne possédez pas encore de stratégie, choisissez Créer une stratégie.

5.    Sur la page Créer une stratégie, entrez un Nom pour votre stratégie. Par exemple : admin.

6.    Sous Ajouter des instructions, procédez comme suit :
Dans le champ Action, saisissez iot:*.
Important : il est utile d'autoriser toutes les actions AWS IoT (iot:*) pour les tests. Il s'agit toutefois d'une bonne pratique pour renforcer la sécurité d'une configuration de production. Pour d'autres exemples de stratégies de sécurité, consultez Exemples de stratégies AWS IoT.
Pour Resource ARN, entrez *.
Pour Effet, cochez la case Autoriser.

7.    Sélectionnez Créer.

Pour plus d'informations, consultez Création d'une stratégie AWS IoT Core et Stratégies AWS IoT Core.

Création d’un objet AWS IoT

Remarque : vous n'avez pas besoin de créer un objet pour vous connecter à AWS IoT. Cependant, les objets vous permettent d'utiliser des contrôles de sécurité supplémentaires et d'autres fonctionnalités d'AWS IoT, telles que Indexation de la flotte, Tâches ou Cliché instantané d'appareil.

1.    Dans la console AWS IoT Core, dans le volet de navigation de gauche, choisissez Gérer.

2.    Si vous disposez d'objets existants, choisissez Créer pour en créer un.
-ou-
Si vous n'avez pas d'objet, sur la page Vous n'avez pas encore d'objets, choisissez Enregistrer un objet.

3.    Sur la page Création d'objets AWS IoT, choisissez Créer un seul objet.

4.    Sur la page Ajouter votre appareil au registre d’objets, procédez comme suit :
Saisissez un Nom pour l'objet. Exemple : Objet-Test.
(Facultatif) Sous Ajouter un type à cet objet, choisissez ou créez un type d'objet.
(Facultatif) Sous Ajouter cet objet à un groupe, choisissez ou créez un groupe. Pour plus d'informations sur les groupes, reportez-vous à Groupes d'objets statiques et Groupes d'objets dynamiques.
(Facultatif) Sous Définir des attributs d'objets pouvant faire l’objet d’une recherche (facultatif), ajoutez des attributs sous forme de paires clé-valeur.
Sélectionnez Suivant.

5.    Sur la page Ajouter un certificat pour votre objet, choisissez Créer un certificat. Vous voyez des notifications confirmant que votre objet et un certificat correspondant ont été créés.

6.    Sur la page Certificat créé, procédez comme suit :
Sous Pour connecter un appareil, vous devez télécharger les éléments suivants, choisissez Télécharger pour le certificat, la clé publique et la clé privée.
Enregistrez chacun des fichiers téléchargés dans le sous-répertoire des certificats que vous avez créé précédemment.
Sous Vous devez également télécharger une autorité de certification racine pour AWS IoT, sélectionnez Télécharger. La page Authentification du serveur s'ouvre sur Certificats d'autorité de certification pour l'authentification du serveur.

7.    Sous Point de terminaison des services de confiance Amazon (préféré), choisissez Autorité de certification Amazon 1. Le certificat s'ouvre dans votre navigateur.

8.    Copiez le certificat (de -----BEGIN CERTIFICATE----- à -----END CERTIFICATE-----) et collez-le dans un éditeur de texte.

9.    Enregistrez le certificat sous la forme d'un fichier .pem nommé root.pem dans le sous-répertoire des certificats.

10.    Sur la page Certificat créé dans la console AWS IoT Core, choisissez Activer. Le bouton devient Désactiver.

11.    Choisissez Joindre une stratégie.

12.    Sur la page Ajouter une stratégie à votre objet, procédez comme suit :
Sélectionnez la stratégie AWS IoT Core que vous avez créée précédemment. Par exemple : admin.
Choisissez Enregistrer l'objet.

Pour plus d'informations, reportez-vous à Création d’un objet.

Copie de l'URL du point de terminaison AWS IoT Core

1.    Dans la console AWS IoT Core, dans le volet de navigation de gauche, choisissez Paramètres.

2.    Sur la page Paramètres, sous Point de terminaison personnalisé, copiez le Point de terminaison. Cette URL de point de terminaison personnalisée AWS IoT Core est propre à votre compte et à votre région AWS.

Création d’un fichier de programme Python

Enregistrez l'un des exemples de code Python suivants dans un fichier de programme Python nommé publish.py.

Si vous avez déjà installé le kit SDK AWS IoT pour Python v2, utilisez l'exemple de code suivant :

Important : remplacez customEndpointUrl par l'URL de votre point de terminaison personnalisé AWS IoT Core. Remplacez certificats par le nom de votre sous-répertoire de certificats. Remplacez a1b23cd45e-certificate.pem.crt par le nom de votre client .crt. Remplacez a1b23cd45e-private.pem.key par le nom de votre clé privée.

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: MIT-0

from awscrt import io, mqtt, auth, http
from awsiot import mqtt_connection_builder
import time as t
import json

# Define ENDPOINT, CLIENT_ID, PATH_TO_CERTIFICATE, PATH_TO_PRIVATE_KEY, PATH_TO_AMAZON_ROOT_CA_1, MESSAGE, TOPIC, and RANGE
ENDPOINT = "customEndpointUrl"
CLIENT_ID = "testDevice"
PATH_TO_CERTIFICATE = "certificates/a1b23cd45e-certificate.pem.crt"
PATH_TO_PRIVATE_KEY = "certificates/a1b23cd45e-private.pem.key"
PATH_TO_AMAZON_ROOT_CA_1 = "certificates/root.pem"
MESSAGE = "Hello World"
TOPIC = "test/testing"
RANGE = 20

# Spin up resources
event_loop_group = io.EventLoopGroup(1)
host_resolver = io.DefaultHostResolver(event_loop_group)
client_bootstrap = io.ClientBootstrap(event_loop_group, host_resolver)
mqtt_connection = mqtt_connection_builder.mtls_from_path(
            endpoint=ENDPOINT,
            cert_filepath=PATH_TO_CERTIFICATE,
            pri_key_filepath=PATH_TO_PRIVATE_KEY,
            client_bootstrap=client_bootstrap,
            ca_filepath=PATH_TO_AMAZON_ROOT_CA_1,
            client_id=CLIENT_ID,
            clean_session=False,
            keep_alive_secs=6
            )
print("Connecting to {} with client ID '{}'...".format(
        ENDPOINT, CLIENT_ID))
# Make the connect() call
connect_future = mqtt_connection.connect()
# Future.result() waits until a result is available
connect_future.result()
print("Connected!")
# Publish message to server desired number of times.
print('Begin Publish')
for i in range (RANGE):
    data = "{} [{}]".format(MESSAGE, i+1)
    message = {"message" : data}
    mqtt_connection.publish(topic=TOPIC, payload=json.dumps(message), qos=mqtt.QoS.AT_LEAST_ONCE)
    print("Published: '" + json.dumps(message) + "' to the topic: " + "'test/testing'")
    t.sleep(0.1)
print('Publish End')
disconnect_future = mqtt_connection.disconnect()
disconnect_future.result()

-ou-

Si vous avez installé le kit SDK AWS IoT Device pour Python (la version précédente du SDK), utilisez l'exemple de code suivant :

Important : remplacez customEndpointUrl par l'URL de votre point de terminaison personnalisé AWS IoT Core. Remplacez certificats par le nom de votre sous-répertoire de certificats. Remplacez a1b23cd45e-certificate.pem.crt par le nom de votre client .crt. Remplacez a1b23cd45e-private.pem.key par le nom de votre clé privée.

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: MIT-0

import time as t
import json
import AWSIoTPythonSDK.MQTTLib as AWSIoTPyMQTT

# Define ENDPOINT, CLIENT_ID, PATH_TO_CERTIFICATE, PATH_TO_PRIVATE_KEY, PATH_TO_AMAZON_ROOT_CA_1, MESSAGE, TOPIC, and RANGE
ENDPOINT = "customEndpointUrl"
CLIENT_ID = "testDevice"
PATH_TO_CERTIFICATE = "certificates/a1b23cd45e-certificate.pem.crt"
PATH_TO_PRIVATE_KEY = "certificates/a1b23cd45e-private.pem.key"
PATH_TO_AMAZON_ROOT_CA_1 = "certificates/root.pem"
MESSAGE = "Hello World"
TOPIC = "test/testing"
RANGE = 20

myAWSIoTMQTTClient = AWSIoTPyMQTT.AWSIoTMQTTClient(CLIENT_ID)
myAWSIoTMQTTClient.configureEndpoint(ENDPOINT, 8883)
myAWSIoTMQTTClient.configureCredentials(PATH_TO_AMAZON_ROOT_CA_1, PATH_TO_PRIVATE_KEY, PATH_TO_CERTIFICATE)

myAWSIoTMQTTClient.connect()
print('Begin Publish')
for i in range (RANGE):
    data = "{} [{}]".format(MESSAGE, i+1)
    message = {"message" : data}
    myAWSIoTMQTTClient.publish(TOPIC, json.dumps(message), 1)
    print("Published: '" + json.dumps(message) + "' to the topic: " + "'test/testing'")
    t.sleep(0.1)
print('Publish End')
myAWSIoTMQTTClient.disconnect()

Test de la configuration

1.    Dans la console AWS IoT Core, dans le volet de navigation de gauche, choisissez Test.

2.    Sur la page du client MQTT, pour Rubrique d’abonnement, saisissez test/testing.

3.    Choisissez S'abonner à la rubrique. Une rubrique de test nommée test/testing est prête pour la publication de message de test. Pour plus d'informations, reportez-vous à Affichage des messages MQTT de l'appareil avec le client AWS IoT MQTT.

4.    Exécutez ce qui suit à partir de votre ligne de commande :

python3 publish.py

Le programme Python publie 20 messages de test sur la rubrique test/testing que vous avez créée dans la console AWS IoT Core. Consultez la rubrique dans la console pour voir les messages publiés.

Astuce : vous pouvez également tester d'autres fonctionnalités des kits SDK, telles que l'abonnement et la connexion via WebSockets, à l'aide des exemples pubsub inclus. Pour plus d'informations, consultez pubsub (kit SDK AWS IoT pour Python v2) ou BasicPubSub (kit SDK AWS IoT Device pour Python) sur GitHub.

(Facultatif) Activation de la journalisation AWS IoT sur Amazon CloudWatch

Vous pouvez surveiller les journaux d'événements pour les messages MQTT que vous publiez sur AWS IoT Core. Pour obtenir des instructions de configuration, consultez Configuration de la journalisation AWS IoT et Surveillance d’AWS IoT à l'aide des journaux CloudWatch.

---

Informations connexes

Bien démarrer avec AWS IoT Core

Approvisionnement des appareils

Questions fréquemment posées (site Web du protocole de messagerie MQTT)