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