Source code for omgeo.services.esri

from datetime import datetime, timedelta
import logging

from omgeo.services.base import GeocodeService
from omgeo.places import Candidate
from omgeo.preprocessors import CancelIfPOBox
from omgeo.postprocessors import (AttrFilter, AttrRename, AttrSorter, UseHighScoreIfAtLeast,
                                  GroupBy, ScoreSorter)

logger = logging.getLogger(__name__)


[docs]class EsriWGS(GeocodeService): """ Class to geocode using the `ESRI World Geocoding service <https://developers.arcgis.com/features/geocoding/>`_. This uses two endpoints -- one for single-line addresses, and one for multi-part addresses. An optional ``key`` parameter can be passed to the :class:`~omgeo.places.PlaceQuery` which will be passed as a ``magicKey`` to the find endpoint if using a single line address/text search. This allows omgeo to be used with the `Esri suggest endpoint <https://developers.arcgis.com/rest/geocode/api-reference/geocoding-suggest.htm>`_. .. warning:: Based on tests using the magicKey parameter, it is recommended that a viewbox not be used with in conjuction with the magicKey. Additionally, address/search text passed via the query may be ignored when using a magicKey. An optional ``for_storage`` flag can be passed to the :class:`~omgeo.places.PlaceQuery` which will cause ``findStorage=true`` to be passed to the find endpoint. The Esri terms of service requires this if the returned point will be stored in a database. If you pass this flag, you `must` set the client_id and client_secret settings. Settings used by the EsriWGS GeocodeService object may include: * client_id -- The Client ID used to access Esri services. * client_secret -- The Client Secret used to access Esri services. """ LOCATOR_MAP = { 'PointAddress': 'rooftop', 'StreetAddress': 'interpolation', 'PostalExt': 'postal_specific', # accept ZIP+4 'Postal': 'postal' } DEFAULT_PREPROCESSORS = [CancelIfPOBox()] DEFAULT_POSTPROCESSORS = [ AttrFilter(['PointAddress', 'StreetAddress', # 'PostalExt', # 'Postal' ], 'locator_type'), # AttrExclude(['USA_Postal'], 'locator'), #accept postal from everywhere but US (need PostalExt) AttrSorter(['PointAddress', 'StreetAddress', # 'PostalExt', # 'Postal' ], 'locator_type'), AttrRename('locator', LOCATOR_MAP), # after filter to avoid searching things we toss out UseHighScoreIfAtLeast(99.8), ScoreSorter(), GroupBy('match_addr'), GroupBy(('x', 'y')), ] _endpoint = 'https://geocode.arcgis.com/arcgis/rest/services/World/GeocodeServer' def __init__(self, preprocessors=None, postprocessors=None, settings=None): preprocessors = EsriWGS.DEFAULT_PREPROCESSORS if preprocessors is None else preprocessors postprocessors = EsriWGS.DEFAULT_POSTPROCESSORS if postprocessors is None else postprocessors settings = {} if settings is None else settings if 'client_id' in settings and 'client_secret' in settings: self._authenticated = True self._client_id = settings['client_id'] self._client_secret = settings['client_secret'] self._token = None elif 'client_id' not in settings and 'client_secret' not in settings: self._authenticated = False else: raise Exception('Must specify both client_id and client_secret to use authentication') GeocodeService.__init__(self, preprocessors, postprocessors, settings) def _geocode(self, pq): """ :arg PlaceQuery pq: PlaceQuery object to use for geocoding :returns: list of location Candidates """ #: List of desired output fields #: See `ESRI docs <https://developers.arcgis.com/rest/geocode/api-reference/geocoding-geocode-addresses.htm>_` for details outFields = ('Loc_name', # 'Shape', 'Score', 'Match_addr', # based on address standards for the country # 'Address', # returned by default # 'Country' # 3-digit ISO 3166-1 code for a country. Example: Canada = "CAN" # 'Admin', # 'DepAdmin', # 'SubAdmin', # 'Locality', # 'Postal', # 'PostalExt', 'Addr_type', # 'Type', # 'Rank', 'AddNum', 'StPreDir', 'StPreType', 'StName', 'StType', 'StDir', # 'Side', # 'AddNumFrom', # 'AddNumTo', # 'AddBldg', 'City', 'Subregion', 'Region', 'Postal', 'Country', # 'Ymax', # 'Ymin', # 'Xmin', # 'Xmax', # 'X', # 'Y', 'DisplayX', 'DisplayY', # 'LangCode', # 'Status', ) outFields = ','.join(outFields) query = dict(f='json', # default HTML. Other options are JSON and KMZ. outFields=outFields, # outSR=WKID, defaults to 4326 maxLocations=20, # default 1; max is 20 ) # Postal-code only searches work in the single-line but not multipart geocoder # Remember that with the default postprocessors, postcode-level results will be eliminated if pq.query == pq.address == '' and pq.postal != '': pq.query = pq.postal if pq.query == '': # multipart query = dict(query, Address=pq.address, # commonly represents the house number and street name of a complete address Neighborhood=pq.neighborhood, City=pq.city, Subregion=pq.subregion, Region=pq.state, Postal=pq.postal, # PostalExt= CountryCode=pq.country, # full country name or ISO 3166-1 2- or 3-digit country code ) else: # single-line magic_key = pq.key if hasattr(pq, 'key') else '' query = dict(query, singleLine=pq.query, # This can be a street address, place name, postal code, or POI. sourceCountry=pq.country, # full country name or ISO 3166-1 2- or 3-digit country code ) if magic_key: query['magicKey'] = magic_key # This is a lookup key returned from the suggest endpoint. if pq.bounded and pq.viewbox is not None: query = dict(query, searchExtent=pq.viewbox.to_esri_wgs_json()) if self._authenticated: if self._token is None or self._token_expiration < datetime.utcnow(): expiration = timedelta(hours=2) self._token = self.get_token(expiration) self._token_expiration = datetime.utcnow() + expiration query['token'] = self._token if getattr(pq, 'for_storage', False): query['forStorage'] = 'true' endpoint = self._endpoint + '/findAddressCandidates' response_obj = self._get_json_obj(endpoint, query) returned_candidates = [] # this will be the list returned try: locations = response_obj['candidates'] for location in locations: c = Candidate() attributes = location['attributes'] c.match_addr = attributes['Match_addr'] c.locator = attributes['Loc_name'] c.locator_type = attributes['Addr_type'] c.score = attributes['Score'] c.x = attributes['DisplayX'] # represents the actual location of the address. c.y = attributes['DisplayY'] c.wkid = response_obj['spatialReference']['wkid'] c.geoservice = self.__class__.__name__ # Optional address component fields. for in_key, out_key in [('City', 'match_city'), ('Subregion', 'match_subregion'), ('Region', 'match_region'), ('Postal', 'match_postal'), ('Country', 'match_country')]: setattr(c, out_key, attributes.get(in_key, '')) setattr(c, 'match_streetaddr', self._street_addr_from_response(attributes)) returned_candidates.append(c) except KeyError: pass return returned_candidates def _street_addr_from_response(self, attributes): """Construct a street address (no city, region, etc.) from a geocoder response. :param attributes: A dict of address attributes as returned by the Esri geocoder. """ # The exact ordering of the address component fields that should be # used to reconstruct the full street address is not specified in the # Esri documentation, but the examples imply that it is this. ordered_fields = ['AddNum', 'StPreDir', 'StPreType', 'StName', 'StType', 'StDir'] result = [] for field in ordered_fields: result.append(attributes.get(field, '')) if any(result): return ' '.join([s for s in result if s]) # Filter out empty strings. else: return ''
[docs] def get_token(self, expires=None): """ :param expires: The time until the returned token expires. Must be an instance of :class:`datetime.timedelta`. If not specified, the token will expire in 2 hours. :returns: A token suitable for use with the Esri geocoding API """ endpoint = 'https://www.arcgis.com/sharing/rest/oauth2/token/' query = {'client_id': self._client_id, 'client_secret': self._client_secret, 'grant_type': 'client_credentials'} if expires is not None: if not isinstance(expires, timedelta): raise Exception('If expires is provided it must be a timedelta instance') query['expiration'] = int(expires.total_seconds() / 60) response_obj = self._get_json_obj(endpoint, query, is_post=True) return response_obj['access_token']