你当前正在访问 Microsoft Azure Global Edition 技术文档网站。 如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

适用于 Python 的 Azure 数字孪生核心客户端库 - 版本 1.2.0

此包包含用于 Azure 数字孪生 API 的 SDK，用于提供对 Azure 数字孪生服务的访问权限，用于管理孪生体、模型、关系等。

免责声明

对 Python 2.7 的 Azure SDK Python 包支持已于 2022 年 1 月 1 日结束。 有关详细信息和问题，请参阅 https://github.com/Azure/azure-sdk-for-python/issues/20691

入门

简介

Azure 数字孪生是下一代 IoT 解决方案的开发人员平台，可用于在云中安全高效地创建、运行和管理业务环境的数字表示形式。 使用 Azure 数字孪生，创建实时操作状态表示形式快速且经济高效，并且数字表示形式与来自 IoT 和其他数据源的实时数据保持最新。 如果你不熟悉 Azure 数字孪生，并且想要了解有关该平台的详细信息，请确保查看 Azure 数字孪生 官方文档页。

有关如何针对 Azure 数字孪生服务进行编程的简介，请访问 编码教程页 ，获取简单的分步指南。 访问 本教程 ，了解如何使用命令行客户端应用程序与 Azure 数字孪生实例交互。 最后，有关如何生成由环境中的实时数据驱动的端到端 Azure 数字孪生解决方案的快速指南，请确保查看 此有用的指南。

上述指南可帮助你开始使用 Azure 数字孪生的关键元素，例如创建 Azure 数字孪生实例、模型、孪生图等。使用以下示例指南熟悉有助于针对 Azure 数字孪生进行编程的各种 API。

安装方法

使用 pip 安装 [azure-digitaltwins-core][pypi_package_keys] 和 azure-identity：

pip install azure-digitaltwins-core azure-identity

azure-identity 用于 Azure Active Directory 身份验证，如下所示。

如何使用

身份验证，权限

若要创建新的数字孪生客户端，需要 Azure 数字孪生实例的终结点和凭据。 对于以下示例， AZURE_URL必须设置 、 AZURE_TENANT_ID、 AZURE_CLIENT_ID和 AZURE_CLIENT_SECRET 环境变量。 客户端需要 TokenCredential 或 ServiceClientCredentials 的实例。 在此示例中，我们演示如何使用一个派生类： DefaultAzureCredentials。

注意：若要访问数字孪生服务的数据平面，必须为实体授予权限。 为此，请使用 Azure CLI 命令： az dt rbac assign-role --assignee '<user-email | application-id>' --role owner -n '<your-digital-twins-instance>'

DefaultAzureCredential 支持不同的身份验证机制，并根据它正在执行的环境确定适当的凭据类型。 它会尝试按顺序使用多个凭据类型，直到找到有效的凭据。

代码示例

# DefaultAzureCredential supports different authentication mechanisms and determines the appropriate credential type based of the environment it is executing in.
# It attempts to use multiple credential types in an order until it finds a working credential.

# - AZURE_URL: The URL to the ADT in Azure
url = os.getenv("AZURE_URL")

# DefaultAzureCredential expects the following three environment variables:
# - AZURE_TENANT_ID: The tenant ID in Azure Active Directory
# - AZURE_CLIENT_ID: The application (client) ID registered in the AAD tenant
# - AZURE_CLIENT_SECRET: The client secret for the registered application
credential = DefaultAzureCredential()
service_client = DigitalTwinsClient(url, credential)

关键概念

Azure 数字孪生是一种 Azure IoT 服务，用于创建物理环境的综合性模型。 它可以创建空间智能图，为人员、空间和设备之间的关系和交互建模。 若要详细了解 Azure 数字孪生，请访问 Azure 数字孪生文档。

示例

