File size: 2,526 Bytes
4fcc331 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | """Temporal compositing, or temporal aggregation, is a method to increase the
quality of data within timesteps by reducing the temporal resolution of a time
series of satellite images.
"""
from typing import Union
import openeo
def median_compositing(
cube: openeo.DataCube, period: Union[str, list]
) -> openeo.DataCube:
"""Perfrom median compositing on the given datacube."""
if isinstance(period, str):
return cube.aggregate_temporal_period(
period=period, reducer="median", dimension="t"
)
elif isinstance(period, list):
return cube.aggregate_temporal(
intervals=period, reducer="median", dimension="t"
)
def mean_compositing(
cube: openeo.DataCube, period: Union[str, list]
) -> openeo.DataCube:
"""Perfrom mean compositing on the given datacube."""
if isinstance(period, str):
return cube.aggregate_temporal_period(
period=period, reducer="mean", dimension="t"
)
elif isinstance(period, list):
return cube.aggregate_temporal(intervals=period, reducer="mean", dimension="t")
def sum_compositing(cube: openeo.DataCube, period: Union[str, list]) -> openeo.DataCube:
"""Perform sum compositing on the given datacube."""
if isinstance(period, str):
return cube.aggregate_temporal_period(
period=period, reducer="sum", dimension="t"
)
elif isinstance(period, list):
return cube.aggregate_temporal(intervals=period, reducer="sum", dimension="t")
def max_ndvi_compositing(cube: openeo.DataCube, period: str) -> openeo.DataCube:
"""Perform compositing by selecting the observation with the highest NDVI value over the
given compositing window."""
def max_ndvi_selection(ndvi: openeo.DataCube):
max_ndvi = ndvi.max()
return ndvi.array_apply(lambda x: x != max_ndvi)
if isinstance(period, str):
ndvi = cube.ndvi(nir="S2-L2A-B08", red="S2-L2A-B04")
rank_mask = ndvi.apply_neighborhood(
max_ndvi_selection,
size=[
{"dimension": "x", "unit": "px", "value": 1},
{"dimension": "y", "unit": "px", "value": 1},
{"dimension": "t", "value": period},
],
overlap=[],
)
cube = cube.mask(mask=rank_mask).aggregate_temporal_period(period, "first")
else:
raise ValueError(
"Custom temporal intervals are not yet supported for max NDVI compositing."
)
return cube
|