In this tutorial, we are going to show you how to build an anomaly detection solution to the vibration data collected from the bearing sensors of two motors step by step. The purpose is to monitor the sensor data continuously and identify any unusual patterns that might indicate motor wear, bearing failure, or other operational issues.

We are going to use an Isolation Forest model, a machine learning technique specifically designed to isolate anomalies by identifying data points that behave differently from the majority. The model processes the sensor data to detect deviations, enabling us to identify potential issues early and take corrective action before they result in equipment failure.

By running this anomaly detection model, we aim to improve the reliability and performance of the motors by predicting failures and scheduling maintenance more effectively. The objective is to calculate in real time the probability to encounter an anomaly on our motors. We would like also to have a unique solution to edit and maintain for all our similar motors.

Prerequisite

The tutorial focuses on a motor's digital twin, specifically its bearing component based on our sample data representing vibration sensors and associated digital twin.

Your environment should have the sample data flowing (mt1_bearing1, mt1_bearing2, ... mt2_bearing4) and your twin structure should look like with all sensors attached respectively to bearings on motor one and two:

If not already the case, please check the following articles to achieve those prerequisite.

How to stream and loop the relevant sample data
Follow how to creating and connecting Twin Units

Create a Template

The first step involves creating a Template with properties such as datapoints, JSON objects, floats, etc. The template helps organise everything in one place.

Since we have two motors to apply the anomaly detection model, the template allows us to use a single pipeline to apply the model to both motors efficiently. A template named "my_example_factory_template" could include the following:

Datapoints: Bearing1, Bearing2, Bearing3, Bearing4, bearing_anomaly
Variable: Threshold
JSON: Bearing_properties

Using the Python Toolkit, you can register a template using the following logic:

wizata_dsapi.api().upsert_template(
    key="my_example_factory_template",
    name="My Example Factory Template"
)

wizata_dsapi.api().add_template_property(
    template='my_example_factory_template',
    property_name='Bearing1'
)

wizata_dsapi.api().add_template_property(
    template='my_example_factory_template',
    property_name='Bearing2'
)

wizata_dsapi.api().add_template_property(
    template='my_example_factory_template',
    property_name='Bearing3'
)

wizata_dsapi.api().add_template_property(
    template='my_example_factory_template',
    property_name='Bearing4',
    property_type='datapoint'
)

wizata_dsapi.api().add_template_property(
    template='my_example_factory_template',
    property_name='bearing_anomaly',
    property_type='datapoint'
)

wizata_dsapi.api().add_template_property(
    template='my_example_factory_template',
    property_name='Threshold',
    property_type='float'
)

wizata_dsapi.api().add_template_property(
    template='my_example_factory_template',
    property_name='Bearing_properties',
    property_type='json'
)

For anomaly detection model we have to create a logical datapoint as "mt1_bearing_anomaly", before registering the twin we should have that datapoint in the platform then we can map it:

datapoint = wizata_dsapi.DataPoint( hardware_id="mt1_bearing_anomaly",name="mt1_bearing_anomaly")
wizata_dsapi.api().create(datapoint)

Once your template is created, the next step is to register a Twin. Each asset will have fixed values for variables and/or mapped datapoints. When querying a registered Twin based on a template, there is no need to know the specific details of each datapoint.

To register the Twin, you must create a mapping between the template property names and their corresponding values:

wizata_dsapi.api().register_twin(
    template="my_example_factory_template",
    twin="motor_1",
    properties={
        "Bearing1": "mt1_bearing1",
        "Bearing2": "mt1_bearing2",
        "Bearing3": "mt1_bearing3",
        "Bearing4": "mt1_bearing4",
        "bearing_anomaly":"mt1_bearing_anomaly",
        "Threshold": 3.1, 
        "Bearing_properties": {"bearing_pressure_threshold": 89.5,
                               "bearing_temperature_threshold": 67}
      
    }
)

We should apply the same registering twin for motor two as well.

wizata_dsapi.api().register_twin(
    template="my_example_factory_template",
    twin="motor_2",
    properties={
        "Bearing1": "mt2_bearing1",
        "Bearing2": "mt2_bearing2",
        "Bearing3": "mt2_bearing3",
        "Bearing4": "mt2_bearing4",
        "bearing_anomaly":"mt2_bearing_anomaly",
        "Threshold": 3.8, 
        "Bearing_properties": {"bearing_pressure_threshold": 88.5,
                               "bearing_temperature_threshold": 66}
      
    }
)

We can create a template directly from the UI. Please see this article to create the template manually:

Create a Template from the UI

At the end of this process, we have a template that includes two Digital Twins: Motor 1 and Motor 2, as shown below.:

Create a Pipeline

After creating the template, the next step is to develop a Pipeline to integrate the solution. A pipeline consists of a series of steps necessary for the project, including querying, transformation, machine learning (ML) models, plots, and writing results.

