mapchete_eo.search package

Submodules

mapchete_eo.search.base module

class mapchete_eo.search.base.CollectionSearcher(collection: str, stac_item_modifiers: List[Callable[[Item], Item]] | None = None)[source]

Bases: ABC

Bridge between a Source and a catalog implementation.

blacklist: Set[str] = {}
abstract property client: CollectionClient[source]
collection: str
property config: BaseModel
config_cls: Type[BaseModel]
property description: str[source]
abstract property eo_bands: List[str][source]
property id: str[source]
abstractmethod search(time: TimeRange | List[TimeRange] | None = None, bounds: Bounds | None = None, area: BaseGeometry | None = None, query: str | None = None, search_kwargs: Dict[str, Any] | None = None) → Generator[Item, None, None][source]
property stac_extensions: List[str][source]
stac_item_modifiers: List[Callable[[Item], Item]] | None = None
class mapchete_eo.search.base.FSSpecStacIO(headers: dict[str, str] | None = None, conformance: list[str] | None = None, parameters: dict[str, Any] | None = None, request_modifier: Callable[[Request], Request | None] | None = None, timeout: float | tuple[float, float] | tuple[float, None] | None = None, max_retries: int | Retry | None = 5)[source]

Bases: StacApiIO

Custom class which allows I/O operations on object storage.

read_text(source: str | PathLike, *args, **kwargs) → str[source]

Read text from the given URI.

Overwrites the default method for reading text from a URL or file to allow urllib.request.Request instances as input. This method also raises any urllib.error.HTTPError exceptions rather than catching them to allow us to handle different response status codes as needed.

static save_json(dest: str | PathLike, json_dict: dict, *args, **kwargs) → None[source]

Write a dict to the given URI as JSON.

See StacIO.write_text for usage of str vs Link as a parameter.

Parameters:

  • dest – The destination file to write the text to.

  • json_dict – The JSON dict to write.

  • *args – Additional positional arguments to be passed to StacIO.json_dumps().

  • **kwargs – Additional keyword arguments to be passed to StacIO.json_dumps().

write_text(dest: str | PathLike, txt: str, *args, **kwargs) → None[source]

A concrete implementation of StacIO.write_text. Converts the dest argument to a string (if it is not already) and delegates to W:meth:~pystac.stac_io.DefaultStacIO.write_text_to_href for opening and reading the file.

class mapchete_eo.search.base.StaticCollectionWriterMixin(collection: str, stac_item_modifiers: List[Callable[[Item], Item]] | None = None)[source]

Bases: CollectionSearcher

write_static_catalog(output_path: str | ~os.PathLike, bounds: ~mapchete.types.Bounds | None = None, area: ~shapely.geometry.base.BaseGeometry | None = None, time: ~mapchete_eo.types.TimeRange | None = None, search_kwargs: ~typing.Dict[str, ~typing.Any] | None = None, name: str | None = None, description: str | None = None, assets: ~typing.List[str] | None = None, assets_dst_resolution: None | float | int = None, assets_convert_profile: ~rasterio.profiles.Profile | None = None, copy_metadata: bool = False, metadata_parser_classes: tuple | None = None, overwrite: bool = False, stac_io: ~pystac.stac_io.DefaultStacIO = <mapchete_eo.search.base.FSSpecStacIO object>, progress_callback: ~typing.Callable | None = None) → MPath[source]

Export a static STAC catalog from the search results.

Parameters:

  • output_path – Destination directory for the static catalog.

  • bounds – Spatial filter bounds.

  • area – Spatial filter geometry.

  • time – Temporal filter range.

  • search_kwargs – Additional search arguments.

  • name – Catalog name.

  • description – Catalog description.

  • assets – List of assets to download.

  • assets_dst_resolution – Sub-sampling resolution for assets.

  • assets_convert_profile – Output profile for assets (e.g. for COG conversion).

  • copy_metadata – Whether to copy sidecar metadata files.

  • metadata_parser_classes – Custom parser classes for metadata.

  • overwrite – Overwrite existing files.

  • stac_io – Custom STAC IO implementation.

  • progress_callback – Optional function for progress reporting.

