Source code for eodag.plugins.crunch.filter_overlap

# -*- coding: utf-8 -*-
# Copyright 2018, CS GROUP - France, https://www.csgroup.eu/
#
# This file is part of EODAG project
#     https://www.github.com/CS-SI/EODAG
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any, List

from eodag.plugins.crunch.base import Crunch
from eodag.utils import get_geometry_from_various

try:
    from shapely.errors import GEOSException
except ImportError:
    # shapely < 2.0 compatibility
    from shapely.errors import TopologicalError as GEOSException

if TYPE_CHECKING:
    from eodag.api.product import EOProduct

logger = logging.getLogger("eodag.crunch.overlap")


[docs] class FilterOverlap(Crunch): """FilterOverlap cruncher Filter products, retaining only those that are overlapping with the search_extent :param config: Crunch configuration, may contain : - `minimum_overlap` : minimal overlap percentage - `contains` : True if product geometry contains the search area - `intersects` : True if product geometry intersects the search area - `within` : True if product geometry is within the search area These configuration parameters are mutually exclusive. :type config: dict """
[docs] def proceed( self, products: List[EOProduct], **search_params: Any ) -> List[EOProduct]: """Execute crunch: Filter products, retaining only those that are overlapping with the search_extent :param products: A list of products resulting from a search :type products: list(:class:`~eodag.api.product._product.EOProduct`) :param search_params: Search criteria that must contain `geometry` :type search_params: dict :returns: The filtered products :rtype: list(:class:`~eodag.api.product._product.EOProduct`) """ logger.debug("Start filtering for overlapping products") filtered: List[EOProduct] = [] add_to_filtered = filtered.append search_geom = get_geometry_from_various(**search_params) if not search_geom: logger.warning( "geometry not found in cruncher arguments, filtering disabled." ) return products minimum_overlap = float(self.config.get("minimum_overlap", "0")) contains = self.config.get("contains", False) intersects = self.config.get("intersects", False) within = self.config.get("within", False) if contains and (within or intersects) or (within and intersects): logger.warning( "contains, intersects and within parameters are mutually exclusive" ) return products elif ( minimum_overlap > 0 and minimum_overlap < 100 and (contains or within or intersects) ): logger.warning( "minimum_overlap will be ignored because of contains/intersects/within usage" ) elif not contains and not within and not intersects: logger.debug("Minimum overlap is: {} %".format(minimum_overlap)) logger.debug("Initial requested extent area: %s", search_geom.area) if search_geom.area == 0: logger.debug( "No product can overlap a requested extent that is not a polygon (i.e with area=0)" ) else: for product in products: logger.debug("Uncovered extent area: %s", search_geom.area) if product.search_intersection: intersection = product.search_intersection product_geometry = product.geometry else: # Product geometry may be invalid if not product.geometry.is_valid: logger.debug( "Trying our best to deal with invalid geometry on product: %r", product, ) product_geometry = product.geometry.buffer(0) try: intersection = search_geom.intersection(product_geometry) except GEOSException: logger.debug( "Product geometry still invalid. Overlap test restricted to containment" ) if search_geom.contains(product_geometry): logger.debug( "Product %r overlaps the search extent. Adding it to filtered results" ) add_to_filtered(product) continue else: product_geometry = product.geometry intersection = search_geom.intersection(product_geometry) if ( (contains and product_geometry.contains(search_geom)) or (within and product_geometry.within(search_geom)) or (intersects and product_geometry.intersects(search_geom)) ): add_to_filtered(product) continue elif contains or within or intersects: continue ipos = (intersection.area / search_geom.area) * 100 ipop = (intersection.area / product_geometry.area) * 100 logger.debug( "Intersection of product extent and search extent covers %f percent of the search extent " "area", ipos, ) logger.debug( "Intersection of product extent and search extent covers %f percent of the product extent " "area", ipop, ) if any( ( search_geom.contains(product.geometry), ipos >= minimum_overlap, ipop >= minimum_overlap, ) ): logger.debug( "Product %r overlaps the search extent by the specified constraint. Adding it to " "filtered results", product, ) add_to_filtered(product) else: logger.debug( "Product %r does not overlaps the search extent by the specified constraint. " "Skipping it", product, ) logger.info("Finished filtering products. %s resulting products", len(filtered)) return filtered