import uuid
# Generate pipeline
pipeline = wizata_dsapi.Pipeline(  # create a pipeline 
    key="bearing_anomaly_pipeline",
    template_id=uuid.UUID("61d63f8b-6142-43bc-bb98-ac146076601b") # this template id is important and it's coming from the template link 
    # the end of the link is giving us the template id
)

Query

First step for the pipeline is having a query :

pipeline.add_query(  #query step
    wizata_dsapi.Request(
        datapoints =["Bearing1","Bearing2","Bearing3","Bearing4"],
        start="now-1d",
        end="now",
         agg_method="mean",
        interval=60000
    ),
    df_name="df"
)

Transformation

For the anomaly detection solution, you will create the transformation, model, and plot scripts, one by one. The process begins with the transformation script :

def bearing_preprocessing(context:wizata_dsapi.Context):
    df = context.dataframe # defines dataframes, this will come from the query step of pipeline
    df = df[(df > 0.05).any(axis=1)] # drop downtimes
    df = df.dropna() # drop null values
    return df 

wizata_dsapi.api().upsert(bearing_preprocessing) # this save the transformation script into the platform

pipeline.add_transformation( # script step
    script="bearing_preprocessing",
    input_df_names=['df'],
    output_df_names=['df_processed']
)

Training Model

The transformation script removes downtimes and eliminates null values, which is the first step in preparing the data for the machine learning model.

So now the next step is creating a model script :

def anomaly_detection_model(context:wizata_dsapi.Context):
    df = context.dataframe
    from sklearn.ensemble import IsolationForest

    # Create and fit the model
    iso_forest = IsolationForest(contamination=0.01, random_state=42)  # Adjust contamination based on your data
    iso_forest = iso_forest.fit(df)
    context.set_model(iso_forest,df.columns)
wizata_dsapi.api().upsert(anomaly_detection_model)

pipeline.add_model(
    config=wizata_dsapi.MLModelConfig(
        train_script='anomaly_detection_model', # this is the model script 
        features=["Bearing1","Bearing2","Bearing3","Bearing4"], # input features for model
        output_append=True, # it will append the prediction results
        output_columns_names=["bearing_anomaly"], # prediction output column's name, it will return model.predict(df["features"])
        function="predict", # it can be, transform, predict etc.
        model_key="ad_model", # this is the name of model as saved in the platform
        by_twin=True # it will save the model seperately by twin. like ad_model_motor1,ad_model_motor1

    ),
    input_df='df_processed',
    output_df='df_predict'
)

For the model step, we have two twins, and we need to run the pipeline for both. This will result in two separate models. If we set by_twin to True, it means the two models will be saved separately, named as ad_model.motor_1 and ad_model.motor_2.

After the model step df_predict will have columns as ["Bearing1", "Bearing2", "Bearing3", "Bearing4", "bearing_anomaly"]

Plot

Next, we will create a plot script to visualse anomalies by marking them with red vertical lines. This step is intended for experimentation purposes only:

def bearing_ad_plot(context:wizata_dsapi.Context):
    df = context.dataframe
    import plotly.graph_objects as go

    # Create the line plot
    fig = go.Figure()

    # Add traces for each sensor
    for column in df.columns[1:-1]:  # Exclude the 'Timestamp' and 'anomaly' columns
        fig.add_trace(go.Scatter(x=df.index, y=df[column], mode='lines', name=column))

    # Add vertical lines for anomalies
    anomaly_times = df[df['bearing_anomaly'] == -1].index
    for anomaly_time in anomaly_times:
        fig.add_vline(x=anomaly_time, line=dict(color='red', width=2))

    # Update layout
    fig.update_layout(
        title="Sensor Data with Anomaly Detection",
        xaxis_title="Timestamp",
        yaxis_title="Sensor Values",
        showlegend=True
    )
    
    context.set_plot(fig, "Bearing Anomaly Detection") #save with the title
wizata_dsapi.api().upsert(bearing_ad_plot)

pipeline.add_plot(df_name="df_predict", script="bearing_ad_plot") # plot step

Write

The final step is to write the results to the platform. For this use case, the result to be written is the "bearing_anomaly."

pipeline.add_writer( #writer step
    config=wizata_dsapi.WriteConfig(
        datapoints={
            "bearing_anomaly":"bearing_anomaly"
        }
    ),
    input_df='df_predict'
)

After completing all these steps, the next action is to upsert the pipeline to the platform.

wizata_dsapi.api().upsert_pipeline(pipeline)

Please see all the steps for the pipeline here:

import uuid
# Generate pipeline
pipeline = wizata_dsapi.Pipeline(  # create a pipeline 
    key="bearing_anomaly_pipeline",
    template_id=uuid.UUID("61d63f8b-6142-43bc-bb98-ac146076601b") # this template is is important and it's coming from the template link 
    # the end of the link is giving us the template id
)

pipeline.add_query(  #query step
    wizata_dsapi.Request(
        datapoints =["Bearing1","Bearing2","Bearing3","Bearing4"],
        start="now-1d",
        end="now",
        agg_method="mean",
        interval=60000,
    ),
    df_name="df"
)

