Azure Digital Twins Management with Python

Introduction

As mentioned in my previous post, Azure Digital Twins (ADT) Service is a new way to build next generation IoT solutions. In the first post I show you in a video how to manage ADT instances with the ADT Explorer. In the second post I show how to do mostly the same but using Postman and the ADT Rest API.

ADT has control plane APIs and data plane APIs. The latest is used to manage elements in the ADT instance. In order to use these APIs Microsoft published a .Net (C#) SDK. And SDK is a convenient way to manage instances, since you can easily create applications for your digital twins. If for any reason you prefer to use another language like java, javascript or python, you need to generate your own SDK.

In this post I describe how to autogenerate a Python SDK using the tool Autorest and a swagger file.

In this repo an example of a generated SDK could be found: https://github.com/pauldj54/ADTDataPlane

Autorest

Autorest is a tool to generate client libraries for accessing RESTFul web services. It is available in github: https://github.com/Azure/autorest

Note: Autorest prerequisite is Node.js and the version 10.x is recommended. If you need multiple versions of Node.js I recommend you the tool nvm-windows that can be downloaded from this link: https://github.com/coreybutler/nvm-windows/releases

I will use PowerShell with admin rights for the next steps.

Now let’s select the desired Node.js version:

Node.js setup using nvm

The steps shown in the figure above are the following:

  • Print the current versions
  • Since only v12.x was available then install the 10.x version
  • List the versions available to confirm
  • Change the used version to 10.x
  • Confirm the changes.

Note the prefix “*” marking the selected Node.js version.

To install autorest I followed the steps from the official Azure documentation https://docs.microsoft.com/en-us/azure/digital-twins/how-to-create-custom-sdks:

npm install -g autorest
# run using command 'autorest'
autorest

Generate our own SDK

  1. In order to generate our own SDK the Swagger file with the ADT data plane APIs definition is needed and can be downloaded from here . Please be aware that the “examples” folder is also required, if not Autorest throws an error
  2. Place the downloads in a directory in your computer. I created a folder under my git directory with the name “adt_dataplane_api”.
  3. Open a console and navigate to the directory created in the previous step. Issue the following command: 
autorest --input-file=digitaltwins.json --python --output-folder=ADTApi --add-credentials --azure-arm --package-name=ADTApi --package-version=1.0

Basically you point to the swagger file (digitaltwins.json), select python as the output language, enter an output folder, package name and other details

Python SDK generation using Autorest

4. If everything ran successfully you should see the following output:

Resulting folder layout

Converting our own SDK to a python package

It is very convenient to convert the generated SDK in a python package and include it in the environments as needed. In order to do so, I followed these steps:

  1. Create a setup.py file in the “ADTDataPlane” directory:
setup(
      name='adtdataplane',
      version='0.1',
      description='ADT data plane python SDK',
      author='Azure Digital Twin autogenerated using autorest by Paul Hernandez',
      url='https://github.com/pauldj54/ADTDataPlane',
      packages=find_packages(exclude=['tests*'])
      )


2. Add the auto-generated code to git. I did it in my github using these directions

3. Now we are ready to install our newly generated package 😊

Installing the generated SDK using pip

4. Verify the installation:

Python packages available in this environment

Manage an ADT Instance

Once we have our SDK python package available is time to test it. For this post I registered an AAD application (app registration) and I am using the same ADT instance of the previous post.

  1. Find your Application (client) ID and Directory (tenant) ID:
client and tenant ID in the App registrations (Azure Portal)

2. Create a client secret and write it down:

Create a client secret to access your application

3. Grant the role “Azure Digital Twins Owner (Preview)” to the registered app:

Grant the correspondent role

4. Create a config file in the root directory (or another directory) within you python project and name it for instance settings.json . Hint: secrets and other sensible information will be stored in this file, so make sure you don’t push it to git or your source control.

The file should look like this:

{
    "client_id" : "<your-client-id>",
    "tenant_id" : "<your-tenant-id>",
    "adt_instance_url" : "https://management.azure.com",
    "secret" : "<your-secret>",
    "endpoint" : "https://<your-adt-instance>.api.neu.digitaltwins.azure.net",
    "scope" : ["https://digitaltwins.azure.net/.default"],
    "authority" : "https://login.microsoftonline.com/<your-tenant-id>"
}

5. Create a dtdl sample file to test the code. I crated the file “SampleModel.json” quite similar to the one in the official documentation:

{
  "@id": "dtmi:com:contoso:SampleModelPy;1",
  "@type": "Interface",
  "displayName": "SampleModelPy",
  "contents": [
    {
      "@type": "Relationship",
      "name": "contains"
    },
    {
      "@type": "Property",
      "name": "data",
      "schema": "string"
    }
  ],
  "@context": "dtmi:dtdl:context;2"
}

6. Import the following modules and install them if required:

import msal 
from msrestazure.azure_active_directory import AADTokenCredentials
import adtdataplane 
import logging
from azure.mgmt.consumption.models.error_response import ErrorResponseException
import json

msal is the Microsoft Authentication Library and is the preferred library according to the documentation. AADTokenCreadentials is the class used to build the credentials, adtdataplane is our generated sdk. Some other packages are required by the code.

7. Load the config file and create a confidential client application as follows:

# Load Config file
with open(r"settings.json") as f:
  config = json.load(f)

# Create a preferably long-lived app instance that maintains a token cache.
app = msal.ConfidentialClientApplication(
    config["client_id"], authority=config["authority"],
    client_credential=config["secret"],
    # token_cache=...  # Default cache is in memory only.
                       # You can learn how to use SerializableTokenCache from
                       # https://msal-python.rtfd.io/en/latest/#msal.SerializableTokenCache
    )


8. I used this code snippet from the azure python sdk examples to obtain a token:

# The pattern to acquire a token looks like this.
result = None

# First, the code looks up a token from the cache.
# Because we're looking for a token for the current app, not for a user,
# use None for the account parameter.
result = app.acquire_token_silent(config["scope"], account=None)

if not result:
    logging.info("No suitable token exists in cache. Let's get a new one from AAD.")
    result = app.acquire_token_for_client(scopes=config["scope"])

if "access_token" in result:
    # Call a protected API with the access token.
    print(result["token_type"], result["access_token"])
else:
    print(result.get("error"))
    print(result.get("error_description"))
    print(result.get("correlation_id"))  # You might need this when reporting a bug. 

9. I transform the acquired toke in AAD token credentials and create an SDK client:

credentials = AADTokenCredentials(result)

try:
    client = adtdataplane.AzureDigitalTwinsAPI(credentials = credentials, base_url = config['endpoint'])
    logging.info("Service client created – ready to go")
except ValueError as err:
    print('Client creation failed with error: {0}'.format(err))

10) Now we can load a dtdl model:

