Comment publier des messages MQTT sur AWS IoT Core à partir de mon appareil lorsque j'utilise Python ?

Brève description

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

Solution

Configurer un annuaire 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 répertoire de travail nouvellement créé. Par exemple : certificats.

3.    À partir de 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 vous ne l'avez pas déjà fait, installez pip pour le packaging Python 3. Pour plus d'informations, veuillez consulter la rubrique Installation sur le site web Python Packaging Authority (PyPA).

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

pip install awsiotsdk

-ou-

Installez le kit SDK pour les appareils AWS IoT pour Python (version précédente du kit 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 pour les appareils AWS IoT pour Python sur GitHub.

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

Créer une stratégie AWS IoT Core

1.    Ouvrez la console AWS IoT Core.

2.    Dans le volet de navigation de gauche, sélectionnez Sécuriser.

3.    Sous Sécuriser, choisissez Stratégies.

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

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

6.    Sous Ajouter des instructions, procédez comme suit :
Pour Action, entrez iot :*.
Remarque : autoriser toutes les actions AWS IoT (iot:*) est utile pour les tests. Cependant, une bonne pratique consiste à renforcer la sécurité d'une configuration de production. Pour obtenir des exemples de stratégies plus sécurisées, veuillez consulter la rubrique Exemple de stratégies AWS IoT.
Pour Ressource ARN, entrez *.
Pour Effet , cochez la case Autoriser.

7.    Sélectionnez Créer.

Pour plus d'informations, veuillez consulter les rubriques Création d'une stratégie AWS IoT Core et Stratégies AWS IoT Core.

Créer 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, ainsi que d'autres fonctionnalités AWS IoT, telles que l'indexation de flotte, les tâches, ou le device shadow.

1.    Dans la console AWS IoT Core, à partir du volet de navigation de gauche, sélectionnez Gérer.

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

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

4.    Sur la page Ajoutez votre appareil au registre des objets procédez comme suit :
Saisissez un nom pour votre objet. Par 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, veuiller consulter les rubriques Groupes d'objets statiques et Groupes d'objets dynamiques.
(Facultatif) Sous Définir les attributs des objets à rechercher (facultatif), ajoutez des attributs en tant que paires clé-valeur.
Sélectionnez Next.

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 pour votre objet sont créés.

6.    Sur la page Certificat créé, procédez comme suit :
Sous le volet 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 le volet Vous devez également télécharger une autorité de certification racine pour AWS IoT, choisissez Télécharger. La page Authentification de serveur s'ouvre sur les certificats CA pour l'authentification de serveur.

7.    Sous le volet Points de terminaison Amazon Trust Services (préférés), choisissez Amazon Root CA 1. Le certificat s'ouvre dans votre navigateur.

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

9.    Enregistrez le certificat en tant que fichier .pem nommé root.pem dans le sous-répertoire certificats.

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

11.    Choisissez Associer une stratégie.

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

Pour plus d'informations, veuillez consulter la rubrique Créer une table.

Copier l'URL du point de terminaison AWS IoT Core

1.    Dans la console principale AWS IoT, dans le volet de navigation de gauche, sélectionnez 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é AWS IoT Core est personnelle pour votre compte et votre région AWS.

Créer un fichier de programme Python

Enregistrez l'un des exemples de code Python suivants sous la forme d'un fichier de programme Python nommé publish.py.

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

Important : Remplacez customEndpointUrl par l'URL de votre point de terminaison personnalisé AWS IoT Core. Remplacez les certificats par le nom du sous-répertoire de vos 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 pour les appareils AWS IoT pour Python (la version précédente du kit SDK), utilisez cet exemple de code :

Important : Remplacez customEndpointUrl par l'URL de votre point de terminaison personnalisé AWS IoT Core. Remplacez les certificats par le nom du sous-répertoire de vos 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()

Tester la configuration

1.    Dans la console principale AWS IoT, dans le volet de navigation de gauche, sélectionnez Test.

2.    Sur la page du client MQTT pour Rubrique d'abonnement, entrez test/tester.

3.    Choisissez S'abonner à la rubrique. Une rubrique de test nommée test/tester est prête pour la publication de messages de test. Pour plus d'informations, consultez Afficher les messages MQTT d'appareil avec le client AWS IoT MQTT.

4.    Exécutez la commande suivante à partir de votre ligne de commande :

python3 publish.py

Le programme Python publie 20 messages de test dans la rubrique test/tester que vous avez créée dans la console AWS IoT Core. Affichez la rubrique dans la console pour afficher les messages publiés.

Conseil : Vous pouvez également tester d'autres fonctionnalités des kits SDK, telles que l'abonnement et la connexion via WebSockets, à l'aide des exemples de pubsub inclus. Pour plus d'informations, veuillez consulter la rubrique pubsub (kit SDK AWS IoT pour Python v2) ou la rubrique BasicPubSub (kit SDK pour les appareils AWS IoT pour Python) sur GitHub.

(Facultatif) Activer la journalisation AWS IoT dans Amazon CloudWatch

Vous pouvez surveiller les journaux d'événements pour les messages MQTT que vous publiez dans AWS IoT Core. Pour savoir comment configurer la fonctionnalité, consultez les articles Configurer la journalisation AWS IoT et Surveiller AWS IoT à l'aide de CloudWatch Logs.

---

Informations connexes

La mise en route avec AWS IoT Core est simple.

Allocation des appareils

Questions fréquentes (site web du protocole de messagerie MQTT)