设备孪生入门 (Python)
设备孪生是存储设备状态信息(包括元数据、配置和条件)的 JSON 文档。 IoT 中心为连接到它的每台设备保留一个设备孪生。
注意
本文所述的功能只能用于 IoT 中心的标准层。 有关 IoT 中心基本层和标准/免费层的详细信息,请参阅选择适合你的解决方案的 IoT 中心层。
使用设备克隆可以:
存储来自解决方案后端的设备元数据。
通过设备应用报告当前状态信息,例如可用功能和条件(例如,使用的连接方法)。
同步设备应用和后端应用之间长时间运行的工作流(例如固件和配置更新)的状态。
查询设备的元数据、配置或状态。
设备孪生旨在执行同步以及查询设备的配置和条件。 有关设备孪生的详细信息(包括何时使用设备孪生),请参阅了解设备孪生。
IoT 中心存储包含以下元素的设备孪生:
标记。 只能由解决方案后端访问的设备元数据。
所需属性。 可以由解决方案后端修改以及由设备应用观察的 JSON 对象。
报告属性。 可以由设备应用修改以及由解决方案后端读取的 JSON 对象。
标记和属性不能包含数组,但可以包含嵌套对象。
下图显示了设备孪生组织:
此外,解决方案后端可以根据上述所有数据查询设备孪生。 有关设备孪生的详细信息,请参阅了解设备孪生。 有关查询的详细信息,请参阅 IoT 中心查询语言。
在本文中,将创建两个 Python 控制台应用:
AddTagsAndQuery.py:一个后端应用,用于添加标记并查询设备孪生。
ReportConnectivity.py:一个模拟设备应用,用于连接到 IoT 中心并报告其连接状况。
备注
有关可用于生成设备和后端应用的 SDK 工具的详细信息,请参阅 Azure IoT SDK。
先决条件
有效的 Azure 帐户。 (如果没有帐户,可以创建一个试用帐户,只需几分钟即可完成。)
已注册的设备。 在 Azure 门户中注册一个。
建议使用 Python 版本 3.7 或更高版本。 请确保根据安装程序的要求,使用 32 位或 64 位安装。 在安装过程中出现提示时,请确保将 Python 添加到特定于平台的环境变量中。
确保已在防火墙中打开端口 8883。 本文中的设备示例使用 MQTT 协议,该协议通过端口 8883 进行通信。 在某些公司和教育网络环境中,此端口可能被阻止。 有关解决此问题的更多信息和方法,请参阅连接到 IoT 中心(MQTT)。
获取 IoT 中心连接字符串
在本文中,你将创建一项后端服务,该服务将所需的属性添加到设备孪生,然后查询标识注册表,从而查找具有已相应更新的报告属性的所有设备。 服务需要“服务连接”权限才能修改设备孪生的所需属性,并且需要“注册表读取”权限才能查询标识注册表。 没有仅包含这两个权限的默认共享访问策略,因此需要创建一个。
若要创建授予“服务连接”和“注册表读取”权限的共享访问策略并获取此策略的连接字符串,请执行以下步骤:
在 Azure 门户中,选择“资源组”。 选择中心所在的资源组,然后从资源列表中选择中心。
在中心的左侧窗格上,选择“共享访问策略”。
在策略列表上方的顶部菜单中,选择“添加共享策略访问策略”。
在右侧的“添加共享访问策略”窗格中,为策略输入一个描述性名称,例如 serviceAndRegistryRead。 在“权限”下,选择“注册表读取”和“服务连接”,然后选择“添加”。
从策略列表中选择新策略。
选择“主连接字符串”的复制图标并保存值。
有关 IoT 中心共享访问策略和权限的详细信息,请参阅访问控制和权限。
创建更新所需属性和查询孪生的服务应用
在本部分中,将创建一个 Python 控制台应用,该应用将位置元数据添加到与 {Device ID} 关联的设备孪生。 该应用在 IoT 中心查询位于美国的设备,然后查询报告已建立移动电话网络连接的设备。
在工作目录中,打开命令提示符并安装安装适用于 Python 的 Azure IoT 中心服务 SDK 。
pip install azure-iot-hub
使用文本编辑器,新建一个 AddTagsAndQuery.py 文件。
添加以下代码,从服务 SDK 导入所需模块:
import sys from time import sleep from azure.iot.hub import IoTHubRegistryManager from azure.iot.hub.models import Twin, TwinProperties, QuerySpecification, QueryResult
添加以下代码。 将
[IoTHub Connection String]
替换为在获取 IoT 中心连接字符串中复制的 IoT 中心连接字符串。 将[Device Id]
替换为你在 IoT 中心注册的设备的设备 ID(名称)。IOTHUB_CONNECTION_STRING = "[IoTHub Connection String]" DEVICE_ID = "[Device Id]"
将以下代码添加到 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 工厂的设备的设备孪生,第二个查询将查询细化为仅选择还要通过手机网络连接的设备。
在 AddTagsAndQuery.py 的末尾添加以下代码来实现 iothub_service_sample_run 函数:
if __name__ == '__main__': print("Starting the Python IoT Hub Device Twin service sample...") print() iothub_service_sample_run()
使用以下方法运行应用程序:
python AddTagsAndQuery.py
在查询位于 Redmond43 的所有设备的查询结果中,应该会看到一个设备,而在将结果限制为使用蜂窝网络的设备的查询结果中没有任何设备。 在下一部分中,你将创建一个将使用手机网络的设备应用,并且你将重新运行此查询来查看其变化情况。
创建更新报告属性的设备应用
本部分将创建一个 Python 控制台应用,该应用以 {Device ID} 身份连接到中心,然后更新其设备孪生的报告属性以确认它已使用手机网络进行连接。
在工作目录中的命令提示符下,安装适用于 Python 的 Azure IoT 中心设备 SDK:
pip install azure-iot-device
使用文本编辑器,新建一个 ReportConnectivity.py 文件。
添加以下代码,从设备 SDK 导入所需模块:
import time from azure.iot.device import IoTHubModuleClient
添加以下代码。 将
[IoTHub Device Connection String]
占位符值替换为你在 IoT 中心注册设备时看到的设备连接字符串:CONNECTION_STRING = "[IoTHub Device Connection String]"
将以下代码添加到 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
在 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()
运行设备应用:
python ReportConnectivity.py
应当会看到关于设备孪生报告属性已更新的确认。
既然设备报告其连接的信息,该信息应显示在两个查询中。 回过头来再次运行查询:
python AddTagsAndQuery.py
这一次,两个查询结果中应当都会显示你的 {Device ID}。
在设备应用中,你将看到有关收到了由服务应用发送的所需属性孪生补丁的确认。
本文内容:
- 从后端应用添加了设备元数据作为标记
- 在设备孪生中报告了设备连接信息
- 使用 IoT 中心查询语言查询设备孪生信息
后续步骤
若要了解操作方法:
若要了解如何从设备发送遥测数据,请参阅快速入门:将遥测数据从 IoT 即插即用设备发送到 Azure IoT 中心一文。
关于如何使用设备孪生的所需属性配置设备,请参阅教程:从后端服务配置设备。
以交互方式控制设备,例如从用户控制的应用打开风扇,请参阅快速入门:控制连接到 IoT 中心的设备。