Compartir a través de

Biblioteca cliente de Azure Digital Twins Core para Python: versión 1.2.0

  • Artículo

Este paquete contiene un SDK para Azure Digital Twins API para proporcionar acceso al servicio Azure Digital Twins para administrar gemelos, modelos, relaciones, etc.

Declinación de responsabilidades

Los paquetes de Python del SDK de Azure para Python 2.7 finalizaron el 01 de enero de 2022. Para más información y preguntas, consulte https://github.com/Azure/azure-sdk-for-python/issues/20691.

Introducción

Introducción

Azure Digital Twins es una plataforma para desarrolladores para soluciones de IoT de próxima generación que le permiten crear, ejecutar y administrar representaciones digitales del entorno empresarial, de forma segura y eficaz en la nube. Con Azure Digital Twins, la creación de representaciones de estado operativo en vivo es rápida y rentable, y las representaciones digitales permanecen al día con datos en tiempo real de IoT y otros orígenes de datos. Si no está familiarizado con Azure Digital Twins y desea obtener más información sobre la plataforma, asegúrese de consultar la página de documentación oficial de Azure Digital Twins.

Para obtener una introducción sobre cómo programar con el servicio Azure Digital Twins, visite la página del tutorial de codificación para obtener una guía paso a paso sencilla. Visite este tutorial para aprender a interactuar con una instancia de Azure Digital Twin mediante una aplicación cliente de línea de comandos. Por último, para obtener una guía rápida sobre cómo crear una solución de Azure Digital Twins de un extremo a otro controlada por datos activos de su entorno, asegúrese de consultar esta guía útil.

Las guías mencionadas anteriormente pueden ayudarle a empezar a trabajar con elementos clave de Azure Digital Twins, como la creación de instancias de Azure Digital Twins, modelos, grafos gemelos, etc. Use esta guía de ejemplos a continuación para familiarizarse con las distintas API que le ayudarán a programar con Azure Digital Twins.

Cómo instalarlo

Instale [azure-digitaltwins-core][pypi_package_keys] y azure-identity con pip:

pip install azure-digitaltwins-core azure-identity

azure-identity se usa para la autenticación de Azure Active Directory, como se muestra a continuación.

Modo de uso

Autenticación, permiso

Para crear un nuevo cliente de gemelos digitales, necesita el punto de conexión a una instancia y credenciales de Azure Digital Twin. Para los ejemplos siguientes, se deben establecer las AZURE_URLvariables de entorno , AZURE_TENANT_ID, AZURE_CLIENT_IDy AZURE_CLIENT_SECRET . El cliente requiere una instancia de TokenCredential o ServiceClientCredentials. En este ejemplo, se muestra cómo usar una clase derivada: DefaultAzureCredentials.

Nota: Para acceder al plano de datos del servicio Digital Twins, la entidad debe tener permisos. Para ello, use el comando de la CLI de Azure: az dt rbac assign-role --assignee '<user-email | application-id>' --role owner -n '<your-digital-twins-instance>'

DefaultAzureCredential admite distintos mecanismos de autenticación y determina el tipo de credencial adecuado en función del entorno en el que se ejecuta. Intenta usar varios tipos de credenciales en un orden hasta que encuentre una credencial de trabajo.

Código de ejemplo
# 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)

Conceptos clave

Azure Digital Twins es un servicio de Azure IoT que crea modelos completos del entorno físico. Puede crear grafos de inteligencia espacial para modelar las relaciones y las interacciones entre personas, espacios y dispositivos. Para más información sobre Azure Digital Twins, visite la documentación de Azure Digital Twins.

Ejemplos

Puede explorar las API de gemelos digitales (mediante la biblioteca cliente) mediante el proyecto de ejemplos.

El proyecto de ejemplos muestra lo siguiente:

  • Creación de una instancia del cliente
  • Creación, obtención y retirada de modelos
  • Creación, consulta y eliminación de un gemelo digital
  • Obtención y actualización de componentes para un gemelo digital
  • Creación, obtención y eliminación de relaciones entre gemelos digitales
  • Creación, obtención y eliminación de rutas de eventos para gemelos digitales
  • Publicación de mensajes de telemetría en un componente de gemelo digital y gemelo digital

Crear, enumerar, retirar y eliminar modelos

Crear modelos

Vamos a crear modelos con el código siguiente. Debe pasar una matriz que contenga una lista de modelos.

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)

Enumerar modelos

Uso list_models de para recuperar todos los modelos creados

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

Obtener modelo

Use get_model con el identificador único del modelo para obtener un modelo específico.

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

Modelo de retirada

Para descommisionar un modelo, pase un identificador de modelo para el modelo que desea descommisionar.

# Decommission a model
service_client.decommission_model(model_id)

Eliminación de un modelo

Para eliminar un modelo, pase un identificador de modelo para el modelo que desea eliminar.

# Delete a model
service_client.delete_model(model_id)

Creación y eliminación de gemelos digitales

Creación de gemelos digitales

