将传感器连接到 aws iot

Question

我们曾尝试将传感器连接到 aws iot。传感器已连接到 aws iot，但我们无法在影子上查看消息。我们的影子没有得到更新。请提出一些方法，以便我们可以更新我们的影子。

Answer 1

下面是我为与 AWS IoT 阴影交互而创建的一个类，它具有读取、更新和删除阴影的方法。这里的主要内容是我使用boto3 client 连接到 AWS IoT，这允许我推送和拉取 JSON 以存储在云中（事物的影子）。

import boto3
import logging
import json

# Initialize logger for CloudWatch logs
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)


class IoTOps:
    """
    Params: iotClient boto3 client used to update thing's shadow and also send 
    messages to the thing
        topic STRING - the MQTT Topic to send a message to
        QoS - INT - Essentially the channel of a specific topic to communicate
        over
    """
    def __init__(self, iotConfig):
        self.iotClient = boto3.client('iot-data', region_name='us-east-1')
        self.topic = iotConfig['topic']
        self.QoS = iotConfig['QoS']
        self.thingName = iotConfig['thingName']

    def publish_mqtt_message(self, payload):
        """                                                                        
        Description: Send MQTT message to MQTT Topic                         

        Return: None                                                          
        """                                                                       
        # Publish a MQTT message to a topic for our thing to ingest               
        logger.info('publishing MQTT message to topic: {}'.format(self.topic))         
        self.iotClient.publish(                                                           
            topic=self.topic,                     
            qos=self.QoS,                                                               
            payload=json.dumps(payload)                                   
        )  
        return

    def update_shadow(self, payload):
        """
        Summary: Updates a things shadow for a specified thing
        Return The state information in a JSON format
        """
        updatedPayload = {'state': {'desired': payload}}
        response = self.iotClient.update_thing_shadow(
            thingName=self.thingName,
            payload=json.dumps(updatedPayload)
        )
        return response

    def get_shadow(self):
        """
        Summary: gets thing shadow for a specified thing
        Return: The state information in JSON format
        """
        response = self.iotClient.get_thing_shadow(
            thingName=self.thingName
        )
        response = json.loads(response['payload'].read().decode('utf-8'))['state']['desired']
        return response

    def delete_shadow(self):
        """
        Summary: deletes thing shadow for a specified thing
        Return: The state information in a JSON format
        """
        response = self.iotClient.delete_thing_shadow(
            thingName=self.thingName
        )
        return response