Publishing an openEO workflow as a User-Defined-Process (UDP)

In this notebook, we want to show how to create an openEO User Defined Process(UDP). Here, we make use of an apply_dimension process that applies a process to all values along a dimension of a data cube.

The notebook involves a section on creating a concrete datacube, inspecting netCDF downloads, and developing a parameterized version stored as a UDP.

import json
import openeo
import xarray
import matplotlib.pyplot as plt
from utils import *

from openeo.processes import array_create, array_concat, ProcessBuilder
from openeo.api.process import Parameter

If you have a local JupyterLab instance running, you need to install the following libraries first: openeo, xarray, ipyleaflet, shapely and matplotlib:

pip install openeo xarray shapely ipyleaflet matplotlib

Make sure to restart the kernel and refresh the webpage.

# Set some defaults for plots
plt.rcParams["figure.figsize"] = [5.0, 3.0]
plt.rcParams["figure.dpi"] = 75

Connect to the openEO Platform backend (at openeo.cloud) and authenticate with OIDC.

connection = openeo.connect("openeofed.dataspace.copernicus.eu").authenticate_oidc()

Authenticated using refresh token.

Inspect raw data

Load initial data cube with raw S1_GRD_SIGMA0_ASCENDING data for a certain spatio-temporal extent.

center = [46.49, 11.35]
zoom = 15

eoMap = openeoMap(center, zoom)
eoMap.map

bbox = eoMap.getBbox()
print("west", bbox[0], "\neast", bbox[2], "\nsouth", bbox[1], "\nnorth", bbox[3])

spatial_extent = {
    "west": 11.3409 ,
    "east": 11.353779 ,
    "south": 46.48772 ,
    "north": 46.493924,
    "crs": 4326,
}
temporal_extent = ["2023-05-01", "2023-07-01"]

spatial_extent = {
    "west": bbox[0],
    "east": bbox[2],
    "south": bbox[1],
    "north": bbox[3],
    "crs": 4326,
}
temporal_extent = ["2023-05-01", "2023-07-01"]

s1_raw = connection.load_collection(
    collection_id="SENTINEL1_GRD",
    temporal_extent=temporal_extent,
    spatial_extent=spatial_extent,
    bands=["VH", "VV"],
)

Let’s download this data cube synchronously as a netCDF file.

This download command triggers the actual processing on the back-end: it sends the process graph to the back-end and waits for the result. It is a synchronous operation (the download() call blocks until the result is fully downloaded) and because we work on a small spatio-temporal extent, this should only take a couple of seconds.

%%time
s1_raw.download("s1sar-raw.nc")

CPU times: user 12.6 ms, sys: 9.67 ms, total: 22.3 ms
Wall time: 46.5 s

However, batch job-based execution is preferred when it is relatively larger spatial/temporal extent and the process may take some time to process.

ds = xarray.load_dataset("s1sar-raw.nc")
ds

sh: line 1: getfattr: command not found

We got these observations dates:

ds.coords["t"].values

array(['2023-05-03T00:00:00.000000000', '2023-05-10T00:00:00.000000000',
       '2023-05-11T00:00:00.000000000', '2023-05-15T00:00:00.000000000',
       '2023-05-22T00:00:00.000000000', '2023-05-23T00:00:00.000000000',
       '2023-05-27T00:00:00.000000000', '2023-06-03T00:00:00.000000000',
       '2023-06-04T00:00:00.000000000', '2023-06-08T00:00:00.000000000',
       '2023-06-15T00:00:00.000000000', '2023-06-16T00:00:00.000000000',
       '2023-06-20T00:00:00.000000000', '2023-06-27T00:00:00.000000000',
       '2023-06-28T00:00:00.000000000'], dtype='datetime64[ns]')

A quick plot for visual inspection.

ds["VH"].isel(t=0).plot(vmin=0, vmax=0.5)

This section presented a straightforward example of retrieving and analyzing a S1_GRD_SIGMA0_ASCENDING data cube from the backend within a defined area of interest during a specified time frame.

Collect statistics

