Merge pull request #439 from stac-utils/pv/aggregations

aggregations endpoint and collection-level aggregate

Author: Phil Varner

CHANGELOG.md

@@ -5,12 +5,17 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
-## Unreleased - TBD
+## [Unreleased] - TBD
 
 ### Changed
 
 - Updated example serverless configuration to use OpenSearch 2.5.
 
+## Added
+
+- Added support for `/aggregations`, `/collections/{collectionId}/aggregations`, and
+  `/collections/{collectionId}/aggregate` endpoints.
+
 ## [0.8.1] - 2023-03-29
 
 ### Added

README.md

@@ -28,6 +28,7 @@
     - [Locking down transaction endpoints](#locking-down-transaction-endpoints)
     - [AWS WAF Rule Conflicts](#aws-waf-rule-conflicts)
   - [Queryables](#queryables)
+  - [Aggregation](#aggregation)
   - [Ingesting Data](#ingesting-data)
     - [Ingesting large items](#ingesting-large-items)
     - [Subscribing to SNS Topics](#subscribing-to-sns-topics)
@@ -835,6 +836,73 @@ from the Collection entity whenever that is returned.
 A non-configurable root-level queryables definition is defined with no named terms but
 `additionalProperties` set to `true`.
 
+## Aggregation
+
+STAC API supports the [Aggregation Extension](https://github.com/stac-api-extensions/aggregation). This allows the definition of per-collection aggregations that can be
+calculated, dependent on the relevant fields being available in the STAC Items in that
+Collection. A field named `aggregations` should be added to the Collection object for
+the collection for which the aggregations are available, e.g.:
+
+```text
+  "aggregations": [
+    {
+      "name": "total_count",
+      "data_type": "integer"
+    },
+    {
+      "name": "datetime_max",
+      "data_type": "datetime"
+    },
+    {
+      "name": "datetime_min",
+      "data_type": "datetime"
+    },
+    {
+      "name": "datetime_frequency",
+      "data_type": "frequency_distribution",
+      "frequency_distribution_data_type": "datetime"
+    },
+    {
+      "name": "grid_code_frequency",
+      "data_type": "frequency_distribution",
+      "frequency_distribution_data_type": "string"
+    },
+    {
+      "name": "grid_geohex_frequency",
+      "data_type": "frequency_distribution",
+      "frequency_distribution_data_type": "string"
+    },
+    {
+      "name": "grid_geohash_frequency",
+      "data_type": "frequency_distribution",
+      "frequency_distribution_data_type": "string"
+    },
+    {
+      "name": "grid_geotile_frequency",
+      "data_type": "frequency_distribution",
+      "frequency_distribution_data_type": "string"
+    }
+  ]
+```
+
+Available aggregations are:
+
+- total_count (count of total items)
+- collection_frequency (Item `collection` field)
+- platform_frequency (Item.Properties.platform)
+- cloud_cover_frequency (Item.Properties.eo:cloud_cover)
+- datetime_frequency (Item.Properties.datetime, monthly interval)
+- datetime_min (earliest Item.Properties.datetime)
+- datetime_max (latest Item.Properties.datetime)
+- grid_code_frequency (Item.Properties.grid:code)
+- grid_code_landsat_frequency (synthesized from Item.Properties.landsat:wrs_path and Item.Properties.landsat:wrs_row)
+- sun_elevation_frequency (Item.Properties.view:sun_elevation)
+- sun_azimuth_frequency (Item.Properties.view:sun_azimuth)
+- off_nadir_frequency (Item.Properties.view:off_nadir)
+- grid_geohex_frequency ([GeoHex grid](https://opensearch.org/docs/2.4/opensearch/geohexgrid-agg/) on Item.Properties.proj:centroid)
+- grid_geohash_frequency ([geohash grid](https://opensearch.org/docs/2.4/opensearch/bucket-agg/#geo_distance-geohash_grid) on Item.Properties.proj:centroid)
+- grid_geotile_frequency (geotile on Item.Properties.proj:centroid)
+
 ## Ingesting Data
 
 STAC Collections and Items are ingested by the `ingest` Lambda function, however this Lambda is not invoked directly by a user, it consumes records from the `stac-server-<stage>-queue` SQS. To add STAC Items or Collections to the queue, publish them to the SNS Topic `stac-server-<stage>-ingest`.

src/lambdas/api/app.js

@@ -77,7 +77,7 @@ app.get('/conformance', async (_req, res, next) => {
 app.get('/queryables', async (req, res, next) => {
   try {
     res.type('application/schema+json')
-    res.json(await api.getQueryables(req.endpoint))
+    res.json(await api.getGlobalQueryables(req.endpoint))
   } catch (error) {
     next(error)
   }
@@ -111,7 +111,7 @@ app.post('/search', async (req, res, next) => {
 
 app.get('/aggregate', async (req, res, next) => {
   try {
-    res.json(await api.aggregate(req.query, database, req.endpoint, 'GET'))
+    res.json(await api.aggregate(null, req.query, database, req.endpoint, 'GET'))
   } catch (error) {
     if (error instanceof ValidationError) {
       next(createError(400, error.message))
@@ -133,6 +133,14 @@ app.post('/aggregate', async (req, res, next) => {
   }
 })
 
+app.get('/aggregations', async (req, res, next) => {
+  try {
+    res.json(await api.getGlobalAggregations(req.endpoint))
+  } catch (error) {
+    next(error)
+  }
+})
+
 app.get('/collections', async (req, res, next) => {
   try {
     res.json(await api.getCollections(database, req.endpoint))
@@ -166,30 +174,52 @@ app.get('/collections/:collectionId', async (req, res, next) => {
   const { collectionId } = req.params
   try {
     const response = await api.getCollection(collectionId, database, req.endpoint)
-    delete response.queryables
     if (response instanceof Error) next(createError(404))
     else res.json(response)
   } catch (error) {
     next(error)
   }
 })
 
-app.get('/collections/:collectionId/items', async (req, res, next) => {
+app.get('/collections/:collectionId/queryables', async (req, res, next) => {
+  const { collectionId } = req.params
+  try {
+    const queryables = await api.getCollectionQueryables(collectionId, database, req.endpoint)
+
+    if (queryables instanceof Error) next(createError(404))
+
+    res.type('application/schema+json')
+    res.json(queryables)
+  } catch (error) {
+    if (error instanceof ValidationError) {
+      next(createError(400, error.message))
+    } else {
+      next(error)
+    }
+  }
+})
+
+app.get('/collections/:collectionId/aggregations', async (req, res, next) => {
+  const { collectionId } = req.params
+  try {
+    res.json(await api.getCollectionAggregations(collectionId, database, req.endpoint))
+  } catch (error) {
+    if (error instanceof ValidationError) {
+      next(createError(400, error.message))
+    } else {
+      next(error)
+    }
+  }
+})
+
+const collectionAggregate = async (req, res, next, httpMethod) => {
   const { collectionId } = req.params
   try {
     const response = await api.getCollection(collectionId, database, req.endpoint)
 
     if (response instanceof Error) next(createError(404))
     else {
-      const items = await api.searchItems(
-        collectionId,
-        req.query,
-        database,
-        req.endpoint,
-        'GET'
-      )
-      res.type('application/geo+json')
-      res.json(items)
+      res.json(await api.aggregate(collectionId, req.query, database, req.endpoint, httpMethod))
     }
   } catch (error) {
     if (error instanceof ValidationError) {
@@ -198,27 +228,31 @@ app.get('/collections/:collectionId/items', async (req, res, next) => {
       next(error)
     }
   }
-})
-
-const DEFAULT_QUERYABLES = {
-  $schema: 'https://json-schema.org/draft/2020-12/schema',
-  type: 'object',
-  properties: {},
-  additionalProperties: true
 }
 
-app.get('/collections/:collectionId/queryables', async (req, res, next) => {
+// are these supposed to be awaited?
+app.get('/collections/:collectionId/aggregate',
+  async (req, res, next) => await collectionAggregate(req, res, next, 'GET'))
+
+app.get('/collections/:collectionId/aggregate',
+  async (req, res, next) => await collectionAggregate(req, res, next, 'POST'))
+
+app.get('/collections/:collectionId/items', async (req, res, next) => {
   const { collectionId } = req.params
   try {
-    const collection = await api.getCollection(collectionId, database, req.endpoint)
+    const response = await api.getCollection(collectionId, database, req.endpoint)
 
-    if (collection instanceof Error) next(createError(404))
+    if (response instanceof Error) next(createError(404))
     else {
-      const queryables = collection.queryables || { ...DEFAULT_QUERYABLES }
-      queryables.$id = `${req.endpoint}/collections/${collectionId}/queryables`
-      queryables.title = `Queryables for Collection ${collectionId}`
-      res.type('application/schema+json')
-      res.json(queryables)
+      const items = await api.searchItems(
+        collectionId,
+        req.query,
+        database,
+        req.endpoint,
+        'GET'
+      )
+      res.type('application/geo+json')
+      res.json(items)
     }
   } catch (error) {
     if (error instanceof ValidationError) {