Event Analytics Client for Python¶

Introduction¶

The Event Analytics Service is used for analyzing event data. The service identifies significant dependencies and helps to get a better understanding of the system's internal processes using statistical analysis.

Further implementation of the IOT EventAnalytics 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

Event Operations services¶

The event operations client analyzes event data statistically to identify the most frequent events.

Client name: EventOperationsClient

Find Top Events¶

This method finds the N most frequently occurring events and returns them sorted by the number of occurrences in descending order.

# 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 EventOperationsClient from eventanalytics module
from eventanalytics import EventOperationsClient

# Import all required models from eventanalytics.models
from eventanalytics import Event, TopEventsInputDataModel, EventsInputModelEventsMetadata, TopEventsRequest

# Create the RestClientConfig and UserToken objects
config = RestClientConfig(proxy_host = "<proxy_host>", proxy_port = <proxy_port>)
credentials = UserToken(authorization = "<bearer_token>")

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

try:
    # Create the Events Input Model Events Metadata object
    metadata = EventsInputModelEventsMetadata("<text>")

    # Create a list of events to be analyzed
    event1 = Event("<time_value_1>", "<event_text_1>", "<text_qc_1>")
    event2 = Event("<time_value_2>", "<event_text_2>", "<text_qc_2>")

    events_list = [event1, event2]

    # Create the TopEventsInputDataModel object to be analyzed
    data = TopEventsInputDataModel(
                events_metadata = metadata,
                events = events_list,
                number_of_top_positions_required = <top_positions_value>
            )

    # Create the request object
    request = TopEventsRequest(data = data)

    # Initiate the API call to find top events
    response = client.top_events(request_object = request)

except MindsphereError as err:
    # Exception Handling

Filter Events¶

This method applies custom filters to simplify the dataset based on the event text.

# Create the EventOperationsClient object as shown above

try:
    # Create the EventsInputModelEventsMetadata object
    metadata = EventsInputModelEventsMetadata("text")

    # Create a list of events to be analyzed
    event1 = Event("2017-10-01T12:00:00.001Z", "INTRODUCING FUEL", 0)
    event2 = Event("2017-10-01T12:02:01.001Z", "Status@Flame On", 0)

    events_list = [event1, event2]

    # Create a list of filter criteria
    filter_list = ["Introduction fuel"]

    # Create the EventSearchInputDataModel object to be analyzed
    data = EventSearchInputDataModel(
                events_metadata = metadata,
                events = events_list,
                filter_list = filter_list
            )

    # Create the request object
    request = FilterEventsRequest(data = data)

    # Initiate the API call to filter events
    response = client.filter_events(request_object = request)

except MindsphereError as err:
    # Exception Handling

Count Events¶

Determines the number of events per time interval for a user defined interval length.

# Create the EventOperationsClient object as shown above

try:
    # Create the EventInputEventsMetadata object
    metadata = EventInputEventsMetadata(event_text_property_name = "text", split_interval = 5000)

    # Create a list of events to be analyzed
    event1 = Event("2017-10-01T12:00:00.001Z", "INTRODUCING FUEL", 0)
    event2 = Event("2017-10-01T12:02:01.001Z", "Status@Flame On", 0)

    events_list = [event1, event2]

    # Create the EventInput object to be analyzed
    data = EventInput(events_metadata = metadata, events = events_list)

    # Create the request object
    request = CountEventsRequest(data = data)

    # Initiate Count Events API call
    response = client.count_events(request_object = request)

except MindsphereError as err:
    # Exception Handling

Remove Duplicate Events¶

This method detects duplicate events within a sliding window of user defined width (example 5,000 ms) and reduces the data set by aggregating duplicate events.

# Create EventOperationsClient object as shown above

try:
    # Create the EventsInputEventsMetadata object
    metadata = EventInputEventsMetadata("text", 5000)

    # Create a list of events to be analyzed
    event1 = Event("2017-10-01T12:00:00.001Z", "INTRODUCING FUEL", 0)
    event2 = Event("2017-10-01T12:02:01.001Z", "Status@Flame On", 0)

    events_list = [event1, event2]

    # Create the EventInput object to be analyzed
    data = EventInput(metadata, events_list)

    # Create the request object
    request = RemoveDuplicateEventsRequest(data = data)

    # Initiate API call to remove events
    response = eventOperationsClient.remove_duplicate_events(request)

except MindsphereError as err:
    # Exception Handling

Pattern Operations Services¶

The Pattern Operations client analyzes the event data statistically to detect event patterns.

Client name: PatternOperationsClient

Find Event Patterns¶

Searches for user defined patterns in a list of events and detects all events matching the specified pattern(s).

# 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 PatternOperationsClient from eventanalytics module
from eventanalytics.clients.pattern_operations_client import PatternOperationsClient

# Import all required models from eventanalytics.models
from eventanalytics.models import PatternMatchingInputDataModel, EventsInputModelEventsMetadata, EventInput, PatternDefinition, MatchingPattern, Event, MatchPatternsOverEventsRequest

# Create the RestClientConfig and UserToken objects
clientConfig = RestClientConfig(proxy_host = "<proxy_host>", proxy_port = <proxy_port>)
credentials = UserToken(authorization = "<bearer_token>")

# Create the PatternOperationsClient object using the RestClientConfig and UserToken objects
client = PatternOperationsClient(rest_client_config = clientConfig, mindsphere_credentials = credentials)

try:
    # Create the EventsInputModelEventsMetadata object
    metadata = EventsInputModelEventsMetadata("<text>")

    # Create a list of events to be analyzed
    event_1 = Event("2017-10-01T12:00:00.001Z", "INTRODUCING FUEL", 0)
    event_2 = Event("2017-10-01T12:02:01.001Z", "Status@Flame On", 0)

    events_list = [event_1, event_2]
    events_input = EventInput(metadata, events_list)

    # Create a list of non events to be analyzed
    non_events = ["Error 2.. occurred", "STOPPING ENGINE"]

    # Create a list of patterns to be analyzed
    matching_pattern_1 = MatchingPattern("INTRODUCING FUEL", 1, 2)
    matching_pattern_2 = MatchingPattern("Status@Flame On", 0, 1)

    pattern_list = [matching_pattern_1, matching_pattern_2]

    pattern_def = PatternDefinition()
    pattern_def.pattern = pattern_list
    pattern_def_list = [pattern_def]

    max_pattern_interval = 200

    # Create the PatternMatchingInputDataModel object to be analyzed
    data = PatternMatchingInputDataModel(max_pattern_interval, pattern_def_list, non_events, events_input)

    # Create the request object
    request = MatchPatternsOverEventsRequest(data = data)

    # Initiate the API call to match patterns
    response = client.match_patterns_over_events(request)

except MindsphereError as err:
    # Exception Handling