As part of more detailed processing within the openEO platform, we’ll gather temporal statistics using the apply_dimension process and a collection of statistical measures (minimum, maximum, quantiles, …).

def get_stats(data: ProcessBuilder) -> ProcessBuilder:
    """
    Collect stats for `data` (to be interpreted as an array of values along the "t" dimension).
    We should return a new array with the stats.
    """
    # Put some scalar stats (`min`, `max`, ... return a scalar value) in a new array
    scalar_stats = array_create(
        [
            data.min(),
            data.max(),
            data.mean(),
            data.sd(),
        ]
    )
    # The `quantiles` process returns an array on its own
    quantile_stats = data.quantiles([0.1, 0.5, 0.9])

    # Combine everything in a single array
    return array_concat(array1=scalar_stats, array2=quantile_stats)

s1_stats = s1_raw.apply_dimension(
    process=get_stats,
    dimension="t",
    target_dimension="bands",
)
# Rename band labels, pairing original band names with stat names
s1_stats = s1_stats.rename_labels(
    "bands",
    [
        f"{b}_{s}"
        for b in s1_raw.metadata.band_names
        for s in ["min", "max", "mean", "sd", "q10", "q50", "q90"]
    ],
)

# %%time
# s1_stats.download("s1grd-stats.nc")

# let's try batch job based execution in this process

job = s1_stats.execute_batch(
    title="Sentinel1_GRD_Statistics", outputfile="S1grd-stats.nc"
)


# # Alternatively if you want to seperately save process metadata
# s1_stats = s1_stats.save_result(format="netcdf")
# job = s1_stats.execute_batch(title="Sentinel 1 Statistics")

# # fetch your results

# results = job.get_results()
# results.download_files("output/batch_job")

0:00:00 Job 'cdse-j-2607131822044818b59ca5d7a2247f8b': send 'start'
0:00:03 Job 'cdse-j-2607131822044818b59ca5d7a2247f8b': queued (progress 0%)
0:00:08 Job 'cdse-j-2607131822044818b59ca5d7a2247f8b': queued (progress 0%)
0:00:15 Job 'cdse-j-2607131822044818b59ca5d7a2247f8b': queued (progress 0%)
0:00:23 Job 'cdse-j-2607131822044818b59ca5d7a2247f8b': queued (progress 0%)
0:00:33 Job 'cdse-j-2607131822044818b59ca5d7a2247f8b': queued (progress 0%)
0:00:46 Job 'cdse-j-2607131822044818b59ca5d7a2247f8b': queued (progress 0%)
0:01:01 Job 'cdse-j-2607131822044818b59ca5d7a2247f8b': queued (progress 0%)
0:01:21 Job 'cdse-j-2607131822044818b59ca5d7a2247f8b': running (progress N/A)
0:01:45 Job 'cdse-j-2607131822044818b59ca5d7a2247f8b': running (progress N/A)
0:02:15 Job 'cdse-j-2607131822044818b59ca5d7a2247f8b': running (progress N/A)
0:02:52 Job 'cdse-j-2607131822044818b59ca5d7a2247f8b': running (progress N/A)
0:03:39 Job 'cdse-j-2607131822044818b59ca5d7a2247f8b': finished (progress 100%)

assets = job.get_results().get_assets()
print(assets[0].href)