pipeline.add_transformation( # script step
    script="bearing_preprocessing",
    input_df_names=['df'],
    output_df_names=['df_processed']
)

pipeline.add_model(
    config=wizata_dsapi.MLModelConfig(
        train_script='anomaly_detection_model', # this is the model script 
        features=["Bearing1","Bearing2","Bearing3","Bearing4"], # input features for model
        output_append=True, # it will append the prediction results
        output_columns_names=["bearing_anomaly"], # prediction output column's name, it will return model.predict(df["features"])
        function="predict", # it can be, transform, predict etc.
        model_key="ad_model", # this is the name of model as saved in the platform
        by_twin=True # it will save the model seperately by twin. like ad_model_motor1,ad_model_motor1

    ),
    input_df='df_processed',
    output_df='df_predict'
)

pipeline.add_plot(df_name="df_predict", script="bearing_ad_plot") # plot step

pipeline.add_writer( #writer step
    config=wizata_dsapi.WriteConfig(
        datapoints={
            "bearing_anomaly":"bearing_anomaly"
        }
    ),
    input_df='df_predict'
)

wizata_dsapi.api().upsert_pipeline(pipeline)

This is how the pipeline looks like in the platform:

Testing and Deploying the Pipeline

After having the pipeline in the platform, the next step is to test whether it runs correctly. To do this, we will run the pipeline in Experiment mode. There are two ways to run a pipeline: Experiment mode and Production mode. The difference between them is as follows:

Experiment mode: This mode is essentially for testing whether the pipeline runs correctly. By default, the write step is not executed in this mode. The pipeline goes through the query, transformation, model, and plot steps, and the experiment returns a plot to verify that the pipeline is functioning properly.
Production mode: This mode is used to run the pipeline in real-time. In production mode, the write step is executed, and the key difference is that the model step uses a pre-trained model for prediction rather than training a new model. At the end of the process, the results are written to the platform.

In summary, you should first run the pipeline in Experiment mode to train the model, save it, and view the plot results. After that, you can run the pipeline in Production mode to make predictions and write the output to the platform.

Testing the Pipeline

Now, we will run the pipeline in Experiment mode for testing. This will save the model and return the plot as part of the experiment. First, we create an experiment, and then we execute the pipeline in Experiment mode.

wizata_dsapi.api().upsert_experiment("bearing_anomaly_experiment", "Bearing Anomaly Experiment",pipeline = "bearing_anomaly_pipeline")
execution = wizata_dsapi.api().experiment(
    experiment='bearing_anomaly_experiment',
    pipeline='bearing_anomaly_pipeline',
    twin='motor_1')

When we have run the pipeline you can check out the result in the platfrom AI Lab- Experiments:

This means the pipeline ran successfully, trained the model, and saved it within the platform. We are ready to deploy the model in real time.

Deploy the Pipeline

The next step is to run the pipeline in Production mode for real-time operations. You can run it either directly through wizata_dsapi or using a trigger. The advantage of using a trigger is that it’s easy to set up and allows you to schedule the pipeline to run at specific intervals, such as every 5 minutes. Additionally, when running the pipeline in Production mode, it will automatically write the output to the platform by default.

Running it by api:

production_mode =wizata_dsapi.api().run(
    pipeline='bearing_anomaly_pipeline',
    twin='motor_2')

If the pipeline is executed via a trigger, allowing real-time operation, we can specify an interval for execution. The pipeline will run automatically at each set interval. For this case we will run this pipeline every 5 minutes. Below is an example of how to create the trigger by wizata_dsapi:

trigger = wizata_dsapi.Trigger()
trigger.interval = 300000
trigger.delay = 0
trigger.pipeline_id = 'c7d569ce-5af5-448a-a37d-164fa19166ff'
trigger.template_id = '61d63f8b-6142-43bc-bb98-ac146076601b'
trigger.twin_ids = [
    "893c5611-887c-45d1-acea-59bf189435b1" , "c3f77812-513d-433b-a6d3-8866df0c1e64"
]
wizata_dapi.api().create(trigger)

We can create the same trigger in the platform:

In the scenario described above, the pipeline is scheduled to run every 5 minutes for both Motor 1 and Motor 2. After this step, we can review the execution results to verify whether the pipeline is running properly for both Motor 1 and Motor 2. For this step, you should check the Execution logs page.

Conclusion

In conclusion, we have successfully developed an end-to-end anomaly detection model, demonstrating each step of the process. This included creating a template, building a pipeline, and running the pipeline in real-time.

By leveraging the template, we efficiently applied the anomaly detection model to both motors, collecting the anomaly_bearing datapoints. This streamlined approach ensures that the same pipeline can be used for multiple assets, allowing for real-time monitoring and detection of anomalies across both Motor 1 and Motor 2.

You can check out the bearing_anomaly outputs with the unique datapoint names in the explorer: