设备孪生入门 (Python)

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

注意

本文所述的功能只能用于 IoT 中心的标准层。 有关 IoT 中心基本层和标准/免费层的详细信息,请参阅选择适合你的解决方案的 IoT 中心层

使用设备克隆可以:

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

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

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

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

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

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

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

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

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

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

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

设备孪生概念图的屏幕截图。

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

在本文中,将创建两个 Python 控制台应用:

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

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

备注

有关可用于生成设备和后端应用的 SDK 工具的详细信息,请参阅 Azure IoT SDK

先决条件

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

  • IoT 中心。 使用 CLIAzure 门户创建一个。

  • 已注册的设备。 在 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 的所有设备的查询结果中,应该会看到一个设备,而在将结果限制为使用蜂窝网络的设备的查询结果中没有任何设备。 在下一部分中,你将创建一个将使用手机网络的设备应用,并且你将重新运行此查询来查看其变化情况。

    第一个显示 Redmond 中所有设备的查询的屏幕截图。

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

本部分将创建一个 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 中心查询语言查询设备孪生信息

后续步骤

若要了解操作方法: