Skip to content

MindConnect Client for Python

Introduction

The MindConnect Python client allows you to interact with MindConnect Service APIs which enables shop floor devices to send data securely and reliably to Insights Hub. Refer to MindConnect for more information about the service.

Further implementation of the MindConnect SDK library has been shown in a sample project that you can download and test in local or on Insights Hub application. Please refer to this repository: industrial-iot-python-sdk-examples

Hint

Placeholders in the following samples are indicated by angular brackets < >.

Mindconnect Operations

Client Name: DiagnosticActivationsClient

Gets diagnostic activations

Get diagnostic activations. Agents are allowed to get their own activation. Users with sufficient scopes are allowed to get all activations in the same tenant as in the token.

# Import the RestClientConfig and UserToken from mindsphere_core module
from mindsphere_core import RestClientConfig
from mindsphere_core import UserToken

# Import the MindsphereError from mindsphere_core.exceptions module
from mindsphere_core.exceptions import MindsphereError

# Import the DiagnosticActivationsClient from mindconnect module
from mindconnect import DiagnosticActivationsClient


# Create the DiagnosticActivationsClient object using the RestClientConfig and UserToken objects
diagnosticActivationsClient = DiagnosticActivationsClient(rest_client_config = config, mindsphere_credentials = credentials)

try:
    # Create the request object
    request_object = DiagnosticActivationsGetRequest()

    # Initiate the API call
    response = diagnosticActivationsClient.diagnostic_activations_get(request_object)

except MindsphereError as err:
    # Exception Handling

Creates a new diagnostic activation

Create a new diagnostic activation. Agents are allowed to create activation for itself only. Users with sufficient scopes are allowed to create activations of the agents in the same tenant as in the token.

try:
    # Create the request object
    diagnosticActivation = DiagnosticActivation(
            agent_id="<agent_id>",
            status="ACTIVE"
        )
    request_object = DiagnosticActivationsPostRequest(
            diagnostic_activation=diagnosticActivation
        )

    # Initiate the API call
    response = diagnosticActivationsClient.diagnostic_activations_post(request_object)

except MindsphereError as err:
    # Exception Handling

Deletes a diagnostic activation

Delete a diagnostic activation. Agents are allowed to delete their own activation. Users with sufficient scopes are allowed to delete any activation in the same tenant as in the token.

try:
    # Create the request object
    request_object = DiagnosticActivationsIdDeleteRequest(
            id="<agent_id>"
        )

    # Initiate the API call
    response = diagnosticActivationsClient.diagnostic_activations_id_delete(request_object)

except MindsphereError as err:
    # Exception Handling

Gets a diagnostic activation by ID

Get a diagnostic activation with given id. Agents are allowed to get their own activation. Users with sufficient scopes are allowed to get any activation in the same tenant as in the token.

try:
    # Create the request object
    request_object = DiagnosticActivationsIdGetRequest(
            id="<agent_id>"
        )

    # Initiate the API call
    response = diagnosticActivationsClient.diagnostic_activations_id_get(request_object)

except MindsphereError as err:
    # Exception Handling

Get a diagnostic messages of specific activation resource

Get a diagnostic messages of specific activation resource. Agents are allowed to get their own activations' messages. Users with sufficient scopes are allowed to get any activations messages in the same tenant as in the token.

try:
    # Create the request object
    request_object = DiagnosticActivationsIdMessagesGetRequest(
            id="<agent_id>"
        )

    # Initiate the API call
    response = diagnosticActivationsClient.diagnostic_activations_id_messages_get(request_object)

except MindsphereError as err:
    # Exception Handling

Update status of Diagnostic Activation

Get a diagnostic messages of specific activation resource. Agents are allowed to get their own activations' messages. Users with sufficient scopes are allowed to get any activations messages in the same tenant as in the token.

try:
    # Create the request object
    diagnosticActivationStatus = DiagnosticActivationStatus(
            status="INACTIVE"
        )
    request_object = DiagnosticActivationsIdPutRequest(
            id="<agent_id>",
            diagnostic_activation_status=diagnosticActivationStatus
        )

    # Initiate the API call
    response = diagnosticActivationsClient.diagnostic_activations_id_put(request_object)

except MindsphereError as err:
    # Exception Handling

