How to build a Compute-to-Data algorithm for the AgrospAI data space

Every algorithm starts the same way: you create an Algorithm instance with its Config, and hang the phases on it.

• Algorithm[Input, Result] — generic typing: input and result type.
• Config(custom_input=...) — declares your parameter model.
• The algorithm object must be accessible from src.algorithm — that is what ocean-execute looks for.

from pathlib import Path
from ocean_runner import Algorithm, Config
from .data import InputParameters

# Result type your algorithm produces
type ResultsT = dict[str, ...]

algorithm = Algorithm[InputParameters, ResultsT].create(
    Config(custom_input=InputParameters)
)

@algorithm.run
def run(_) -> ResultsT:
    params = algorithm.job_details.input_parameters
    ...                     # your logic
    return result

@algorithm.save_results
def save(_, result: ResultsT, base: Path):
    ...                     # write to `base` (= /data/outputs)

The user sends parameters as JSON. Turn them into a typed model: that gives you automatic validation and clear error messages before running anything.

• Each field is an attribute with its type.
• With @field_validator you add business rules.
• An invalid input fails fast, without spending compute or downloads.

from typing import Any
from pydantic import BaseModel, field_validator

class InputParameters(BaseModel):
    refcat: str

    @field_validator("refcat", mode="before")
    @classmethod
    def is_valid_refcat(cls, v: Any) -> str | None:
        if v is None:
            return v
        stripped = str(v).strip()
        if stripped and not stripped.isalnum():
            raise ValueError(f"'{stripped}' must be alpha-numeric")
        return stripped

{ "refcat": "25168A010000060000IZ" }

Before running the heavy logic, the runner calls the validation phase. By default it checks that the dataset metadata and files exist.

You can override it with @algorithm.validate to add your own preconditions: are the credentials present? is there a connection to the external service? does the dataset have the expected format?

@algorithm.validate
def validate(_) -> None:
    jd = algorithm.job_details

    # Did the dataset metadata arrive?
    assert jd.metadata, "Missing the dataset DDOs"

    # Are there files to process?
    assert jd.files, "The dataset contains no files"

    # Your own business preconditions...
    algorithm.logger.info("Input validated successfully")

If you don't register this phase, the runner applies a reasonable default validation. In clip-raster the data comes from external services, so the key validation lives inside run itself.

The run function should read like a summary of the algorithm: a few lines that tell the story. The detail lives in specialized modules.

• Read parameters from job_details.input_parameters.
• Delegate each step to a single-purpose function.
• Return an in-memory result; don't write files here.

import geopandas as gpd
from .download import download_product

@algorithm.run
def run(_) -> ResultsT:
    params = algorithm.job_details.input_parameters
    refcat = params.refcat

    # ① parcel geometry (Cadastre service)
    url = f"{CATASTRO_URL_PREFIX}{refcat}"
    gdf = gpd.read_file(url)
    geometry = gdf.geometry.item()

    # ②③ search and download the Sentinel-2 product
    download_product(geometry)

    # ④⑤⑥ clip bands and compute indices
    return compute_indices(geometry)

🔐 Secrets

Never embed credentials in the code or the image. The runner exposes them in /data/transformations/algorithm as key/value pairs you read at runtime.

🧯 Errors

Raise Algorithm.Error for expected failures. The runner catches it in the on_error phase and ends the job cleanly, with a useful message in the logs.

from dotenv import get_key
from ocean_runner import Algorithm

def get_copernicus_token() -> str:
    user = get_key('/data/transformations/algorithm', 'username')
    pwd  = get_key('/data/transformations/algorithm', 'password')
    if not user or not pwd:
        raise Algorithm.Error('Missing credentials')
    return request_copernicus_token(user, pwd)

The save_results phase receives the result from run and the output base path. It is the only place where you write files the user will receive.

• Receives (_, result, base) — base already points to /data/outputs.
• Choose a clear format: image, CSV, JSON, GeoTIFF…
• Keep the separation: run computes, save persists.

from pathlib import Path
from .raster import save_as_img

@algorithm.save_results
def save(_, result: ResultsT, base: Path):
    out_path = base / 'indices.png'   # /data/outputs/indices.png
    save_as_img(result, out_path)

Why separate run and save? Because it lets you change the output format without touching the logic, and test the computation without writing to disk.

Before publishing, test the image exactly as it will run on the portal. The trick is to mount your _data/ folder at the paths the algorithm expects.

• Inputs are mounted read-only (:ro).
• outputs and logs read/write (:rw).
• Mounting src/ lets you iterate without rebuilding the image.

services:
  algorithm:
    build: .
    entrypoint: 'ocean-execute --base-dir /data'
    volumes:
      - ./_data/inputs:/data/inputs:ro
      - ./_data/ddos:/data/ddos:ro
      - ./_data/transformations:/data/transformations:ro
      - ./_data/outputs:/data/outputs:rw
      - ./_data/logs:/data/logs:rw
      - ./algorithm/src:/algorithm/src/   # iterate without rebuild

docker compose up --build   # first time
docker compose up           # changes only in src/