Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Apply a function to each group as an `ImageCollection` and combine the results.
``func`` must take two arguments: the group, and an `ImageCollection` of all Images in that group.
If it returns an `ImageCollection`, the ImageCollections for all groups will be concatenated together.
If it returns an `Image`, those Images will be combined into a single ImageCollection.
If it returns any other type, `map` will return a `Dict`, where keys are groups
and values are results of ``func``.
Note that every `Image` in every `ImageCollecion` gets a ``"group"`` field added to its `~.Image.properties`,
so that field will be merged/dropped according to normal metadata broadcasting rules (i.e. will still be present
on an `Image` composited from a group's `ImageCollection`, since it's the same for every `Image`.)
"""
proxy_func = Function.from_callable(func, self.key_type, ImageCollection)
result_type = proxy_func._type_params[-1]
if result_type in (ImageCollection, Image):
out_type = ImageCollection
func = "ImageCollectionGroupby.map_ic"
else:
out_type = Dict[self.key_type, result_type]
func = "ImageCollectionGroupby.map"
return out_type._from_apply(func, self, proxy_func)
def proxify(obj):
from ..core import Proxytype, ProxyTypeError
from ..function import Function
from ..containers import Tuple, List
from ..primitives import Int, Float, Bool, Str, NoneType, Any
from ..datetimes import Datetime, Timedelta
if isinstance(obj, Proxytype):
return obj
elif callable(obj):
return Function.from_callable(obj)
elif isinstance(obj, (tuple, list)):
contents = [proxify(x) for x in obj]
types = tuple(type(x) for x in contents)
if (
isinstance(obj, list)
and len(types) > 0
and all(t is types[0] for t in types[1:])
):
return List[types[0]](contents)
else:
return Tuple[types](contents)
elif isinstance(obj, int):
return Int(obj)
elif isinstance(obj, float):
return Float(obj)
elif isinstance(obj, bool):
from .groupby import ImageCollectionGroupby
if func and dates:
raise TypeError("Only one of `func` or `dates` may be given")
if dates:
if isinstance(dates, six.string_types):
dates = (dates,)
# consider implementing this on the backend instead; may be more performant
def func(img):
date = img.properties["date"]
fields = tuple(getattr(date, field) for field in dates)
return fields[0] if len(fields) == 1 else fields
delayed_func = Function.from_callable(func, Image)
key_type = delayed_func._type_params[-1]
return ImageCollectionGroupby[key_type](self, delayed_func)
Parameters
----------
back: Int, optional, default 0
Number of previous Images to pass as ``back`` to the function.
fwd: Int, optional, default 0
Number of subsequent Images to pass as ``fwd`` to the function.
Returns
-------
mapped: ImageCollection or List
If ``func`` returns an `ImageCollection` or `Image`,
all of them are concatenated together and returned as one `ImageCollection`.
Otherwise, returns a `List` of the values returned by ``func``.
"""
delayed_func = Function.from_callable(
func, ImageCollection, Image, ImageCollection
)
return_type = delayed_func._type_params[-1]
out_type, func = (
(ImageCollection, "ImageCollection.map_window_ic")
if return_type in (Image, ImageCollection)
else (List[return_type], "ImageCollection.map_window")
)
return out_type._from_apply(func, self, delayed_func, back, fwd)