Client Name: DiagnosticInformationClient

Get Diagnostic Information

This operation is deprecated! Use \"/diagnosticActivations/{id}/messages\" instead. Gets Diagnostic Information.

# Import the DiagnosticInformationClient from mindconnect module
from mindconnect.clients.diagnostic_information_client import DiagnosticInformationClient


# Create the DiagnosticInformationClient object using the RestClientConfig and UserToken objects
diagnosticInformationClient = DiagnosticInformationClient(rest_client_config = config, mindsphere_credentials = credentials)

try:
    # Create the request object
    requestObject = DiagnosticInformationGetRequest()

    # Initiate the API call
    response = diagnosticInformationClient.diagnostic_information_get(request_object)

except MindsphereError as err:
    # Exception Handling

Client Name: MappingsClient

Get mappings

Retrieves a list of mappings.

# Import the MappingsClient from mindconnect module
from mindconnect.clients.mapping_client import MappingsClient


# Create the MappingsClient object using the RestClientConfig and UserToken objects
mappingsClient = MappingsClient(rest_client_config = config, mindsphere_credentials = credentials)

try:
    # Create the request object
    request_object = DataPointMappingsGetRequest()

    # Initiate the API call
    response = mappingsClient.data_point_mappings_get(request_object)

except MindsphereError as err:
    # Exception Handling

Get a mapping by id

Gets a mapping by given id.

try:
    # Create the request object
    request_object = DataPointMappingsIdGetRequest(
            id="<data_point_mapping_id>"
        )

    # Initiate the API call
    response = mappingsClient.data_point_mappings_id_get(request_object)

except MindsphereError as err:
    # Exception Handling

Create single mapping

Creates a single mapping.

try:
    # Create the request object
    mapping = Mapping(
            agent_id="<agent_id>",
            data_point_id="SDKDP13",
            entity_id="<entity_id>",
            property_set_name="TyreTemperature",
            property_name="FLWheel",
            keep_mapping=False
        )
    request_object = DataPointMappingsPostRequest(
            mapping=mapping
        )

    # Initiate the API call
    response = mappingsClient.data_point_mappings_post(request_object)

except MindsphereError as err:
    # Exception Handling

Delete a mapping

Deletes a mapping.

try:
    # Create the request object
    request_object = DataPointMappingsIdDeleteRequest(
            id="<data_point_mapping_id>"
        )

    # Initiate the API call
    response = mappingsClient.data_point_mappings_id_delete(request_object)

except MindsphereError as err:
    # Exception Handling

Client Name: RecordRecoveryClient

Re-play a recoverable record

Re-play a recoverable record.

# Import the RecordRecoveryClient from mindconnect module
from mindconnect.clients.record_recovery_client import RecordRecoveryClient


# Create the RecordRecoveryClient object using the RestClientConfig and UserToken objects
recordRecoveryClient = RecordRecoveryClient(rest_client_config = config, mindsphere_credentials = credentials)

try:
    # Create the request object
    request_object = RecoverableRecordsIdReplayPostRequest(
            id="<agent_id>"
        )

    # Initiate the API call
    response = recordRecoveryClient.recoverable_records_id_replay_post(request_object)

except MindsphereError as err:
    # Exception Handling

Get all recoverable records

Gets all recoverable records.

try:
    # Create the request object
    request_object = RecoverableRecordsGetRequest()

    # Initiate the API call
    response = recordRecoveryClient.recoverable_records_get(request_object)

except MindsphereError as err:
    # Exception Handling

Gets download link of record payload.

try:
    # Create the request object
    request_object = RecoverableRecordsIdDownloadLinkGetRequest(
            id="<agent_id>"
        )

    # Initiate the API call
    response = recordRecoveryClient.recoverable_records_id_download_link_get(request_object)

except MindsphereError as err:
    # Exception Handling

Delete a recoverable record

Deletes a recoverable record.

try:
    # Create the request object
    request_object = RecoverableRecordsIdDeleteRequest(
            id="<agent_id>"
        )

    # Initiate the API call
    response = recordRecoveryClient.recoverable_records_id_delete(request_object)

except MindsphereError as err:
    # Exception Handling

Last update: February 23, 2024

Except where otherwise noted, content on this site is licensed under the Development License Agreement.