Source code for openeo.api.process

from __future__ import annotations

import textwrap
import warnings
from typing import List, Optional, Union


[docs] class Parameter: """ A (process) parameter to build parameterized :ref:`user-defined processes<user-defined-processes>`. Parameter objects can be :ref:`defined <udp-declaring-parameters>` with at least a name and expected schema (e.g. is the parameter a placeholder for a string, a bounding box, a date, ...) and can then be :ref:`used <build_and_store_udp>` with various functions and classes, like :py:class:`~openeo.rest.datacube.DataCube`, to build parameterized user-defined processes. Apart from the generic :py:class:`Parameter` constructor, this class also provides various helpers (class methods) to easily create parameters for common parameter types. :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. :param schema: JSON schema describing the expected data type and structure of the parameter. :param default: default value for the parameter when it's optional. :param optional: toggle to indicate whether the parameter is optional or required. """ # TODO unify with openeo.internal.processes.parse.Parameter? __slots__ = ("name", "description", "schema", "default", "optional") _DEFAULT_UNDEFINED = object() def __init__( self, name: str, description: Optional[str] = None, schema: Union[list, dict, str, None] = None, default=_DEFAULT_UNDEFINED, optional: Optional[bool] = None, ): self.name = name if description is None: # Description is required in openEO API, we are a bit more permissive here. warnings.warn("Parameter without description: using name as description.") description = name self.description = description self.schema = {"type": schema} if isinstance(schema, str) else (schema or {}) # TODO: automatically set `optional` when `default` is set? self.default = default self.optional = optional
[docs] def to_dict(self) -> dict: """ Convert to dictionary for JSON-serialization. """ d = {"name": self.name, "description": self.description, "schema": self.schema} if self.optional is not None: d["optional"] = self.optional if self.default is not self._DEFAULT_UNDEFINED: d["default"] = self.default d["optional"] = True return d
[docs] @classmethod def raster_cube(cls, name: str = "data", description: str = "A data cube.", **kwargs) -> Parameter: """ Helper to easily create a 'raster-cube' parameter. :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. See the generic :py:class:`Parameter` constructor for information on additional arguments (except ``schema``). """ schema = {"type": "object", "subtype": "raster-cube"} return cls(name=name, description=description, schema=schema, **kwargs)
[docs] @classmethod def datacube(cls, name: str = "data", description: str = "A data cube.", **kwargs) -> Parameter: """ Helper to easily create a 'datacube' parameter. :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. See the generic :py:class:`Parameter` constructor for information on additional arguments (except ``schema``). .. versionadded:: 0.22.0 """ schema = {"type": "object", "subtype": "datacube"} return cls(name=name, description=description, schema=schema, **kwargs)
[docs] @classmethod def string( cls, name: str, description: Optional[str] = None, *, values: Optional[List[str]] = None, subtype: Optional[str] = None, format: Optional[str] = None, **kwargs, ) -> Parameter: """ Helper to easily create a 'string' parameter. :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. :param values: Optional list of allowed string values to make this an "enum". :param subtype: Optional subtype of the 'string' schema. :param format: Optional format of the 'string' schema. See the generic :py:class:`Parameter` constructor for information on additional arguments (except ``schema``). """ schema = {"type": "string"} if values is not None: schema["enum"] = values if subtype: schema["subtype"] = subtype if format: schema["format"] = format return cls(name=name, description=description, schema=schema, **kwargs)
[docs] @classmethod def integer(cls, name: str, description: Optional[str] = None, **kwargs) -> Parameter: """ Helper to create an 'integer' parameter. :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. See the generic :py:class:`Parameter` constructor for information on additional arguments (except ``schema``). """ return cls(name=name, description=description, schema={"type": "integer"}, **kwargs)
[docs] @classmethod def number(cls, name: str, description: Optional[str] = None, **kwargs) -> Parameter: """ Helper to easily create a 'number' parameter. :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. See the generic :py:class:`Parameter` constructor for information on additional arguments (except ``schema``). """ return cls(name=name, description=description, schema={"type": "number"}, **kwargs)
[docs] @classmethod def boolean(cls, name: str, description: Optional[str] = None, **kwargs) -> Parameter: """ Helper to easily create a 'boolean' parameter. :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. See the generic :py:class:`Parameter` constructor for information on additional arguments (except ``schema``). """ return cls(name=name, description=description, schema={"type": "boolean"}, **kwargs)
[docs] @classmethod def array( cls, name: str, description: Optional[str] = None, *, item_schema: Optional[Union[str, dict]] = None, **kwargs, ) -> Parameter: """ Helper to easily create parameter with an 'array' schema. :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. :param item_schema: Schema of the array items given in JSON Schema style, e.g. ``{"type": "string"}``. Simple schemas can also be specified as single string: e.g. ``"string"`` will be expanded to ``{"type": "string"}``. See the generic :py:class:`Parameter` constructor for information on additional arguments (except ``schema``). .. versionchanged:: 0.23.0 Added ``item_schema`` argument. """ schema = {"type": "array"} if item_schema: if isinstance(item_schema, str): item_schema = {"type": item_schema} schema["items"] = item_schema return cls(name=name, description=description, schema=schema, **kwargs)
[docs] @classmethod def object( cls, name: str, description: Optional[str] = None, *, subtype: Optional[str] = None, **kwargs ) -> Parameter: """ Helper to create an 'object' type parameter :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. :param subtype: subtype of the 'object' schema See the generic :py:class:`Parameter` constructor for information on additional arguments (except ``schema``). .. versionadded:: 0.26.0 """ schema = {"type": "object"} if subtype: schema["subtype"] = subtype return cls(name=name, description=description, schema=schema, **kwargs)
[docs] @classmethod def bounding_box( cls, name: str, description: str = "Spatial extent specified as a bounding box with 'west', 'south', 'east' and 'north' fields.", **kwargs, ) -> Parameter: """ Helper to easily create a 'bounding box' parameter, which allows to specify a spatial extent with "west", "south", "east" and "north" bounds (and optionally a CRS identifier). :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. See the generic :py:class:`Parameter` constructor for information on additional arguments (except ``schema``). .. versionadded:: 0.30.0 """ schema = { "type": "object", "subtype": "bounding-box", "required": ["west", "south", "east", "north"], "properties": { "west": { "type": "number", "description": "West (lower left corner, coordinate axis 1).", }, "south": { "type": "number", "description": "South (lower left corner, coordinate axis 2).", }, "east": { "type": "number", "description": "East (upper right corner, coordinate axis 1).", }, "north": { "type": "number", "description": "North (upper right corner, coordinate axis 2).", }, "crs": { "description": "Coordinate reference system of the extent, specified as as [EPSG code](http://www.epsg-registry.org/) or [WKT2 CRS string](http://docs.opengeospatial.org/is/18-010r7/18-010r7.html). Defaults to `4326` (EPSG code 4326) unless the client explicitly requests a different coordinate reference system.", "anyOf": [ { "type": "integer", "subtype": "epsg-code", "title": "EPSG Code", "minimum": 1000, }, { "type": "string", "subtype": "wkt2-definition", "title": "WKT2 definition", }, ], "default": 4326, }, # TODO: support base and height? }, } return cls(name=name, description=description, schema=schema, **kwargs)
[docs] @classmethod def spatial_extent( cls, name: str = "spatial_extent", description: Optional[str] = None, **kwargs, ) -> Parameter: """ Helper to easily create a 'spatial_extent' parameter, which is compatible with the ``load_collection`` argument of the same name. This allows to conveniently create user-defined processes that can be applied to a bounding box and vector data for spatial filtering. It is also possible for users to set to null, and define spatial filtering using other processes. :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. See the generic :py:class:`Parameter` constructor for information on additional arguments (except ``schema``). .. versionadded:: 0.32.0 """ if description is None: description = textwrap.dedent( """ Limits the data to process to the specified bounding box or polygons. For raster data, the process loads the pixel into the data cube if the point at the pixel center intersects with the bounding box or any of the polygons (as defined in the Simple Features standard by the OGC). For vector data, the process loads the geometry into the data cube if the geometry is fully within the bounding box or any of the polygons (as defined in the Simple Features standard by the OGC). Empty geometries may only be in the data cube if no spatial extent has been provided. Empty geometries are ignored. Set this parameter to null to set no limit for the spatial extent. """ ).strip() schema = [ { "title": "Bounding Box", "type": "object", "subtype": "bounding-box", "required": ["west", "south", "east", "north"], "properties": { "west": {"description": "West (lower left corner, coordinate axis 1).", "type": "number"}, "south": {"description": "South (lower left corner, coordinate axis 2).", "type": "number"}, "east": {"description": "East (upper right corner, coordinate axis 1).", "type": "number"}, "north": {"description": "North (upper right corner, coordinate axis 2).", "type": "number"}, "base": { "description": "Base (optional, lower left corner, coordinate axis 3).", "type": ["number", "null"], "default": None, }, "height": { "description": "Height (optional, upper right corner, coordinate axis 3).", "type": ["number", "null"], "default": None, }, "crs": { "description": "Coordinate reference system of the extent, specified as as [EPSG code](http://www.epsg-registry.org/) or [WKT2 CRS string](http://docs.opengeospatial.org/is/18-010r7/18-010r7.html). Defaults to `4326` (EPSG code 4326) unless the client explicitly requests a different coordinate reference system.", "anyOf": [ { "title": "EPSG Code", "type": "integer", "subtype": "epsg-code", "minimum": 1000, "examples": [3857], }, {"title": "WKT2", "type": "string", "subtype": "wkt2-definition"}, ], "default": 4326, }, }, }, { "title": "Vector data cube", "description": "Limits the data cube to the bounding box of the given geometries in the vector data cube. For raster data, all pixels inside the bounding box that do not intersect with any of the polygons will be set to no data (`null`). Empty geometries are ignored.", "type": "object", "subtype": "datacube", "dimensions": [{"type": "geometry"}], }, { "title": "No filter", "description": "Don't filter spatially. All data is included in the data cube.", "type": "null", }, ] return cls(name=name, description=description, schema=schema, **kwargs)
[docs] @classmethod def date(cls, name: str, description: str = "A date.", **kwargs) -> Parameter: """ Helper to easily create a 'date' parameter. :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. See the generic :py:class:`Parameter` constructor for information on additional arguments (except ``schema``). .. versionadded:: 0.30.0 """ schema = {"type": "string", "subtype": "date", "format": "date"} return cls(name=name, description=description, schema=schema, **kwargs)
[docs] @classmethod def date_time(cls, name: str, description: str = "A date with time.", **kwargs) -> Parameter: """ Helper to easily create a 'date-time' parameter. :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. See the generic :py:class:`Parameter` constructor for information on additional arguments (except ``schema``). .. versionadded:: 0.30.0 """ schema = {"type": "string", "subtype": "date-time", "format": "date-time"} return cls(name=name, description=description, schema=schema, **kwargs)
[docs] @classmethod def geojson(cls, name: str, description: str = "Geometries specified as GeoJSON object.", **kwargs) -> Parameter: """ Helper to easily create a 'geojson' parameter, which allows to specify geometries as an inline GeoJSON object. :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. See the generic :py:class:`Parameter` constructor for information on additional arguments (except ``schema``). .. versionadded:: 0.30.0 """ schema = {"type": "object", "subtype": "geojson"} return cls(name=name, description=description, schema=schema, **kwargs)
[docs] @classmethod def temporal_interval( cls, name: str = "temporal_extent", description: str = "Temporal extent specified as two-element array with start and end date/date-time.", **kwargs, ) -> Parameter: """ Helper to easily create a 'temporal-interval' parameter, which allows to specify a temporal extent as a two-element array with start and end date/date-time. :param name: parameter name, which will be used to assign concrete values to. It is recommended to stick to the convention of snake case naming (using lowercase with underscores). :param description: human-readable description of the parameter. See the generic :py:class:`Parameter` constructor for information on additional arguments (except ``schema``). .. versionadded:: 0.30.0 """ schema = { "type": "array", "subtype": "temporal-interval", "uniqueItems": True, "minItems": 2, "maxItems": 2, "items": { "anyOf": [ {"type": "string", "subtype": "date-time", "format": "date-time"}, {"type": "string", "subtype": "date", "format": "date"}, {"type": "null"}, ] }, } return cls(name=name, description=description, schema=schema, **kwargs)
def schema_supports(schema: Union[dict, List[dict]], type: str, subtype: Optional[str] = None) -> bool: """Helper to check if parameter schema supports given type/subtype""" # TODO: support checking item type in arrays if isinstance(schema, dict): actual_type = schema.get("type") if isinstance(actual_type, str): if actual_type != type: return False elif isinstance(actual_type, list): if type not in actual_type: return False else: raise ValueError(actual_type) if subtype: if schema.get("subtype") != subtype: return False return True elif isinstance(schema, list): return any(schema_supports(s, type=type, subtype=subtype) for s in schema) else: raise ValueError(schema)