Skip to content

blockutils.stac

Utilities for handling SpatioTemporal Asset Catalog (STAC) queries.

STACQuery dataclass

Object representing a STAC query as used by data blocks. Standardized arguments are:

Parameters:

Name Type Description Default
ids

Unique identifiers to search for.

required
bbox

A bounding box type geometry.

required
intersects

A geometry to use for searching with intersection operator.

required
contains

A geometry to use for searching with contains operator.

required
time

Time string in RFC3339 (for points in time) or RFC3339/RFC3339 (for datetime ranges).

required
limit

Maximum number of returned elements.

required
time_series

List of time strings in RFC3339 (for points in time) or RFC3339/RFC3339 (for datetime ranges).

required

Additional arguments can be passed to the constructor.

__new__(cls, **kwargs) special staticmethod

Inspired by https://stackoverflow.com/a/63291704/14236301 Adds attributes to the data class if they are passed in input dict in init, even if they are not explicitly defined as field attributes.

Source code in blockutils/stac.py
def __new__(cls, **kwargs):
    """
    Inspired by https://stackoverflow.com/a/63291704/14236301
    Adds attributes to the data class if they are passed in
    input dict in __init__, even if they are not explicitly defined as
    field attributes.
    """
    try:
        initializer = cls.__initializer
    except AttributeError:
        # Store the original init on the class in a different place
        cls.__initializer = initializer = cls.__init__
        # replace init with something harmless
        cls.__init__ = lambda *a, **k: None

    # code from adapted from Arne
    added_args = {}
    for name in list(kwargs.keys()):
        if name not in cls.__annotations__:
            added_args[name] = kwargs.pop(name)

    ret = object.__new__(cls)
    initializer(ret, **kwargs)
    # ... and add the new ones by hand
    for new_name, new_val in added_args.items():
        setattr(ret, new_name, new_val)

    return ret

__post_init__(self) special

Post processes the passed arguments to make sure they are valid, i.e. only one of "bbox", "intersects" or "contains" is passed.

Source code in blockutils/stac.py
def __post_init__(self):
    """
    Post processes the passed arguments to make sure they are valid,
    i.e. only one of "bbox", "intersects" or "contains" is passed.
    """
    if (
        self.bbox
        and self.intersects
        or self.intersects
        and self.contains
        or self.contains
        and self.bbox
    ):
        raise UP42Error(
            SupportedErrors.WRONG_INPUT_ERROR,
            """Only one of the following query parameters is
            allowed at a query at any given time:
            * bbox
            * intersects
            * contains.""",
        )

    if self.time and self.time_series:
        raise UP42Error(
            SupportedErrors.WRONG_INPUT_ERROR,
            """Only one of the following query parameters is
            allowed at a query at any given time:
            * time
            * time_series
            """,
        )

    if not self.validate_datetime_str(self.time):
        raise UP42Error(
            SupportedErrors.WRONG_INPUT_ERROR,
            """Time string could not be validated.
            It must be one of the following formats:
            <RFC3339> (for points in time)
            <RFC3339>/<RFC3339> (for datetime ranges)""",
        )

    if self.limit is not None and self.limit < 1:
        logger.warning(
            "WARNING: limit parameter cannot be < 1, and has been automatically set to 1"
        )
        self.limit = 1

    if self.time_series is not None:
        for datestr in self.time_series:
            if not self.validate_datetime_str(datestr):
                raise UP42Error(
                    SupportedErrors.WRONG_INPUT_ERROR,
                    """Time string from time_series could not be validated.
                    It must be one of the following formats:
                    <RFC3339> (for points in time)
                    <RFC3339>/<RFC3339> (for datetime ranges)""",
                )

bounds(self)

Get the bounding box of the query AOI(s) as a BoundingBox object.

Returns:

Type Description
Tuple[Union[int, decimal.Decimal, float], Union[int, decimal.Decimal, float], Union[int, decimal.Decimal, float], Union[int, decimal.Decimal, float]]

A bounding box object.

Source code in blockutils/stac.py
def bounds(self) -> BoundingBox:
    """
    Get the bounding box of the query AOI(s) as a BoundingBox object.

    Returns:
        A bounding box object.
    """
    if self.bbox:
        return self.bbox
    elif self.intersects:
        return shapely.geometry.shape(self.intersects).bounds
    elif self.contains:
        return shapely.geometry.shape(self.contains).bounds
    else:
        raise UP42Error(
            SupportedErrors.WRONG_INPUT_ERROR,
            """STACQuery does not contain any of the following query parameters:
            * bbox
            * intersects
            * contains.""",
        )

from_dict(dict_data, validator=<function STACQuery.<lambda> at 0x11aa70ee0>) classmethod

Get a STACQuery from a dict representation of a query.

Parameters:

Name Type Description Default
dict_data dict

Dictionary representation of the query.

required
validator Callable[[dict], bool]

Function used to validate the query.

<function STACQuery.<lambda> at 0x11aa70ee0>

