forked from hans/Nominatim
fbff4fa218
Parse code now checks presence of all required fields and exports all fields for inspection.
301 lines
11 KiB
Python
301 lines
11 KiB
Python
# SPDX-License-Identifier: GPL-2.0-only
|
|
#
|
|
# This file is part of Nominatim. (https://nominatim.org)
|
|
#
|
|
# Copyright (C) 2023 by the Nominatim developer community.
|
|
# For a full list of authors see the git log.
|
|
"""
|
|
Classes wrapping HTTP responses from the Nominatim API.
|
|
"""
|
|
import re
|
|
import json
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from check_functions import Almost, check_for_attributes
|
|
|
|
OSM_TYPE = {'N' : 'node', 'W' : 'way', 'R' : 'relation',
|
|
'n' : 'node', 'w' : 'way', 'r' : 'relation',
|
|
'node' : 'n', 'way' : 'w', 'relation' : 'r'}
|
|
|
|
def _geojson_result_to_json_result(geojson_result):
|
|
result = geojson_result['properties']
|
|
result['geojson'] = geojson_result['geometry']
|
|
if 'bbox' in geojson_result:
|
|
# bbox is minlon, minlat, maxlon, maxlat
|
|
# boundingbox is minlat, maxlat, minlon, maxlon
|
|
result['boundingbox'] = [geojson_result['bbox'][1],
|
|
geojson_result['bbox'][3],
|
|
geojson_result['bbox'][0],
|
|
geojson_result['bbox'][2]]
|
|
return result
|
|
|
|
class BadRowValueAssert:
|
|
""" Lazily formatted message for failures to find a field content.
|
|
"""
|
|
|
|
def __init__(self, response, idx, field, value):
|
|
self.idx = idx
|
|
self.field = field
|
|
self.value = value
|
|
self.row = response.result[idx]
|
|
|
|
def __str__(self):
|
|
return "\nBad value for row {} field '{}'. Expected: {}, got: {}.\nFull row: {}"""\
|
|
.format(self.idx, self.field, self.value,
|
|
self.row[self.field], json.dumps(self.row, indent=4))
|
|
|
|
|
|
class GenericResponse:
|
|
""" Common base class for all API responses.
|
|
"""
|
|
def __init__(self, page, fmt, errorcode=200):
|
|
fmt = fmt.strip()
|
|
if fmt == 'jsonv2':
|
|
fmt = 'json'
|
|
|
|
self.page = page
|
|
self.format = fmt
|
|
self.errorcode = errorcode
|
|
self.result = []
|
|
self.header = dict()
|
|
|
|
if errorcode == 200 and fmt != 'debug':
|
|
getattr(self, '_parse_' + fmt)()
|
|
|
|
def _parse_json(self):
|
|
m = re.fullmatch(r'([\w$][^(]*)\((.*)\)', self.page)
|
|
if m is None:
|
|
code = self.page
|
|
else:
|
|
code = m.group(2)
|
|
self.header['json_func'] = m.group(1)
|
|
self.result = json.JSONDecoder().decode(code)
|
|
if isinstance(self.result, dict):
|
|
if 'error' in self.result:
|
|
self.result = []
|
|
else:
|
|
self.result = [self.result]
|
|
|
|
|
|
def _parse_geojson(self):
|
|
self._parse_json()
|
|
if self.result:
|
|
geojson = self.result[0]
|
|
# check for valid geojson
|
|
check_for_attributes(geojson, 'type,features')
|
|
assert geojson['type'] == 'FeatureCollection'
|
|
assert isinstance(geojson['features'], list)
|
|
|
|
self.result = []
|
|
for result in geojson['features']:
|
|
check_for_attributes(result, 'type,properties,geometry')
|
|
assert result['type'] == 'Feature'
|
|
new = result['properties']
|
|
check_for_attributes(new, 'geojson', 'absent')
|
|
new['geojson'] = result['geometry']
|
|
if 'bbox' in result:
|
|
check_for_attributes(new, 'boundingbox', 'absent')
|
|
# bbox is minlon, minlat, maxlon, maxlat
|
|
# boundingbox is minlat, maxlat, minlon, maxlon
|
|
new['boundingbox'] = [result['bbox'][1],
|
|
result['bbox'][3],
|
|
result['bbox'][0],
|
|
result['bbox'][2]]
|
|
for k, v in geojson.items():
|
|
if k not in ('type', 'features'):
|
|
check_for_attributes(new, '__' + k, 'absent')
|
|
new['__' + k] = v
|
|
self.result.append(new)
|
|
|
|
|
|
def _parse_geocodejson(self):
|
|
self._parse_geojson()
|
|
if self.result:
|
|
for r in self.result:
|
|
assert set(r.keys()) == {'geocoding', 'geojson', '__geocoding'}, \
|
|
f"Unexpected keys in result: {r.keys()}"
|
|
check_for_attributes(r['geocoding'], 'geojson', 'absent')
|
|
r |= r.pop('geocoding')
|
|
|
|
|
|
def assert_field(self, idx, field, value):
|
|
""" Check that result row `idx` has a field `field` with value `value`.
|
|
Float numbers are matched approximately. When the expected value
|
|
starts with a carat, regular expression matching is used.
|
|
"""
|
|
assert field in self.result[idx], \
|
|
"Result row {} has no field '{}'.\nFull row: {}"\
|
|
.format(idx, field, json.dumps(self.result[idx], indent=4))
|
|
|
|
if isinstance(value, float):
|
|
assert Almost(value) == float(self.result[idx][field]), \
|
|
BadRowValueAssert(self, idx, field, value)
|
|
elif value.startswith("^"):
|
|
assert re.fullmatch(value, self.result[idx][field]), \
|
|
BadRowValueAssert(self, idx, field, value)
|
|
elif isinstance(self.result[idx][field], dict):
|
|
assert self.result[idx][field] == eval('{' + value + '}'), \
|
|
BadRowValueAssert(self, idx, field, value)
|
|
else:
|
|
assert str(self.result[idx][field]) == str(value), \
|
|
BadRowValueAssert(self, idx, field, value)
|
|
|
|
|
|
def assert_subfield(self, idx, path, value):
|
|
assert path
|
|
|
|
field = self.result[idx]
|
|
for p in path:
|
|
assert isinstance(field, dict)
|
|
assert p in field
|
|
field = field[p]
|
|
|
|
if isinstance(value, float):
|
|
assert Almost(value) == float(field)
|
|
elif value.startswith("^"):
|
|
assert re.fullmatch(value, field)
|
|
elif isinstance(field, dict):
|
|
assert field, eval('{' + value + '}')
|
|
else:
|
|
assert str(field) == str(value)
|
|
|
|
|
|
def assert_address_field(self, idx, field, value):
|
|
""" Check that result rows`idx` has a field `field` with value `value`
|
|
in its address. If idx is None, then all results are checked.
|
|
"""
|
|
if idx is None:
|
|
todo = range(len(self.result))
|
|
else:
|
|
todo = [int(idx)]
|
|
|
|
for idx in todo:
|
|
assert 'address' in self.result[idx], \
|
|
"Result row {} has no field 'address'.\nFull row: {}"\
|
|
.format(idx, json.dumps(self.result[idx], indent=4))
|
|
|
|
address = self.result[idx]['address']
|
|
assert field in address, \
|
|
"Result row {} has no field '{}' in address.\nFull address: {}"\
|
|
.format(idx, field, json.dumps(address, indent=4))
|
|
|
|
assert address[field] == value, \
|
|
"\nBad value for row {} field '{}' in address. Expected: {}, got: {}.\nFull address: {}"""\
|
|
.format(idx, field, value, address[field], json.dumps(address, indent=4))
|
|
|
|
def match_row(self, row, context=None):
|
|
""" Match the result fields against the given behave table row.
|
|
"""
|
|
if 'ID' in row.headings:
|
|
todo = [int(row['ID'])]
|
|
else:
|
|
todo = range(len(self.result))
|
|
|
|
for i in todo:
|
|
for name, value in zip(row.headings, row.cells):
|
|
if name == 'ID':
|
|
pass
|
|
elif name == 'osm':
|
|
assert 'osm_type' in self.result[i], \
|
|
"Result row {} has no field 'osm_type'.\nFull row: {}"\
|
|
.format(i, json.dumps(self.result[i], indent=4))
|
|
assert self.result[i]['osm_type'] in (OSM_TYPE[value[0]], value[0]), \
|
|
BadRowValueAssert(self, i, 'osm_type', value)
|
|
self.assert_field(i, 'osm_id', value[1:])
|
|
elif name == 'osm_type':
|
|
assert self.result[i]['osm_type'] in (OSM_TYPE[value[0]], value[0]), \
|
|
BadRowValueAssert(self, i, 'osm_type', value)
|
|
elif name == 'centroid':
|
|
if ' ' in value:
|
|
lon, lat = value.split(' ')
|
|
elif context is not None:
|
|
lon, lat = context.osm.grid_node(int(value))
|
|
else:
|
|
raise RuntimeError("Context needed when using grid coordinates")
|
|
self.assert_field(i, 'lat', float(lat))
|
|
self.assert_field(i, 'lon', float(lon))
|
|
elif '+' in name:
|
|
self.assert_subfield(i, name.split('+'), value)
|
|
else:
|
|
self.assert_field(i, name, value)
|
|
|
|
def property_list(self, prop):
|
|
return [x[prop] for x in self.result]
|
|
|
|
|
|
class SearchResponse(GenericResponse):
|
|
""" Specialised class for search and lookup responses.
|
|
Transforms the xml response in a format similar to json.
|
|
"""
|
|
|
|
def _parse_xml(self):
|
|
xml_tree = ET.fromstring(self.page)
|
|
|
|
self.header = dict(xml_tree.attrib)
|
|
|
|
for child in xml_tree:
|
|
assert child.tag == "place"
|
|
self.result.append(dict(child.attrib))
|
|
|
|
address = {}
|
|
for sub in child:
|
|
if sub.tag == 'extratags':
|
|
self.result[-1]['extratags'] = {}
|
|
for tag in sub:
|
|
self.result[-1]['extratags'][tag.attrib['key']] = tag.attrib['value']
|
|
elif sub.tag == 'namedetails':
|
|
self.result[-1]['namedetails'] = {}
|
|
for tag in sub:
|
|
self.result[-1]['namedetails'][tag.attrib['desc']] = tag.text
|
|
elif sub.tag == 'geokml':
|
|
self.result[-1][sub.tag] = True
|
|
else:
|
|
address[sub.tag] = sub.text
|
|
|
|
if address:
|
|
self.result[-1]['address'] = address
|
|
|
|
|
|
class ReverseResponse(GenericResponse):
|
|
""" Specialised class for reverse responses.
|
|
Transforms the xml response in a format similar to json.
|
|
"""
|
|
|
|
def _parse_xml(self):
|
|
xml_tree = ET.fromstring(self.page)
|
|
|
|
self.header = dict(xml_tree.attrib)
|
|
self.result = []
|
|
|
|
for child in xml_tree:
|
|
if child.tag == 'result':
|
|
assert not self.result, "More than one result in reverse result"
|
|
self.result.append(dict(child.attrib))
|
|
elif child.tag == 'addressparts':
|
|
address = {}
|
|
for sub in child:
|
|
address[sub.tag] = sub.text
|
|
self.result[0]['address'] = address
|
|
elif child.tag == 'extratags':
|
|
self.result[0]['extratags'] = {}
|
|
for tag in child:
|
|
self.result[0]['extratags'][tag.attrib['key']] = tag.attrib['value']
|
|
elif child.tag == 'namedetails':
|
|
self.result[0]['namedetails'] = {}
|
|
for tag in child:
|
|
self.result[0]['namedetails'][tag.attrib['desc']] = tag.text
|
|
elif child.tag == 'geokml':
|
|
self.result[0][child.tag] = True
|
|
else:
|
|
assert child.tag == 'error', \
|
|
"Unknown XML tag {} on page: {}".format(child.tag, self.page)
|
|
|
|
|
|
class StatusResponse(GenericResponse):
|
|
""" Specialised class for status responses.
|
|
Can also parse text responses.
|
|
"""
|
|
|
|
def _parse_text(self):
|
|
pass
|