Para Crear gemelo, deberá proporcionar el identificador de un gemelo digital, como my_twin y el gemelo digital application/json basado en el modelo creado anteriormente. Puede ver la aplicación o json de ejemplo aquí.

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)

Obtención de un gemelo digital

La obtención de un gemelo digital es extremadamente fácil.

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

Consulta de los gemelos digitales

Consulte la instancia de Azure Digital Twins para gemelos digitales mediante Azure Digital Twins Almacén de consultas lanaguage. Las llamadas de consulta admiten la paginación. Este es un ejemplo de cómo consultar gemelos digitales y cómo iterar los resultados.

Tenga en cuenta que puede haber un retraso entre antes de que los cambios en la instancia se reflejen en las consultas. Para más información sobre las limitaciones de las consultas, consulte (/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)

Eliminación de gemelos digitales

Elimine un gemelo digital simplemente proporcionando el identificador de un gemelo digital como se indica a continuación.

service_client.delete_digital_twin(digital_twin_id)

Obtención y actualización de componentes de gemelos digitales

Actualización de componentes de gemelos digitales

Para actualizar un componente o en otras palabras para reemplazar, quitar o agregar una propiedad de componente o una subpropiedad dentro de Digital Twin, necesitaría el identificador de un gemelo digital, el nombre del componente y las operaciones application/json-patch+json que se realizarán en el componente del gemelo digital especificado. Este es el código de ejemplo sobre cómo hacerlo.

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

Obtención de componentes de gemelos digitales

Para obtener un componente, proporcione el nombre de un componente y el identificador del gemelo digital al que pertenece.

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

Creación y enumeración de relaciones de gemelos digitales

Creación de relaciones de gemelos digitales

upsert_relationship crea una relación en un gemelo digital proporcionado con el identificador de un gemelo digital, el nombre de la relación como "contains", el identificador de una relación como "FloorContainsRoom" y una relación application/json que se va a crear. Debe contener la propiedad con la clave "$targetId" para especificar el destino de la relación. Las cargas de ejemplo para las relaciones se pueden encontrar aquí.

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
    )

Enumeración de relaciones de gemelos digitales

list_relationships y list_incoming_relationships enumera todas las relaciones y todas las relaciones entrantes respectivamente de un gemelo digital.

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)

Creación, enumeración y eliminación de rutas de eventos de gemelos digitales

Creación de rutas de eventos

Para crear una ruta de eventos, proporcione un identificador de una ruta de eventos, como "myEventRouteId" y datos de ruta de eventos que contengan el punto de conexión y un filtro opcional, como el ejemplo que se muestra a continuación.

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)

Para obtener más información sobre el lenguaje de filtro de ruta de eventos, consulte la documentación sobre cómo administrar rutas de los eventos de filtro.

Enumerar rutas de eventos

Enumere una ruta de eventos específica dada el identificador de ruta de evento o todas las opciones de configuración de rutas de eventos con list_event_routes.

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

Eliminar rutas de eventos

Elimine una ruta de eventos dada el identificador de ruta de evento.

service_client.delete_event_route(event_route_id)

Publicación de mensajes de telemetría para un gemelo digital

Para publicar un mensaje de telemetría para un gemelo digital, debe proporcionar el identificador de gemelo digital, junto con la carga en la que la telemetría necesita la actualización.

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

También puede publicar un mensaje de telemetría para un componente específico en un gemelo digital. Además del identificador y la carga del gemelo digital, debe especificar el identificador del componente de destino.

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
)

Solución de problemas

Registro

Esta biblioteca usa la biblioteca de registro estándar para el registro. La información básica sobre las sesiones HTTP (direcciones URL, encabezados, etc.) se registra en el nivel INFO.

El registro detallado de nivel DEBUG, incluidos los cuerpos de solicitud/respuesta y los encabezados no aprobados, se puede habilitar en un cliente con el argumento de palabra clave logging_enable:

Registro de nivel de cliente

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)

Registro de nivel por operación

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)

Configuración opcional

Los argumentos de palabra clave opcionales se pueden pasar en el nivel de cliente y por operación. En la documentación de referencia de azure-core se describen las configuraciones disponibles para reintentos, registro, protocolos de transporte, etc.

Pasos siguientes

Envío de comentarios

Si encuentra errores o tiene sugerencias, cree una incidencia.

Contribuciones

Este proyecto agradece las contribuciones y sugerencias. La mayoría de las contribuciones requieren que acepte un Contrato de licencia para el colaborador (CLA) que declara que tiene el derecho a concedernos y nos concede los derechos para usar su contribución. Para más detalles, visite https://cla.microsoft.com.

Cuando se envía una solicitud de incorporación de cambios, un bot de CLA determinará de forma automática si tiene que aportar un CLA y completar la PR adecuadamente (por ejemplo, la etiqueta, el comentario). Solo siga las instrucciones que le dará el bot. Solo será necesario que lo haga una vez en todos los repositorios con nuestro CLA.

Este proyecto ha adoptado el Código de conducta de Microsoft Open Source. Para más información, consulte las preguntas más frecuentes del código de conducta o póngase en contacto con opencode@microsoft.com si tiene cualquier otra pregunta o comentario.