Source code for descarteslabs.catalog.search

# Copyright 2018-2023 Descartes Labs.
#
# 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 collections.abc import Mapping
import copy
import json
import warnings

from strenum import StrEnum

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

from .attributes import serialize_datetime





[docs]class Interval(StrEnum): """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(StrEnum): """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"
class GeoSearch(Search): """A search request that supports an :py:meth:`intersects` method for searching geometries.""" def __init__( self, model, client=None, url=None, includes=True, request_params=None ): super(GeoSearch, self).__init__( model, client, url, includes, request_params=request_params ) self._intersects = None def intersects(self, geometry): """Filter images or blobs 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.GeoSearch` 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 class SummarySearchMixin(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 add support for summary methods. The `SummarySearch` is identical to `Search` but with a couple of summary methods: :py:meth:`summary` and :py:meth:`summary_interval`. """ _unsupported_summary_params = ["sort"] # must be set in derived class SummaryResult = None DEFAULT_AGGREGATE_DATE_FIELD = None 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 def summary(self): """Get summary statistics about the current `Search` 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 = Image.search().filter( ... 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 self.SummaryResult(**response["data"]["attributes"]) def summary_interval( self, aggregate_date_field=None, 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`. Field must be defined for the class. 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 = ( ... Image.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 or self.DEFAULT_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 [self.SummaryResult(**d["attributes"]) for d in response["data"]]