add STAC API + CQL2 JSON merging

Author: tomkralidis

pycsw/core/pygeofilter_evaluate.py

@@ -29,10 +29,12 @@
 
 import logging
 
+from shapely import box
 from shapely.geometry import shape
 from sqlalchemy import text
 
 from pygeofilter import ast
+from pygeofilter.values import Envelope
 from pygeofilter.backends.evaluator import handle
 from pygeofilter.backends.sqlalchemy import filters
 from pygeofilter.backends.sqlalchemy.evaluate import SQLAlchemyFilterEvaluator
@@ -43,7 +45,8 @@
 
 
 class PycswFilterEvaluator(SQLAlchemyFilterEvaluator):
-    def __init__(self, field_mapping=None, dbtype='sqlite', undefined_as_null=None):
+    def __init__(self, field_mapping=None, dbtype='sqlite',
+                 undefined_as_null=None):
         super().__init__(field_mapping, undefined_as_null=undefined_as_null)
         self._pycsw_dbtype = dbtype
 
@@ -56,7 +59,11 @@ def intersects(self, node, lhs, rhs):
         except AttributeError:
             crs = 4326
 
-        wkt = shape(node.rhs.geometry).wkt
+        if isinstance(node.rhs, Envelope):
+            wkt = box(node.rhs.x1, node.rhs.x2,
+                      node.rhs.y1, node.rhs.y2, ccw=False).wkt
+        else:
+            wkt = shape(node.rhs.geometry).wkt
 
         if self._pycsw_dbtype == 'postgresql+postgis+native':
             return text(f"ST_Intersects({geometry}, 'SRID={crs};{wkt}')")

pycsw/ogc/pubsub/__init__.py

@@ -39,6 +39,7 @@ def publish_message(pubsub_client, action: str, collection: str = None,
     """
     Publish broker message
 
+    :param pubsub_client: `paho.mqtt.client.Client` instance
     :param action: `str` of action trigger name (create, update, delete)
     :param collection: `str` of collection identifier
     :param item: `str` of item identifier

pycsw/stac/api.py

@@ -419,6 +419,8 @@ def items(self, headers_, json_post_data, args, collection='metadata:main'):
         :returns: tuple of headers, status code, content
         """
 
+        cql_ops = []
+
         if collection not in self.get_all_collections():
             msg = 'Invalid collection'
             LOGGER.exception(msg)
@@ -430,32 +432,93 @@ def items(self, headers_, json_post_data, args, collection='metadata:main'):
             LOGGER.debug('Empty JSON payload')
             json_post_data = {}
 
-        if 'filter' in json_post_data:
-            LOGGER.debug('Detected filter query parameter')
-            json_post_data = json_post_data.pop('filter')
-
         if 'bbox' in json_post_data:
             LOGGER.debug('Detected bbox query parameter')
-            bbox_ = json_post_data.pop('bbox')
-            bbox_ = ','.join([str(b) for b in bbox_])
-            args['bbox'] = bbox_
+
+            cql_ops.append({
+                'op': 's_intersects', 'args': [{
+                    'property': 'geometry'
+                    },
+                    {'bbox': json_post_data.pop('bbox')}]
+            })
 
         if 'limit' in json_post_data:
             LOGGER.debug('Detected limit query parameter')
             args['limit'] = json_post_data.pop('limit')
 
         if 'collections' in json_post_data:
-            LOGGER.debug('Detected limit query parameter')
-            args['collections'] = ','.join(json_post_data.pop('collections'))
+            LOGGER.debug('Detected collections query parameter')
+            cql_ops.append({
+                'op': 'in',
+                'args': [{
+                    'property': 'collections',
+                    },
+                    json_post_data.pop('collections')
+                ]
+            })
 
         if 'ids' in json_post_data:
             LOGGER.debug('Detected ids query parameter')
-            args['ids'] = ','.join(json_post_data.pop('ids'))
+            cql_ops.append({
+                'op': 'in',
+                'args': [{
+                    'property': 'identifier',
+                    },
+                    json_post_data.pop('ids')
+                ]
+            })
 
         if 'intersects' in json_post_data:
             LOGGER.debug('Detected intersects query parameter')
             # TODO
 
+        if 'datetime' in json_post_data:
+            if '/' not in json_post_data['datetime']:
+                cql_ops.append({
+                    'op': '=',
+                    'args': [
+                        {'property': 'date'},
+                        json_post_data.pop('datetime')
+                    ]
+                })
+            else:
+                begin, end = json_post_data.pop('datetime').split('/')
+                if begin != '..':
+                    cql_ops.append({
+                        'op': '>=',
+                        'args': [
+                            {'property': 'time_begin'},
+                            begin
+                        ]
+                    })
+                if end != '..':
+                    cql_ops.append({
+                        'op': '<=',
+                        'args': [
+                            {'property': 'time_end'},
+                            end
+                        ]
+                    })
+
+        if 'filter' in json_post_data:
+            LOGGER.debug('Detected filter query parameter')
+            json_post_data = json_post_data.pop('filter')
+
+        if not json_post_data and not cql_ops:
+            LOGGER.debug('No JSON POST data or CQL ops')
+        elif not json_post_data and cql_ops:
+            LOGGER.debug('No JSON POST data left')
+            json_post_data = {
+                'op': 'and',
+                'args': cql_ops
+            }
+        else:
+            LOGGER.debug('JSON POST data is CQL2 JSON')
+            if cql_ops:
+                print("JJ2", json_post_data)
+                LOGGER.debug('Adding STAC API query parameters to CQL2 JSON')
+                json_post_data['args'].extend(cql_ops)
+
         headers, status, response = super().items(headers_, json_post_data, args, collection)
 
         response = json.loads(response)

tests/functionaltests/suites/stac_api/test_stac_api_functional.py

@@ -287,6 +287,7 @@ def test_items(config):
     assert content['numberMatched'] == 1
     assert content['features'][0]['properties']['view:off_nadir'] == 3.8
 
+    # test post CQL2 JSON requests
     cql_json = {
         'filter-lang': 'cql2-json',
         'filter': {