# load models
with open(r"models\SampleModel.json") as f:
  dtdl = json.load(f)
dtdl_list = []
dtdl_list.append(dtdl)
try:
  response = client.digital_twin_models.add(model = dtdl_list, raw=True)
  print(response)
except adtdataplane.models.ErrorResponseException as e:
  print(e)

Please notice the model location and modify it accordingly

11. Verify if the model was created

# Verify the model was created
response = client.digital_twin_models.get_by_id('dtmi:com:contoso:SampleModelPy;1')
print(response)

You should see something like this:

DTDL model retrieved

12. We could also verify that the model was correctly upload using the ADT Explorer:

ADT Explorer available DTDL models
ADT Explorer sample model definition

The entire python code:

import msal 
from msrestazure.azure_active_directory import AADTokenCredentials
import adtdataplane 
import logging
from azure.mgmt.consumption.models.error_response import ErrorResponseException
import json

# Load Config file
with open(r"settings.json") as f:
  config = json.load(f)

# Create a preferably long-lived app instance that maintains a token cache.
app = msal.ConfidentialClientApplication(
    config["client_id"], authority=config["authority"],
    client_credential=config["secret"],
    # token_cache=...  # Default cache is in memory only.
                       # You can learn how to use SerializableTokenCache from
                       # https://msal-python.rtfd.io/en/latest/#msal.SerializableTokenCache
    )

# The pattern to acquire a token looks like this.
result = None

# First, the code looks up a token from the cache.
# Because we're looking for a token for the current app, not for a user,
# use None for the account parameter.
result = app.acquire_token_silent(config["scope"], account=None)

if not result:
    logging.info("No suitable token exists in cache. Let's get a new one from AAD.")
    result = app.acquire_token_for_client(scopes=config["scope"])

if "access_token" in result:
    # Call a protected API with the access token.
    print(result["token_type"], result["access_token"])
else:
    print(result.get("error"))
    print(result.get("error_description"))
    print(result.get("correlation_id"))  # You might need this when reporting a bug. 

credentials = AADTokenCredentials(result)

try:
    client = adtdataplane.AzureDigitalTwinsAPI(credentials = credentials, base_url = config['endpoint'])
    logging.info("Service client created – ready to go")
except ValueError as err:
    print('Client creation failed with error: {0}'.format(err))

# load models
with open(r"models\SampleModel.json") as f:
  dtdl = json.load(f)
dtdl_list = []
dtdl_list.append(dtdl)
try:
  response = client.digital_twin_models.add(model = dtdl_list, raw=True)
  print(response)
except adtdataplane.models.ErrorResponseException as e:
  print(e)

# Verify the model was created
response = client.digital_twin_models.get_by_id('dtmi:com:contoso:SampleModelPy;1')
print(response)

Next steps

Even when the scenario presented in this post is extremely basic, now you have a python SDK to manage ADT instances. The benefit is more obvious when you have some data sets and want to populate your twins. In a next post I would like to show you how to write an ingestion program using the python SDK and a data source, most probably a CSV or a JSON file, let’s see what the open data world offers to us.

References

About Paul Hernandez

I'm an Electronic Engineer and Computer Science professional, specialized in Data Analysis and Business Intelligence Solutions. Also a father, swimmer and music lover.
This entry was posted in azure, python and tagged , , , , . Bookmark the permalink.

1 Response to Azure Digital Twins Management with Python

  1. Pingback: Managing Azure IoT Digital Twins with Python – Curated SQL

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out /  Change )

Google photo

You are commenting using your Google account. Log Out /  Change )

Twitter picture

You are commenting using your Twitter account. Log Out /  Change )

Facebook photo

You are commenting using your Facebook account. Log Out /  Change )

Connecting to %s