可以使用客户端库) 示例项目浏览数字孪生 API (。

示例项目演示了以下内容：

• 对客户端进行实例化
• 创建、获取和停用模型
• 创建、查询和删除数字孪生体
• 获取和更新数字孪生体的组件
• 创建、获取和删除数字孪生体之间的关系
• 创建、获取和删除数字孪生体的事件路由
• 将遥测消息发布到数字孪生体和数字孪生体组件

创建、列出、解除授权和删除模型

创建模型

让我们使用以下代码创建模型。 需要传递包含模型列表的数组。

temporary_component = {
    "@id": component_id,
    "@type": "Interface",
    "@context": "dtmi:dtdl:context;2",
    "displayName": "Component1",
    "contents": [
    {
        "@type": "Property",
        "name": "ComponentProp1",
        "schema": "string"
    },
    {
        "@type": "Telemetry",
        "name": "ComponentTelemetry1",
        "schema": "integer"
    }
    ]
}

temporary_model = {
    "@id": model_id,
    "@type": "Interface",
    "@context": "dtmi:dtdl:context;2",
    "displayName": "TempModel",
    "contents": [
    {
        "@type": "Property",
        "name": "Prop1",
        "schema": "string"
    },
    {
        "@type": "Component",
        "name": "Component1",
        "schema": component_id
    },
    {
        "@type": "Telemetry",
        "name": "Telemetry1",
        "schema": "integer"
    }
    ]
}

new_models = [temporary_component, temporary_model]
models = service_client.create_models(new_models)
print('Created Models:')
print(models)

列出模型

使用 list_models 检索所有创建的模型

listed_models = service_client.list_models()
for model in listed_models:
    print(model)

获取模型

将 与模型的唯一标识符一起使用 get_model 以获取特定模型。

# Get a model
get_model = service_client.get_model(model_id)
print('Get Model:')
print(get_model)

解除模型授权

若要取消提交模型，请传入要取消提交的模型的模型 ID。

# Decommission a model
service_client.decommission_model(model_id)

删除模型

若要删除模型，请传入要删除的模型的模型 ID。

# Delete a model
service_client.delete_model(model_id)

创建和删除数字孪生体

创建数字孪生

对于创建孪生体，需要提供数字孪生体的 ID，例如 my_twin 和基于之前创建的模型的应用程序/json 数字孪生体。 可 在此处查看示例 application/json。

digital_twin_id = 'digitalTwin-' + str(uuid.uuid4())
temporary_twin = {
    "$metadata": {
        "$model": model_id
    },
    "$dtId": digital_twin_id,
    "Prop1": 42
}

created_twin = service_client.upsert_digital_twin(digital_twin_id, temporary_twin)
print('Created Digital Twin:')
print(created_twin)

获取数字孪生体

获取数字孪生体非常简单。

get_twin = service_client.get_digital_twin(digital_twin_id)
print('Get Digital Twin:')
print(get_twin)

查询数字孪生

使用 Azure 数字孪生查询存储语言查询数字孪生的 Azure 数字孪生实例。 查询调用支持分页。 下面是如何查询数字孪生体以及如何循环访问结果的示例。

请注意，在实例中的更改反映在查询中之前，可能会有延迟。 有关查询限制的更多详细信息，请参阅 (/azure/digital-twins/how-to-query-graph#query-limitations)

query_expression = 'SELECT * FROM digitaltwins'
query_result = service_client.query_twins(query_expression)
print('DigitalTwins:')
for twin in query_result:
    print(twin)

删除数字孪生体

只需提供数字孪生体的 ID 即可删除数字孪生体，如下所示。

service_client.delete_digital_twin(digital_twin_id)

获取和更新数字孪生组件

更新数字孪生组件

若要更新组件或替换、删除和/或添加数字孪生中的组件属性或子属性，需要对指定数字孪生的组件执行数字孪生的 ID、组件名称和 application/json-patch+json 操作。 下面是有关如何执行此操作的示例代码。

component_name = "Component1"
patch = [
    {
        "op": "replace",
        "path": "/ComponentProp1",
        "value": "value2"
    }
]
service_client.update_component(digital_twin_id, component_name, patch)

获取数字孪生组件

通过提供组件的名称和它所属的数字孪生体的 ID 来获取组件。

get_component = service_client.get_component(digital_twin_id, component_name)
print('Get Component:')
print(get_component)

创建和列出数字孪生关系

创建数字孪生关系

upsert_relationship 在提供数字孪生体的 ID、关系名称（如“contains”）、关系 ID（如“FloorContainsRoom”）和要创建的 application/json 关系的数字孪生体上创建关系。 必须包含键为“$targetId”的属性，才能指定关系的目标。 可 在此处找到关系的示例有效负载。

hospital_relationships = [
    {
        "$relationshipId": "BuildingHasFloor",
        "$sourceId": building_twin_id,
        "$relationshipName": "has",
        "$targetId": floor_twin_id,
        "isAccessRestricted": False
    },
    {
        "$relationshipId": "BuildingIsEquippedWithHVAC",
        "$sourceId": building_twin_id,
        "$relationshipName": "isEquippedWith",
        "$targetId": hvac_twin_id
    },
    {
        "$relationshipId": "HVACCoolsFloor",
        "$sourceId": hvac_twin_id,
        "$relationshipName": "controlsTemperature",
        "$targetId": floor_twin_id
    },
    {
        "$relationshipId": "FloorContainsRoom",
        "$sourceId": floor_twin_id,
        "$relationshipName": "contains",
        "$targetId": room_twin_id
    }
]

for relationship in hospital_relationships:
    service_client.upsert_relationship(
        relationship["$sourceId"],
        relationship["$relationshipId"],
        relationship
    )

列出数字孪生关系

list_relationships 和 list_incoming_relationships 分别列出数字孪生体的所有关系和所有传入关系。

relationships = service_client.list_relationships(digital_twint_id)
for relationship in relationships:
    print(relationship)

incoming_relationships = service_client.list_incoming_relationships(digital_twin_id)
for incoming_relationship in incoming_relationships:
    print(incoming_relationship)

创建、列出和删除数字孪生的事件路由

创建事件路由

若要创建事件路由，请提供事件路由的 ID（如“myEventRouteId”）以及包含终结点和可选筛选器的事件路由数据，如下所示。

event_route_id = 'eventRoute-' + str(uuid.uuid4())
event_filter = "$eventType = 'DigitalTwinTelemetryMessages' or $eventType = 'DigitalTwinLifecycleNotification'"
route = DigitalTwinsEventRoute(
    endpoint_name=event_hub_endpoint_name,
    filter=event_filter
)
service_client.upsert_event_route(event_route_id, route)

有关事件路由筛选器语言的详细信息，请参阅“如何管理路由” 筛选器事件文档。

列出事件路由

使用 列出给定事件路由 ID 的特定事件路由或所有事件路由设置选项 list_event_routes。

event_routes = service_client.list_event_routes()
for event_route in event_routes:
    print(event_route)

删除事件路由

删除给定事件路由 ID 的事件路由。

service_client.delete_event_route(event_route_id)

发布数字孪生体的遥测消息

若要发布数字孪生体的遥测消息，需要提供数字孪生体 ID 以及需要更新的遥测的有效负载。

digita_twin_id = "<DIGITAL TWIN ID>"
telemetry_payload = '{"Telemetry1": 5}'
service_client.publish_telemetry(
    digita_twin_id,
    telemetry_payload
)

还可以为数字孪生体中的特定组件发布遥测消息。 除了数字孪生体 ID 和有效负载外，还需要指定目标组件 ID。

digita_twin_id = "<DIGITAL TWIN ID>"
component_name = "<COMPONENT_NAME>"
telemetry_payload = '{"Telemetry1": 5}'
service_client.publish_component_telemetry(
    digita_twin_id,
    component_name,
    telemetry_payload
)

疑难解答

日志记录

此库使用标准日志记录库进行日志记录。 有关 HTTP 会话 (URL、标头等的基本信息，) 在 INFO 级别记录。

可以使用 logging_enable 关键字参数在客户端上启用详细的调试级别日志记录，包括请求/响应正文和未处理标头：

客户端级别日志记录

import sys
import logging

# Create logger
logger = logging.getLogger('azure')
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(stream=sys.stdout)
logger.addHandler(handler)

# Create service client and enable logging for all operations
service_client = DigitalTwinsClient(url, credential, logging_enable=True)

按操作级别日志记录

import sys
import logging

# Create logger
logger = logging.getLogger('azure')
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(stream=sys.stdout)
logger.addHandler(handler)

# Get model with logging enabled
model = service_client.get_model(model_id, logging_enable=True)

可选配置

可选关键字参数可以在客户端和每个操作级别传入。 azure-core 参考文档 介绍了重试、日志记录、传输协议等的可用配置。

后续步骤

提供反馈

如果遇到 bug 或有建议，请创建问题。

贡献

本项目欢迎贡献和建议。 大多数贡献要求你同意贡献者许可协议 (CLA)，并声明你有权（并且确实有权）授予我们使用你的贡献的权利。 有关详细信息，请访问 https://cla.microsoft.com 。

提交拉取请求时，CLA 机器人将自动确定你是否需要提供 CLA，并相应地修饰 PR（例如标签、注释）。 直接按机器人提供的说明操作。 只需使用 CLA 对所有存储库执行一次这样的操作。

此项目采用了 Microsoft 开放源代码行为准则。 有关详细信息，请参阅行为准则常见问题解答，或如果有任何其他问题或意见，请与 联系。