https://s3.waw3-1.openeo.v1.dataspace.copernicus.eu/openeo-data-prod-waw4-1/batch_jobs/j-2607131822044818b59ca5d7a2247f8b/openEO.nc?X-Proxy-Head-As-Get=true&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=92e5a22641de41939bd6e0089c8504c2%2F20260713%2Fwaw4-1%2Fs3%2Faws4_request&X-Amz-Date=20260713T182545Z&X-Amz-Expires=86400&X-Amz-SignedHeaders=host&X-Amz-Security-Token=eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJyb2xlX2FybiI6ImFybjpvcGVuZW93czppYW06Ojpyb2xlL29wZW5lby1kYXRhLXByb2Qtd2F3NC0xLXdvcmtzcGFjZSIsImluaXRpYWxfaXNzdWVyIjoib3BlbmVvLnByb2Qud2F3My0xLm9wZW5lby1pbnQudjEuZGF0YXNwYWNlLmNvcGVybmljdXMuZXUiLCJodHRwczovL2F3cy5hbWF6b24uY29tL3RhZ3MiOnsicHJpbmNpcGFsX3RhZ3MiOnsiam9iX2lkIjpbImotMjYwNzEzMTgyMjA0NDgxOGI1OWNhNWQ3YTIyNDdmOGIiXSwidXNlcl9pZCI6WyIzZTI0ZTI1MS0yZTlhLTQzOGYtOTBhOS1kNDUwMGU1NzY1NzQiXX0sInRyYW5zaXRpdmVfdGFnX2tleXMiOlsidXNlcl9pZCIsImpvYl9pZCJdfSwiaXNzIjoic3RzLndhdzMtMS5vcGVuZW8udjEuZGF0YXNwYWNlLmNvcGVybmljdXMuZXUiLCJzdWIiOiJvcGVuZW8tZHJpdmVyIiwiZXhwIjoxNzg0MDEwMzQ0LCJuYmYiOjE3ODM5NjcxNDQsImlhdCI6MTc4Mzk2NzE0NCwianRpIjoiNjNmMTliMTAtYzIyNC00YzBlLTk4YzktYmUzNDk3OTI1ODZhIiwiYWNjZXNzX2tleV9pZCI6IjkyZTVhMjI2NDFkZTQxOTM5YmQ2ZTAwODljODUwNGMyIn0.VT7D-agN_LsEmOWF2CYhT5X5KzVPbjvNLDps35XhhgMUKJr8DkmV1g6nNkE0Hwn5Drb0AV46UlNy5qpBrmkeEMe0-w_z5YEGZmpe7TmrobVGTergXODcF_nBIpxJS-YyzNFfLMNjQmX7KKG5ft_4tAQ3nMMzWNJVsIa25-nmpg41qHZYUDj2pO9IQFqBi3o22G29l8vAWBh92Z6FmgDO__7_3Nu43iGwG0KMYXQO8IlgvwoEhFskBExXFHr0aGLVZzB_pK3n3zukhCpn7m1ZAWfV8LUObuNutRMr_VEx89AAy_gZHWEBZ5VlfPuNin3UXFL3ycJjTJ752tsFA9_ATQ&X-Amz-Signature=11f0a73b55323cc5567ca0751f1c804fd6ec9c8600ea5eed28e8de96f28c077d

ds = xarray.load_dataset("S1grd-stats.nc").drop_vars("crs")
ds

sh: line 1: getfattr: command not found

Build S1 SAR stats UDP

Suppose we want to save the above-described algorithm as a User-Defined-Process(UDP). Therefore, in this section, we define the input parameters, define the earlier workflow and then save it as a process.

The only limitation of this approach, is that your workflow needs to be defined as a single process graph. So workflows that require multiple openEO invocations or complex parameter preprocessing won’t work yet. However, thanks to the flexibility of openEO and the ability to include custom code as a UDF, a lot of algorithms can already be defined in a single openEO graph.

import openeo
from openeo.api.process import Parameter
from openeo.processes import array_create, array_concat

Let us define the UDP parameters to allow specifying the spatio-temporal extent.

To make a service available to users, we might want to replace certain fixed values in your process graph with parameters that can be set by the user of your process. This provides you with a parameterised UDP.

temporal_extent = Parameter(
    name="temporal_extent",
    description="The time window to calculate the stats for.",
    schema={"type": "array", "subtype": "temporal-interval"},
    default=["2023-05-01", "2023-07-30"],
)
spatial_extent = Parameter(
    name="spatial_extent",
    description="The spatial extent to calculate the stats for.",
    schema={"type": "object", "subtype": "bounding-box"},
    default={"west": 8.82, "south": 44.40, "east": 8.92, "north": 44.45},
)

