Accelerate Digital Twin Data Collection with Azure Digital Twins SDK and Weights & Biases

                      
                     
import os
from azure.identity import DefaultAzureCredential
from azure.digitaltwins.core import DigitalTwinsClient
from wandb import init, log
from typing import Any, Dict

# Configuration
AD_TENANT_ID = os.getenv('AD_TENANT_ID')  # Azure AD Tenant ID
AD_CLIENT_ID = os.getenv('AD_CLIENT_ID')  # Azure AD Client ID
AD_CLIENT_SECRET = os.getenv('AD_CLIENT_SECRET')  # Azure AD Client Secret
DTC_URL = os.getenv('DTC_URL')  # Digital Twins URL

# Initialize Azure Digital Twins Client
credential = DefaultAzureCredential()
client = DigitalTwinsClient(DTC_URL, credential)

# Initialize Weights & Biases
init(project='digital_twins_project', entity='data_collection')

def log_data(twin_id: str, telemetry: Dict[str, Any]) -> None:
    try:
        # Log telemetry data to Azure Digital Twins
        client.update_component(twin_id, {'telemetry': telemetry})
        # Log data to Weights & Biases
        log(telemetry)
        print(f'Successfully logged data for twin ID: {twin_id}')
    except Exception as e:
        print(f'Error logging data for twin ID: {twin_id}: {str(e)}')

if __name__ == '__main__':
    # Example twin ID and telemetry data
    twin_id = 'twin-12345'
    telemetry = {'temperature': 22.5, 'humidity': 60}
    log_data(twin_id, telemetry)