设备孪生入门 (Python)

设备孪生是存储设备状态信息（包括元数据、配置和条件）的 JSON 文档。 IoT 中心为连接到它的每台设备保留一个设备孪生。

使用设备克隆可以：

• 存储来自解决方案后端的设备元数据。

• 通过设备应用报告当前状态信息，例如可用功能和条件（例如，使用的连接方法）。

• 同步设备应用和后端应用之间长时间运行的工作流（例如固件和配置更新）的状态。

• 查询设备的元数据、配置或状态。

设备孪生旨在执行同步以及查询设备的配置和条件。 有关设备孪生的详细信息（包括何时使用设备孪生），请参阅了解设备孪生。

IoT 中心存储包含以下元素的设备孪生：

• 标记。 只能由解决方案后端访问的设备元数据。

• 所需属性。 可以由解决方案后端修改以及由设备应用观察的 JSON 对象。

• 报告属性。 可以由设备应用修改以及由解决方案后端读取的 JSON 对象。

标记和属性不能包含数组，但可以包含嵌套对象。

下图显示了设备孪生组织：

此外，解决方案后端可以根据上述所有数据查询设备孪生。 有关设备孪生的详细信息，请参阅了解设备孪生。 有关查询的详细信息，请参阅 IoT 中心查询语言。

在本文中，将创建两个 Python 控制台应用：

• AddTagsAndQuery.py：一个后端应用，用于添加标记并查询设备孪生。

• ReportConnectivity.py：一个模拟设备应用，用于连接到 IoT 中心并报告其连接状况。

先决条件

• 有效的 Azure 帐户。 （如果没有帐户，可以创建一个试用帐户，只需几分钟即可完成。）

• IoT 中心。 使用 CLI 或 Azure 门户创建一个。

• 已注册的设备。 在 Azure 门户中注册一个。

• 建议使用 Python 版本 3.7 或更高版本。 请确保根据安装程序的要求，使用 32 位或 64 位安装。 在安装过程中出现提示时，请确保将 Python 添加到特定于平台的环境变量中。

• 确保已在防火墙中打开端口 8883。 本文中的设备示例使用 MQTT 协议，该协议通过端口 8883 进行通信。 在某些公司和教育网络环境中，此端口可能被阻止。 有关解决此问题的更多信息和方法，请参阅连接到 IoT 中心(MQTT)。

获取 IoT 中心连接字符串

在本文中，你将创建一项后端服务，该服务将所需的属性添加到设备孪生，然后查询标识注册表，从而查找具有已相应更新的报告属性的所有设备。 服务需要“服务连接”权限才能修改设备孪生的所需属性，并且需要“注册表读取”权限才能查询标识注册表。 没有仅包含这两个权限的默认共享访问策略，因此需要创建一个。

若要创建授予“服务连接”和“注册表读取”权限的共享访问策略并获取此策略的连接字符串，请执行以下步骤：

1. 在 Azure 门户中，选择“资源组”。 选择中心所在的资源组，然后从资源列表中选择中心。

2. 在中心的左侧窗格上，选择“共享访问策略”。

3. 在策略列表上方的顶部菜单中，选择“添加共享策略访问策略”。

4. 在右侧的“添加共享访问策略”窗格中，为策略输入一个描述性名称，例如 serviceAndRegistryRead。 在“权限”下，选择“注册表读取”和“服务连接”，然后选择“添加”。

   

5. 从策略列表中选择新策略。

6. 选择“主连接字符串”的复制图标并保存值。

   

有关 IoT 中心共享访问策略和权限的详细信息，请参阅访问控制和权限。

创建更新所需属性和查询孪生的服务应用

在本部分中，将创建一个 Python 控制台应用，该应用将位置元数据添加到与 {Device ID} 关联的设备孪生。 该应用在 IoT 中心查询位于美国的设备，然后查询报告已建立移动电话网络连接的设备。

1. 在工作目录中，打开命令提示符并安装安装适用于 Python 的 Azure IoT 中心服务 SDK 。

   pip install azure-iot-hub
   

2. 使用文本编辑器，新建一个 AddTagsAndQuery.py 文件。

3. 添加以下代码，从服务 SDK 导入所需模块：

   import sys
   from time import sleep
   from azure.iot.hub import IoTHubRegistryManager
   from azure.iot.hub.models import Twin, TwinProperties, QuerySpecification, QueryResult
   

4. 添加以下代码。 将 [IoTHub Connection String] 替换为在获取 IoT 中心连接字符串中复制的 IoT 中心连接字符串。 将 [Device Id] 替换为你在 IoT 中心注册的设备的设备 ID（名称）。

   IOTHUB_CONNECTION_STRING = "[IoTHub Connection String]"
   DEVICE_ID = "[Device Id]"
   

