Source code for harvester.filter

from datetime import datetime
from typing import Dict, Iterator, Optional

from pygeofilter.backends.native.evaluate import NativeEvaluator
from pygeofilter.parsers.cql_json import parse as json_parse


CQL_ATTR_MAP = {"point_attr": "geometry", "*": "properties.*"}


def cql_filter(
    _filter: Dict,
    data: Iterator[Dict],
    context: Optional[Dict] = None,
) -> Iterator[Dict]:
    if not _filter:
        yield from data
        return
    _filter = json_parse(_filter)
    context = context or {}
    nat_eval = NativeEvaluator(
        function_map={
            "now": datetime.now,
            "context": lambda name, default=None: context.get(name, default),
        },
        attribute_map=CQL_ATTR_MAP,
        use_getattr=False,
    )
    evaluator = nat_eval.evaluate(_filter)
    yield from filter(evaluator, data)