GOR User Task Tutorial#

In this tutorial we will create a complete implementation of a user task that computes Gas-Oil Ratio (GOR) based on gas and oil rate inputs. First you need to create three project files: gor_task_simulation.py, gor_task.yaml and gor_task.py, that should be placed in the same folder.

1. Preparing the YAML file#

When creating a new user task, first thing to do is to design its inputs and outputs. Let’s populate the gor_task.yaml with a YAML definition of input and output parameters.

We are going to add three input parameters (gas rate gauge, oil rate gauge and a flag that specifies the way we treat not-a-number values) and one output parameter (for GOR data that we are going to compute) as follows:

gor_task.yaml

inputs:
  Gas Rate:
    type: dataset
    mandatory: true
    dataType: qg
    isByStep: true
  Oil Rate:
    type: dataset
    mandatory: true
    dataType: qo
    isByStep: true
  Remove NaN:
    type: boolean
    mandatory: true
    currentValue: true

outputs:
  Gas/Oil Ratio:
    type: dataset
    dataType: GOR
    isByStep: true

2. Preparing the test data#

To make a real simulation run we need some test data on KAPPA-Automate that we will use as input for our user task. Since we have two input gauges, we need a well with gas and oil production data. Let’s assume that we have one and the field hierarchy is as follows:

  • There is a field named GOR Tutorial Field

  • This field contains a well named GOR Tutorial Well

  • This well contains two production gauges (imported as step data), one for the gas rate and one for the oil rate, with data-types properly assigned to qg and qo respectfully.

3. Binding the inputs and the outputs#

Now, as we have a definition for our input and output parameters and we also have data on KAPPA-Automate server, we can start setting up the execution (simulation) of our user task. First thing would be to define connection parameters and context of your user task. Let’s introduce and set the ka_server_address variable to point to the URL of your KAPPA-Automate instance,

import kappa_sdk as ka

# Specify the connection configuration
ka_server_address = "https://your.kappa-automate.instance.address.com/"

Next thing is to setup a user task context, by at least specifying field and well ids. Those could be taken from the URL in your web browser when you navigate to the page of your test well (GOR Tutorial Well from the the GOR Tutorial Field field).

import kappa_sdk as ka

# Specify the connection configuration
ka_server_address = "https://your.kappa-automate.instance.address.com/"

# Specify the context (field and well ids)
context = ka.user_tasks.Context()
context.field_id = "55e9a45a-5327-4c44-9dc1-9500c5e95854"  # id of your field
context.well_id = "da92c7f2-a771-4a13-a56b-e53d78ebd4ac"   # if of your well

After defining the context, the next step is to initialize (bind) task inputs, enabling their use in our user task code. While this binding process occurs through the KAPPA-Automate UI during normal deployment and task creation, here we’re implementing it programmatically. To accomplish this, we first create a connection object and then instantiate a user task object using that connection and our previously created context. The user task object provides two important mechanisms for input handling:

  • A list of inputs (kappa_sdk.user_tasks.simulation.UserTaskInstance.inputs) for manually assigning input values

  • A helper method kappa_sdk.user_tasks.simulation.UserTaskInstance.bind_input() that automatically matches and assigns dataset-typed inputs, functioning similarly to the KAPPA-Automate UI when creating instances of deployed user tasks

import kappa_sdk as ka

# Specify the connection configuration
ka_server_address = "https://your.kappa-automate.instance.address.com/"

# Specify the context (field and well ids)
context = ka.user_tasks.Context()
context.field_id = "55e9a45a-5327-4c44-9dc1-9500c5e95854"  # id of your field
context.well_id = "da92c7f2-a771-4a13-a56b-e53d78ebd4ac"   # if of your well

connection = ka.Connection(ka_server_address, verify_ssl=False)
user_task = ka.user_tasks.simulation.UserTaskInstance(connection, context)

# Assign the inputs
user_task.bind_input("Gas Rate", "qg")  # Automatically binds the "Gas Rate" input to the reference gauge of "qg" type
user_task.bind_input("Oil Rate", "qo")  # Automatically binds the "Oil Rate" input to the reference gauge of "qo" type
user_task.inputs["Remove NaN"] = True   # Manually sets the input value

After initializing the inputs we can finally call the kappa_sdk.user_tasks.simulation.UserTaskInstance.run() method that will execute the task. The resulting code of gor_task_simulation.py should be as follows:

import kappa_sdk as ka

# Specify the connection configuration
ka_server_address = "https://your.kappa-automate.instance.address.com/"

# Specify the context (field and well ids)
context = ka.user_tasks.Context()
context.field_id = "55e9a45a-5327-4c44-9dc1-9500c5e95854"  # id of your field
context.well_id = "da92c7f2-a771-4a13-a56b-e53d78ebd4ac"   # if of your well

connection = ka.Connection(ka_server_address, verify_ssl=False)
user_task = ka.user_tasks.simulation.UserTaskInstance(connection, context)

# Assign the inputs
user_task.bind_input("Gas Rate", "qg")  # Automatically binds the "Gas Rate" input to the reference gauge of "qg" type
user_task.bind_input("Oil Rate", "qo")  # Automatically binds the "Oil Rate" input to the reference gauge of "qo" type
user_task.inputs["Remove NaN"] = True   # Manually sets the input value

# Execute the task
user_task.run()

With your empty gor_task.py file in place, you can execute the gor_task_simulation.py script. Upon visiting your GOR Tutorial Well in the KAPPA-Automate UI, you will observe a newly created gauge labeled Gas/Oil Ratio. This gauge creation happens automatically through the kappa_sdk during user task execution. The simulation framework intelligently generates all required outputs for your simulated user task instance. An additional benefit is that these outputs are automatically refreshed each time you run your user task, ensuring clean data for each execution.

