適用于 Python 的 Azure Azure Digital Twins Core 用戶端程式庫 - 1.2.0 版

發行項
04/05/2023

此套件包含適用于 Azure Digital Twins API 的 SDK，可存取 Azure Digital Twins 服務來管理對應項、模型、關聯性等。

免責聲明

Python 2.7 的 Azure SDK Python 套件支援已于 2022 年 1 月 1 日結束。如需詳細資訊和問題，請參閱 https://github.com/Azure/azure-sdk-for-python/issues/20691

開始使用

簡介

Azure Digital Twins 是新一代 IoT 解決方案的開發人員平臺，可讓您在雲端中安全地且有效率地建立、執行及管理商務環境的數位標記法。使用 Azure Digital Twins，建立即時作業狀態標記法快速且符合成本效益，而數位標記法會隨時掌握來自 IoT 和其他資料來源的即時資料。如果您不熟悉 Azure Digital Twins，而且想要深入瞭解平臺，請確定您已查看 Azure Digital Twins 官方檔頁面。

如需如何針對 Azure Digital Twins 服務進行程式的簡介，請流覽程式碼撰寫教學課程頁面，以取得簡單的逐步指南。請流覽本教學課程，瞭解如何使用命令列用戶端應用程式與 Azure Digital Twin 實例互動。最後，如需如何建置端對端 Azure Digital Twins 解決方案的快速指南，該解決方案是由您環境中的即時資料所驅動，請確定您已查看此實用指南。

上述指南可協助您開始使用 Azure Digital Twins 的重要元素，例如建立 Azure Digital Twins 實例、模型、對應項圖形等。使用下列範例指南，熟悉可協助您針對 Azure Digital Twins 進行程式設計的各種 API。

如何安裝

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

pip install azure-digitaltwins-core azure-identity

azure-identity 用於 Azure Active Directory 驗證，如下所示。

如何使用

驗證，許可權

若要建立新的數位對應項用戶端，您需要 Azure Digital Twin 實例和認證的端點。針對下列範例， AZURE_URL 必須設定、 AZURE_TENANT_IDAZURE_CLIENT_ID 、和 AZURE_CLIENT_SECRET 環境變數。用戶端需要 TokenCredential 或 ServiceClientCredentials 的實例。在此範例中，我們會說明如何使用一個衍生類別： DefaultAzureCredentials。

注意：若要存取 Digital Twins 服務的資料平面，實體必須獲得許可權。若要這樣做，請使用 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 Digital Twins 是一項 Azure IoT 服務，其可建立完整的實體環境模型。此服務可建立空間智慧圖形，用模型來呈現人員、空間和裝置之間的關聯性與互動情形。您可以流覽 Azure Digital Twins 檔，以深入瞭解 Azure Digital Twins。

範例

您可以使用用戶端程式庫) 範例專案來探索數位對應項 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)

解除委任模型

若要取消認可模型，請傳入您要取消認可之模型的模型識別碼。

# Decommission a model
service_client.decommission_model(model_id)

刪除模型

若要刪除模型，請傳入您要刪除之模型的模型識別碼。

# Delete a model
service_client.delete_model(model_id)

建立和刪除數位對應項

建立數位分身

針對建立對應項，您必須根據稍早建立的模型，提供數位對應項的識別碼，例如 my_twin 和 application/json 數位對應項。您可以在這裡查看範例應用程式/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 Digital Twins 查詢存放區語言查詢 Azure Digital Twins的數位對應項實例。查詢呼叫支援分頁。以下是如何查詢數位對應項以及如何逐一查看結果的範例。

請注意，在實例中的變更反映在查詢中之前，可能會有延遲。如需查詢限制的詳細資訊，請參閱 (/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)

刪除數位對應項

只要提供數位對應項的識別碼，即可刪除數位對應項，如下所示。

service_client.delete_digital_twin(digital_twin_id)

取得和更新數位對應項元件

更新數位對應項元件

若要更新元件，或換句話說，若要在 Digital Twin 中取代、移除和/或新增元件屬性或子屬性，您需要在指定的數位對應項元件上執行數位對應項、元件名稱和應用程式/json-patch+json 作業的識別碼。以下是作法的範例程式碼。

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

取得數位對應項元件

提供其所屬數位對應項的元件名稱和識別碼，以取得元件。

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

建立和列出數位對應項關聯性

建立數位對應項關聯性

upsert_relationship 會在數位對應項識別碼提供的數位對應項上建立關聯性、關聯性名稱，例如「contains」、關聯性的識別碼，例如「FloorContainsRoom」和要建立的應用程式/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)

建立、列出和刪除數位對應項的事件路由

建立事件路由

若要建立事件路由，請提供事件路由的識別碼，例如「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)

如需事件路由篩選語言的詳細資訊，請參閱篩選事件檔。

列出事件路由

使用列出特定事件路由指定的事件路由識別碼或所有事件路由設定選項 list_event_routes 。

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

刪除事件路由

刪除指定事件路由識別碼的事件路由。

service_client.delete_event_route(event_route_id)

發佈數位對應項的遙測訊息

若要發佈數位對應項的遙測訊息，您必須提供數位對應項識別碼，以及需要更新之遙測的承載。

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

您也可以在數位對應項中發佈特定元件的遙測訊息。除了數位對應項識別碼和承載之外，您還需要指定目標群組件識別碼。

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 關鍵字引數的用戶端上啟用詳細的 DEBUG 層級記錄，包括要求/回應主體和未回應標頭：

用戶端層級記錄

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 核心參考檔說明重試、記錄、傳輸通訊協定等可用的組態。

下一步

提供意見反應

如果您遇到錯誤或有任何建議，請提出問題。

參與

此專案歡迎參與和提供建議。大部分的參與都要求您同意「參與者授權合約 (CLA)」，宣告您有權且確實授與我們使用投稿的權利。如需詳細資料，請前往 https://cla.microsoft.com 。

當您提交提取要求時，CLA Bot 會自動判斷您是否需要提供 CLA，並適當地裝飾 PR (例如標籤、註解)。請遵循 bot 提供的指示。您只需要使用我們的 CLA 在所有存放庫上執行此動作一次。

此專案採用 Microsoft Open Source Code of Conduct (Microsoft 開放原始碼管理辦法)。如需詳細資訊，請參閱管理辦法常見問題集，如有任何其他問題或意見請連絡 opencode@microsoft.com。