s1_raw = connection.load_collection(
    collection_id="SENTINEL1_GRD",
    temporal_extent=temporal_extent,
    spatial_extent=spatial_extent,
    bands=["VH", "VV"],
)
s1_raw = s1_raw.sar_backscatter(coefficient="sigma0-ellipsoid")

# Unlike above, where we defined the `apply_dimension` process
# through a regular python function, we do it here compactily with a single "lambda".
s1_stats = s1_raw.apply_dimension(
    process=lambda data: array_concat(
        array1=array_create([data.min(), data.max(), data.mean(), data.sd()]),
        array2=data.quantiles([0.1, 0.5, 0.9]),
    ),
    dimension="t",
    target_dimension="bands",
)
# Rename band labels, pairing original band names with stat names
s1_stats = s1_stats.rename_labels(
    "bands",
    [
        f"{b}_{s}"
        for b in s1_stats.metadata.band_names
        for s in ["min", "max", "mean", "sd", "q10", "q50", "q90"]
    ],
)

Store this parameterized data cube as a UDP

udp_sar = connection.save_user_defined_process(
    user_defined_process_id="s1_stats",
    process_graph=s1_stats,
    parameters=[temporal_extent, spatial_extent],
    summary="S1 SAR stats",
    description="Calculate S1 SAR stats (min, max, mean, sd, q10, q50, q90). This service can cost an approximate of 3-5 credits per sq km. This cost is based on resource consumpltion only and added-value cost has not been included.",
    public=True,
)

Preflight process graph validation raised: [UpstreamValidationInfo] Backend 'cdse' reported validation errors [ProcessParameterRequired] Process 'n/a' parameter 'spatial_extent' is required.

When saving a process, please note that saved processes are private by default, nonetheless can be used multiple times by an individual. Therefore, to share with a large audience, you will need a public URL that can be achieved once the process is saved as public.

Use the saved UDP in the Python Client

Now, let’s evaluate our freshly created user-defined processes “s1_stats”. We can use datacube_from_process() to create a DataCube from this process and only have to provide concrete temporal and spatial extents

Note: Since the spatial_extent and temporal_extent variable were re-assigned as a paramter definition, you might have lost their value, so please don’t forget to re-define your interested extent in the cell below.

sar = connection.datacube_from_process(
    "s1_stats",
    namespace=public_url,
    temporal_extent=["2023-05-01", "2023-07-30"],
    spatial_extent={"west": 8.82, "south": 44.40, "east": 8.92, "north": 44.45},
)

sar.download("sar_udp.nc")

ds = xarray.load_dataset("sar_udp.nc").drop_vars("crs")
ds

sh: line 1: getfattr: command not found

Publishing your service online

Once the UDP defined above is saved within the openEO platform, a user also has the option to add this service to the openEO Marketplace. To register a User Defined Process (UDP), you must have a public URL for your service. You’ll also need to provide the saved process ID, which can be located within the public URL.

A detailed documentation on the process can be followed here: https://documentation.dataspace.copernicus.eu/Applications/PlazaDetails/ManageService.html#register-and-publish-your-service

Credit Usage

Every openEO user is provided with a specific amount of credits. It’s important to understand that examining data, processes, or creating process graphs like UDP doesn’t cost any credits. However, executing these operations (synchronous or batch) which requires authentication does consume credits based on:

CPU usage (measured in cores per second)
Memory usage (measured in gigabytes per second)
Storage usage (measured in gigabytes per day)
Accessing data from specific layers (e.g., Sentinel Hub or commercial sources)
Additional costs may apply if there’s value-added content, typically provided by third-party services like ‘s1_stats.’ For this reason, when publishing such services online, it’s advisable to include information about their credit consumption in the service description.

You can estimate the credits your service might use by reviewing the job information in the web editor.

With refernce to the documentation available here, you can calculate the possible service usage per square kilometer. Suppose, in my case, for 1 square kilometer, it amounted to 2273 CPU seconds and 5,457,138 megabytes-seconds, equivalent to approximately 0.9 and 1.45 credits, respectively. Hence, the total credits consumed by this process come to approximately 2.35 credits.