Returns:

Type Description

A STACQuery object.

Source code in blockutils/stac.py
@classmethod
def from_dict(
    cls, dict_data: dict, validator: Callable[[dict], bool] = lambda x: True
):
    """
    Get a STACQuery from a dict representation of a query.

    Arguments:
        dict_data: Dictionary representation of the query.
        validator: Function used to validate the query.

    Returns:
        A STACQuery object.
    """

    if not validator(dict_data):
        raise UP42Error(
            SupportedErrors.WRONG_INPUT_ERROR,
            """Input Query did not pass validation. Please refer
            to the official block documentation or block specification.""",
        )

    if ("intersects" or "contains") in dict_data:
        dict_data = cls.handle_z_coordinate(dict_data)

    return STACQuery(**dict_data)

from_json(json_data) classmethod

Get a STACQuery from a json string representation of a query.

Parameters:

Name Type Description Default
json_data str

String representation of the query.

required

Returns:

Type Description

A STACQuery object/

Source code in blockutils/stac.py
@classmethod
def from_json(cls, json_data: str):
    """
    Get a STACQuery from a json string representation of a query.

    Arguments:
        json_data: String representation of the query.

    Returns:
        A STACQuery object/
    """
    return STACQuery.from_dict(json.loads(json_data))

geometry(self)

Get the geometry of the query AOI(s) as a GeoJson Polygon object.

Returns:

Type Description
Geometry

A GeoJson Polygon object

Source code in blockutils/stac.py
def geometry(self) -> Geometry:
    """
    Get the geometry of the query AOI(s) as a GeoJson Polygon object.

    Returns:
        A GeoJson Polygon object
    """
    if self.bbox:
        return json.loads(
            json.dumps(shapely.geometry.mapping(shapely.geometry.box(*self.bbox)))
        )
    elif self.intersects:
        return self.intersects
    elif self.contains:
        return self.contains
    else:
        raise UP42Error(
            SupportedErrors.WRONG_INPUT_ERROR,
            """STACQuery does not contain any of the following query parameters:
            * bbox
            * intersects
            * contains.""",
        )

get_param_if_exists(self, key, default_value=None)

Get a query parameter if exists in he STAC query object. Returns default_value if not in the query.

Parameters:

Name Type Description Default
key str

An identifier of the parameter

required
default_value Any

The default return value if the parameter does not exist.

None
Source code in blockutils/stac.py
def get_param_if_exists(self, key: str, default_value: Any = None) -> Any:
    """
    Get a query parameter if exists in he STAC query object. Returns
    default_value if not in the query.

    Arguments:
        key: An identifier of the parameter
        default_value: The default return value if the parameter does not exist.
    """
    return getattr(self, key, default_value)

handle_z_coordinate(query) staticmethod

Checks if a geometry of a qury has z coordinate or not. if they exist, they will be removed.

Parameters:

Name Type Description Default
query dict

Dictionary representation of the query.

required

Returns:

Type Description

A new dictionary without z coordinates (if they exist).

Source code in blockutils/stac.py
@staticmethod
def handle_z_coordinate(query: dict):
    """
    Checks if a geometry of a qury has z coordinate or not. if they exist, they will be removed.
    Args:
        query: Dictionary representation of the query.

    Returns:
        A new dictionary without z coordinates (if they exist).

    """
    query_geom = query[("intersects" or "contains")]
    if query_geom:
        geom_drop_z = geojson.loads(
            geojson.dumps(
                geojson.utils.map_tuples(lambda c: [c[0], c[1]], query_geom)
            )
        )

        query[("intersects" or "contains")] = geom_drop_z
    return query

set_param_if_not_exists(self, key, value)

Set a query parameter if it does not exist in the STAC query object.

Parameters:

Name Type Description Default
key str

An identifier of the parameter

required
value Any

The parameter value to set.

required
Source code in blockutils/stac.py
def set_param_if_not_exists(self, key: str, value: Any):
    """
    Set a query parameter if it does not exist in the STAC query object.

    Arguments:
        key: An identifier of the parameter
        value: The parameter value to set.
    """
    if self.get_param_if_exists(key) is None:
        setattr(self, key, value)

validate_datetime_str(string) staticmethod

Validate a datetime string.

Parameters:

Name Type Description Default
string Optional[str]

A datetime string representation.

required

Returns:

Type Description
bool

If datetime string is valid.

Source code in blockutils/stac.py
@staticmethod
def validate_datetime_str(string: Optional[str]) -> bool:
    """
    Validate a datetime string.

    Arguments:
        string: A datetime string representation.

    Returns:
        If datetime string is valid.
    """
    try:
        if string is None:
            return True
        elif len(str(string).split("/")) == 2:
            ciso8601.parse_datetime(str(string).split("/", maxsplit=1)[0])
            ciso8601.parse_datetime(str(string).split("/")[1])
        else:
            ciso8601.parse_datetime(str(string))
    except ValueError:
        return False
    return True