Source code for

from import Mapping
import copy
import json
import warnings

from enum import Enum

from .catalog_client import CatalogClient
from ..common.property_filtering.filtering import AndExpression
from ..common.property_filtering.filtering import Expression  # noqa: F401

from .attributes import parse_iso_datetime, serialize_datetime

V1_COMPATIBILITY = "v1_compatibility"

[docs]class Interval(str, Enum): """An interval for the :py:meth:`ImageSearch.summary_interval` method. Attributes ---------- YEAR : enum Aggregate on a yearly basis QUARTER : enum Aggregate on a quarterly basis MONTH : enum Aggregate on a monthly basis WEEK : enum Aggregate on a weekly basis DAY : enum Aggregate on a daily basis HOUR : enum Aggregate on a hourly basis MINUTE : enum Aggregate per minute """ YEAR = "year" QUARTER = "quarter" MONTH = "month" WEEK = "week" DAY = "day" HOUR = "hour" MINUTE = "minute"
[docs]class AggregateDateField(str, Enum): """A date field to use for aggragation for the :py:meth:`ImageSearch.summary_interval` method. Attributes ---------- ACQUIRED : enum Aggregate on the `Image.acquired` field. CREATED : enum Aggregate on the `Image.created` field. MODIFIED : enum Aggregate on the `Image.modified` field. PUBLISHED : enum Aggregate on the `Image.published` field. """ ACQUIRED = "acquired" CREATED = "created" MODIFIED = "modified" PUBLISHED = "published"
[docs]class ImageSearch(Search): # Be aware that the `|` characters below add whitespace. The first one is needed # avoid the `Inheritance` section from appearing before the auto summary. """A search request that iterates over its search results for images. The `ImageSearch` is identical to `Search` but with a couple of summary methods: :py:meth:`summary` and :py:meth:`summary_interval`; and an additional method :py:meth:`intersects` for geometry. """ _unsupported_summary_params = ["sort"] def __init__( self, model, client=None, url=None, includes=True, request_params=None ): super(ImageSearch, self).__init__( model, client, url, includes, request_params=request_params ) self._intersects = None
[docs] def intersects(self, geometry): """Filter images to those that intersect the given geometry. Successive calls to `intersects` override the previous intersection geometry. Parameters ---------- geometry : shapely.geometry.base.BaseGeometry, ~descarteslabs.common.geo.GeoContext, geojson-like Geometry that found images must intersect. Returns ------- Search A new instance of the :py:class:`~descarteslabs.catalog.ImageSearch` class that includes geometry filter. """ # noqa: E501 s = copy.deepcopy(self) name, value = self._model_cls._serialize_filter_attribute("geometry", geometry) s._request_params["intersects"] = json.dumps( value, separators=(",", ":"), ) s._intersects = copy.deepcopy(geometry) return s
def _summary_request(self): # don't modify existing search params params = copy.deepcopy(self._request_params) for p in self._unsupported_summary_params: params.pop(p, None) filters = self._serialize_filters() if filters: # urlencode encodes spaces in the json object which create an invalid filter value when # the server tries to parse it, so we have to remove spaces prior to encoding. params["filter"] = json.dumps(filters, separators=(",", ":")) return params
[docs] def summary(self): """Get summary statistics about the current `ImageSearch` query. Returns ------- SummaryResult The summary statistics as a `SummaryResult` object. Raises ------ ~descarteslabs.exceptions.ClientError or ~descarteslabs.exceptions.ServerError :ref:`Spurious exception <network_exceptions>` that can occur during a network request. Example ------- >>> from descarteslabs.catalog import Image, properties as p >>> search = ... p.product_id=="landsat:LC08:01:RT:TOAR" ... ) >>> s = search.summary() # doctest: +SKIP >>> print(s.count, s.bytes) # doctest: +SKIP """ s = copy.deepcopy(self) summary_url = s._url + "/summary/all" r = self._client.session.put(summary_url, json=self._summary_request()) response = r.json() return SummaryResult(**response["data"]["attributes"])
[docs] def summary_interval( self, aggregate_date_field="acquired", interval="year", start_datetime=None, end_datetime=None, ): """Get summary statistics by specified datetime intervals about the current `ImageSearch` query. Parameters ---------- aggregate_date_field : str or AggregateDateField, optional The date field to use for aggregating summary results over time. Valid inputs are `~AggregateDateField.ACQUIRED`, `~AggregateDateField.CREATED`, `~AggregateDateField.MODIFIED`, `~AggregateDateField.PUBLISHED`. The default is `~AggregateDateField.ACQUIRED`. interval : str or Interval, optional The time interval to use for aggregating summary results. Valid inputs are `~Interval.YEAR`, `~Interval.QUARTER`, `~Interval.MONTH`, `~Interval.WEEK`, `~Interval.DAY`, `~Interval.HOUR`, `~Interval.MINUTE`. The default is `~Interval.YEAR`. start_datetime : str or datetime, optional Beginning of the date range over which to summarize data in ISO format. The default is least recent date found in the search result based on the `aggregate_date_field`. The start_datetime is included in the result. To set it as unbounded, use the value ``0``. end_datetime : str or datetime, optional End of the date range over which to summarize data in ISO format. The default is most recent date found in the search result based on the `aggregate_date_field`. The end_datetime is included in the result. To set it as unbounded, use the value ``0``. Returns ------- list(SummaryResult) The summary statistics for each interval, as a list of `SummaryResult` objects. Raises ------ ~descarteslabs.exceptions.ClientError or ~descarteslabs.exceptions.ServerError :ref:`Spurious exception <network_exceptions>` that can occur during a network request. Example ------- >>> from descarteslabs.catalog import Image, AggregateDateField, Interval, properties >>> search = ( ... ... .filter(properties.product_id == "landsat:LC08:01:RT:TOAR") ... ) >>> interval_results = search.summary_interval( ... aggregate_date_field=AggregateDateField.ACQUIRED, interval=Interval.MONTH ... ) # doctest: +SKIP >>> print([(i.interval_start, i.count) for i in interval_results]) # doctest: +SKIP """ s = copy.deepcopy(self) summary_url = "{}/summary/{}/{}".format(s._url, aggregate_date_field, interval) # The service will calculate start/end if not given if start_datetime is not None: if start_datetime: s._request_params["_start"] = serialize_datetime(start_datetime) else: s._request_params["_start"] = "" # Unbounded if end_datetime is not None: if end_datetime: s._request_params["_end"] = serialize_datetime(end_datetime) else: s._request_params["_end"] = "" # Unbounded r = self._client.session.put(summary_url, json=s._summary_request()) response = r.json() return [SummaryResult(**d["attributes"]) for d in response["data"]]
[docs] def collect(self, geocontext=None, **kwargs): """ Execute the search query and return the collection of the appropriate type. Parameters ---------- geocontext : shapely.geometry.base.BaseGeometry, descarteslabs.common.geo.Geocontext, geojson-like, default None # noqa: E501 AOI for the ImageCollection. Returns ------- ~descarteslabs.catalog.ImageCollection ImageCollection of Images returned from the search. Raises ------ BadRequestError If any of the query parameters or filters are invalid ~descarteslabs.exceptions.ClientError or ~descarteslabs.exceptions.ServerError :ref:`Spurious exception <network_exceptions>` that can occur during a network request. """ if geocontext is None: geocontext = self._intersects if geocontext is not None: kwargs["geocontext"] = geocontext return super(ImageSearch, self).collect(**kwargs)
[docs]class SummaryResult(object): """ The readonly data returned by :py:meth:`ImageSearch.summary` or :py:meth:`ImageSearch.summary_interval`. Attributes ---------- count : int Number of images in the summary. bytes : int Total number of bytes of data across all images in the summary. products : list(str) List of IDs for the products included in the summary. interval_start: datetime For interval summaries only, a datetime representing the start of the interval period. """ def __init__( self, count=None, bytes=None, products=None, interval_start=None, **kwargs ): self.count = count self.bytes = bytes self.products = products self.interval_start = ( parse_iso_datetime(interval_start) if interval_start else None ) def __repr__(self): text = [ "\nSummary for {} images:".format(self.count), " - Total bytes: {:,}".format(self.bytes), ] if self.products: text.append(" - Products: {}".format(", ".join(self.products))) if self.interval_start: text.append(" - Interval start: {}".format(self.interval_start)) return "\n".join(text)