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
Except where otherwise noted, content on this site is licensed under the Development License Agreement.