Source code for pystac.summaries

from __future__ import annotations

import importlib.resources
import json
import numbers
from abc import abstractmethod
from import Iterable
from copy import deepcopy
from enum import Enum
from functools import lru_cache
from typing import (

import pystac
from pystac.utils import get_required

    from pystac.collection import Collection
    from pystac.item import Item

def __getattr__(name: str) -> Any:
    if name == "FIELDS_JSON_URL":
        import warnings

            "FIELDS_JSON_URL is deprecated and will be removed in v2",
        return (
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

class _Comparable_x(Protocol):
    """Protocol for annotating comparable types.

    For matching __lt__ that takes an 'x' parameter
    (e.g. float)

    def __lt__(self: T, x: T) -> bool:
        return NotImplemented

class _Comparable_other(Protocol):
    """Protocol for annotating comparable types.

    For matching __lt___ that takes an 'other' parameter
    (e.g. datetime)

    def __lt__(self: T, other: T) -> bool:
        return NotImplemented

T = TypeVar("T", bound=Union[_Comparable_x, _Comparable_other])

[docs]class RangeSummary(Generic[T]): minimum: T maximum: T def __init__(self, minimum: T, maximum: T): self.minimum = minimum self.maximum = maximum
[docs] def to_dict(self) -> dict[str, Any]: return {"minimum": self.minimum, "maximum": self.maximum}
[docs] def update_with_value(self, v: T) -> None: self.minimum = min(self.minimum, v) self.maximum = max(self.maximum, v)
[docs] @classmethod def from_dict(cls, d: dict[str, Any]) -> RangeSummary[T]: minimum: T = get_required(d.get("minimum"), "RangeSummary", "minimum") maximum: T = get_required(d.get("maximum"), "RangeSummary", "maximum") return cls(minimum=minimum, maximum=maximum)
def __eq__(self, o: object) -> bool: if not isinstance(o, RangeSummary): return NotImplemented return self.to_dict() == o.to_dict() def __repr__(self) -> str: return self.to_dict().__repr__()
@lru_cache(maxsize=None) def _get_fields_json(url: str | None) -> dict[str, Any]: if url is None: # Every time pystac is released this file gets pulled from # jsonfields: dict[str, Any] = json.loads( importlib.resources.files("pystac.static") .joinpath("fields-normalized.json") .read_text() ) return jsonfields return pystac.StacIO.default().read_json(url)
[docs]class SummaryStrategy(Enum): ARRAY = "v" RANGE = "r" SCHEMA = "s" DONT_SUMMARIZE = False DEFAULT = True
[docs]class Summarizer: """The Summarizer computes summaries from values, following the definition of fields to summarize. The fields to summarize can be provided as a JSON file or as a dictionary of field names and SummaryStrategys. If nothing is provided, a default JSON file will be used. Only fields that are in the Item `properties` can be summarized. Thus it is not possible to summarize the top-level fields such as `id` or `assets`. For more information about the structure of the fields JSON file, see: The default JSON file used is a snapshot of the following file at the time of the pystac release: Args: fields: A string containing the path to the json file with field descriptions. Alternatively, a dict with the field names as keys and SummaryStrategys as values. If nothing is passed, a default file with field descriptions will be used. """ summaryfields: dict[str, SummaryStrategy] def __init__(self, fields: str | dict[str, SummaryStrategy] | None = None): if isinstance(fields, dict): self._set_field_definitions(fields) else: jsonfields = _get_fields_json(fields) self._set_field_definitions(jsonfields["metadata"]) def _set_field_definitions(self, fields: dict[str, Any]) -> None: self.summaryfields = {} for name, desc in fields.items(): strategy: SummaryStrategy = SummaryStrategy.DEFAULT if isinstance(desc, SummaryStrategy): strategy = desc elif isinstance(desc, dict): strategy_value = desc.get("summary", True) try: strategy = SummaryStrategy(strategy_value) except ValueError: pass if strategy != SummaryStrategy.DONT_SUMMARIZE: self.summaryfields[name] = strategy def _update_with_item(self, summaries: Summaries, item: Item) -> None: for k, v in if k in self.summaryfields: strategy = self.summaryfields[k] if strategy == SummaryStrategy.RANGE or ( strategy == SummaryStrategy.DEFAULT and isinstance(v, numbers.Number) and not isinstance(v, bool) ): rangesummary: RangeSummary[Any] | None = summaries.get_range(k) if rangesummary is None: summaries.add(k, RangeSummary(v, v)) else: rangesummary.update_with_value(v) elif strategy == SummaryStrategy.ARRAY or ( strategy == SummaryStrategy.DEFAULT and isinstance(v, list) ): listsummary: list[Any] = summaries.get_list(k) or [] if not isinstance(v, list): v = [v] for element in v: if element not in listsummary: listsummary.append(element) summaries.add(k, listsummary) else: summary: list[Any] = summaries.get_list(k) or [] if v not in summary: summary.append(v) summaries.add(k, summary)
[docs] def summarize(self, source: Collection | Iterable[Item]) -> Summaries: """Creates summaries from items""" summaries = Summaries.empty() if isinstance(source, pystac.Collection): for item in source.get_items(recursive=True): self._update_with_item(summaries, item) else: for item in source: self._update_with_item(summaries, item) return summaries
[docs]class Summaries: _summaries: dict[str, Any] lists: dict[str, list[Any]] other: dict[str, Any] ranges: dict[str, RangeSummary[Any]] schemas: dict[str, dict[str, Any]] maxcount: int def __init__( self, summaries: dict[str, Any], maxcount: int = DEFAULT_MAXCOUNT ) -> None: self._summaries = summaries self.maxcount = maxcount self.lists = {} self.ranges = {} self.schemas = {} self.other = {} for prop_key, summary in summaries.items(): self.add(prop_key, summary)
[docs] def get_list(self, prop: str) -> list[Any] | None: return self.lists.get(prop)
[docs] def get_range(self, prop: str) -> RangeSummary[Any] | None: return self.ranges.get(prop)
[docs] def get_schema(self, prop: str) -> dict[str, Any] | None: return self.schemas.get(prop)
[docs] def add( self, prop_key: str, summary: list[Any] | RangeSummary[Any] | dict[str, Any], ) -> None: if isinstance(summary, list): self.lists[prop_key] = summary elif isinstance(summary, dict): if "minimum" in summary: self.ranges[prop_key] = RangeSummary[Any].from_dict(summary) else: self.schemas[prop_key] = summary elif isinstance(summary, RangeSummary): self.ranges[prop_key] = summary else: self.other[prop_key] = summary
[docs] def remove(self, prop_key: str) -> None: self.lists.pop(prop_key, None) self.ranges.pop(prop_key, None) self.schemas.pop(prop_key, None) self.other.pop(prop_key, None)
[docs] def update(self, summaries: Summaries) -> None: self.lists.update(summaries.lists) self.ranges.update(summaries.ranges) self.schemas.update(summaries.schemas) self.other.update(summaries.other)
[docs] def combine(self, summaries: Summaries) -> None: for listname, listvalue in summaries.lists.items(): if listname in self.lists: self.lists[listname].extend(listvalue) else: self.lists[listname] = listvalue for rangename, rang in summaries.ranges.items(): if rangename in self.ranges: self.ranges[rangename].update_with_value(rang.minimum) self.ranges[rangename].update_with_value(rang.maximum) else: self.ranges[rangename] = rang for schemaname, schema in summaries.schemas.items(): if schemaname in self.schemas: self.schemas[schemaname].update(schema) else: self.schemas[schemaname] = schema for k, v in summaries.other.items(): if k in self.other: self.other[k].update(v) else: self.other[k] = v
[docs] def is_empty(self) -> bool: return not ( any(self.lists) or any(self.ranges) or any(self.schemas) or any(self.other) )
[docs] def clone(self) -> Summaries: """Clones this object. Returns: Summaries: The clone of this object """ cls = self.__class__ summaries = cls(summaries=deepcopy(self._summaries), maxcount=self.maxcount) summaries.lists = deepcopy(self.lists) summaries.other = deepcopy(self.other) summaries.ranges = deepcopy(self.ranges) summaries.schemas = deepcopy(self.schemas) return summaries
[docs] def to_dict(self) -> dict[str, Any]: return { **{k: v for k, v in self.lists.items() if len(v) < self.maxcount}, **{k: v.to_dict() for k, v in self.ranges.items()}, **self.schemas, **self.other, }
[docs] @classmethod def empty(cls) -> Summaries: return Summaries({})