How to Create Event-Based Alerts

Create the alert on Corva

An alert has to be first created on the Corva UI which will be triggered from the backend app.

1. Login to Corva and click the alerts tab in the top left part of the page

   

2. Go to alert definitions page by clicking on the pencil icon

   

3. Create a new alert “From Blank”

   

4. Toggle to advanced mode and fill in the alert details

   1. Alert name - Name to be displayed on the alert fired
   2. Alert description - Description shown on the alert fired
   3. Check Type - Select “Event Based”. This is what lets you trigger the alert from a backend app.
   4. Level - The criticality of the alert. Only used for filtering and UI
   5. Identifier - A UNIQUE text value which will be used by the backend to tell Corva which alert to fire
   6. Time Range & Reopen Interval - Used to control if subsequent alerts are merged or a new notification is triggered.

   

This document provides a step-by-step guide on how to create simple event-based alerts using Python and the Corva API. The example code demonstrates how to implement Yellow and Red alert conditions for drilling operations.

Table of Contents

• How to Create Event-Based Alerts
  • Create the alert on Corva
  • Table of Contents
  • Introduction
  • Prerequisites
  • Lambda Function
  • Yellow Alert Logic
  • Red Alert Logic
  • Trigger Alert Function
  • Conclusion

Introduction

This guide explains how to implement a scheduled lambda function that checks simple conditions in drilling data and triggers alerts based on those conditions. The implementation includes:

• Fetching data using the get_dataset method.
• Checking conditions for Yellow and Red alerts.
• Triggering alerts via the Corva API.

Prerequisites

Before you begin, ensure you have the following:

• Access to the Corva API.
• Python environment set up with necessary libraries.
• Basic understanding of Python and event-driven programming.
• Special permission for your DC app from Corva admin.

Lambda Function

The entry point for the scheduled event is the lambda_handler function. This function fetches data, checks alert conditions, and triggers alerts if conditions are met.

from corva import Api, Cache, Logger, ScheduledDataTimeEvent, scheduled
import json

@scheduled
def lambda_handler(event: ScheduledDataTimeEvent, api: Api, cache: Cache):
    """Scheduled lambda function to check and trigger alerts."""
    identifier_yellow = "example_alert_yellow"
    identifier_red = "example_alert_red"
    asset_id = event.asset_id
    start_time = event.start_time
    end_time = event.end_time

    Logger.info(f"Fetching data for asset_id: {asset_id} from {start_time} to {end_time}")

    # Fetch data from the dataset
    records = api.get_dataset(
        provider="corva",
        dataset="wits",
        query={
            'asset_id': asset_id,
            'timestamp': {
                '$gte': start_time,
                '$lte': end_time,
            }
        },
        sort={'timestamp': 1},
        limit=500
    )

    Logger.info(f"Fetched {len(records)} records")

    # Print the first record for inspection
    if records:
        Logger.info(f"First record: {json.dumps(records[0], indent=2)}")

    # Iterate through records and trigger alerts if conditions are met
    for i, record in enumerate(records):
        Logger.info(f"Processing record {i + 1}/{len(records)} with timestamp {record['timestamp']}")
        if check_yellow_alert(record):
            Logger.info(f"Yellow alert condition met for record {i + 1}")
            trigger_alert(api=api, identifier=identifier_yellow, asset_id=asset_id)
            if check_red_alert(record):
                Logger.info(f"Red alert condition met for record {i + 1}")
                trigger_alert(api=api, identifier=identifier_red, asset_id=asset_id)

Yellow Alert Logic

The check_yellow_alert function checks the conditions for triggering a Yellow alert. It ensures the data state contains "In Slips".

def check_yellow_alert(data: dict) -> bool:
    """
    Check the conditions for the Yellow alert based on the event data.

    :param data: Dictionary containing the event data
    :return: True if the conditions for the Yellow alert are met, otherwise False
    """
    try:
        # Simple condition: Check if state contains 'DRILLING'
        if "In Slips" in data["data"]["state"]:
            Logger.info("Yellow alert condition: 'In Slips' state found")
            return True
        else:
            Logger.info("Yellow alert condition not met: 'In Slips' state not found")
    except KeyError as e:
        Logger.error(f"Missing key in event data for Yellow alert: {e}")
    
    return False

Red Alert Logic

The check_red_alert function checks the conditions for triggering a Red alert. It ensures the rop value is greater than 100. This check is only performed if the Yellow alert conditions are met.

def check_red_alert(data: dict) -> bool:
    """
    Check the conditions for the Red alert based on the event data.

    :param data: Dictionary containing the event data
    :return: True if the conditions for the Red alert are met, otherwise False
    """
    try:
        # Simple condition: Check if rop is greater than 100
        if data["data"]["rop"] > 100:
            Logger.info("Red alert condition: 'rop' > 100 found")
            return True
        else:
            Logger.info("Red alert condition not met: 'rop' <= 100")
    except KeyError as e:
        Logger.error(f"Missing key in event data for Red alert: {e}")
    
    return False

Trigger Alert Function

The trigger_alert function sends a POST request to the Corva API to trigger an alert when conditions are met.

def trigger_alert(api: Api, identifier: str, asset_id: int) -> None:
    """
    At this point we know that we want to trigger an alert, so we just need to send all
    custom data to the API.
    Ensure that we have the 2 special keys in place: identifier, asset_id

    :param identifier: A unique string that will trigger an alert in Corva
    :param asset_id: Asset id of the well
    :param api: API connection
    :return:
    """

    # Building dict to be posted
    alert_body = {
        "identifier": identifier,
        "asset_id": asset_id,
    }

    # Log the alert body and its type for debugging
    Logger.info(f"Triggering alert with body: {alert_body}")
    Logger.info(f"Type of alert body: {type(alert_body)}")

    # Sending a POST request to trigger the alert.
    res = api.post("/v1/alerts/definitions/trigger/", data=alert_body)

    # Log the status and response content
    Logger.info(f"Response status: {res.status_code}")
    Logger.info(f"Response content: {res.text}")

    # Check for HTTP status
    res.raise_for_status()
    Logger.info(f"Alert triggered successfully: {res.json()}")

Conclusion

This guide provided a comprehensive approach to creating simple event-based alerts using Python and the Corva API. By following the steps and code examples, you can implement Yellow and Red alerts based on specific conditions in drilling data. Customize the logic as needed to fit your specific use case and ensure effective monitoring and alerting in your drilling operations.