Como uso o Python para publicar mensagens MQTT no AWS IoT Core a partir do meu dispositivo?

Breve descrição

Verifique se sua coisa do AWS IoT está configurada corretamente e se seus certificados estão devidamente anexados. Para testar sua configuração, é possível usar o cliente do MQTT do AWS IoT e o exemplo de código Python fornecido neste artigo.

Resolução

Configure um diretório para testar a publicação do MQTT

1. Crie um diretório de trabalho no seu ambiente de desenvolvimento. Por exemplo, crie um diretório chamado iot-test-publish.
2. Crie um subdiretório para certificados no seu novo diretório de trabalho. Por exemplo, crie um subdiretório chamado certificados.
3. Na linha de comandos, altere o diretório para seu novo diretório de trabalho.

Instale o pip e o SDK do AWS IoT para Python

Observação: antes de começar, instale o pip para o pacote Python 3. Para obter mais informações, consulte Installation (Instalação) no site da Python Packaging Authority (PyPA).

1. Para instalar o SDK do AWS IoT para Python v2, use o seguinte comando:

   pip install awsiotsdk

   -ou-

   Para instalar o Device SDK do AWS IoT para Python (a versão anterior do SDK, v1) execute o seguinte comando:

   pip install AWSIoTPythonSDK

Observação: é uma prática recomendada usar a atualização mais recente para sua versão do SDK do AWS IoT para Python. Para obter mais informações, consulte AWS IoT SDK for Python v2 (SDK do AWS IoT para Python v2) ou AWS IoT Device SDK for Python (Device SDK do AWS IoT para Python) no GitHub. Também é uma prática recomendada usar esses SDKs para se conectar ao AWS IoT Core, mas esses SDKs não são obrigatórios — é possível usar qualquer outro cliente MQTT terceirizado compatível para se conectar ao AWS IoT Core.

Crie uma política do AWS IoT Core

1. Abra o console do AWS IoT Core.
2. No painel de navegação esquerdo, clique em Seguro.
3. Em Seguro, selecione Políticas.
4. Se você tiver políticas existentes do AWS IoT Core, clique em Criar para criar uma nova política.
   -ou-
   Se você não tiver políticas existentes, na página Você ainda não tem políticas, selecione Criar uma política.
5. Na página Criar uma política, insira um Nome para sua política. Por exemplo, digite o nome admin.
6. Em Adicionar instruções, conclua as seguintes ações:
   Em Ação, insira iot:*.
   Importante: é possível permitir todas as ações do AWS IoT (iot:*) para fins de teste. No entanto, é uma prática recomendada aumentar a segurança de uma configuração de produção. Para obter exemplos de políticas mais seguras, consulte AWS IoT Core policy examples (Exemplos de políticas do AWS IoT Core).
   Em ARN do recurso, insira *.
   Em Efeito, marque a caixa de seleção Permitir.
7. Selecione Criar.

Para obter mais informações, consulte Create an AWS IoT policy (Crie uma política do AWS IoT) e AWS IoT Core policies (Políticas do AWS IoT Core).

Crie uma coisa do AWS IoT

Observação: não é necessário criar uma coisa para se conectar ao AWS IoT. No entanto, coisas permitem que você use controles de segurança adicionais e outros recursos do AWS IoT, como Indexação de frotas, Trabalhos ou Sombra do dispositivo.

1. No console do AWS IoT Core, no painel de navegação esquerdo, clique em Gerenciar.
2. Se você tiver coisas existentes, selecione Criar para criar uma coisa nova.
   -ou-
   Se você não tiver coisas existentes, na página Você ainda não tem coisas, clique em Registrar uma coisa.
3. Na página Criar coisas do AWS IoT, selecione Criar uma única coisa.
4. Na página Adicionar seu dispositivo ao registro de coisas, conclua as seguintes ações:
   Insira um Nome para sua coisa. Por exemplo, insira Teste-Coisa.
   (Opcional) Em Adicionar um tipo a essa coisa, selecione ou crie um tipo de coisa.
   (Opcional) Em Adicionar essa coisa a um grupo, selecione ou crie um grupo. Para obter mais informações sobre grupos, consulte Grupos de objetos estáticos e Grupos de objetos dinâmicos.
   (Opcional) Em Definir atributos pesquisáveis da coisa (opcional), adicione atributos como pares de valor-chave.
   Clique em Próximo.
   Observação: é uma prática recomendada combinar o nome da coisa com o clientId. Se os nomes corresponderem, será possível usar os recursos do AWS IoT Core, como Indexação de frota e variáveis de política da coisa.
5. Na página Adicionar um certificado à sua coisa, clique em Criar certificado. Você vê notificações confirmando que sua coisa e um certificado para sua coisa foram criados.
6. Na página Certificado criado, conclua as seguintes ações:
   Em Para conectar um dispositivo, você precisa fazer download do seguinte, selecione Fazer download para o certificado, a chave pública e a chave privada.
   Salve cada arquivo baixado no subdiretório de certificados que você criou anteriormente.
   Em Você também precisa fazer download de um CA raiz no AWS IoT, selecione Fazer download. A página Autenticação do servidor é aberta em Certificados CA para autenticação do servidor.
7. Em Endpoints de serviços de confiança da Amazon (preferencial), clique em CA raiz 1 da Amazon. O certificado é aberto no navegador.
8. Copie o certificado (desde -----BEGIN CERTIFICATE----- to -----END CERTIFICATE-----) e cole-o em um editor de texto.
9. Salve o certificado como um arquivo .pem denominado root.pem no subdiretório de certificados.
10. Na página Certificado criado do console do AWS IoT Core, selecione Ativar. O botão muda para Desativar.
11. Clique em Anexar uma política.
12. Na página Adicionar uma política para sua coisa, conclua as seguintes ações:
    Selecione a política do AWS IoT Core que você criou anteriormente. Por exemplo, selecione admin.
    Clique em Registrar coisa.

Para obter mais informações, consulte Criar um objeto.

Copie o URL do endpoint do AWS IoT Core

1. No console do AWS IoT Core, no painel de navegação, selecione Configurações.
2. Na página Configurações, em Endpoint personalizado, copie o Endpoint. Esse URL de endpoint personalizado do AWS IoT Core é único para sua conta e região da AWS.

Crie um arquivo de programa Python

Salve um dos exemplos de código Python a seguir como um arquivo de programa Python denominado publish.py.

No SDK do AWS IoT para Python v2, use o seguinte código de exemplo:

Importante: substitua customEndpointUrl pelo seu URL de endpoint personalizado do AWS IoT Core. Substitua certificates pelo nome do seu subdiretório de certificados. Substitua a1b23cd45e-certificate.pem.crt pelo nome do seu cliente .crt. Substitua a1b23cd45e-private.pem.key pelo nome da sua chave privada.

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

No Device SDK do AWS IoT para Python v1, use o seguinte código de exemplo:

Importante: substitua customEndpointUrl pelo seu URL de endpoint personalizado do AWS IoT Core. Substitua certificates pelo nome do seu subdiretório de certificados. Substitua a1b23cd45e-certificate.pem.crt pelo nome do seu cliente .crt. Substitua a1b23cd45e-private.pem.key pelo nome da sua chave privada.

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

Observação: as versões mais antigas do Device SDK do AWS IoT para Python não são compatíveis com o Python 3.12 ou posterior. Certifique-se de usar a versão mais recente do Device SDK do AWS IoT para Python.

Teste a configuração

1. No console do AWS IoT Core, no painel de navegação à esquerda, selecione Testar.
2. Na página Cliente MQTT, em Tópico de assinatura, insira teste/testar.
3. Clique em Assinar um tópico. Um tópico de teste chamado teste/testar está pronto para publicação da mensagem de teste. Para obter mais informações, consulte Exibir mensagens MQTT com o cliente AWS IoT MQTT.
4. Execute o seguinte comando na sua linha de comandos:

   python3 publish.py

O programa Python publica 20 mensagens de teste no tópico teste/testar que você criou no console do AWS IoT Core. Visualize o tópico no console para ver as mensagens publicadas.

Dica: também é possível usar os exemplos do pubsub incluídos para testar outros recursos dos SDKs, como assinaturas e conexões do WebSockets. Para obter mais informações, consulte pubsub (SDK do AWS IoT para Python v2) ou BasicPubSub (Device SDK do AWS IoT para Python) no site da GitHub.

(Opcional) Ative o registro em log do AWS IoT no Amazon CloudWatch

É possível monitorar logs de eventos para mensagens MQTT que você publica no AWS IoT Core. Para obter instruções de configuração, consulte Configure AWS IoT logging (Configurar o registro em log do AWS IoT) e Monitor AWS IoT using CloudWatch Logs (Monitorar o AWS IoT usando CloudWatch Logs).

Informações relacionadas

Introdução aos AWS IoT Core tutoriais

Provisionamento de dispositivos

FAQ (Perguntas frequentes) (site do protocolo de mensagens MQTT)