Wie verwende ich Python, um MQTT-Nachrichten von meinem Gerät auf AWS IoT Core zu veröffentlichen?

Lesedauer: 8 Minute

Ich kann keine Message Queuing Telemetry Transport (MQTT)-Nachrichten zwischen AWS IoT Core und meinem Gerät oder MQTT-Client senden oder empfangen.

Kurzbeschreibung

Stelle sicher, dass das AWS-IoT-Objekt richtig konfiguriert ist und die Zertifikate ordnungsgemäß angefügt sind. Um das Setup zu testen, kannst du den AWS-IoT-MQTT-Client und den in diesem Artikel bereitgestellten Python-Beispielcode verwenden.

Lösung

Ein Verzeichnis einrichten, um die MQTT-Veröffentlichung zu testen

Erstelle ein Arbeitsverzeichnis in deiner Entwicklungsumgebung. Erstelle beispielsweise ein Verzeichnis mit dem Namen iot-test-publish.
Erstelle im neuen Arbeitsverzeichnis ein Unterverzeichnis für Zertifikate. Erstelle beispielsweise ein Unterverzeichnis mit dem Namen certificates.
Ändere das Verzeichnis über die Befehlszeile in dein neues Arbeitsverzeichnis.

pip und das AWS-IoT-SDK für Python installieren

Hinweis: Bevor du beginnst, installiere pip für die Python-3-Paketierung. Weitere Informationen findest du unter Installation auf der Python Packaging Authority (PyPA)-Website.

Verwende den folgenden Befehl, um das AWS-IoT-SDK für Python v2 zu installieren:
```
pip install awsiotsdk
```
-oder-

Installiere das AWS-IoT-Geräte-SDK für Python (die vorherige SDK-Version, v1), indem du den folgenden Befehl ausführst:
```
pip install AWSIoTPythonSDK
```

Hinweis: Es hat sich bewährt, das neueste Update für deine Version des AWS-IoT-SDK für Python zu verwenden. Weitere Informationen findest du unter AWS-IoT-SDK für Python v2 oder AWS-IoT-Geräte-SDK für Python auf GitHub. Es ist auch eine bewährte Methode, diese SDKs für die Verbindung mit AWS IoT Core zu verwenden, aber diese SDKs sind nicht erforderlich. Du kannst jeden anderen kompatiblen MQTT-Client eines Drittanbieters verwenden, um eine Verbindung zu AWS IoT Core herzustellen.

Erstellen einer AWS-IoT-Core-Richtlinie

Öffne die AWS-IoT-Core-Konsole.
Wähle im linken Navigationsbereich Sicher aus.
Wähle unter Sicher die Option Richtlinien aus.
Wenn du bereits über AWS-IoT-Core-Richtlinien verfügst, wähle Erstellen, um eine neue Richtlinie zu erstellen.
-oder-
Wenn du noch keine Richtlinien hast, wähle auf der Seite Du hast noch keine Richtlinien die Option Eine Richtlinie erstellen aus.
Gib auf der Seite Eine Richtlinie erstellen einen Namen für deine Richtlinie ein. Gib beispielsweise den Namen admin ein.
Führe unter Anweisungen hinzufügen die folgenden Aktionen aus:
Gib für Aktion iot:* ein.
Wichtig: Du kannst alle AWS-IoT-Aktionen (*iot:\ ) zu Testzwecken zulassen. Es ist jedoch eine bewährte Methode, die Sicherheit in einem Produktions-Setup zu erhöhen. Weitere Beispiele für Sicherheitsrichtlinien findest du unter Beispiele für AWS-IoT-Core-Richtlinien.
Gib für Ressourcen-ARN* ein.
Wähle unter Effekt das Kontrollkästchen Erlauben aus.
Wähle Erstellen.

Weitere Informationen findest du unter Erstellen einer AWS-IoT-Richtlinie und unter AWS-IoT-Core-Richtlinien.

Erstellen eines AWS-IoT-Objekts

Hinweis: Du musst kein Objekt erstellen, um eine Verbindung zu AWS IoT herzustellen. Objekte ermöglichen es dir jedoch, zusätzliche Sicherheitskontrollen und andere AWS-IoT-Funktionen wie Flottenindizierung, Aufgaben oder Geräteschatten zu verwenden.

Wähle in der AWS-IoT-Core-Konsole im linken Navigationsbereich Verwalten aus.
Wenn du bereits Objekte hast, wähleErstellen, um ein neues Objekt zu erstellen.
-oder-
Wenn du noch keine Objekte hast, wähle auf der Seite Du hast noch keine Objekte die Option Ein Objekt registrieren aus.
Wähle auf der Seite Erstellen von AWS IoT-Objekten die Option Ein einzelnes Objekt erstellen aus.
Führe auf der Seite Füge dein Gerät zur Objektregistrierung hinzu die folgenden Aktionen aus:
Gib einen Namen für das Objekt ein. Gib beispielsweise Test-Thing ein.
(Optional) Wähle unter Diesem Objekt eine Art hinzufügen einen Objekttypen aus oder erstelle einen.
(Optional) Wähle unter Dieses Objekt einer Gruppe hinzufügen eine Gruppe aus oder erstelle eine Gruppe. Weitere Informationen zu Gruppen findest du unter Statische Objektgruppen und Dynamische Objektgruppen.
(Optional) Füge unter Suchbare Objektattribute festlegen (optional) Attribute als Schlüssel-Wert-Paare hinzu.
Wähle Weiter aus.
Hinweis: Es hat sich bewährt, den Namen des Objekts der clientId zuzuordnen. Wenn die Namen übereinstimmen, kannst du AWS-IoT-Core-Funktionen wie Flottenindizierung und Objektrichtlinien-Variablen verwenden.
Wähle auf der Seite Füge ein Zertifikat für dein Objekt hinzu die Option Zertifikat erstellen aus. Du siehst Benachrichtigungen, die bestätigen, dass das Objekt und ein Zertifikat für das Objekt erstellt wurden.
Führe auf der Seite Zertifikat erstellt die folgenden Aktionen aus:
Wähle unter Um ein Gerät zu verbinden, musst du Folgendes herunterladen Herunterladen für das Zertifikat, den öffentlichen Schlüssel und den privaten Schlüssel aus.
Speichere jede heruntergeladene Datei im Unterverzeichnis für Zertifikate, das du zuvor erstellt hast.
Wähle unter Du musst auch eine Root-CA für AWS IoT herunterladen die Option Herunterladen aus. Die Seite Server authentication (Serverauthentifizierung) wird bei CA-Zertifikate für die Serverauthentifizierung geöffnet.
Wähle unter Amazon-Trust-Services-Endpunkte (bevorzugt) Amazon Root-CA 1 aus. Das Zertifikat wird in deinem Browser geöffnet.
Kopiere das Zertifikat (alles von -----BEGIN CERTIFICATE----- bis -----END CERTIFICATE-----) und füge es in einen Texteditor ein.
Speichere das Zertifikat als PEM-Datei mit dem Namen root.pem im Unterverzeichnis für Zertifikate.
Wähle auf der Seite Zertifikat erstellt in der AWS-IoT-Core-Konsole die Option Aktivieren aus. Die Schaltfläche ändert sich zu Deaktivieren.
Wähle Eine Richtlinie anfügen aus.
Führe auf der Seite Füge eine Richtlinie für dein Objekt hinzu die folgenden Aktionen aus:
Wähle die AWS-IoT-Core-Richtlinie aus, die du zuvor erstellt hast. Wähle beispielsweise admin aus.
Wähle Objekt registrieren aus.

Weitere Informationen findest du unter Ein Dingobjekt erstellen.

Die URL des AWS-IoT-Core-Endpunkts kopieren

Wähle in der AWS-IoT-Core-Konsole im Navigationsbereich Einstellungen aus.
Kopiere auf der Seite Einstellungen unter Benutzerdefinierter Endpunkt den Endpunkt. Diese benutzerdefinierte AWS-IoT-Core-Endpunkt-URL ist für dein AWS-Konto und deine Region eindeutig.

Erstellen einer Python-Programmdatei

Speichere eines der folgenden Python-Codebeispiele als Python-Programmdatei mit dem Namen publish.py.

Verwende für das AWS-IoT-SDK für Python v2 den folgenden Beispielcode:

Wichtig: Ersetze customEndpointUrl durch deine benutzerdefinierte AWS-IoT-Core-Endpunkt-URL. Ersetze certificates durch den Namen deines Unterverzeichnisses für Zertifikate. Ersetze a1b23cd45e-certificate.pem.crt durch den Namen deines Clients .crt. Ersetze a1b23cd45e-private.pem.key durch den Namen deines privaten Schlüssels.

# 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()

-oder-

Verwende für das AWS-IoT-Geräte-SDK für Python v1 den folgenden Beispielcode:

# 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()

Hinweis: Ältere Versionen von AWS IoT Device SDK für Python werden für Python 3.12 oder höher nicht unterstützt. Stelle sicher, dass du die neueste Version von AWS IoT Device SDK für Python verwendest.

Setup testen

Wähle in der AWS-IoT-Core-Konsole im linken Navigationsbereich Testen aus.
Gib auf der Seite MQTT-Client unter Abonnementthema den Text test/testing ein.
Wähle Thema abonnieren aus. Ein Testthema mit dem Namen test/testing ist bereit für die Veröffentlichung der Testnachricht. Weitere Informationen findest du unter MQTT-Nachrichten mit dem AWS-IoT-MQTT-Client anzeigen.
Führe den folgenden Befehl von der Befehlszeile aus:
```
python3 publish.py
```

Das Python-Programm veröffentlicht 20 Testnachrichten zum Thema test/testing, das du in der AWS-IoT-Core-Konsole erstellt hast. Sieh dir das Thema in der Konsole an, um die veröffentlichten Nachrichten zu sehen.

Tipp: Du kannst auch die mitgelieferten Pubsub-Beispiele verwenden, um andere Funktionen der SDKs zu testen, z. B. WebSockets-Abonnements und -Verbindungen. Weitere Informationen findest du unter pubsub (AWS IoT SDK für Python v2) oder BasicPubSub (AWS IoT Device SDK für Python) auf GitHub.

(Optional) Die AWS-IoT-Protokollierung auf Amazon CloudWatch aktivieren

Du kannst Ereignisprotokolle für MQTT-Nachrichten überwachen, die du in AWS IoT Core veröffentlichst. Anweisungen zur Einrichtung findest du unter Konfigurieren der AWS-IoT-Protokollierung und Überwachen von AWS IoT mithilfe von CloudWatch Logs.

Relevanter Inhalt

Wie konfiguriere ich die Protokollierungsebenen manuell für bestimmte Ressourcen in AWS IoT Core?
AWS OFFICIALAktualisiert vor 3 Jahren
Wie konfiguriere ich die Protokollierungsebenen dynamisch für bestimmte Ressourcen in AWS IoT Core?
AWS OFFICIALAktualisiert vor 3 Jahren
Wie kann ich JITP mit AWS IoT Core verwenden?
AWS OFFICIALAktualisiert vor einem Jahr
Wie verwalte ich die Protokollebenen meiner AWS-IoT-Protokolle in AWS IoT Core am besten?
AWS OFFICIALAktualisiert vor 2 Jahren