4. Writing the user task code#

Now that we have everything ready, we can write code of the GOR user task inside gor_task.py. First thing would be to import a built-in value services of kappa_sdk.user_tasks.Services type, it will allow us to get our input’s values and use different objects and services available for user tasks. Let’s also add some logging.

from kappa_sdk.user_tasks.user_task_environment import services

services.log.info("[GOR task] Executing task..")
services.log.info(f"  [->] Gas Rate = {services.parameters.input['Gas Rate']}")
services.log.info(f"  [->] Oil Rate = {services.parameters.input['Oil Rate']}")
services.log.info(f"  [->] Remove NaN = {services.parameters.input['Remove NaN']}")
services.log.info(f"  [<-] Gas/Oil Ratio = {services.parameters.output['Gas/Oil Ratio']}")

Now if you run this file, you will see the following in the console output (gauge ids will be different and will point to data in your test well):

[INFO] [GOR Task] Executing task..
[INFO]   [->] Gas Rate = 9e2a4fa8-62c3-45f0-98d4-472e7470620e
[INFO]   [->] Oil Rate = a35e6ad4-48f9-49d9-984e-bc2245798fec
[INFO]   [->] Remove NaN = True
[INFO]   [<-] Gas/Oil Ratio = c774aadb-dceb-43c3-9bba-ef3f8e91477e

5. Implementing the GOR calculation#

At this stage, we can implement the core functionality of our task: reading data from input gauges, calculating Gas/Oil Ratio (GOR) values, and writing results to the output gauge.

The initial step requires reading data from two input rate gauges using a synchronized time index. For this purpose, our user task provides access to the kappa_sdk.DataEnumerator service, which handles this synchronization process.

Add the following implementation to your gor_task.py file:

gas_rate_id = services.parameters.input['Gas Rate']
oil_rate_id = services.parameters.input['Oil Rate']
points = services.data_enumerator.to_enumerable([gas_rate_id, oil_rate_id])

points variable now contains a multi-dimensional array that gives you access to interpolated data from both gauges. We can now iterate through the common time index using the following code:

for point in points[1:]:
    gas_rate = point.values[0]
    oil_rate = point.values[1]
    date = point.date

For the sake of this tutorial, let’s add the GOR calculation as a separate function as shown below:

def calculate_gor(gas_rate_value, oil_rate_value):
    return gas_rate_value / oil_rate_value if oil_rate_value != 0 else None

After that we can use it to calculate GOR value for each time index. The loop statement should now look as follows:

for point in points[1:]:
    gas_rate = point.values[0]
    oil_rate = point.values[1]
    date = point.date
    gas_oil_ratio = calculate_gor(gas_rate, oil_rate)

Next, let’s add two lists for x and y values that will store the resulting GOR vector and let’s populate it in the loop (honoring the Remove NaN setting that should instruct us to ignore any points with non-existing gas rate values):

x = list()
y = list()

for point in points[1:]:
    gas_rate = point.values[0]
    oil_rate = point.values[1]
    date = point.date
    gas_oil_ratio = calculate_gor(gas_rate, oil_rate)
    if not services.parameters.input['Remove NaN'] or gas_rate is not None:
        x.append(date)
        y.append(gas_oil_ratio)

After the loop completes, we can use the kappa_sdk.user_tasks.DataCommandService to write the resulting data into the output gauge:

services.data_command_service.replace_data(services.parameters.output['Gas/Oil Ratio'], x, y, points[0].date)

Since we are operating with the rates that are step data, we are skipping the first point when iterating over the list of points (points[1:]), and we are passing it (points[0].date) to the kappa_sdk.user_tasks.DataCommandService.replace_data() method to properly create that first step in the output gauge.

After completing all the implementation steps, your gor_task.py file should contain the following complete solution:

from kappa_sdk.user_tasks.user_task_environment import services

services.log.info("[GOR task] Executing task..")
services.log.info(f"  [->] Gas Rate = {services.parameters.input['Gas Rate']}")
services.log.info(f"  [->] Oil Rate = {services.parameters.input['Oil Rate']}")
services.log.info(f"  [->] Remove NaN = {services.parameters.input['Remove NaN']}")
services.log.info(f"  [<-] Gas/Oil Ratio = {services.parameters.output['Gas/Oil Ratio']}")


def calculate_gor(gas_rate_value, oil_rate_value):
    return gas_rate_value / oil_rate_value if oil_rate_value != 0 else None


gas_rate_id = services.parameters.input['Gas Rate']
oil_rate_id = services.parameters.input['Oil Rate']
points = services.data_enumerator.to_enumerable([gas_rate_id, oil_rate_id])

x = list()
y = list()

for point in points[1:]:
    gas_rate = point.values[0]
    oil_rate = point.values[1]
    date = point.date
    gas_oil_ratio = calculate_gor(gas_rate, oil_rate)
    if not services.parameters.input['Remove NaN'] or gas_rate is not None:
        x.append(date)
        y.append(gas_oil_ratio)

services.data_command_service.replace_data(services.parameters.output['Gas/Oil Ratio'], x, y, points[0].date)

You can now run the user task, then go to the GOR Tutorial Well in KAPPA-Automate UI and see that the newly created output gauge contains the calculated GOR data. After running the user task, go to the GOR Tutorial Well in KAPPA-Automate UI where the newly created output gauge will display the calculated GOR data.