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

Get download link of record payload¶

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