Skip to content

Instantly share code, notes, and snippets.

@tomkralidis tomkralidis/csw2.py.diff Secret
Last active Jun 14, 2017

Embed
What would you like to do?
OGC filter to Elasticsearch query syntax
diff --git a/pycsw/ogc/csw/csw2.py b/pycsw/ogc/csw/csw2.py
index 3249dab..067d78f 100644
--- a/pycsw/ogc/csw/csw2.py
+++ b/pycsw/ogc/csw/csw2.py
@@ -42,6 +42,7 @@ from pycsw.ogc.csw.cql import cql2fes1
from pycsw.plugins.profiles import profile as pprofile
import pycsw.plugins.outputschemas
from pycsw.core import config, log, metadata, util
+from pycsw.core.formats.fmt_json import xml2dict
from pycsw.ogc.fes import fes1
import logging
@@ -1550,6 +1551,7 @@ class Csw2(object):
query['where'], query['values'] = fes1.parse(tmp,
self.parent.repository.queryables['_all'], self.parent.repository.dbtype,
self.parent.context.namespaces, self.parent.orm, self.parent.language['text'], self.parent.repository.fts)
+ query['_dict'] = xml2dict(etree.tostring(tmp), self.parent.context.namespaces)
except Exception as err:
return 'Invalid Filter request: %s' % err
@@ -1563,6 +1565,7 @@ class Csw2(object):
query['where'], query['values'] = fes1.parse(cql,
self.parent.repository.queryables['_all'], self.parent.repository.dbtype,
self.parent.context.namespaces, self.parent.orm, self.parent.language['text'], self.parent.repository.fts)
+ query['_dict'] = xml2dict(etree.tostring(cql), self.parent.context.namespaces)
except Exception as err:
LOGGER.exception('Invalid CQL request: %s', tmp.text)
LOGGER.exception('Error message: %s', err)
import json
es_query = {'query': {}}
MODEL = {
'GeometryOperands': {
'values': ['gml3.TYPES']
},
'SpatialOperators': {
'values': ['BBOX', 'Beyond', 'Contains', 'Crosses', 'Disjoint',
'DWithin', 'Equals', 'Intersects', 'Overlaps', 'Touches', 'Within']
},
'ComparisonOperators': {
'ogc:PropertyIsBetween': {'opname': 'Between', 'opvalue': 'and'},
'ogc:PropertyIsEqualTo': {'opname': 'EqualTo', 'opvalue': '='},
'ogc:PropertyIsGreaterThan': {'opname': 'GreaterThan', 'opvalue': '>'},
'ogc:PropertyIsGreaterThanOrEqualTo': {
'opname': 'GreaterThanEqualTo', 'opvalue': '>='},
'ogc:PropertyIsLessThan': {'opname': 'LessThan', 'opvalue': '<'},
'ogc:PropertyIsLessThanOrEqualTo': {
'opname': 'LessThanEqualTo', 'opvalue': '<='},
'ogc:PropertyIsLike': {'opname': 'Like', 'opvalue': 'like'},
'ogc:PropertyIsNotEqualTo': {'opname': 'NotEqualTo', 'opvalue': '!='},
'ogc:PropertyIsNull': {'opname': 'NullCheck', 'opvalue': 'is null'},
},
'Functions': {
'length': {'args': '1'},
'lower': {'args': '1'},
'ltrim': {'args': '1'},
'rtrim': {'args': '1'},
'trim': {'args': '1'},
'upper': {'args': '1'},
},
'Ids': {
'values': ['EID', 'FID']
}
}
def _get_typed_value(value):
if '.' in value: # try float
return float(value)
else:
try:
return int(value)
except ValueError:
return value
def _get_comparison_operator(element):
comp_op_fes = element.keys()[0]
pname = element[comp_op_fes]['ogc:PropertyName'] # TODO: validate propertyname
try:
literal = _get_typed_value(element[comp_op_fes]['ogc:Literal'])
except KeyError: # is ogc:PropertyIsBetween
upper_ = _get_typed_value(element[comp_op_fes]['ogc:UpperBoundary']['ogc:Literal'])
lower_ = _get_typed_value(element[comp_op_fes]['ogc:LowerBoundary']['ogc:Literal'])
if 'AnyText' in pname:
pname = '_all'
if comp_op_fes == 'ogc:PropertyIsEqualTo':
return '{}:{}'.format(pname, literal)
elif comp_op_fes == 'ogc:PropertyIsNotEqualTo':
return 'NOT {}:{}'.format(pname, literal)
elif comp_op_fes == 'ogc:PropertyIsGreaterThan':
return '{}:>{}'.format(pname, literal)
elif comp_op_fes == 'ogc:PropertyIsGreaterThanOrEqualTo':
return '{}:>={}'.format(pname, literal)
elif comp_op_fes == 'ogc:PropertyIsLessThan':
return '{}:<{}'.format(pname, literal)
elif comp_op_fes == 'ogc:PropertyIsLessThanOrEqualTo':
return '{}:<={}'.format(pname, literal)
elif comp_op_fes == 'ogc:PropertyIsBetween':
return '{}:[{} TO {}]'.format(pname, lower_, upper_)
elif comp_op_fes == 'ogc:PropertyIsLike':
if '@wildCard' in element[comp_op_fes]:
literal = literal.replace(element[comp_op_fes]['@wildCard'], '*')
if '@singleChar' in element[comp_op_fes]:
literal = literal.replace(element[comp_op_fes]['@singleChar'], '?')
if '@escapeChar' in element[comp_op_fes]:
literal = literal.replace(element[comp_op_fes]['@escapeChar'], '\\\\')
return '{}:{}'.format(pname, literal)
with open('filter.json') as ff:
data = json.load(ff)['ogc:Filter']
print(_get_comparison_operator(data))
import json
import sys
from pycsw.core.formats.fmt_json import xml2dict
from pycsw.core.config import StaticContext
namespaces = StaticContext().namespaces
es_query = {'query': {}}
MODEL = {
'GeometryOperands': {
'values': ['gml3.TYPES']
},
'SpatialOperators': {
'values': ['BBOX', 'Beyond', 'Contains', 'Crosses', 'Disjoint',
'DWithin', 'Equals', 'Intersects', 'Overlaps', 'Touches',
'Within']
},
'ComparisonOperators': {
'ogc:PropertyIsBetween': {'opname': 'Between', 'opvalue': 'and'},
'ogc:PropertyIsEqualTo': {'opname': 'EqualTo', 'opvalue': '='},
'ogc:PropertyIsGreaterThan': {'opname': 'GreaterThan', 'opvalue': '>'},
'ogc:PropertyIsGreaterThanOrEqualTo': {
'opname': 'GreaterThanEqualTo', 'opvalue': '>='},
'ogc:PropertyIsLessThan': {'opname': 'LessThan', 'opvalue': '<'},
'ogc:PropertyIsLessThanOrEqualTo': {
'opname': 'LessThanEqualTo', 'opvalue': '<='},
'ogc:PropertyIsLike': {'opname': 'Like', 'opvalue': 'like'},
'ogc:PropertyIsNotEqualTo': {'opname': 'NotEqualTo', 'opvalue': '!='},
'ogc:PropertyIsNull': {'opname': 'NullCheck', 'opvalue': 'is null'},
},
'Functions': {
'length': {'args': '1'},
'lower': {'args': '1'},
'ltrim': {'args': '1'},
'rtrim': {'args': '1'},
'trim': {'args': '1'},
'upper': {'args': '1'},
},
'Ids': {
'values': ['EID', 'FID']
}
}
def _get_typed_value(value):
try:
value2 = float(value)
except ValueError:
try:
value2 = int(value)
except ValueError: # string
value2 = value
if isinstance(value2, float) and '.' not in value: # it's an int
value2 = int(value)
if isinstance(value2, int) and '.' in value: # it's a float
value2 = float(value)
return value2
def _parse_filter(element):
compound = []
comp_op_fes = element.keys()[0]
logic_ops = ['ogc:And', 'ogc:Or', 'ogc:Not']
print(json.dumps(element, indent=4))
if comp_op_fes in logic_ops:
return _parse_binary_logical_operator(comp_op_fes, element[comp_op_fes])
else:
return _parse_comparison_operator(comp_op_fes, element[comp_op_fes])
def _parse_binary_logical_operator(logic_op, comparisons):
compound = []
logic_op_es = None
if logic_op == 'ogc:And':
logic_op_es = 'must'
elif logic_op == 'ogc:Or':
logic_op_es = 'should'
elif logic_op == 'ogc:Not':
logic_op_es = 'must_not'
for key, value in comparisons.items():
if key in ['ogc:And', 'ogc:Or', 'ogc:Not']:
compound.append(_parse_binary_logical_operator(key, value))
elif isinstance(value, list): # list of same comparisons
for value2 in value:
compound.append(_parse_comparison_operator(key, value2))
else: # dict of various comparisons
compound.append(_parse_comparison_operator(key, value))
return {'bool': {logic_op_es: compound}}
def _parse_comparison_operator(comparison, expression):
pname = expression['ogc:PropertyName'] # TODO: validate propertyname
pname_es = '{}.keyword'.format(pname)
try:
literal = _get_typed_value(expression['ogc:Literal'])
except KeyError: # is ogc:PropertyIsBetween
upper = _get_typed_value(expression['ogc:UpperBoundary']['ogc:Literal'])
lower = _get_typed_value(expression['ogc:LowerBoundary']['ogc:Literal'])
if 'AnyText' in pname:
pname_es = pname = '_all'
if comparison in MODEL['SpatialOperators']['values']:
print("SPATIAL")
if comparison == 'ogc:PropertyIsEqualTo':
return {'match': {pname_es: literal}}
elif comparison == 'ogc:PropertyIsNotEqualTo':
return {'must_not': {'match': {pname_es: literal}}}
elif comparison == 'ogc:PropertyIsGreaterThan':
return {'range': {pname_es: {'gt': literal}}}
elif comparison == 'ogc:PropertyIsGreaterThanOrEqualTo':
return {'range': {pname_es: {'gte': literal}}}
elif comparison == 'ogc:PropertyIsLessThan':
return {'range': {pname_es: {'lt': literal}}}
elif comparison == 'ogc:PropertyIsLessThanOrEqualTo':
return {'range': {pname_es: {'lte': literal}}}
elif comparison == 'ogc:PropertyIsBetween':
return {'range': {pname_es: {'gte': upper, 'lte': lower}}}
elif comparison == 'ogc:PropertyIsLike':
return {'match': {pname: literal}}
def _parse_spatial_comparison_operator(spatial_op, expression):
pass
if __name__ == '__main__':
if len(sys.argv) < 2:
print('Usage: {} <filter.json>'.format(sys.argv[0]))
sys.exit(1)
with open(sys.argv[1]) as ff:
data = xml2dict(ff.read(), namespaces)
query = {'query': _parse_filter(data['ogc:Filter'])}
print(json.dumps(query, indent=4))
{
"ogc:Filter": {
"ogc:PropertyIsLike": {
"@wildCard": "%",
"@singleChar": "_",
"@escapeChar": "\\",
"ogc:PropertyName": "csw:AnyText",
"ogc:Literal": "ozone"
}
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.