geowatch.stac.util_stac module

geowatch.stac.util_stac.parallel_map_items(catalog, mapper_func, max_workers=4, mode='process', drop_on_error=True, extra_args=[], extra_kwargs={})[source]

Functions similarly to pystac.Catalog.map_items function but in parallel, and allows the mapper function to return None indicating that the mapped STAC item should be dropped from the output catalog

Parameters:

  • catalog (pystac.Catalog) – catalog to apply transform to

  • mapper_func (callable) – function to be applied to each STAC item. The first positional argument must accept a pystac.Item, and may take any arbitrary additional positional or keyword arguments.

  • max_workers (int) – number of jobs

  • mode (str) – process, thread, or serial

  • extra_args (List | Tuple) – extra positional args passed to mapper_func

  • extra_kwargs (Dict[str, object]) – extra keyword args passed to mapper_func

Returns:

modified catalog

Return type:

pystac.Catalog

Example

>>> # xdoctest: +SKIP("too many https errors")
>>> from geowatch.stac.util_stac import *  # NOQA
>>> from geowatch.demo import stac_demo
>>> catalog_fpath = stac_demo.demo()
>>> catalog = pystac.Catalog.from_file(catalog_fpath)
>>> def demo_mapper_func(item):
>>>     import copy
>>>     print('Process: item = {}'.format(ub.urepr(item, nl=1)))
>>>     print('item.assets = {}'.format(ub.urepr(item.assets, nl=1)))
>>>     if 'data' not in item.assets:
>>>         print('Drop asset without data')
>>>         return None  # drop assets without data
>>>     # Pretend we do some image operation and write to a new path
>>>     in_fpath = item.assets['data'].href
>>>     out_fpath = ub.augpath(in_fpath, suffix='_demo_process')
>>>     print('in_fpath = {!r}'.format(in_fpath))
>>>     out_fpath = in_fpath
>>>     new_item = copy.deepcopy(item)
>>>     new_item.assets['data'].href = out_fpath
>>>     return item
>>> out_catalog = parallel_map_items(
>>>     catalog, demo_mapper_func, mode='serial')
>>> assert len(list(catalog.get_all_items())) == 2, 'two items in'
>>> assert len(list(out_catalog.get_all_items())) == 1, 'one item out'
geowatch.stac.util_stac.maps(_item_map=None, history_entry=None)[source]

General-purpose wrapper for STAC _item_maps.

An _item_map should take in a STAC item and return a STAC item, an iterable of STAC items or None. To support this, it should have an arg ‘stac_item’ and an arg or kwarg ‘outdir’.

This decorator handles the following tasks:

  • add original item link

  • create an item_outdir from a base outdir, and pass it to _item_map.

  • update item’s self_href

  • add an entry to ‘watch:process_history’

References

https://pybit.es/articles/decorator-optional-argument/ https://docs.python.org/3/library/inspect.html#inspect.BoundArguments

geowatch.stac.util_stac.associate_msi_pan(stac_catalog)[source]

Match up WorldView multispectral and panchromatic items.

Returns a dict {msi_item.id: pan_item}, where pan_item can be nonunique.