Source code for eodag.plugins.crunch.filter_property

# -*- coding: utf-8 -*-
# Copyright 2018, CS GROUP - France,
# This file is part of EODAG project
# 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

import logging
import operator

from eodag.plugins.crunch.base import Crunch

logger = logging.getLogger("eodag.plugins.crunch.filter_property")

[docs]class FilterProperty(Crunch): """FilterProperty cruncher Filter products, retaining only those whose property match criteria :param config: Crunch configuration, should contain : - `property=value` : property key from, associated to its filter value - `operator` : (optional) Operator used for filtering (one of `lt,le,eq,ne,ge,gt`). Default is `eq` :type config: dict """
[docs] def proceed(self, products, **search_params): """Execute crunch: Filter products, retaining only those that match property filtering :param products: A list of products resulting from a search :type products: list(:class:`~eodag.api.product._product.EOProduct`) :returns: The filtered products :rtype: list(:class:`~eodag.api.product._product.EOProduct`) """ operator_name = self.config.pop("operator", "eq") try: operator_method = getattr(operator, operator_name) except AttributeError: logger.warning( "Unknown operator `%s`, should be one of `lt,le,eq,ne,ge,gt`", operator_name, ) return products if len(self.config.keys()) != 1: logger.warning("One property is needed for filtering, filtering disabled.") return products property_key = next(iter(self.config)) property_value = self.config.get(property_key, None) logger.debug( "Start filtering for products matching operator.%s(['%s'], %s)", operator_name, property_key, property_value, ) filtered = [] add_to_filtered = filtered.append for product in products: if property_key not in logger.warning( "%s not found in, filtering disabled.", property_key, ) return products if operator_method([property_key], property_value): add_to_filtered(product)"Finished filtering products. %s resulting products", len(filtered)) return filtered