Returns:

Path to the generated catalog.json.

Return type:

MPath

mapchete_eo.search.base.filter_items(items: Generator[Item, None, None], query: str | None = None) → Generator[Item, None, None][source]

Only for cloudcover now, this can and should be adapted for filter field and value the field and value for the item filter would be defined in search.config.py corresponding configs and passed down to the individual search approaches via said config and this Function.

mapchete_eo.search.config module

class mapchete_eo.search.config.StacSearchConfig(*, max_cloud_cover: float = 100.0, query: str | None = None, catalog_chunk_threshold: int = 10000, catalog_chunk_zoom: int = 5, catalog_pagesize: int = 100, footprint_buffer: float = 0)[source]

Bases: BaseModel

catalog_chunk_threshold: int
catalog_chunk_zoom: int
catalog_pagesize: int
classmethod deprecate_max_cloud_cover(values: Dict[str, Any]) → Dict[str, Any][source]
footprint_buffer: float
max_cloud_cover: float
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

query: str | None
class mapchete_eo.search.config.StacStaticConfig[source]

Bases: BaseModel

classmethod deprecate_max_cloud_cover(values: Dict[str, Any]) → Dict[str, Any][source]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class mapchete_eo.search.config.UTMSearchConfig(*, sinergise_aws_collections: dict = {'S1_GRD': {'endpoint': 's3://sentinel-s1-l1c-stac', 'id': 'sentinel-s1-l1c', 'path': <mapchete.io.MPath object: https://sentinel-s1-l1c-stac.s3.amazonaws.com/sentinel-s1-l1c.json, storage_options={'asynchronous': False, 'timeout': None}>}, 'S2_L1C': {'endpoint': 's3://sentinel-s2-l1c-stac', 'id': 'sentinel-s2-l1c', 'path': <mapchete.io.MPath object: https://sentinel-s2-l1c-stac.s3.amazonaws.com/sentinel-s2-l1c.json, storage_options={'asynchronous': False, 'timeout': None}>}, 'S2_L2A': {'endpoint': 's3://sentinel-s2-l2a-stac', 'id': 'sentinel-s2-l2a', 'path': <mapchete.io.MPath object: https://sentinel-s2-l2a-stac.s3.amazonaws.com/sentinel-s2-l2a.json, storage_options={'asynchronous': False, 'timeout': None}>}}, search_index: str | ~os.PathLike | None = None)[source]

Bases: BaseModel

classmethod deprecate_max_cloud_cover(values: Dict[str, Any]) → Dict[str, Any][source]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

search_index: str | PathLike | None
sinergise_aws_collections: dict
mapchete_eo.search.config.patch_invalid_assets()[source]

Context manager/decorator to fix pystac crash on malformed assets (strings instead of dicts).

mapchete_eo.search.s2_mgrs module

exception mapchete_eo.search.s2_mgrs.InvalidMGRSSquare[source]

Bases: Exception

Raised when an invalid square index has been given

class mapchete_eo.search.s2_mgrs.MGRSCell(utm_zone: 'str', latitude_band: 'str')[source]

Bases: object

property crs: CRS[source]
property hemisphere: Literal['S', 'N'][source]
latitude_band: str
property latlon_bounds: Bounds[source]
property latlon_geometry: BaseGeometry[source]
tile(grid_square: str, column_index: int | None = None, row_index: int | None = None) → S2Tile[source]
tiles() → List[S2Tile][source]
utm_zone: str
class mapchete_eo.search.s2_mgrs.S2Tile(utm_zone: 'str', latitude_band: 'str', grid_square: 'str', global_column_index: 'Optional[int]' = None, global_row_index: 'Optional[int]' = None)[source]

Bases: object

property bounds: Bounds[source]
property crs: CRS[source]
static from_grid_code(grid_code: str) → S2Tile[source]
static from_tile_id(tile_id: str) → S2Tile[source]
global_column_index: int | None = None
global_row_index: int | None = None
grid_square: str
property hemisphere: Literal['S', 'N'][source]
latitude_band: str
property latlon_bounds: Bounds[source]
property latlon_geometry: BaseGeometry[source]
property mgrs_cell: MGRSCell[source]
property square_column: int[source]
property square_row: int[source]
property tile_id: str[source]
utm_zone: str
mapchete_eo.search.s2_mgrs.s2_tiles_from_bounds(left: float, bottom: float, right: float, top: float) → List[S2Tile][source]

mapchete_eo.search.stac_static module

class mapchete_eo.search.stac_static.STACStaticCollection(collection: str, stac_item_modifiers: List[Callable[[Item], Item]] | None = None)[source]

Bases: StaticCollectionWriterMixin, CollectionSearcher

Search implementation for static STAC collections.

property client: CollectionClient[source]
config_cls

alias of StacStaticConfig

property eo_bands: List[str][source]
search(time: TimeRange | List[TimeRange] | None = None, bounds: List[float] | Tuple[float, float, float, float] | dict | Polygon | None = None, area: BaseGeometry | None = None, query: str | None = None, search_kwargs: Dict[str, Any] | None = None) → Generator[Item, None, None][source]

Module contents

A catalog is an instance with a specific endpoint and a specific collection.

The catalog class aims to abstract product search as well as homogenization of product metadata.

It helps the InputData class to find the input products and their metadata.

class mapchete_eo.search.STACSearchCollection(collection: str, stac_item_modifiers: List[Callable[[Item], Item]] | None = None)[source]

Bases: StaticCollectionWriterMixin, CollectionSearcher

Search implementation for STAC APIs.

property client: CollectionClient[source]
collection: str
config_cls

alias of StacSearchConfig

property default_search_params[source]
property eo_bands: List[str][source]
search(time: TimeRange | List[TimeRange] | None = None, bounds: List[float] | Tuple[float, float, float, float] | dict | Polygon | None = None, area: BaseGeometry | None = None, query: str | None = None, search_kwargs: Dict[str, Any] | None = None) → Generator[Item, None, None][source]
property search_client: Client[source]
class mapchete_eo.search.STACStaticCollection(collection: str, stac_item_modifiers: List[Callable[[Item], Item]] | None = None)[source]

Bases: StaticCollectionWriterMixin, CollectionSearcher

Search implementation for static STAC collections.

property client: CollectionClient[source]
config_cls

alias of StacStaticConfig

property eo_bands: List[str][source]
search(time: TimeRange | List[TimeRange] | None = None, bounds: List[float] | Tuple[float, float, float, float] | dict | Polygon | None = None, area: BaseGeometry | None = None, query: str | None = None, search_kwargs: Dict[str, Any] | None = None) → Generator[Item, None, None][source]
class mapchete_eo.search.UTMSearchCatalog(collection: str, stac_item_modifiers: List[Callable[[Item], Item]] | None = None)[source]

Bases: StaticCollectionWriterMixin, CollectionSearcher

Search implementation for UTM-grid based catalogs (e.g. Sentinel-2 on AWS).

property client: CollectionClient[source]
collection: str
config_cls

alias of UTMSearchConfig

day_subdir_schema: str = '{year}/{month:02d}/{day:02d}'
property endpoint: str | None[source]
property eo_bands: List[str][source]
get_collections()[source]
yeild transformed collection from:

https://sentinel-s2-l2a-stac.s3.amazonaws.com/sentinel-s2-l2a.json, or https://sentinel-s2-l1c-stac.s3.amazonaws.com/sentinel-s2-l1c.json, etc.

search(time: TimeRange | List[TimeRange] | None = None, bounds: List[float] | Tuple[float, float, float, float] | dict | Polygon | None = None, area: BaseGeometry | None = None, query: str | None = None, search_kwargs: Dict[str, Any] | None = None) → Generator[Item, None, None][source]
stac_json_endswith: str = 'T{tile_id}.json'