5. 将以下代码添加到 AddTagsAndQuery.py 文件：

   def iothub_service_sample_run():
       try:
           iothub_registry_manager = IoTHubRegistryManager(IOTHUB_CONNECTION_STRING)
   
           new_tags = {
                   'location' : {
                       'region' : 'US',
                       'plant' : 'Redmond43'
                   }
               }
   
           twin = iothub_registry_manager.get_twin(DEVICE_ID)
           twin_patch = Twin(tags=new_tags, properties= TwinProperties(desired={'power_level' : 1}))
           twin = iothub_registry_manager.update_twin(DEVICE_ID, twin_patch, twin.etag)
   
           # Add a delay to account for any latency before executing the query
           sleep(1)
   
           query_spec = QuerySpecification(query="SELECT * FROM devices WHERE tags.location.plant = 'Redmond43'")
           query_result = iothub_registry_manager.query_iot_hub(query_spec, None, 100)
           print("Devices in Redmond43 plant: {}".format(', '.join([twin.device_id for twin in query_result.items])))
   
           print()
   
           query_spec = QuerySpecification(query="SELECT * FROM devices WHERE tags.location.plant = 'Redmond43' AND properties.reported.connectivity = 'cellular'")
           query_result = iothub_registry_manager.query_iot_hub(query_spec, None, 100)
           print("Devices in Redmond43 plant using cellular network: {}".format(', '.join([twin.device_id for twin in query_result.items])))
   
       except Exception as ex:
           print("Unexpected error {0}".format(ex))
           return
       except KeyboardInterrupt:
           print("IoT Hub Device Twin service sample stopped")
   

   IoTHubRegistryManager 对象公开从该服务与设备孪生交互所需的所有方法。 此代码将首先初始化 IoTHubRegistryManager 对象，然后更新 DEVICE_ID 的设备孪生，最后运行两个查询。 第一个查询仅选择位于 Redmond43 工厂的设备的设备孪生，第二个查询将查询细化为仅选择还要通过手机网络连接的设备。

6. 在 AddTagsAndQuery.py 的末尾添加以下代码来实现 iothub_service_sample_run 函数：

   if __name__ == '__main__':
       print("Starting the Python IoT Hub Device Twin service sample...")
       print()
   
       iothub_service_sample_run()
   

7. 使用以下方法运行应用程序：

   python AddTagsAndQuery.py
   

   在查询位于 Redmond43 的所有设备的查询结果中，应该会看到一个设备，而在将结果限制为使用蜂窝网络的设备的查询结果中没有任何设备。 在下一部分中，你将创建一个将使用手机网络的设备应用，并且你将重新运行此查询来查看其变化情况。

   

创建更新报告属性的设备应用

本部分将创建一个 Python 控制台应用，该应用以 {Device ID} 身份连接到中心，然后更新其设备孪生的报告属性以确认它已使用手机网络进行连接。

1. 在工作目录中的命令提示符下，安装适用于 Python 的 Azure IoT 中心设备 SDK：

   pip install azure-iot-device
   

2. 使用文本编辑器，新建一个 ReportConnectivity.py 文件。

3. 添加以下代码，从设备 SDK 导入所需模块：

   import time
   from azure.iot.device import IoTHubModuleClient
   

4. 添加以下代码。 将 [IoTHub Device Connection String] 占位符值替换为你在 IoT 中心注册设备时看到的设备连接字符串：

   CONNECTION_STRING = "[IoTHub Device Connection String]"
   

5. 将以下代码添加到 ReportConnectivity.py 文件以实例化客户端并实现设备孪生功能：

   def create_client():
       # Instantiate client
       client = IoTHubModuleClient.create_from_connection_string(CONNECTION_STRING)
   
       # Define behavior for receiving twin desired property patches
       def twin_patch_handler(twin_patch):
           print("Twin patch received:")
           print(twin_patch)
   
       try:
           # Set handlers on the client
           client.on_twin_desired_properties_patch_received = twin_patch_handler
       except:
           # Clean up in the event of failure
           client.shutdown()
   
       return client
   

6. 在 ReportConnectivity.py 的末尾添加以下代码来运行应用程序：

   def main():
       print ( "Starting the Python IoT Hub Device Twin device sample..." )
       client = create_client()
       print ( "IoTHubModuleClient waiting for commands, press Ctrl-C to exit" )
   
       try:
           # Update reported properties with cellular information
           print ( "Sending data as reported property..." )
           reported_patch = {"connectivity": "cellular"}
           client.patch_twin_reported_properties(reported_patch)
           print ( "Reported properties updated" )
   
           # Wait for program exit
           while True:
               time.sleep(1000000)
       except KeyboardInterrupt:
           print ("IoT Hub Device Twin device sample stopped")
       finally:
           # Graceful exit
           print("Shutting down IoT Hub Client")
           client.shutdown()
   
   if __name__ == '__main__':
       main()
   

7. 运行设备应用：

   python ReportConnectivity.py
   

   应当会看到关于设备孪生报告属性已更新的确认。

   

8. 既然设备报告其连接的信息，该信息应显示在两个查询中。 回过头来再次运行查询：

   python AddTagsAndQuery.py
   

   这一次，两个查询结果中应当都会显示你的 {Device ID}。

   

   在设备应用中，你将看到有关收到了由服务应用发送的所需属性孪生补丁的确认。

   

本文内容：

• 从后端应用添加了设备元数据作为标记
• 在设备孪生中报告了设备连接信息
• 使用 IoT 中心查询语言查询设备孪生信息

后续步骤

若要了解操作方法：

• 若要了解如何从设备发送遥测数据，请参阅快速入门：将遥测数据从 IoT 即插即用设备发送到 Azure IoT 中心一文。

• 关于如何使用设备孪生的所需属性配置设备，请参阅教程：从后端服务配置设备。

• 以交互方式控制设备，例如从用户控制的应用打开风扇，请参阅快速入门：控制连接到 IoT 中心的设备。